第 1 篇埋过一条主线：到本篇为止，你在使用别人的抽象；本篇之后，你理解每个抽象为什么存在。现在就走到这条主线的分水岭——不用任何 prebuilt，徒手把一个完整的 ReAct Agent 从 State 定义一路写到编译运行，Agent 本体的代码不到 40 行。四块积木今天全部合体：第 5 篇的工具调用协议（bind_tools、ToolMessage）、第 6 篇的 StateGraph、第 7 篇的 State 与 reducer、第 8 篇的条件边。
本篇是《AI Agent 开发》系列第 9 篇，代码基于 langgraph 1.2.x 与 langchain-core 1.6.x，环境沿用第 2 篇的 uv 项目与 API Key 配置。全文按"拆解 → State → 模型节点 → 工具节点 → 路由 → 组装运行 → 对照反思"的顺序推进，写完你会得到一个能跑多轮工具调用的 Agent，以及一把拆开一切 Agent 框架的螺丝刀。
1、ReAct 循环的形状
ReAct 是 Reasoning 与 Acting 的合成词，出自 Yao 等人 2022 年的论文 ReAct: Synergizing Reasoning and Acting in Language Models：模型先推理出下一步该做什么，再通过工具调用付诸行动，看到执行结果后继续推理，交替循环直到给出最终回答。这套"想一步、做一步"的范式是当今 Agent 的事实标准，第 1 篇的全景地图里它占据正中央。
把 ReAct 落到图上只需要三个部件，每个都对应前面某一篇的积木：
- 模型节点：读消息历史，输出决策——最终回答或工具调用
- 工具节点：执行调用，把结果回填进消息历史
- 条件回边：根据模型是否发起调用，决定继续还是结束
先画图后写码，目标拓扑如下：
┌─────────────────────────────────┐
│ 条件回边（执行完回到模型） │
▼ │
START ──► 模型节点 ───有 tool_calls───► 工具节点 ──┘
│
└────无 tool_calls────► END
图的读法：用户消息从 START 进入模型节点；模型发起调用就走下路进工具节点，工具节点执行完回到模型节点，这条闭合回路就是条件回边；模型不再发起调用则直接到 END。对照第 5 篇手写的 while 循环，形状完全一致，只是控制流从命令式的 while True 搬到了声明式的图上。
这一搬是质变而不是语法糖：循环的每一步变成图上可检视、可中断、可持久化的节点。第 11 篇在人机协同里打断执行、第 12 篇把状态存盘再恢复，都以"控制流长在图上"为前提——散落在 while 循环里的逻辑，框架无处下手。
Agent 的骨架就是一条带循环的消息流水线：模型节点负责想，工具节点负责做，条件边决定还要不要再来一轮。
2、State 与 add_messages
第 7 篇立过纪律：节点只返回增量，由 reducer 负责合并进共享 State。ReAct Agent 的 State 简单得出奇——全部状态就是消息历史，一个键：
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
Annotated[list, add_messages] 声明 messages 键用 add_messages 这个 reducer 合并，规则只有两条：不带 ID 的新消息追加到列表尾部；ID 与已有消息相同的消息原地替换。第 7 篇讲过，没有 reducer 的键走"整体覆盖"——对消息历史而言那是灾难：模型节点刚返回一条 AIMessage，工具节点回填结果时会把整段历史抹掉，循环永远只有最后一批消息。
所以这个注解不是仪式，而是循环得以成立的前提：每一轮"决策 + 执行"的产物都累积进同一条历史，模型每轮看到的是越来越长的完整上下文。顺带一提，LangGraph 内置了等价的 MessagesState 快捷类，展开就是上面这几行；本篇坚持手写，因为这份 State 就是 Agent 的全部家底，值得逐行看清。
3、工具与模型节点
先把本篇示例的两件数据工具备好，写法是第 5 篇的 @tool 装饰器——docstring 即描述、类型注解即 schema：
from langchain_core.tools import tool
@tool
def get_population(city: str) -> str:
"""查询指定城市的常住人口，单位万人。"""
data = {"北京": "2185", "上海": "2487", "深圳": "1798"}
return f"{city}常住人口 {data.get(city, '未知')} 万人"
@tool
def get_area(city: str) -> str:
"""查询指定城市的陆地面积，单位平方公里。"""
data = {"北京": "16410", "上海": "6341", "深圳": "1997"}
return f"{city}陆地面积 {data.get(city, '未知')} 平方公里"
两件工具返回写死的真实统计数据，足够演示多轮调用，也免去了外部 API 的不确定性。接着是模型节点，全文最短的一个：
from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-sonnet-5")
tools = [get_population, get_area]
llm_rt = llm.bind_tools(tools) # 第 5 篇：绑定工具的新实例
def call_model(state: AgentState) -> dict:
response = llm_rt.invoke(state["messages"])
return {"messages": [response]}
逐行拆开看。bind_tools 把两件工具的 schema 附到模型实例上并返回新实例，绑定一次即可反复使用，这是第 5 篇的老朋友。函数体只有两步：入参 state["messages"] 是 reducer 合并后的全量历史——节点读全量、写增量，第 7 篇的规矩在这里落地；返回值只声明"新增了一条消息"，合并交给 add_messages。至于模型这一步输出的是带 tool_calls 的 AIMessage 还是纯文本的最终回答，节点本身不关心——区分是路由的事。
4、手写工具节点
到了本篇的核心。工具节点承担第 5 篇循环里"执行 + 回填"两步：取出模型最后一条消息里的 tool_calls，逐个执行，把每个结果包成 ToolMessage 回填。基础版如下：
from langchain_core.messages import ToolMessage
def call_tools(state: AgentState) -> dict:
tools_by_name = {t.name: t for t in tools}
results = []
last = state["messages"][-1] # 路由保证：末条必是 AIMessage
for tc in last.tool_calls: # 并行调用在此展开为列表遍历
output = tools_by_name[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(
content=str(output),
tool_call_id=tc["id"], # 结果与调用配对的唯一线索
))
return {"messages": results}
先看取数。state["messages"][-1] 取末条消息——能进入工具节点，说明路由已确认它是带 tool_calls 的 AIMessage，这是工具节点与路由之间的隐式契约，也是手写版最值得警惕的耦合点，防御性更强的写法可以先断言 last.type == "ai" 再往下走。tool_calls 是列表：第 5 篇的并行调用（一次响应发起多个调用）在这里天然覆盖，for 循环逐个处理即可。
再看执行与回填。tools_by_name 按名字查表，tc["args"] 就是参数字典，直接传给 invoke——真正执行工具的只有这一行。每个结果包成 ToolMessage 时，tool_call_id 必须原样写回 tc["id"]：这是模型把结果与请求配对的唯一线索，OpenAI 接口对不上号直接返回 400，第 5 篇踩过的坑在这里原样生效。最后返回增量：一轮的所有结果打包成列表，交给 add_messages 追加进历史。
基础版有个生产隐患：任何一件工具抛出异常，整张图当场崩掉，用户的提问以报错收场。第 5 篇 6.2 节的结论在节点层面同样成立——把异常转成错误字符串回传，而不是抛出去。模型读到可读的错误信息，下一轮通常自己修正参数重试：
def call_tools(state: AgentState) -> dict:
tools_by_name = {t.name: t for t in tools}
results = []
for tc in state["messages"][-1].tool_calls:
try:
tool = tools_by_name[tc["name"]] # 查表也进 try
content = str(tool.invoke(tc["args"]))
except Exception as e:
content = f"工具执行出错：{e!r}，请检查参数后重试"
results.append(ToolMessage(content=content,
tool_call_id=tc["id"]))
return {"messages": results}
相比基础版只多一层 try，但注意查表也挪进了保护范围——模型幻觉出不存在的工具名时，KeyError 同样被兜住而不是炸图。错误信息带上异常内容与"请检查参数"的提示，等于给模型一个明确的修正方向。这个"错误回传、模型自愈"的机制不是本篇的发明，prebuilt ToolNode 的 handle_tool_errors 参数做的就是这件事，第 7 节对照时回收这个伏笔，细节见 ToolNode 参考文档。
5、路由函数
图往哪走由一个函数说了算。第 8 篇的条件边需要一个 path 函数，返回值就是下一个节点的名字：
from langgraph.graph import END
def should_continue(state: AgentState) -> str:
last = state["messages"][-1]
if last.tool_calls: # 还有调用意图 → 去执行
return "tools"
return END # 已是最终回答 → 结束
判断的信息量小到奢侈：只看末条消息有没有 tool_calls。有，说明模型还想动手，把执行交给工具节点；没有，模型已经给出最终回答，返回内置常量 END 结束运行。should_continue 这个名字沿自官方示例，条件边的完整参数见 LangGraph Graph API 文档。
值得体会的是决策权的分配：路由函数自己不做决策，它只翻译模型的决策。要不要调用、调用什么、传什么参数，全部由模型决定；图只负责"模型说要调，就去执行；模型说完了，就收工"。路由保持愚蠢，正是 ReAct 得以简洁的原因。
6、组装与运行
6.1、六个调用拼成图
零件备齐，组装只要六个调用：
from langgraph.graph import StateGraph, START
builder = StateGraph(AgentState)
builder.add_node("model", call_model) # 决策节点
builder.add_node("tools", call_tools) # 执行节点
builder.add_edge(START, "model") # 入口：先思考再行动
builder.add_conditional_edges(
"model", should_continue, ["tools", END]) # 条件出口
builder.add_edge("tools", "model") # 回边：执行完回到决策
graph = builder.compile()
逐行看装配。add_edge(START, "model") 定入口：用户消息进图第一站是模型节点，Agent 的第一个动作永远是"想"而不是"做"。模型的出边用 add_conditional_edges 挂上路由函数，第三个参数列出全部可能的目的地，get_graph() 画拓扑时才能画完整。工具节点的出边是普通 add_edge：执行完无条件回到模型节点——工具从不决定"接下来干什么"，与第 5 节的分工呼应。
最后 compile() 把 builder 变成可执行的图，invoke、stream、挂 checkpointer（第 12 篇）都从这一步开始。回边让图能循环，也让图能死循环——模型若反复调用失败的工具，圈会一直转。LangGraph 的安全阀是 recursion_limit，默认 25：图每执行一个节点计一步，超限抛出 GraphRecursionError 终止整次运行。这正是第 5 篇 max_turns 的图版、第 8 篇递归上限的实战现场；需要更长循环时传 config={"recursion_limit": 50}。
6.2、多轮调用实录
写一个紧凑的打印函数按角色展示消息历史，然后带着问题跑一次：
from langchain_core.messages import HumanMessage
def show(messages):
for m in messages:
if m.type == "human":
print("用户:", m.content)
elif m.type == "ai" and m.tool_calls:
calls = ", ".join(f"{c['name']}({c['args']})"
for c in m.tool_calls)
print("模型调用:", calls)
elif m.type == "ai":
print("模型回答:", m.content)
else:
print("工具结果:", m.content)
question = "北京和上海哪座城市人口密度更高？"
show(graph.invoke(
{"messages": [HumanMessage(content=question)]})["messages"])
输出如下，数据的分轮策略是模型自己的选择：
用户: 北京和上海哪座城市人口密度更高？
模型调用: get_population({'city': '北京'}), get_population({'city': '上海'})
工具结果: 北京常住人口 2185 万人
工具结果: 上海常住人口 2487 万人
模型调用: get_area({'city': '北京'}), get_area({'city': '上海'})
工具结果: 北京陆地面积 16410 平方公里
工具结果: 上海陆地面积 6341 平方公里
模型回答: 上海更高：北京约 1332 人/平方公里，上海约 3922 人/平方公里，约为北京的 3 倍。
对着第 1 节的图数圈数：模型节点进了 3 次，工具节点进了 2 次，条件边判了 3 次。第一轮模型并行查两地人口，第二轮补查面积——分两轮取数是它自己的决策，一轮并行发四个调用同样合法；第三轮信息集齐，模型不再发起调用，should_continue 返回 END，图终止。两次"模型 → 工具 → 模型"的往返，就是那条回边转的两圈，而这一切对调用方只是一次同步的 invoke。
7、对照 create_agent
7.1、prebuilt 替你做的事
先看对照组。langchain 1.x 的统一入口 create_agent（create_react_agent 的继任者）只要一行，第 2 篇的第一个 Agent 用的就是它，见 create_agent 文档：
from langchain.agents import create_agent
agent = create_agent(llm, tools) # 与第 6 节的整张图等价
一行换掉本篇几十行，换来的是一整层工程化能力：
| 能力 | 手写版 | create_agent |
|---|---|---|
| 工具执行与回填 | 第 4 节的 15 行 | 内置 ToolNode 开箱即用 |
| 工具异常处理 | 自己写 try/except | handle_tool_errors 可配置 |
| 失败自我修复 | 错误回传靠模型自觉 | 同机制加错误格式化 |
| 流式语义标注 | 无，前端自行判断 | 区分最终回答与中间步骤 |
| 扩展方式 | 直接改节点代码 | middleware 挂标准钩子 |
展开说两件手写版没有的。一是流式：编译后的图本就能 stream（第 13 篇展开），但 create_agent 额外给消息打上语义标签——哪段 token 属于最终回答、哪段是工具调用前的中间思考，前端渲染不用自己猜，这对聊天产品的体验是决定性的。二是扩展模型：middleware 在模型调用前后、工具调用前后提供标准挂钩，日志、护栏、动态插工具都往这层挂，不用动图的结构，机制见 LangChain v1 发布说明。
但最值得记住的是一件事：create_agent 返回的也是一张编译好的图。它没有另一个运行时，内部结构与本篇手写的图同构——模型节点、工具节点、条件回边一样不少，agent.get_graph() 可以把拓扑画出来亲自验证。所谓 prebuilt，是"替你写图的语法糖"；分层设计的意义就在这里，你在本篇理解的每个节点，在 prebuilt 里同名同位。
7.2、手写的适用场景
那什么时候值得手写？三类场景：
- 理解与排障：读懂别人的 Agent、排查死循环、面试讲清原理
- 非标准拓扑：多模型节点、工具结果分级、在回边里插审批
- 教学与演示：每一步都要可打印、可断点、可讲解
反过来，标准 ReAct 场景、要快速接入 LangSmith 追踪与 middleware 生态时，用 create_agent 起步，别重复造轮子。判断标准一句话：要改的是图的形状，手写；要加的是节点前后的行为，prebuilt 加 middleware。第 10 篇会把 create_agent 的参数逐个拆开，届时你会认出里面每一个概念——因为本篇你都亲手写过。
8、总结
本篇把第 5 到 8 篇的四块积木拼成了完整的 ReAct Agent：add_messages 让消息历史在节点间累积，模型节点 bind_tools 后输出决策，手写的工具节点遍历 tool_calls、以 tool_call_id 配对回填，should_continue 路由接上条件回边，六个调用组装成图。多轮调用的实录验证了闭环——回边转了两圈后，模型自行收敛到最终回答。不到 60 行的代码（含示例工具），就是一切 Agent 框架共享的内核。
留一个练习，把这张图升级成多跳问答 Agent：
- 加一件
search_wiki工具，输入关键词，返回两三条预置的百科摘要 - 提问："《三体》的作者拿过哪个科幻最高奖？"
- 验收：至少两轮搜索，且第二轮的 query 用到了第一轮的结果
- 进阶：把 recursion_limit 调到 3，看报错在第几步触发
对于正在选型的团队，建议先用 create_agent 验证业务闭环，撞到标准拓扑装不下的需求再回到本篇的手写骨架——骨架你已经亲手搭过一遍，回去的路随时畅通。下一篇拆解 create_agent 的完整参数表与 middleware 体系，把"一行"背后的旋钮逐个拧给你看。
