Day 53 / 共 60 天 · 阶段 9 预制件与流式
create_react_agent 总览:一行函数背后的一张图
前 52 天我们把 LangGraph 的底盘全拆过了:StateGraph、Pregel、通道、检查点、中断、函数式 API。今天开始进入预制件(prebuilt)——官方用底盘拼好、开箱即用的成品。最有名的就是 create_react_agent:你只给它「一个模型 + 一堆工具」,它还你一个能「想→调工具→再想→回答」循环的 Agent。今天先鸟瞰:这一行函数,究竟在幕后帮你 add_node / add_edge 了什么。
📍 阶段 9 · 预制件与流式(6 天)你在这里
D53 总览→
D54 逐行①→
D55 ToolNode→
D56 校验→
D57 流式5模式→
D58 流式底层
💡 用一个类比先兜住今天
create_react_agent 像一家「装修队」。你交给它毛坯房(一个聊天模型)和家具清单(工具列表),它按一套标准户型给你装好:客厅(agent 节点,负责让模型思考)、工具间(tools 节点,负责真正执行工具)、以及一条走廊(should_continue 条件边,决定"还要用工具"就去工具间、"想好了"就出门)。你搬进去(invoke)就能住。今天我们不看装修细节(那是 D54/D55),先看整张户型图长啥样、由哪几块拼成。L01
它到底给你搭了什么?先看"骨架"
🤔 痛点ReAct(Reason + Act)循环人人会讲:模型先推理、要用工具就调、拿到结果再推理、直到能回答。但自己用 StateGraph 手搓一遍,要写状态、写模型节点、写工具节点、写条件边、处理并行工具调用、处理死循环……几十行且容易错。有没有"一行搞定"?
💡 本质
create_react_agent 就是把「手搓 ReAct 图」的那几十行封装成一个工厂函数。它内部还是老老实实用 StateGraph / add_node / add_conditional_edges / compile——没有任何黑魔法。看懂它,等于把前 52 天的知识串成一个真实产品。整个函数定义在 chat_agent_executor.py:278,最核心的"搭图"动作集中在函数尾部 chat_agent_executor.py:860-1003。先看它建图的三行主干:
# chat_agent_executor.py:860
workflow = StateGraph(
state_schema=state_schema or AgentState, context_schema=context_schema
)
# :864 两个核心节点
workflow.add_node("agent", RunnableCallable(call_model, acall_model), input_schema=input_schema)
workflow.add_node("tools", tool_node)
# :884 入口
workflow.set_entry_point(entrypoint) # 默认 entrypoint = "agent"
StateGraph(state_schema=...)用一个 StateGraph(D03 学过的)当画布。状态默认是 AgentState(下一讲讲)。add_node("agent", ...)第一个节点叫 agent:里面是 call_model(同步)/acall_model(异步),负责"让模型思考"。add_node("tools", tool_node)第二个节点叫 tools:就是一个 ToolNode(D55 主角),负责"真正执行工具"。set_entry_point("agent")入口是 agent——每次运行先让模型思考。(若配了 pre_model_hook,入口会前移,见 L06)👶 一句话两个节点(agent 想、tools 做)+ 一条会拐弯的走廊,就是整个 ReAct Agent。剩下全是"锦上添花"的可选件(结构化输出、钩子、断点)。
L02
AgentState:Agent 的"记忆表"
图要跑,得先有"状态 schema"。默认状态定义在 chat_agent_executor.py:57:
# chat_agent_executor.py:57
class AgentState(TypedDict):
"""The state of the agent."""
messages: Annotated[Sequence[BaseMessage], add_messages]
remaining_steps: NotRequired[RemainingSteps]
# :69 Pydantic 版本(同字段,默认 remaining_steps=25)
class AgentStatePydantic(BaseModel):
messages: Annotated[Sequence[BaseMessage], add_messages]
remaining_steps: RemainingSteps = 25
messages: Annotated[..., add_messages]核心字段:一串消息,reducer 用 add_messages(D08 学过:按 id 去重/更新、支持删除)。整个对话历史就存这里。remaining_steps: RemainingSteps剩余步数配额。NotRequired 表示可不传——LangGraph 运行时会自动注入一个"还能走几步"的数(防死循环,见 D54 的 _are_more_steps_needed)。AgentStatePydantic同一张表的 Pydantic 写法,字段一样,只是默认 remaining_steps=25。你想要运行时校验就用它。注意源码里
AgentState 顶上有 @deprecated(:53):官方把它迁到了 langchain.agents。但底层机制不变,读它仍是理解 ReAct Agent 状态设计的最佳样本。💡 设计取舍①:为什么状态只放 messages(+步数),不放"当前在想什么/工具结果单独存"?朴素设计会想给 Agent 加一堆字段:
thoughts、tool_results、scratchpad……但 ReAct 的精髓是一切皆消息:模型的思考和工具调用请求是 AIMessage,工具结果是 ToolMessage,用户输入是 HumanMessage。全塞进一条 messages 列表,模型下一轮直接拿整条历史当上下文——无需额外拼装。字段越少,状态越好序列化、好持久化(D33 的检查点)、好流式(D57)。这是"用统一的消息协议换掉一堆专用字段"的极简主义。📝 真实值跑一轮"北京天气"后,
state["messages"] 大概是:[HumanMessage("北京天气?"), AIMessage(tool_calls=[{name:"weather",args:{city:"北京"}}]), ToolMessage("晴 25℃"), AIMessage("北京今天晴,25℃")]。四条消息就是完整一次 ReAct。L03
函数签名:所有"旋钮"一次看清
入口签名 chat_agent_executor.py:278,裁剪掉类型噪音看关键参数:
# chat_agent_executor.py:278
def create_react_agent(
model, # 模型:实例 / "openai:gpt-4" 字符串 / 动态可调用
tools, # 工具:list[BaseTool|Callable|dict] 或一个 ToolNode
*,
prompt=None, # 系统提示:字符串 / SystemMessage / 函数 / Runnable
response_format=None, # 结构化输出 schema(额外加一个节点)
pre_model_hook=None, # 调模型前的钩子节点
post_model_hook=None, # 调模型后的钩子节点
state_schema=None, # 自定义状态(默认 AgentState)
context_schema=None, # 运行时上下文类型
checkpointer=None, # 检查点(记忆,D33)
store=None, # 长期记忆 Store(D40)
interrupt_before=None, # 静态断点(D43)
interrupt_after=None,
version="v2", # v1=一次性去 tools;v2=Send 并行分发
name=None,
...
) -> CompiledStateGraph:
model / tools唯二必填。其余全是可选"旋钮"。prompt系统提示的四种形态(字符串/SystemMessage/函数/Runnable),D54 会看 _get_prompt_runnable 如何统一它们。checkpointer / store直接透传给 compile——所以前面学的记忆机制"免费"接上了。version="v2"关键开关:v1 一次性把整条 AIMessage 丢给 tools 节点;v2 用 Send(D16)把每个 tool_call 并行分发成多个 tools 任务。默认 v2。💡 本质:这个签名就是"ReAct 的所有可配置维度"你会发现每个参数都对应前面某一天:
checkpointer→D33、interrupt_before→D43、version→D16 的 Send、state_schema→D05。预制件不是新知识,是老知识的组合套餐。L04
两个节点从哪来:tool_node 与 call_model
建图前,函数先把 tools 参数规整成一个 ToolNode(chat_agent_executor.py:553):
# chat_agent_executor.py:553
llm_builtin_tools: list[dict] = []
if isinstance(tools, ToolNode):
tool_classes = list(tools.tools_by_name.values())
tool_node = tools # 你直接传了 ToolNode,就用它
else:
llm_builtin_tools = [t for t in tools if isinstance(t, dict)] # dict = 模型内建工具
tool_node = ToolNode([t for t in tools if not isinstance(t, dict)]) # 其余包成 ToolNode
tool_classes = list(tool_node.tools_by_name.values())
isinstance(tools, ToolNode)贴心:你可以直接传一个自己配好的 ToolNode(比如自定义了错误处理),它就原样采用。isinstance(t, dict)list 里的 dict 是"模型内建工具"(如 OpenAI 的 web_search),它们不进 ToolNode,而是 llm_builtin_tools,稍后随 bind_tools 一起给模型。ToolNode([...])普通函数/BaseTool 全塞进一个 ToolNode——这就是图里那个 "tools" 节点的本体。而 "agent" 节点里是 call_model(chat_agent_executor.py:661),它做三件事,逐行 D54 讲,这里只看骨架:
# chat_agent_executor.py:661(裁剪)
def call_model(state, runtime, config):
model_input = _get_model_input_state(state) # ① 取消息、校验历史
response = static_model.invoke(model_input, config) # ② 调模型(含 prompt+bind_tools)
response.name = name
if _are_more_steps_needed(state, response): # ③ 步数不够就返回"抱歉"
return {"messages": [AIMessage(id=response.id, content="Sorry, need more steps...")]}
return {"messages": [response]} # 正常:把 AIMessage 追加进状态
👶 一句话agent 节点 = "把历史丢给模型,拿回一条 AIMessage 追加进 messages"。tools 节点 = "把 AIMessage 里的 tool_calls 执行了,产出 ToolMessage"。两者交替,就是 ReAct。
L05
should_continue:那条会拐弯的走廊
两个节点靠一条条件边连起来。路由函数 chat_agent_executor.py:831:
# chat_agent_executor.py:831
def should_continue(state) -> str | list[Send]:
messages = _get_state_value(state, "messages")
last_message = messages[-1]
# 模型没要求调工具 → 结束(或去钩子/结构化输出节点)
if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
if post_model_hook is not None:
return "post_model_hook"
elif response_format is not None:
return "generate_structured_response"
else:
return END
# 模型要求调工具 → 去 tools
else:
if version == "v1":
return "tools"
elif version == "v2":
if post_model_hook is not None:
return "post_model_hook"
return [ # ★ v2:每个 tool_call 一个 Send
Send("tools", ToolCallWithContext(
__type="tool_call_with_context", tool_call=call, state=state))
for call in last_message.tool_calls
]
last_message = messages[-1]看最后一条消息:它一定是 agent 节点刚产出的 AIMessage。not last_message.tool_calls → END模型没要求调工具(直接回答了)→ 走出门(END)。这就是循环的出口。version=="v1": return "tools"v1:整条 AIMessage 一次性丢给 tools 节点,节点内部串行/并行处理所有 tool_call。version=="v2": return [Send(...)]v2 核心:给每个 tool_call 发一个 Send(D16 的动态扇出),tools 节点被并行实例化多份,每份跑一个工具。💡 设计取舍②:v1 vs v2——为什么默认换成 Send 扇出?v1 把所有 tool_call 塞进一个 tools 节点,节点内用线程池并行(D55 会看到)。看似也并行,但整个节点是一个超步里的一个任务:任意一个工具触发
interrupt()(人在环,D41),整批都得停、恢复时容易搞混谁执行过了。v2 用 Send 把每个工具变成独立任务:单个工具中断/重试互不影响,检查点能精确记录"哪几个 Send 完成了"。代价是图的执行记录更碎(更多任务事件)。官方权衡后默认 v2——为了人在环和容错的正确性,牺牲一点执行记录的简洁。L06
编译成图:可选件如何"插进"骨架
骨架有了,函数再把可选件按需接上,最后 compile。看连边与编译(chat_agent_executor.py:872-1003):
# chat_agent_executor.py:872 pre_model_hook 让入口前移
if pre_model_hook is not None:
workflow.add_node("pre_model_hook", pre_model_hook)
workflow.add_edge("pre_model_hook", "agent")
entrypoint = "pre_model_hook"
else:
entrypoint = "agent"
# :966 agent 出来走 should_continue
workflow.add_conditional_edges("agent", should_continue, path_map=agent_paths)
# :988 tools 跑完回到入口(无 return_direct 时)
workflow.add_edge("tools", entrypoint)
# :995 编译,透传所有记忆/中断配置
return workflow.compile(
checkpointer=checkpointer, store=store,
interrupt_before=interrupt_before, interrupt_after=interrupt_after,
debug=debug, name=name,
)
pre_model_hook → entrypoint 前移配了前置钩子,入口就变成 hook,hook 跑完再 add_edge 到 agent。用来做"调模型前压缩历史"等。add_conditional_edges("agent", should_continue)把 L05 的路由挂在 agent 出口。path_map 提前声明所有可能去处(画图/校验用)。add_edge("tools", entrypoint)工具跑完回到入口——形成"agent→tools→agent"的循环。这就是 ReAct 的"环"。compile(checkpointer=...)最后编译。你传的记忆、断点全在这里生效——预制件本身不发明记忆,只是转交给底盘。⚠️ 边界:return_direct 的工具会"抄近道"出门如果某工具设了
return_direct=True(工具结果直接当最终答案,不再让模型总结),tools 后就不能无脑回 agent,否则模型会对着结果再啰嗦一轮。源码为此在 chat_agent_executor.py:970 加了 route_tool_responses:若最后的 ToolMessage 来自 return_direct 工具,直接 END。这是"并行工具里混着 return_direct"的经典坑——它甚至处理了"return_direct 工具在另一个 Send 里执行"的竞态。L07
全景图 + 今日小结
图注:两节点都读写同一条 messages 通道,靠 add_messages 累加成完整对话。
图注:agent→判断→tools→回 agent 是"环",判断为"否"时才从 END 出门。
🧠 今天你应该能回答
- create_react_agent 内部用了哪几个底盘 API?(StateGraph / add_node / add_conditional_edges / compile)
- 它建了哪两个节点?各干啥?(agent 调模型产 AIMessage;tools 执行工具产 ToolMessage)
- AgentState 为什么只有 messages + remaining_steps?(一切皆消息,极简好序列化)
- should_continue 何时结束、何时去 tools?(无 tool_calls→END;有→tools 或 Send 扇出)
- v1 与 v2 的区别?(v1 整条丢 tools;v2 用 Send 每个 tool_call 并行独立任务)
- checkpointer/interrupt 是它发明的吗?(不是,只是透传给 compile)
✋ 10 分钟动手
# 1. 通读建图段(骨架就这 140 行)
sed -n '553,561p' libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py
sed -n '831,1003p' libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py
# 2. 亲手看它画出来的图
python - <<'PY'
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
@tool
def add(a:int,b:int)->int:
"加法"
return a+b
# 用假模型也行,这里只看图结构
agent = create_react_agent("openai:gpt-4o-mini", [add])
print(agent.get_graph().draw_ascii()) # 打印 START/agent/tools/END 拓扑
PY
明天预告 · Day 54:钻进 agent 节点内部——
_get_prompt_runnable 如何把四种 prompt 统一成 Runnable、_should_bind_tools 如何判断"要不要绑工具"、call_model 逐行、以及 remaining_steps 防死循环的门槛判断。