爱回收
75.46M · 2026-04-05
复杂问题之所以“复杂”，往往不是因为某一步很难，而是因为它由许多步骤组成：需要先澄清任务目标、拆分子问题、选择合适工具、在本地与网络信息之间做取舍、把中间结果串起来，最后再产出符合格式的交付物。对 Agent 来说，planning（规划）能力就是把这些“人类默认会做的步骤”显式化、程序化的关键。
本章目标是把第二章“本地知识库问答 Agent”升级为一个会自主规划的智能 Agent：它不仅能调用工具回答问题，还能在多轮工具调用中维持状态、判断是否继续、并把最终产出写入文件，形成可复用的工作流。
我们在遇到多步推理/多源检索的时候，常规的workflow很难满足我们的要求，比如我们要做一个Claude Code类的coding agent的时候，我们几乎不可能提供一个预先的执行路径。而LLM提供的思考、规划、调整的能力就给了我们一个全新的解法。这也是Agent往人工智能方向迈进的一个新的步骤。
到目前为止，我们已经有了三类关键工具（具体实现会因项目而异，这里先按能力维度回顾）：
第二章的“本地知识库问答 Agent”已经能做到：优先本地检索、必要时联网、最后生成答案。但它通常仍有一个明显限制：缺少显式的执行图（Graph）与可循环的控制逻辑，从而很难把“检索→阅读→再检索→写入结果”变成一个稳定的自治流程。本章会用 LangGraph 把这些能力组织成可控的工作流，让 Agent 具备“自我驱动地决定下一步”的 planning 形态。
在“能回答”之外，真实场景往往需要“能交付”：例如把一组问答写入报告、把检索结果落盘、把最终结论追加到实验记录里。因此，我们先补齐一个最实用的工具：write_doc，用于把 Agent 的答案写入文件。
本章使用的 write_doc 工具实现位于：它的行为很简单：确保父目录存在，然后以追加模式（append）把内容写入指定文件，并返回写入结果（success、路径、长度等）。核心签名如下：
@tool
async def write_doc(file_path: str, content: str) -> dict:
...
在我们的 Agent 里，write_doc 通常用于把最终答案保存成可持续积累的“工作记录”。比如约定输出格式：
Q: {question}
A: {answer}
这样每次运行都可以把结果追加进同一个文件，形成可检索的历史。
LangChain 擅长把 LLM 调用与工具调用串成“链”，但当你想要一个可循环、可分支、可终止的自治 Agent 时，“链”会逐渐变得笨重：你要自己维护状态、自己写 while 循环、自己处理工具调用的路由与错误恢复。
LangGraph 的定位更接近“可控的 Agent 工作流引擎” 他可以使用图来表达控制流，并且最重要的是状态是显式的一等公民 GraphState 统一承载 messages、路径、问题、结果位置等上下文，跨节点传递更稳定。这点非常重要，我们在agent的plan过程中经常会用到各种全局变量，我们可以将全局变量保存至graph state中。
ReAct（Reason + Act）是一种经典的 Agent 工作方式：模型并不一次性给出最终答案，而是在一个循环中不断执行：
用流程图表示就是一个“LLM↔Tools”的回路：
flowchart TD
A[输入: 文件路径/问题/结果路径] --> B[LLM 节点: 决定下一步或给最终答案]
B -->|调用工具| C[Tools 节点: grep/read_lines/tavily/write_doc]
C --> D[Observation: 工具返回结果]
D --> B
B -->|无需工具调用| E[结束: 输出最终答案]
ReAct 的关键不在于“想得更久”，而在于把复杂任务拆成多个可验证的小步骤：每一步都通过工具观测来降低不确定性。
我们把Agent工作流拆成三个核心部分：
用“节点连接图”表示如下（Start/LLM/Tool/End）：
flowchart LR
S([Start]) --> L[llm]
L -->|有 tool_calls| T[tools]
T --> L
L -->|无 tool_calls| E([End])
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
class GraphState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
path: str
questions: str
result_path: str
async def llm_node(state: GraphState, tools: list) -> GraphState:
llm_with_tools = llm.bind_tools(tools)
system_message = SystemMessage(content='''
你是一个AI researcher，使用 Thought-Action-Observation 模式解决问题。
**工作流程：**
1. **Thought（思考）**：分析问题，决定下一步行动
2. **Action（行动）**：调用可用工具获取信息
3. **Observation（观察）**：观察工具返回的结果
4. 重复以上步骤直到获得足够信息，然后给出最终答案
**行动规则：**
- 优先使用本地文件中的知识库，如果没有检索到关键词则不要读取文件
- 如果本地知识库没有相关内容则进行网络查询
- 每次回复必须先输出 Thought，说明你的思考过程
- 最终结果保存至文件，格式为：Q: question n A: answer
**输出格式示例：**
Thought: 我需要先搜索文件中的相关内容...
Action: 调用工具搜索关键词
Observation: 工具返回的结果...
...
最终答案：
''')
user_message = HumanMessage(
content=f"文件路径: {state['path']}n问题: {state['questions']}n保存结果路径: {state['result_path']}"
)
messages = state.get("messages", [])
if not messages:
messages = [system_message, user_message]
else:
messages = [system_message] + list(messages) + [user_message]
response = await llm_with_tools.ainvoke(messages)
return {"messages": [response], "path": state["path"], "questions": state["questions"], "result_path": state["result_path"]}
async def tool_node(state: GraphState, tools: list) -> GraphState:
last_message = state["messages"][-1]
if not last_message.tool_calls:
return state
tool_messages = []
for tool_call in last_message.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
tool_call_id = tool_call["id"]
selected_tool = next((t for t in tools if t.name == tool_name), None)
tool_result = await selected_tool.ainvoke(tool_args) if selected_tool else f"错误：未找到工具 {tool_name}"
tool_messages.append(ToolMessage(content=str(tool_result), tool_call_id=tool_call_id, name=tool_name))
return {"messages": tool_messages, "path": state["path"], "questions": state["questions"]}
def should_continue(state: GraphState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
def create_agent_graph(tools: list):
workflow = StateGraph(GraphState)
workflow.add_node("llm", lambda state: asyncio.run(llm_node(state, tools)))
workflow.add_node("tools", lambda state: asyncio.run(tool_node(state, tools)))
workflow.set_entry_point("llm")
workflow.add_conditional_edges("llm", should_continue, {"tools": "tools", END: END})
workflow.add_edge("tools", "llm")
return workflow.compile()
看到这里，你可以把 ReAct Agent 理解为：
以问题“三体3中的三则童话故事中隐含了什么技术信息”为例，ReAct 的运行过程通常是“先本地、后联网、再写入文件”的循环：
Agent 对话记录:
==================================================
**[ai]**
Thought: 需要先搜索文件中关于三体3中三则童话故事的内容，找出其中隐含的技术信息。
Action: 搜索文件中关于三则童话故事的内容。
functions.async_grep({
pattern: "三则童话故事",
path: "santi.txt",
case_sensitive: true
})
工具调用: [{'name': 'async_grep', 'args': {'pattern': '三则童话故事', 'path': 'santi.txt', 'case_sensitive': True}, 'id': 'call_mSMWCroNzr6ewijrboiVm2Ca', 'type': 'tool_call'}]
**[tool]**
{
"pattern": "三则童话故事",
"total_matches": 0,
"files_matched": 0,
"results": []
}
**[ai]**
Thought: 本地文件中没有找到关于三则童话故事的内容，需要进行网络搜索来获取相关信息。
Action: 使用网络搜索工具搜索关于三体3中三则童话故事的技术信息。
工具调用: [{'name': 'tavily_search', 'args': {'query': 'Three-Body Problem 3 three fairy tales technical information', 'max_results': 5}, 'id': 'call_OWUSxMXvLXXAQohPE6Sh9FB0', 'type': 'tool_call'}]
**[tool]**
[{'type': 'text', 'text': '...略...(网络上搜索的三体内容)']
**[ai]**
工具调用: [{'name': 'write_doc', 'args': {'file_path': 'result.txt', 'content': 'Q: 三体3中的三则童话故事中隐含了什么技术信息nA: 三体3中的三则童话故事实际上是技术寓言，分别代表着人类生存所依赖的技术，包括：n1. 皇家画师的故事n2. 无名氏的故事n3. 云天明的故事'}, 'id': 'call_f13iBmh3ErlJQLi5ba8tei7c', 'type': 'tool_call'}]
**[tool]**
{
"success": true,
"file_path": "result.txt",
"message": "内容已成功追加到文件: result.txt",
"content_length": 98
}
**[ai]**
已成功保存结果至文件中：
Q: 三体3中的三则童话故事中隐含了什么技术信息
A:《三体3》中的三则童话故事由云天明讲述，分别是《王国的新画师》、《饕餮海》和《深水王子》。这些故事对三体文明的先进技术作了诸多隐喻，是小说的重要线索。
1. **《王国的新画师》**：
- 故事中的“画进画里”隐喻二维化，直到太阳系被二维化时才明白其隐喻。
- 黑伞象征光速，可以逃脱二维化。
2. **《饕餮海》**：
- 饕餮鱼象征黑域或安全声明，能隔断信息交换。
- 香皂隐喻光，捕捉魔树泡象征降低光速或安全声明。
- 香皂放在玩具船尾部象征曲率驱动。
3. **《深水王子》**：
- 故事中涉及的曲率驱动引擎和四维空间碎片等技术隐喻了高维空间的概念。
async_grep 在本地知识库文件中定位相关段落tavily_search 补齐背景与解读write_doc 追加写入 result_path用一张序列图把这一轮循环压缩表达出来：
sequenceDiagram
participant L as LLM
participant T as Tools
participant F as File(result_path)
L->>T: async_grep(本地检索)
T-->>L: 命中行号/或 0 matches
alt 本地不足
L->>T: tavily_search(联网补充)
T-->>L: 搜索结果
end
L->>T: write_doc(写入答案)
T-->>F: append content
L-->>L: 输出最终回答
ReAct 的优势是灵活，但当任务变长（例如 5～10 个步骤）时会出现两个常见问题：
Plan-Execution 的做法是把“想清楚怎么做”和“真正去做”分成两段：
这样做的好处是：执行过程更可控、日志更可读、失败点更容易定位，尤其适合把 Agent 做成可重复运行的“工作流”。
传统的 Plan-Execution 还有一个典型痛点：计划一旦走偏，模型往往不知道该怎么“承认失败并纠正”。
例如下面这种情况非常常见：
因此我们在 Plan-Execution 的基础上增加一个 Evaluator（评估器）：
replan，让 Planner 基于已有观测重新制定计划；continue，推进到下一步；done，结束流程。这本质上是在工作流里加入了“质量闸门”：让 Agent 具备最基本的自我纠错能力，显著减少“明明失败了却继续往下走”的问题。
我们首先需要在GraphState中定义状态：
plan: List[str]：Planner 输出的步骤列表current_step: int：正在执行第几步（0-based）evaluation_result: "continue" | "replan" | "done"：Evaluator 的路由结果messages：贯穿全流程的消息历史（用于把工具观测、执行上下文带到下一轮）节点定义如下：
replan 时，会把历史消息带上，并要求“新计划不能与旧计划相同”对应的 LangGraph 连线结构如下：
flowchart TD
S([Start]) --> P[planner]
P --> X[executor]
X -->|有 tool_calls| T[tools]
X -->|无 tool_calls| E[evaluator]
T --> E
E -->|continue| X
E -->|replan| P
E -->|done| END([End])
我们同样以问题“三体3中的三则童话故事中隐含了什么技术信息”为例，看一下Agent是否可以进行Plan-RePlan的循环：
============================================================
生成的 Plan:
============================================================
1. 使用async_grep在文件santi.txt中搜索关键词“三则童话故事”
2. 使用async_read_line读取搜索到的关键词所在行及其上下文行，以获取完整的童话故事内容
3. 使用tavily_search查询三体3中的三则童话故事的技术信息分析
4. 将查询到的技术信息分析结果保存到文件result.txt
============================================================
▶️ 执行步骤 1/4: 使用async_grep在文件santi.txt中搜索关键词“三则童话故事”
调用工具: async_grep
参数: {'pattern': '三则童话故事', 'path': 'santi.txt'}
结果: {
"pattern": "三则童话故事",
"total_matches": 0,
"files_matched": 0,
"results": []
}...
评估结果: REPLAN
============================================================
生成的 Plan:
============================================================
1. 使用tavily_search在网络上查询“三则童话故事”的相关信息
2. 将查询到的信息整理并保存到文件/result.txt
============================================================
▶️ 执行步骤 1/2: 使用tavily_search在网络上查询“三则童话故事”的相关信息
调用工具: tavily_search
参数: {'query': '三体3 三则童话故事 技术信息', 'max_results': 5, 'search_depth': 'advanced'}
结果: [{'type': 'text', 'text': '{"query":"三体3 三则童话故事 技术信息","follow_up_questions":null,"answer":null,"images":[],"results":[{"url":"https://zh.wikiversity.org/wiki/%E3%80%8A%E4%B8%89%E4%BD%93%E3%80%8B%E4%B8...
评估结果: CONTINUE
▶️ 执行步骤 2/2: 将查询到的信息整理并保存到文件result.txt
调用工具: write_doc
参数: {'file_path': 'result.txt', 'content': '《三体3》中的三则童话故事由云天明讲述，分别是《王国的新画师》、《饕餮海》和《深水王子》。这些故事对三体文明的先进技术作了诸多隐喻，是小说的重要线索。nn1. **《王国的新画师》**：n - 故事中的“画进画里”隐喻二维化，直到太阳系被二维化时才明白其隐喻。n - 黑伞象征光速，可以逃脱二维化。nn2. **《饕餮海》**：n - 饕餮鱼象征黑域或安全声明，能隔断信息交换。n - 香皂隐喻光，捕捉魔树泡象征降低光速或安全声明。n - 香皂放在玩具船尾部象征曲率驱动。nn3. **《深水王子》**：n - 故事中涉及的曲率驱动引擎和四维空间碎片等技术隐喻了高维空间的概念。nn这些童话故事不仅对小说的发展有重要作用，也对三体文明的技术进行了隐晦的描述。解读这些隐喻需要结合小说的背景和情节发展。'}
结果: {
"success": true,
"file_path": "result.txt",
"message": "内容已成功追加到文件: result.txt"
}
评估结果: DONE
============================================================
Agent 运行完成
============================================================
可以看到，在本地Grep没有在本地搜索到信息的时候，Agent进行了Re-plan，改用tavily进行网络搜索。
把你这次运行的关键分支抽象成序列图如下：
sequenceDiagram
participant U as User
participant P as Planner
participant X as Executor
participant T as Tools
participant E as Evaluator
U->>P: 输入(文件路径/问题/结果路径)
P-->>X: Plan#1(本地优先)
X->>T: async_grep("三则童话故事")
T-->>X: matches=0
X-->>E: 提交执行与观测
E-->>P: REPLAN(本地失败)
P-->>X: Plan#2(改用联网)
X->>T: tavily_search(...)
T-->>X: 搜索结果
X-->>E: 提交执行与观测
E-->>X: CONTINUE
X->>T: write_doc(result.txt, answer)
T-->>X: success=true
X-->>E: 提交执行与观测
E-->>U: DONE(流程完成)
这张图的核心信息只有一句话：执行不是盲走，而是每一步都要被评估；评估不通过就回到 planner 改路线。
