生成结果以后，系统需要知道哪里出了问题，以及修改是否真的改善了结果。反馈流程应围绕具体标准推进，并限制修订次数。以解释RAG为例，第一版可能缺少证据ID，检查阶段指出问题，第二版补齐，再把这次失败类型保存为后续任务的策略提示。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 生成具备有效证据ID的答复，在有限修订内达到结构要求。 |
| 状态 | strategy保存当前策略；draft保存草稿；issues保存问题；attempt保存生成次数；status保存完成或人工处理结果。 |
| 执行边界 | 最多生成两次；只使用给定证据；结构检查由程序执行；经验仅保存教学任务的缺证据标记；不修改模型参数。 |
| 验收标准 | 发现缺证据后进入修订；合格草稿结束；连续失败在第二次后停止；后续任务可根据已保存经验选择另一策略。 |
方法与设计思想
反思：让反馈具体到可执行修改
反思由生成、检查和修订组成。检查条件应先写清楚，再决定是否需要模型批评者。本任务要求非空答案、非空证据ID和有效ID，因此先用确定性规则检查。
程序能判断结构是否合格，却不能仅凭ID证明事实正确。若任务需要检查语义支持关系，可以增加人工标注或独立评审，但要记录判断理由并验证其误判率。
推理：把复杂判断拆成可验证步骤
推理可以表现为问题分解、候选比较、计算和证据核对。系统应检查可观察的计划、工具结果和结论支持关系。模型给出的解释不能自动证明其内部计算过程。
这里把“检查证据 → 定位问题 → 修改答复”组织成显式步骤。修订是否成功由相同的验收规则决定，不能因为文字更长就判定更好。
学习与适应：区分策略变化和参数变化
成功修订以后，系统保存一次“缺少证据”的经验标记。后续任务优先采用“先列证据再作答”的策略。这是基于外部记忆的适应，没有训练权重，也不构成强化学习。
真实经验需要按任务、用户和版本隔离，并定期验证策略是否有帮助。一个成功样例不足以证明新策略具有普遍优势。
探索与发现：限定候选空间与验证预算
当前例子只包含两种预定义策略。它展示有限候选选择，不等同于开放式科学发现。需要探索新假设时，应保存候选、实验和结果，设置实验预算，并区分已验证结论与待检验假设。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| 反思循环 | 修复第一次输出中的明确缺陷 | 检查第一轮问题和第二轮修复结果 |
| 规则检查 | 让结构要求可重复判断 | 空内容、空ID和未知ID分别进入失败分支 |
| 经验存储 | 让后续任务调整策略 | 第二次任务加载不同策略；另外评测其效果 |
| 次数上限 | 避免无期限优化和费用增长 | 持续失败时恰好两次生成后转人工 |
LangChain与LangGraph代码实现
主代码在Python 3.12.14、LangChain 1.4.3、langchain-core 1.6.6、LangGraph 1.2.12、Pydantic 2.13.5下运行验证。建议使用独立虚拟环境：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install langchain==1.4.3 langchain-core==1.6.6 langgraph==1.2.12 pydantic==2.13.5
将下面完整代码保存为flow04.py，运行python flow04.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import json
import os
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, START, END
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel, Field
EVIDENCE = {"d1": "RAG通过检索外部资料为生成答案补充依据。"}
store = InMemoryStore()
class Draft(BaseModel):
answer: str = Field(min_length=1)
evidence_ids: list[str] = Field(default_factory=list)
class State(TypedDict, total=False):
goal: str
strategy: str
attempt: int
draft: dict
issues: list[str]
status: str
def choose_strategy(state: State) -> dict:
item = store.get(("experience", "demo"), "missing_evidence")
return {"strategy": "先列证据再作答" if item else "直接作答", "attempt": 0}
def build(always_bad=False):
offline = [
'{"answer":"RAG需要外部资料","evidence_ids":[]}',
json.dumps({"answer": EVIDENCE["d1"], "evidence_ids": ["d1"]}, ensure_ascii=False),
]
if always_bad:
offline = [offline[0], offline[0]]
model = (init_chat_model(os.environ["MODEL_ID"]) if os.getenv("MODEL_ID")
else FakeListChatModel(responses=offline))
prompt = ChatPromptTemplate.from_messages([
("system", '根据证据回答，只输出JSON：{{"answer":"...",'
'"evidence_ids":["d1"]}}。修正反馈指出的问题。'),
("human", "目标：{goal}\n策略：{strategy}\n证据：{evidence}\n反馈：{issues}"),
])
chain = prompt | model | StrOutputParser()
def generate(state):
text = chain.invoke({**state, "evidence": json.dumps(EVIDENCE, ensure_ascii=False),
"issues": state.get("issues", [])})
try:
draft = Draft.model_validate_json(text).model_dump()
except ValueError:
draft = {"answer": "", "evidence_ids": []}
return {"draft": draft, "attempt": state["attempt"] + 1}
def critique(state):
draft = state["draft"]
issues = []
if not draft["answer"]:
issues.append("内容为空或格式无效")
if not draft["evidence_ids"]:
issues.append("缺少证据ID")
if not set(draft["evidence_ids"]) <= set(EVIDENCE):
issues.append("证据ID不存在")
return {"issues": issues}
def route(state):
if not state["issues"]:
return "accept"
return "generate" if state["attempt"] < 2 else "fail"
def accept(state):
if state["attempt"] > 1:
store.put(("experience", "demo"), "missing_evidence", {"count": 1})
return {"status": "completed"}
graph = StateGraph(State)
graph.add_node("strategy", choose_strategy)
graph.add_node("generate", generate)
graph.add_node("critique", critique)
graph.add_node("accept", accept)
graph.add_node("fail", lambda _: {"status": "needs_human"})
graph.add_edge(START, "strategy")
graph.add_edge("strategy", "generate")
graph.add_edge("generate", "critique")
graph.add_conditional_edges("critique", route)
graph.add_edge("accept", END)
graph.add_edge("fail", END)
return graph.compile()
if __name__ == "__main__":
result = build().invoke({"goal": "说明RAG的作用"}, {"recursion_limit": 15})
if not os.getenv("MODEL_ID"):
assert result["status"] == "completed" and result["attempt"] == 2
next_run = build().invoke({"goal": "再次说明RAG"})
assert next_run["strategy"] == "先列证据再作答"
failed = build(always_bad=True).invoke({"goal": "持续失败案例"})
assert failed["status"] == "needs_human" and failed["attempt"] == 2
print(json.dumps(result, ensure_ascii=False, indent=2))
strategy节点选择初始策略，generate通过LangChain提示链产出草稿，critique给出问题列表。条件边根据问题和次数决定结束、修订或转人工。accept只在成功且经历修订时保存经验。
接入真实模型
代码通过init_chat_model读取MODEL_ID。安装所选供应商的集成包，配置其认证环境变量，再把MODEL_ID设为供应商:模型名称即可切换。供应商和模型需支持相应功能；第二篇需要工具调用能力。敏感参数放在环境变量中，不能写入正文或源码。真实模型输出允许变化，结构与边界检查保持相同。
验收与验证结果
固定模型第一版缺少证据，第二版补齐。脚本检查完成状态、两次生成、后续策略变化，以及always_bad模式的次数上限。真实模型的修订成功率，需要在同一任务集上比较有反馈和无反馈两种流程。
Google ADK实现与LangChain对照
Google ADK使用LoopAgent(max_iterations=2)组织生成与检查。检查通过时，EventActions.escalate结束循环；持续失败则依靠次数上限停止。这里的escalate表示模板工作流控制信号，不等同于创建人工工单。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 修订循环 | StateGraph条件边决定返回generate | LoopAgent重复执行generate与review |
| 停止条件 | 问题为空或达到两次上限 | 检查通过时escalate，或max_iterations耗尽 |
| 策略经验 | store保存失败类型标记 | user:strategy保存用户作用范围的策略提示 |
| 语义检查 | 当前检查格式与ID | 当前检查固定证据ID；同样需要额外事实评估 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk04.py后运行。默认不调用真实模型。
import asyncio
import inspect
import json
from collections.abc import Callable
from typing import Any
from google.adk.agents import BaseAgent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class Step(BaseAgent):
action: Callable[..., Any]
async def _run_async_impl(self, ctx):
delta = self.action(ctx.session.state)
if inspect.isawaitable(delta):
delta = await delta
yield Event(author=self.name, actions=EventActions(state_delta=delta))
async def run_state(agent, state, session_id="s1", user_id="u1", service=None):
service = service or InMemorySessionService()
await service.create_session(app_name="demo", user_id=user_id,
session_id=session_id, state=state)
runner = Runner(agent=agent, app_name="demo", session_service=service)
async for _ in runner.run_async(user_id=user_id, session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text="执行任务")])):
pass
session = await service.get_session(app_name="demo", user_id=user_id,
session_id=session_id)
return dict(session.state)
class Review(BaseAgent):
async def _run_async_impl(self, ctx):
state = ctx.session.state
valid = state["draft"].get("evidence_ids") == ["d1"]
delta = {"issues": [] if valid else ["缺少有效证据"],
"status": "completed" if valid else "needs_human"}
if valid and state["attempt"] > 1:
delta["user:strategy"] = "先列证据"
yield Event(author=self.name, actions=EventActions(state_delta=delta, escalate=valid))
def generate(state):
attempt = state.get("attempt", 0) + 1
ids = ["d1"] if attempt == 2 and not state.get("always_bad") else []
return {"attempt": attempt, "draft": {"answer": "RAG使用外部资料。", "evidence_ids": ids}}
def build():
return LoopAgent(name="reflection", max_iterations=2, sub_agents=[
Step(name="generate", action=generate), Review(name="review"),
])
async def main():
service = InMemorySessionService()
result = await run_state(build(), {}, "good", "u1", service)
assert result["status"] == "completed" and result["attempt"] == 2
next_session = await service.create_session(app_name="demo", user_id="u1", session_id="next")
assert next_session.state["user:strategy"] == "先列证据"
bad = await run_state(build(), {"always_bad": True})
assert bad["status"] == "needs_human" and bad["attempt"] == 2
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
两种实现均覆盖第一次不合格、第二次合格以及持续失败的两次上限。ADK示例验证经验字段可跨会话取得，没有声称训练模型权重。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
自动检查只覆盖格式和证据ID。代码没有语义事实判定、在线训练或开放式发现能力。不能把该样例的通过结果转化成模型质量提升比例。