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 加一堆字段:thoughtstool_resultsscratchpad……但 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→D33interrupt_before→D43version→D16 的 Sendstate_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_modelchat_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

全景图 + 今日小结

数据结构:AgentState 在两节点间流动 agent 节点(call_model) 读 messages → 调模型 产出 AIMessage 写回 {"messages":[response]} tools 节点(ToolNode) 读 AIMessage.tool_calls 执行工具 产出 ToolMessage 列表 messages 通道 add_messages 累加 写 AIMessage 读消息 回写 ToolMessage
图注:两节点都读写同一条 messages 通道,靠 add_messages 累加成完整对话。
控制流:ReAct 循环(should_continue 是分叉点) START agent 有tool_calls? tools END tools→回 agent(循环)
图注: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 防死循环的门槛判断。
← Day 52 节点级重试 Day 54 · 逐行① 模型节点 →