当工具调用调用结果返回给模型后，模型会进入到Reflection阶段，在这个阶段模型将回答交给Critic进行反思，评价任务是否被正确执行。
举一个例子，当询问”帮我查找三个适合日本摇滚乐队使用的 Bass 音箱插件”，模型会做如下判断
1. 搜索插件
2. 找到三个
3. Reflection：
- 我是否真的找到了三个？
- 三个插件是不是都属于 Bass Amp？
- 有没有把 Guitar Amp 当成 Bass Amp？
- 用户说的是“日系摇滚”，我的推荐是否符合？
4. 发现某个插件其实是 Guitar Amp,结束Reflection进入主Agent
5. 再搜索一个 Bass Amp
6. 重新整理
7. 输出最终答案
SYSTEM_PROMPT = """
你是一个 ReAct 文件整理助手。
目标：检查 inbox，移动 inbox/a.txt 到 archive/a.txt，查看整理后的目录，然后总结。
每一轮最多调用一个工具。
如果有 reflection note，请优先参考它决定下一步。
""".strip()
REFLECTION_PROMPT = """
你是 reviewer。根据最近的工具结果，给 agent 一句下一步建议。
目标：检查 inbox -> 移动 inbox/a.txt 到 archive/a.txt -> 查看整理后的目录 -> 总结。
如果已经看到 archive/a.txt，请提醒 agent 停止调用工具并总结。
只输出一句 reflection note。
""".strip()
def main() -> None:
task = " ".join(sys.argv[1:]).strip() or DEFAULT_TASK
reset_workspace()
print("=== 03. LangGraph ReAct + Reflection Node ===")
print("\n用户任务:")
print(task)
print("\n运行前 workspace:")
print(show_workspace())
tools = [list_files, move_file]
tool_map = {item.name: item for item in tools}
base_llm = load_llm()
tool_llm = base_llm.bind_tools(tools)
def agent_node(state: AgentState) -> AgentState:
print("\n[agent] 模型思考")
prompt = SYSTEM_PROMPT
if state["reflection"]:
prompt += f"\n\nReflection note: {state['reflection']}"
response = tool_llm.invoke([SystemMessage(content=prompt), *state["messages"]])
return {"messages": [*state["messages"], response], "reflection": state["reflection"]}
def tools_node(state: AgentState) -> AgentState:
response = state["messages"][-1]
new_messages = list(state["messages"])
for tool_call in response.tool_calls:
print("\n[tools] 执行工具")
print(f"tool_name = {tool_call['name']}")
print(f"tool_args = {tool_call['args']}")
result = tool_map[tool_call["name"]].invoke(tool_call["args"])
print(result)
new_messages.append(
ToolMessage(
content=str(result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": new_messages, "reflection": state["reflection"]}
def reflection_node(state: AgentState) -> AgentState:
print("\n[reflection] 复盘工具结果")
transcript = "\n".join(str(message.content) for message in state["messages"][-4:])
note = base_llm.invoke(
[
SystemMessage(content=REFLECTION_PROMPT),
HumanMessage(content=transcript),
]
).content
print(note)
return {"messages": state["messages"], "reflection": str(note)}
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
return "tools" if getattr(last_message, "tool_calls", None) else END
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)
graph.add_node("reflection", reflection_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "reflection")
graph.add_edge("reflection", "agent")
result = graph.compile().invoke(
{"messages": [HumanMessage(content=task)], "reflection": ""},
config={"recursion_limit": 12},
)
print("\n最终回答:")
print(result["messages"][-1].content)
print("\n运行后 workspace:")
print(show_workspace())
上面例子中，我们将reflection_node的输出结果传递给state上下文的reflection字段，并由agent_node节点单独处理这个字段（不过实现方式是追加到System Prompt中），这种做法并不优雅。
我们可以让reflection节点直接处理messages记录，而不是专门设计reflection字段传递给Agent节点，让agent节点负责其他职责的工作
def reflection_node(state):
note = ...
return {
"messages": [
*state["messages"],
HumanMessage(content=f"[Reflection] {note}")
]
}
评论（0）
暂无评论