LangChain 源码学习路线：第 5 篇 create_agent 与 ReAct Agent 运行机制源码解剖#

核心问题： create_agent 如何把模型、工具、中间件和结构化输出策略装配成可持续运行的 ReAct Agent Runtime？

源码主线： create_agent → CompiledStateGraph → model node → ToolNode → route → END

前置文章： 第 1 篇 Runnable、第 2 篇 Prompt / Message / ChatModel、第 3 篇 OutputParser、第 4 篇 Tool Calling

依赖基线： langchain==1.3.11、langchain-core==1.4.8、langgraph==1.2.7

源码基线： LangChain commit 83e824922ac9bf3bf882347fecabd09614bccc37，LangGraph commit 5931a5f0b313feff24e2516a586c55601b868ac1

阅读边界： 本篇覆盖标准 ReAct Agent Harness，不展开自定义复杂业务图、持久化实现和分布式部署。

0. 本篇在源码学习主线中的位置#

前四篇分别解决了统一执行协议、消息与模型对象流转、结构化输出，以及工具调用协议。本篇把这些能力收束成一个可以持续运行的 Agent Runtime：

1
Runnable 执行协议
2
  + Prompt / Message / ChatModel
3
  + Structured Output
4
  + Tool Calling
5
  ↓
6
create_agent
7
  ↓
8
CompiledStateGraph
9
  ↓
10
model → tools → model

阅读时不要把 create_agent 理解成“帮你写好一个 while 循环”。更准确的心智模型是：

create_agent 是一个图工厂：它把模型、工具、中间件、状态 Schema、结构化输出策略和运行时能力装配成可执行的 LangGraph 状态图。

全文按四条源码链路展开：

1
构建期：参数归一化 → State schema → 节点与边 → compile
2
模型链：before_model → ModelRequest → wrap_model_call → ChatModel → after_model
3
工具链：ToolNode 解析 → 并行调度 → 工具执行 → ToolMessage / Command
4
控制链：条件路由 → structured_response → interrupt → recursion_limit

1. 本篇问题、学习目标与能力边界#

本篇目标是理解：

LangChain Agent 不是一个神秘的“智能体对象”，而是由 create_agent 构造出来的 LangGraph 编译状态图。它通过模型节点、工具节点、条件边、回边和中间件，把 ReAct 类的“模型推理—工具调用—观察反馈—再次推理”循环工程化。

你需要掌握：

create_agent
CompiledStateGraph
AgentState
model node
tools node
ToolNode
middleware
before_agent
before_model
after_model
after_agent
wrap_model_call
wrap_tool_call
tool loop
stop condition
recursion_limit
iteration limit
structured response
response_format
ProviderStrategy
ToolStrategy

完成本篇后，你应该能回答：

create_agent 返回的到底是什么对象？
它内部是否构建了 LangGraph graph？
model node 和 tools node 如何连接？
AIMessage.tool_calls 如何被识别？
工具执行结果如何重新进入 messages？
middleware 在模型调用前后能做什么？
wrap_model_call 和 wrap_tool_call 为什么比普通 callback 更强？
stop condition 在哪里判断？
iteration limit 与 LangGraph 的 recursion_limit 是什么关系？
structured_response 如何进入最终 Agent State？
为什么现代 LangChain Agent 不一定显式暴露 Thought？
什么时候应该使用 create_agent，什么时候应该自己写 LangGraph？

2. 核心概念与最小心智模型#

经典 ReAct 范式#

经典 ReAct 可以抽象成：

1
Thought
2
  ↓
3
Action
4
  ↓
5
Observation
6
  ↓
7
Thought
8
  ↓
9
Action
10
  ↓
11
Observation
12
  ↓
13
Final Answer

其中：

ReAct 概念	含义
Thought	模型对下一步的推理
Action	模型决定调用工具
Action Input	模型生成工具参数
Observation	工具返回结果
Final Answer	模型最终回答

LangChain / LangGraph 现代映射#

现代 LangChain 不一定强制模型输出显式的 Thought: 文本，而是使用结构化 message trace 表达 ReAct 行为：

1
ReAct
2
→ model call
3
→ AIMessage(tool_calls=[...])
4
→ tools node
5
→ ToolMessage
6
→ model call
7
→ final AIMessage

对应关系如下：

ReAct 经典概念	LangChain / LangGraph 实现
Thought	不一定显式暴露；可通过 trace、reasoning block、metadata 观察
Action	`AIMessage.tool_calls[i]["name"]`
Action Input	`AIMessage.tool_calls[i]["args"]`
Observation	`ToolMessage.content`
Action ID	`ToolCall.id`
Observation ID	`ToolMessage.tool_call_id`
Loop	`model node → tools node → model node`
Stop Condition	最后一条 `AIMessage` 不再包含 `tool_calls`
Final Answer	`AIMessage(content=...)`
Structured Final	`state["structured_response"]`

为什么生产中更推荐结构化 trace#

旧式 ReAct 常要求模型输出：

1
Thought: 我需要搜索东京热门景点。
2
Action: search_city_info
3
Action Input: 东京热门景点
4
Observation: ...
5
Final Answer: ...

现代 Agent 工程更推荐观察：

1
AIMessage
2
AIMessage.tool_calls
3
ToolMessage
4
final AIMessage
5
LangSmith / LangGraph trace

原因：

tool_calls 是结构化字段，比自然语言 Action: 更稳定；
ToolMessage.tool_call_id 可以精确关联工具请求与工具结果；
多工具并行调用时，结构化 ID 比文本解析可靠；
显式 Thought 可能增加 token 成本，也可能带来安全与隐私风险；
生产排查更关心“调用了什么工具、传了什么参数、返回了什么结果、为什么停止”。

3. 完整执行链路#

最小代码并入执行链，用于观察 create_agent 返回对象和 ReAct 循环。

最小代码#

1
from langchain.agents import create_agent
2
from langchain.tools import tool
3

4
@tool
5
def search_city_info(query: str) -> str:
6
    """Search city information for travel planning."""
7
    return f"Search result for {query}"
8

9
agent = create_agent(
10
    model="openai:gpt-4.1-mini",
11
    tools=[search_city_info],
12
    system_prompt="你是一个旅行规划助手。"
13
)
14

15
result = agent.invoke({
16
    "messages": [
17
        {
18
            "role": "user",
19
            "content": "帮我规划东京 5 天行程，需要查一下热门景点"
20
        }
21
    ]
22
})
23

24
print(result)

你应该观察什么#

不要只看最终回答，要打印完整 messages：

1
for i, message in enumerate(result["messages"]):
2
    print("=" * 20, i)
3
    print(type(message))
4
    print(message)
5
    print("tool_calls:", getattr(message, "tool_calls", None))
6
    print("tool_call_id:", getattr(message, "tool_call_id", None))

你应该重点观察：

1
HumanMessage
2
  ↓
3
AIMessage(tool_calls=[...])
4
  ↓
5
ToolMessage(tool_call_id=...)
6
  ↓
7
AIMessage(content=最终回答)

create_agent 的输出不是字符串#

agent.invoke(...) 的返回值是最终 Agent State，常见结构是：

1
{
2
    "messages": [
3
        HumanMessage(...),
4
        AIMessage(tool_calls=[...]),
5
        ToolMessage(...),
6
        AIMessage(content="...")
7
    ]
8
}

如果配置了结构化输出：

1
{
2
    "messages": [...],
3
    "structured_response": TravelPlan(...)
4
}

这说明：

1
create_agent 返回的是可执行状态图，
2
不是 prompt | model | parser 那种简单线性链。

执行流程#

高层执行流程#

1
用户 messages
2
  ↓
3
model node
4
  ↓
5
AIMessage 是否包含 tool_calls？
6
  ↓
7
如果有：tools node 执行工具
8
  ↓
9
ToolMessage 加入 messages
10
  ↓
11
回到 model node
12
  ↓
13
直到模型输出 final answer 或达到停止条件

细粒度执行流程#

1
agent.invoke(input_state)
2
  ↓
3
CompiledStateGraph.invoke
4
  ↓
5
初始化 AgentState
6
  ↓
7
进入 before_agent middleware
8
  ↓
9
进入 before_model middleware
10
  ↓
11
组装 model request
12
  ↓
13
注入 system_prompt
14
  ↓
15
绑定 tools / response_format
16
  ↓
17
wrap_model_call middleware 包裹模型调用
18
  ↓
19
ChatModel.invoke(messages)
20
  ↓
21
得到 AIMessage
22
  ↓
23
after_model middleware
24
  ↓
25
把 AIMessage 追加到 state["messages"]
26
  ↓
27
route_after_model 判断是否有 tool_calls
28
  ↓
29
有 tool_calls：进入 tools node
30
  ↓
31
ToolNode 读取最后一条 AIMessage.tool_calls
32
  ↓
33
根据 name 找到 BaseTool
34
  ↓
35
构造 ToolRuntime / 注入 state、store、config、tool_call_id
36
  ↓
37
wrap_tool_call middleware 包裹工具调用
38
  ↓
39
BaseTool.invoke(full_tool_call)
40
  ↓
41
返回 ToolMessage
42
  ↓
43
ToolMessage 追加到 state["messages"]
44
  ↓
45
tools node 回到 model node
46
  ↓
47
模型根据 ToolMessage 继续推理
48
  ↓
49
最后一条 AIMessage 没有 tool_calls
50
  ↓
51
END
52
  ↓
53
after_agent middleware
54
  ↓
55
返回最终 Agent State

4. 源码地图、关键文件与阅读顺序#

核心目录#

1
langchain/agents/
2
langchain/agents/factory.py
3
langchain/agents/middleware/
4
langchain/agents/structured_output.py
5

6
langgraph/prebuilt/
7
langgraph/prebuilt/tool_node.py
8

9
langchain_core/messages/
10
langchain_core/tools/
11
langchain_core/language_models/

重点文件#

1
libs/langchain/langchain/agents/factory.py
2
libs/langchain/langchain/agents/middleware/
3
libs/langchain/langchain/agents/structured_output.py
4

5
libs/prebuilt/langgraph/prebuilt/tool_node.py
6

7
libs/core/langchain_core/messages/ai.py
8
libs/core/langchain_core/messages/tool.py
9
libs/core/langchain_core/tools/base.py
10
libs/core/langchain_core/language_models/chat_models.py

重点类与函数#

1
create_agent
2
AgentState
3
AgentMiddleware
4
ModelRequest
5
ModelResponse
6
ToolCallRequest
7

8
before_agent
9
before_model
10
after_model
11
after_agent
12
wrap_model_call
13
wrap_tool_call
14

15
ToolNode
16
ToolRuntime
17
ToolMessage
18
AIMessage
19
ToolCall
20

21
ProviderStrategy
22
ToolStrategy
23
AutoStrategy
24

25
StateGraph
26
CompiledStateGraph

5. 对象模型、继承关系与协议边界#

核心对象关系#

1
create_agent
2
  ↓
3
StateGraph
4
  ├─ model node
5
  ├─ tools node / ToolNode
6
  ├─ middleware nodes and wrappers
7
  └─ conditional edges
8
  ↓
9
CompiledStateGraph

协议边界#

对象	负责什么	不负责什么
`create_agent`	装配标准 Agent 图	承载单次运行状态
model node	构造模型请求并保存 AIMessage	执行本地工具
`ToolNode`	执行 ToolCall 并返回状态更新	决定最终回答
middleware	包裹或拦截执行阶段	替代图拓扑

6. 源码阅读策略与证据标准#

按执行边界追踪源码#

前面的执行流程解决“Agent 如何运行”，接下来要解决“源码为什么这样组织”。不建议从 factory.py 第一行开始顺序阅读，那样很容易陷入参数分支和兼容逻辑。更高效的方式是围绕四个边界追踪：

阅读边界	起点	终点	需要回答的问题
工厂边界	`create_agent(...)`	`graph.compile(...)`	输入能力如何被装配成图
模型边界	`state["messages"]`	`AIMessage`	模型请求如何构造、包装与路由
工具边界	`AIMessage.tool_calls`	`ToolMessage / Command`	工具如何查找、注入、并行执行与归一化
控制边界	`route_after_model`	`END / interrupt / error`	循环如何继续、暂停或终止

每一节都使用同一套阅读模板：

1
职责 → 输入/输出 → 细粒度伪代码 → 分支解释 → 工程含义

伪代码只保留影响理解的控制流，不追求逐字复刻源码。真正阅读本地版本时，应同时对照函数签名、类型定义和调用方，避免把内部实现细节误当成稳定 API。

阅读顺序#

1
公开入口
2
  ↓
3
输入输出类型
4
  ↓
5
构建期对象
6
  ↓
7
运行时主链
8
  ↓
9
分支、异常与停止条件
10
  ↓
11
扩展接口

证据标准#

标记	使用条件
源码事实	当前正式版源码可以直接证明
官方契约	官方文档或 API Reference 明确承诺
简化伪代码	压缩真实控制流，且明确不是逐字源码
作者推断	根据调用关系得出，必须标注为推断
工程建议	说明适用条件，不写成框架保证

7. 构建期源码解剖#

构建期的职责不是执行 Agent，而是把用户提供的能力转换成一张结构明确、可以被 LangGraph 驱动的状态图：

1
model / tools / middleware / response_format
2
  ↓
3
参数与能力归一化
4
  ↓
5
合成 State schema
6
  ↓
7
构造 model node 与 tools node
8
  ↓
9
注册条件边、回边与 middleware 节点
10
  ↓
11
compile
12
  ↓
13
CompiledStateGraph

`create_agent` 顶层入口伪代码#

伪代码#

1
def create_agent(
2
    model,
3
    tools=None,
4
    *,
5
    system_prompt=None,
6
    middleware=(),
7
    response_format=None,
8
    state_schema=None,
9
    context_schema=None,
10
    checkpointer=None,
11
    store=None,
12
    interrupt_before=None,
13
    interrupt_after=None,
14
    debug=False,
15
    name=None,
16
    cache=None,
17
):
18
    # 1. 标准化模型
19
    chat_model = _init_chat_model(model)
20

21
    # 2. 标准化工具
22
    tool_objects = _normalize_tools(tools)
23

24
    # 3. 标准化 response_format
25
    response_strategy = _normalize_response_format(
26
        response_format,
27
        chat_model,
28
    )
29

30
    # 4. 标准化 middleware
31
    middleware_list = _normalize_middleware(middleware)
32

33
    # 5. 合成最终 State schema
34
    final_state_schema = _resolve_state_schema(
35
        state_schema,
36
        middleware_list,
37
        response_strategy,
38
    )
39

40
    # 6. 创建 StateGraph builder
41
    graph = StateGraph(
42
        final_state_schema,
43
        context_schema=context_schema,
44
    )
45

46
    # 7. 添加 middleware node
47
    _add_middleware_nodes(graph, middleware_list)
48

49
    # 8. 添加 model node
50
    graph.add_node("model", _make_model_node(...))
51

52
    # 9. 添加 tools node
53
    if tool_objects:
54
        graph.add_node("tools", _make_tools_node(...))
55

56
    # 10. 添加条件边和回边
57
    _add_agent_edges(graph, has_tools=bool(tool_objects), ...)
58

59
    # 11. 编译为 CompiledStateGraph
60
    return graph.compile(
61
        checkpointer=checkpointer,
62
        store=store,
63
        interrupt_before=interrupt_before,
64
        interrupt_after=interrupt_after,
65
        debug=debug,
66
        name=name,
67
        cache=cache,
68
    )

逐行解释#

第 1 步：标准化模型#

model 可以是字符串，也可以是已经实例化的 BaseChatModel。

1
model="openai:gpt-4.1-mini"

会被转换成：

1
ChatOpenAI(...)

或其他供应商对应的 BaseChatModel 子类。

第 2 步：标准化工具#

tools 中可能是：

1
BaseTool
2
普通 Python function
3
dict schema

它们会被统一成工具对象或模型可绑定的工具 schema。

第 3 步：标准化结构化输出#

response_format 可能是：

1
None
2
Pydantic model
3
TypedDict
4
JSON Schema
5
ProviderStrategy
6
ToolStrategy

源码会把它归一成内部策略对象。

第 4 步：标准化 middleware#

middleware 可能来自：

1
装饰器函数
2
AgentMiddleware 子类实例
3
预置 middleware

内部会统一成 middleware 实例，并按 hook 类型注册。

第 5 步：合成 State schema#

最终 Agent State 至少需要：

1
messages

如果有结构化输出，还需要：

1
structured_response

如果 middleware 声明额外 state 字段，也要合并进来。

第 6—11 步：构图与编译#

这部分说明 create_agent 的本质：

1
不是创建一个 while-loop 对象，
2
而是构建 StateGraph 并 compile。

模型初始化源码拆解#

伪代码#

1
def _init_chat_model(model):
2
    if isinstance(model, str):
3
        return init_chat_model(model)
4

5
    if isinstance(model, BaseChatModel):
6
        return model
7

8
    raise TypeError(
9
        "model must be a model identifier or BaseChatModel"
10
    )

解释#

create_agent 支持：

1
create_agent(model="openai:gpt-4.1-mini", tools=[...])

也支持：

1
from langchain.chat_models import init_chat_model
2

3
model = init_chat_model("openai:gpt-4.1-mini")
4
agent = create_agent(model=model, tools=[...])

源码这样设计是为了同时支持：

1
快速原型：传字符串
2
生产配置：传已经配置好的 model 实例

生产中更推荐传显式 model 实例，因为你可以设置：

1
timeout
2
max_retries
3
temperature
4
model_kwargs
5
base_url
6
api_key
7
profile
8
callbacks
9
tags
10
metadata

工具归一化源码拆解#

伪代码#

1
def _normalize_tools(tools):
2
    normalized = []
3

4
    for tool in tools or []:
5
        if isinstance(tool, BaseTool):
6
            normalized.append(tool)
7
            continue
8

9
        if callable(tool):
10
            normalized.append(create_tool(tool))
11
            continue
12

13
        if isinstance(tool, dict):
14
            normalized.append(tool)
15
            continue
16

17
        raise TypeError("Unsupported tool type")
18

19
    return normalized

解释#

这一步连接第 4 篇 Tool Calling：

1
普通函数
2
  ↓
3
@tool / create_tool
4
  ↓
5
StructuredTool
6
  ↓
7
BaseTool

例如：

1
def search_city_info(query: str) -> str:
2
    """Search city information."""
3
    return "..."

在工具归一化阶段会变成：

1
StructuredTool(
2
    name="search_city_info",
3
    description="Search city information.",
4
    args_schema=...
5
)

最终进入：

1
model.bind_tools([...])
2
ToolNode([...])

注意：

1
工具归一化不代表工具会被执行。

此时只是将工具登记到 Agent 能力表。

response_format 归一化源码拆解#

伪代码#

1
def _normalize_response_format(response_format, model):
2
    if response_format is None:
3
        return None
4

5
    if isinstance(response_format, ProviderStrategy):
6
        return response_format
7

8
    if isinstance(response_format, ToolStrategy):
9
        return response_format
10

11
    if is_schema_type(response_format):
12
        if model_supports_native_structured_output(model):
13
            return ProviderStrategy(schema=response_format)
14

15
        return ToolStrategy(schema=response_format)
16

17
    raise TypeError("Unsupported response_format")

解释#

response_format=TravelSummary 看起来只是传了一个 Pydantic 类，但源码会做策略选择：

1
模型支持原生结构化输出
2
  → ProviderStrategy
3

4
模型不支持原生结构化输出
5
  → ToolStrategy

这与第 3 篇结构化输出一致。

ProviderStrategy 心智模型#

1
TravelSummary schema
2
  ↓
3
转换为 provider native schema
4
  ↓
5
模型 API 原生约束输出
6
  ↓
7
框架校验
8
  ↓
9
state["structured_response"]

ToolStrategy 心智模型#

1
TravelSummary schema
2
  ↓
3
转换为“虚拟工具”
4
  ↓
5
模型通过 tool_call 输出结构化参数
6
  ↓
7
框架解析 tool call args
8
  ↓
9
Pydantic 校验
10
  ↓
11
state["structured_response"]

注意#

如果同时有业务工具和结构化输出：

1
create_agent(
2
    model=model,
3
    tools=[search_city_info],
4
    response_format=TravelSummary,
5
)

模型需要同时支持：

1
工具调用
2
结构化输出

如果供应商能力不一致，源码需要选择 fallback 策略。

middleware 归一化源码拆解#

伪代码#

1
def _normalize_middleware(middleware):
2
    normalized = []
3
    seen_names = set()
4

5
    for m in middleware:
6
        instance = _coerce_to_middleware(m)
7

8
        if instance.name in seen_names:
9
            raise ValueError("Duplicate middleware name")
10

11
        seen_names.add(instance.name)
12
        normalized.append(instance)
13

14
    return normalized

解释#

middleware 的作用不是简单 callback，而是 Agent loop 的扩展点。

一个 middleware 可能声明：

1
before_agent
2
before_model
3
after_model
4
after_agent
5
wrap_model_call
6
wrap_tool_call
7
state_schema
8
tools

源码会将它拆到不同位置：

1
node-style hook
2
  → 编译成图节点或节点前后逻辑
3

4
wrap-style hook
5
  → 包裹 model call 或 tool call
6

7
middleware.tools
8
  → 合并进 Agent 工具列表
9

10
middleware.state_schema
11
  → 合并进 Agent State

State schema 合成源码拆解#

最小 AgentState#

1
class AgentState(TypedDict):
2
    messages: Annotated[list[AnyMessage], add_messages]

加 structured_response 后#

1
class AgentStateWithStructuredResponse(TypedDict):
2
    messages: Annotated[list[AnyMessage], add_messages]
3
    structured_response: TravelSummary | None

加 middleware state 后#

1
class CustomAgentState(TypedDict):
2
    messages: Annotated[list[AnyMessage], add_messages]
3
    structured_response: TravelSummary | None
4
    tool_call_count: int
5
    risk_level: str

伪代码#

1
def _resolve_state_schema(
2
    user_state_schema,
3
    middleware_list,
4
    response_strategy,
5
):
6
    if user_state_schema is not None:
7
        base_schema = user_state_schema
8
    else:
9
        base_schema = AgentState
10

11
    if response_strategy is not None:
12
        base_schema = _add_structured_response_field(
13
            base_schema
14
        )
15

16
    for middleware in middleware_list:
17
        if middleware.state_schema is not None:
18
            base_schema = _merge_state_schema(
19
                base_schema,
20
                middleware.state_schema,
21
            )
22

23
    return base_schema

解释#

messages 字段最关键，因为它承载整个 ReAct trace：

1
HumanMessage
2
AIMessage(tool_calls)
3
ToolMessage
4
AIMessage(final)

messages 不能简单覆盖，必须用 reducer 追加。

构建期总结、伪代码与源码证据#

以下为保留关键控制流的简化伪代码，不是源码逐字复制：

1
def create_agent(model, tools, middleware, response_format):
2
    state_schema = compose_state_schema(middleware, response_format)
3
    graph = StateGraph(state_schema)
4
    graph.add_node("model", build_model_node(model, middleware))
5
    graph.add_node("tools", ToolNode(tools))
6
    graph.add_conditional_edges("model", route_after_model)
7
    graph.add_edge("tools", "model")
8
    return graph.compile()

源码证据：

https://github.com/langchain-ai/langchain/blob/83e824922ac9bf3bf882347fecabd09614bccc37/libs/langchain/langchain/agents/factory.py

8. 运行时主链源码解剖#

本章合并 model node 与 ToolNode，形成完整的一次循环主链。

Model 调用链：从 State 到 AIMessage#

model node 是 Agent Runtime 的决策中心。它不只是调用一次 ChatModel，而是负责把当前 State 转换为模型请求，让 middleware 有机会介入，再把模型输出转换为可路由的状态更新。

阅读这一链路时，始终跟踪三个对象：

1
Agent State
2
  ↓
3
ModelRequest
4
  ↓
5
AIMessage / structured response

model node 构造源码拆解#

model node 的职责#

1
读取 state["messages"]
2
  ↓
3
注入 system_prompt
4
  ↓
5
执行 before_model middleware
6
  ↓
7
构造 ModelRequest
8
  ↓
9
绑定 tools / response_format
10
  ↓
11
执行 wrap_model_call 链
12
  ↓
13
调用 ChatModel
14
  ↓
15
执行 after_model middleware
16
  ↓
17
返回 {"messages": [AIMessage]}

伪代码：model node 外壳#

1
def _make_model_node(
2
    model,
3
    tools,
4
    system_prompt,
5
    middleware_list,
6
    response_strategy,
7
):
8
    def model_node(state, runtime, config):
9
        # 1. 执行 before_model hooks
10
        state = _run_before_model_hooks(
11
            state,
12
            runtime,
13
            middleware_list,
14
        )
15

16
        # 2. 构造模型输入 messages
17
        messages = _prepare_messages(
18
            state=state,
19
            system_prompt=system_prompt,
20
        )
21

22
        # 3. 构造 ModelRequest
23
        request = ModelRequest(
24
            model=model,
25
            messages=messages,
26
            tools=tools,
27
            response_format=response_strategy,
28
            state=state,
29
            runtime=runtime,
30
            config=config,
31
        )
32

33
        # 4. 调用模型，经过 wrap_model_call 链
34
        response = _call_model_with_middleware(
35
            request,
36
            middleware_list,
37
        )
38

39
        # 5. 将 response.message 转成 state update
40
        update = {
41
            "messages": [response.message]
42
        }
43

44
        # 6. 执行 after_model hooks
45
        update = _run_after_model_hooks(
46
            state=state,
47
            update=update,
48
            runtime=runtime,
49
            middleware_list=middleware_list,
50
        )
51

52
        return update
53

54
    return model_node

before_model hooks 细拆#

伪代码#

1
def _run_before_model_hooks(state, runtime, middleware_list):
2
    current_state = state
3

4
    for middleware in middleware_list:
5
        hook = middleware.before_model
6

7
        if hook is None:
8
            continue
9

10
        update = hook(current_state, runtime)
11

12
        if update is None:
13
            continue
14

15
        if "jump_to" in update:
16
            return _apply_jump(update)
17

18
        current_state = _merge_state_update(
19
            current_state,
20
            update,
21
        )
22

23
    return current_state

每一步解释#

读取 hook#

1
hook = middleware.before_model

有些 middleware 只实现工具 wrapper，不实现 before_model。

执行 hook#

1
update = hook(current_state, runtime)

它可以返回：

1
None
2
state update dict
3
带 jump_to 的 dict

合并 state#

1
current_state = _merge_state_update(...)

如果更新的是 messages，会走 reducer 追加或移除逻辑。

典型 before_model 用途#

1
消息裁剪
2
上下文摘要
3
注入用户偏好
4
动态 system prompt
5
安全检查
6
超过消息上限直接跳到 END

system_prompt 注入源码拆解#

伪代码#

1
def _prepare_messages(state, system_prompt):
2
    messages = state["messages"]
3

4
    if system_prompt is None:
5
        return messages
6

7
    if isinstance(system_prompt, str):
8
        system_message = SystemMessage(
9
            content=system_prompt
10
        )
11
    elif isinstance(system_prompt, SystemMessage):
12
        system_message = system_prompt
13
    else:
14
        raise TypeError("Invalid system_prompt")
15

16
    return [system_message, *messages]

解释#

system_prompt 通常不是永久写入 state["messages"] 的历史消息，而是模型调用时注入。

这意味着：

1
每次 model node 调用
2
  都能看到 system_prompt

但 Agent State 中的 trace 仍主要记录：

1
HumanMessage
2
AIMessage
3
ToolMessage

ModelRequest 构造源码拆解#

伪代码#

1
request = ModelRequest(
2
    model=model,
3
    messages=messages,
4
    tools=available_tools,
5
    response_format=response_strategy,
6
    state=state,
7
    runtime=runtime,
8
    config=config,
9
)

字段解释#

字段	作用
`model`	当前要调用的 ChatModel
`messages`	传给模型的上下文
`tools`	当前模型可见工具
`response_format`	当前结构化输出策略
`state`	当前 Agent State
`runtime`	LangGraph 运行时上下文
`config`	RunnableConfig，含 tags、metadata、callbacks 等

为什么要有 ModelRequest#

因为 middleware 需要能够修改模型调用：

1
request = request.override(
2
    model=better_model,
3
    tools=filtered_tools,
4
    messages=trimmed_messages,
5
)

如果直接调用：

1
model.invoke(messages)

中间件就很难优雅地修改模型、工具和消息。

wrap_model_call 链源码拆解#

目标#

wrap_model_call 不是普通日志钩子，它可以决定：

1
是否调用 handler
2
调用几次 handler
3
用哪个 model 调用
4
用哪些 tools 调用
5
失败后是否 fallback
6
是否直接返回自定义 ModelResponse

伪代码：包装链构造#

1
def _call_model_with_middleware(request, middleware_list):
2
    def base_handler(req):
3
        return _invoke_model(req)
4

5
    handler = base_handler
6

7
    for middleware in reversed(middleware_list):
8
        wrapper = middleware.wrap_model_call
9

10
        if wrapper is None:
11
            continue
12

13
        previous_handler = handler
14

15
        def handler(req, wrapper=wrapper, previous_handler=previous_handler):
16
            return wrapper(req, previous_handler)
17

18
    return handler(request)

解释#

这类似洋葱模型：

1
wrap_model_call_1(
2
  wrap_model_call_2(
3
    base_model_invoke
4
  )
5
)

base_handler 伪代码#

1
def _invoke_model(request):
2
    model = request.model
3

4
    if request.tools:
5
        model = model.bind_tools(
6
            request.tools,
7
            **_tool_binding_kwargs(request),
8
        )
9

10
    if request.response_format:
11
        model = _bind_response_format(
12
            model,
13
            request.response_format,
14
        )
15

16
    message = model.invoke(
17
        request.messages,
18
        config=request.config,
19
    )
20

21
    return ModelResponse(message=message)

典型 wrap_model_call 示例#

1
def dynamic_model_selection(request, handler):
2
    if len(request.messages) > 20:
3
        request = request.override(model=strong_model)
4
    else:
5
        request = request.override(model=fast_model)
6

7
    return handler(request)

源码理解重点#

wrap_model_call 的核心不是“模型调用前后打日志”，而是：

1
对一次模型调用拥有控制权。

它可以：

1
短路
2
重试
3
fallback
4
改模型
5
改工具
6
改 messages
7
改 response_format

ChatModel 调用与 AIMessage 生成#

伪代码#

1
def _invoke_model(request):
2
    bound_model = request.model
3

4
    if request.tools:
5
        bound_model = bound_model.bind_tools(
6
            request.tools,
7
            tool_choice=_resolve_tool_choice(request),
8
        )
9

10
    if request.response_format:
11
        bound_model = _apply_structured_output(
12
            bound_model,
13
            request.response_format,
14
        )
15

16
    ai_message = bound_model.invoke(
17
        request.messages,
18
        config=request.config,
19
    )
20

21
    return ModelResponse(message=ai_message)

模型返回的两种典型 AIMessage#

需要工具#

1
AIMessage(
2
    content="",
3
    tool_calls=[
4
        {
5
            "name": "search_city_info",
6
            "args": {"query": "东京 热门景点"},
7
            "id": "call_001",
8
            "type": "tool_call",
9
        }
10
    ]
11
)

最终回答#

1
AIMessage(
2
    content="下面是东京 5 天行程建议...",
3
    tool_calls=[]
4
)

这就是 stop condition 能工作的基础：

1
最后一条 AIMessage 有没有 tool_calls？

after_model hooks 细拆#

伪代码#

1
def _run_after_model_hooks(
2
    state,
3
    update,
4
    runtime,
5
    middleware_list,
6
):
7
    current_update = update
8

9
    for middleware in middleware_list:
10
        hook = middleware.after_model
11

12
        if hook is None:
13
            continue
14

15
        hook_update = hook(
16
            _preview_state_after_update(state, current_update),
17
            runtime,
18
        )
19

20
        if hook_update is None:
21
            continue
22

23
        if "jump_to" in hook_update:
24
            return _merge_updates(
25
                current_update,
26
                hook_update,
27
            )
28

29
        current_update = _merge_updates(
30
            current_update,
31
            hook_update,
32
        )
33

34
    return current_update

after_model 能做什么#

1
检查模型是否生成了危险工具调用
2
检查 AIMessage.content 是否包含敏感信息
3
记录 usage_metadata
4
强制改写输出
5
在不允许工具调用时清空 tool_calls
6
jump_to="end"

示例#

1
def block_dangerous_tool(state, runtime):
2
    last = state["messages"][-1]
3

4
    if not getattr(last, "tool_calls", None):
5
        return None
6

7
    for call in last.tool_calls:
8
        if call["name"] in {"create_booking", "send_email"}:
9
            return {
10
                "messages": [
11
                    AIMessage(content="该操作需要人工确认。")
12
                ],
13
                "jump_to": "end",
14
            }
15

16
    return None

route_after_model 源码拆解#

最小伪代码#

1
def route_after_model(state):
2
    messages = state["messages"]
3
    last_message = messages[-1]
4

5
    if isinstance(last_message, AIMessage):
6
        if last_message.tool_calls:
7
            return "tools"
8

9
    return END

加 structured_response 后的伪代码#

1
def route_after_model(state):
2
    last_message = state["messages"][-1]
3

4
    if _has_business_tool_calls(last_message):
5
        return "tools"
6

7
    if _has_structured_response_tool_call(last_message):
8
        parsed = _try_parse_structured_response(last_message)
9

10
        if parsed.ok:
11
            return END
12

13
        return "model"  # 让模型修复结构化输出
14

15
    if _response_format_required(state):
16
        if state.get("structured_response") is None:
17
            return "model"
18

19
    return END

解释#

实际源码会比这个复杂，原因是需要区分：

1
业务工具调用
2
结构化输出工具调用
3
最终自然语言回答
4
模型错误调用
5
多 structured output 调用
6
Schema validation error

但核心判断仍然围绕：

1
最后一条 AIMessage

和：

1
tool_calls 是否存在、属于哪一类。

ToolNode 执行链：从 ToolCall 到状态更新#

ToolNode 负责把模型产生的调用意图变成真实执行结果。这里的重点不是“调用一个函数”，而是执行协议的闭环：

1
解析 ToolCall
2
  ↓
3
查找工具并注入 Runtime
4
  ↓
5
经过 wrap_tool_call
6
  ↓
7
同步或异步并行执行
8
  ↓
9
归一化为 ToolMessage / Command
10
  ↓
11
合并为 State update

ToolNode 初始化源码拆解#

源码心智模型#

ToolNode 是 tools node 的核心执行器。

伪代码#

1
class ToolNode(RunnableCallable):
2
    name = "tools"
3

4
    def __init__(
5
        self,
6
        tools,
7
        *,
8
        name="tools",
9
        tags=None,
10
        handle_tool_errors=_default_handle_tool_errors,
11
        messages_key="messages",
12
        wrap_tool_call=None,
13
        awrap_tool_call=None,
14
    ):
15
        super().__init__(self._func, self._afunc, name=name, tags=tags)
16

17
        self._tools_by_name = {}
18
        self._injected_args = {}
19
        self._handle_tool_errors = handle_tool_errors
20
        self._messages_key = messages_key
21
        self._wrap_tool_call = wrap_tool_call
22
        self._awrap_tool_call = awrap_tool_call
23

24
        for tool in tools:
25
            if not isinstance(tool, BaseTool):
26
                tool = create_tool(tool)
27

28
            self._tools_by_name[tool.name] = tool
29
            self._injected_args[tool.name] = _get_all_injected_args(tool)

逐行解释#

`RunnableCallable`#

ToolNode 本身是一个 Runnable 节点：

1
输入 state
2
  ↓
3
输出 partial state update

`_tools_by_name`#

用于根据模型输出的：

1
call["name"]

找到真实工具对象。

`_injected_args`#

提前分析每个工具是否需要注入：

1
state
2
store
3
runtime
4
config
5
tool_call_id

这样执行时不用重复反射。

`wrap_tool_call`#

如果存在中间件 wrapper，ToolNode 不直接执行工具，而是调用 wrapper：

1
wrap_tool_call(request, execute)

ToolNode._parse_input 源码拆解#

目标#

ToolNode 支持三种输入：

1
1. Graph state dict: {"messages": [...]}
2
2. Message list: [AIMessage(...)]
3
3. Direct tool call list: [{"name": ..., "args": ..., "id": ..., "type": "tool_call"}]

伪代码#

1
def _parse_input(self, input):
2
    if isinstance(input, dict):
3
        messages = input[self._messages_key]
4
        input_type = "dict"
5

6
    elif isinstance(input, list) and _is_message_list(input):
7
        messages = input
8
        input_type = "list"
9

10
    elif isinstance(input, list) and _is_tool_call_list(input):
11
        return input, "tool_calls"
12

13
    else:
14
        raise ValueError("Invalid input to ToolNode")
15

16
    if not messages:
17
        raise ValueError("No messages found in input")
18

19
    last_message = messages[-1]
20

21
    if not isinstance(last_message, AIMessage):
22
        raise ValueError("Last message is not an AIMessage")
23

24
    tool_calls = last_message.tool_calls
25

26
    return tool_calls, input_type

逐行解释#

dict 输入#

在 create_agent 中最常见：

1
{
2
    "messages": [
3
        HumanMessage(...),
4
        AIMessage(tool_calls=[...])
5
    ]
6
}

message list 输入#

适合手动测试 ToolNode：

1
ToolNode([tool]).invoke([
2
    AIMessage(tool_calls=[...])
3
])

direct tool_calls 输入#

适合跳过 message 解析，直接执行工具调用：

1
ToolNode([tool]).invoke([
2
    {
3
        "name": "search_city_info",
4
        "args": {"query": "东京"},
5
        "id": "call_001",
6
        "type": "tool_call",
7
    }
8
])

最后一条必须是 AIMessage#

因为 tool_calls 是模型输出的一部分，只有 AIMessage 才应该携带：

1
tool_calls

ToolNode._func 同步执行源码拆解#

伪代码#

1
def _func(self, input, config, runtime):
2
    # 1. 解析 tool_calls
3
    tool_calls, input_type = self._parse_input(input)
4

5
    # 2. 为每个 tool_call 分配 config
6
    config_list = get_config_list(
7
        config,
8
        len(tool_calls),
9
    )
10

11
    # 3. 构造每个工具调用的 ToolRuntime
12
    tool_runtimes = []
13

14
    for call, cfg in zip(tool_calls, config_list):
15
        state = self._extract_state(input, cfg)
16

17
        tool_runtime = ToolRuntime(
18
            state=state,
19
            tool_call_id=call["id"],
20
            config=cfg,
21
            context=runtime.context,
22
            store=runtime.store,
23
            stream_writer=runtime.stream_writer,
24
            tools=list(self.tools_by_name.values()),
25
            execution_info=runtime.execution_info,
26
            server_info=runtime.server_info,
27
        )
28

29
        tool_runtimes.append(tool_runtime)
30

31
    # 4. 并行执行多个工具调用
32
    with get_executor_for_config(config) as executor:
33
        outputs = list(
34
            executor.map(
35
                self._run_one,
36
                tool_calls,
37
                [input_type] * len(tool_calls),
38
                tool_runtimes,
39
            )
40
        )
41

42
    # 5. 合并输出
43
    return self._combine_tool_outputs(
44
        outputs,
45
        input_type,
46
    )

逐段解释#

第 1 段：解析 tool_calls#

1
tool_calls, input_type = self._parse_input(input)

这一步决定：

1
要执行哪些工具
2
输出应该按 dict 还是 list 格式返回

第 2 段：配置列表#

1
config_list = get_config_list(config, len(tool_calls))

如果有多个工具调用，每个工具调用都应该拿到一份 config。

第 3 段：构造 ToolRuntime#

ToolRuntime 让工具可以访问：

1
state
2
context
3
store
4
config
5
stream_writer
6
tool_call_id
7
execution_info
8
server_info

这解释了为什么工具不只是一个普通函数调用，而是在图运行时上下文中执行。

第 4 段：并行执行#

1
executor.map(...)

这意味着：

1
一条 AIMessage 中的多个 tool_calls 可以并行执行。

例如模型一次输出：

1
query_weather
2
search_attractions

ToolNode 可以同时执行这两个工具。

第 5 段：合并输出#

工具输出可能是：

1
ToolMessage
2
Command
3
list[ToolMessage | Command]

所以必须统一合并。

ToolNode._afunc 异步执行源码拆解#

伪代码#

1
async def _afunc(self, input, config, runtime):
2
    tool_calls, input_type = self._parse_input(input)
3
    config_list = get_config_list(config, len(tool_calls))
4

5
    tool_runtimes = []
6

7
    for call, cfg in zip(tool_calls, config_list):
8
        state = self._extract_state(input, cfg)
9
        tool_runtimes.append(
10
            ToolRuntime(
11
                state=state,
12
                tool_call_id=call["id"],
13
                config=cfg,
14
                context=runtime.context,
15
                store=runtime.store,
16
                stream_writer=runtime.stream_writer,
17
                tools=list(self.tools_by_name.values()),
18
                execution_info=runtime.execution_info,
19
                server_info=runtime.server_info,
20
            )
21
        )
22

23
    coros = []
24

25
    for call, tool_runtime in zip(tool_calls, tool_runtimes):
26
        coros.append(
27
            self._arun_one(call, input_type, tool_runtime)
28
        )
29

30
    outputs = await asyncio.gather(*coros)
31

32
    return self._combine_tool_outputs(outputs, input_type)

解释#

同步版用线程池或执行器并行：

1
executor.map(...)

异步版用：

1
asyncio.gather(...)

这要求工具最好实现原生异步：

1
async def

否则可能退化为线程执行。

ToolNode._run_one 源码拆解#

伪代码#

1
def _run_one(self, call, input_type, tool_runtime):
2
    # 1. 根据工具名查找工具
3
    tool = self.tools_by_name.get(call["name"])
4

5
    # 2. 构造 ToolCallRequest
6
    request = ToolCallRequest(
7
        tool_call=call,
8
        tool=tool,
9
        state=tool_runtime.state,
10
        runtime=tool_runtime,
11
    )
12

13
    config = tool_runtime.config
14

15
    # 3. 没有 wrapper，直接执行
16
    if self._wrap_tool_call is None:
17
        return self._execute_tool_sync(
18
            request,
19
            input_type,
20
            config,
21
        )
22

23
    # 4. 有 wrapper，提供 execute 函数
24
    def execute(req):
25
        return self._execute_tool_sync(
26
            req,
27
            input_type,
28
            config,
29
        )
30

31
    # 5. wrapper 决定如何执行
32
    try:
33
        return self._wrap_tool_call(
34
            request,
35
            execute,
36
        )
37
    except Exception as e:
38
        if not self._handle_tool_errors:
39
            raise
40

41
        content = _handle_tool_error(
42
            e,
43
            flag=self._handle_tool_errors,
44
        )
45

46
        return ToolMessage(
47
            content=content,
48
            name=request.tool_call["name"],
49
            tool_call_id=request.tool_call["id"],
50
            status="error",
51
        )

逐段解释#

工具查找#

1
tool = self.tools_by_name.get(call["name"])

模型输出的是字符串工具名，执行器必须从注册表里找到真实工具。

ToolCallRequest#

ToolCallRequest 是传给 wrap_tool_call 的对象，包含：

1
tool_call
2
tool
3
state
4
runtime

execute 函数#

execute(req) 是真正执行工具的函数。

wrapper 可以选择：

1
不调用 execute：短路 / 缓存命中 / 拒绝执行
2
调用一次 execute：正常执行
3
调用多次 execute：重试
4
修改 request 后调用 execute：参数改写

wrap_tool_call 源码语义细拆#

wrapper 示例#

1
def cache_tool_call(request, execute):
2
    cache_key = build_cache_key(request.tool_call)
3

4
    if cached := cache.get(cache_key):
5
        return ToolMessage(
6
            content=cached,
7
            name=request.tool_call["name"],
8
            tool_call_id=request.tool_call["id"],
9
        )
10

11
    result = execute(request)
12
    cache.set(cache_key, result.content)
13
    return result

源码语义#

1
request
2
  是当前工具调用请求
3

4
execute
5
  是继续执行原始工具调用的函数

wrapper 可以做什么#

1
缓存
2
重试
3
熔断
4
限流
5
参数改写
6
工具调用审计
7
高风险操作人工确认
8
调用前权限校验
9
调用后结果校验

与 callback 的区别#

callback 通常只能观察：

1
on_tool_start
2
on_tool_end
3
on_tool_error

wrapper 可以控制：

1
是否执行
2
执行几次
3
修改什么参数
4
返回什么结果

这就是 middleware 在 v1 中更强的原因。

ToolNode._execute_tool_sync 源码拆解#

伪代码#

1
def _execute_tool_sync(self, request, input_type, config):
2
    call = request.tool_call
3
    tool = request.tool
4

5
    # 1. 工具不存在时返回错误 ToolMessage
6
    if tool is None:
7
        invalid_tool_message = self._validate_tool_call(call)
8
        if invalid_tool_message:
9
            return invalid_tool_message
10

11
        raise TypeError(f"Tool {call['name']} is not registered")
12

13
    # 2. 注入 state / store / runtime
14
    injected_call = self._inject_tool_args(
15
        call,
16
        request.runtime,
17
        tool,
18
    )
19

20
    # 3. 补上 type=tool_call，让 BaseTool 能识别完整 ToolCall
21
    call_args = {
22
        **injected_call,
23
        "type": "tool_call",
24
    }
25

26
    try:
27
        try:
28
            # 4. 调用工具
29
            response = tool.invoke(call_args, config)
30

31
        except ValidationError as exc:
32
            # 5. 参数校验失败，过滤掉系统注入参数错误
33
            injected = self._injected_args.get(call["name"])
34
            filtered_errors = _filter_validation_errors(
35
                exc,
36
                injected,
37
            )
38

39
            raise ToolInvocationError(
40
                call["name"],
41
                exc,
42
                call["args"],
43
                filtered_errors,
44
            ) from exc
45

46
        # 6. 归一化工具返回
47
        return self._normalize_tool_response(
48
            response,
49
            request.tool_call,
50
            input_type,
51
        )
52

53
    except GraphBubbleUp:
54
        # 7. interrupt 等图控制异常必须向上冒泡
55
        raise
56

57
    except Exception as e:
58
        # 8. 判断是否处理异常
59
        if not self._should_handle_error(e):
60
            raise
61

62
        # 9. 转成 error ToolMessage
63
        content = _handle_tool_error(
64
            e,
65
            flag=self._handle_tool_errors,
66
        )
67

68
        return ToolMessage(
69
            content=content,
70
            name=call["name"],
71
            tool_call_id=call["id"],
72
            status="error",
73
        )

核心细节 1：补 `type="tool_call"`#

第 4 篇讲过：

1
tool.invoke({
2
    "name": "search_city_info",
3
    "args": {...},
4
    "id": "call_001",
5
    "type": "tool_call",
6
})

只有传完整 ToolCall，BaseTool 才能保留：

1
tool_call_id

并返回：

1
ToolMessage(...)

所以 ToolNode 在执行前会确保传给工具的是完整 ToolCall。

核心细节 2：注入发生在真正执行前#

1
injected_call = self._inject_tool_args(...)

这一步会注入：

1
state
2
store
3
runtime
4
tool_call_id

模型不能控制这些参数。

核心细节 3：过滤 validation errors#

如果工具有隐藏注入参数，模型不应该看到这些参数的错误。

所以源码会过滤：

1
只保留模型可控参数的校验错误。

这样模型收到的错误更可修复。

ToolNode._normalize_tool_response 源码语义#

伪代码#

1
def _normalize_tool_response(response, original_call, input_type):
2
    if isinstance(response, ToolMessage):
3
        return response
4

5
    if isinstance(response, Command):
6
        return response
7

8
    if isinstance(response, list):
9
        return [
10
            _normalize_tool_response(item, original_call, input_type)
11
            for item in response
12
        ]
13

14
    return ToolMessage(
15
        content=msg_content_output(response),
16
        name=original_call["name"],
17
        tool_call_id=original_call["id"],
18
    )

解释#

工具可能返回：

1
字符串
2
dict
3
list
4
ToolMessage
5
Command
6
list[ToolMessage | Command]

ToolNode 需要统一输出成图能合并的格式。

为什么支持 Command#

工具返回 Command 时，不只是给模型一个 observation，而是可以：

1
更新 state
2
触发 goto
3
发送 Send
4
控制父图

这已经超出普通 ReAct，进入 LangGraph 的高级控制流。

ToolNode._combine_tool_outputs 源码拆解#

伪代码#

1
def _combine_tool_outputs(outputs, input_type):
2
    # 1. 展平 list 输出
3
    flat_outputs = []
4

5
    for output in outputs:
6
        if isinstance(output, list):
7
            flat_outputs.extend(output)
8
        else:
9
            flat_outputs.append(output)
10

11
    # 2. 如果没有 Command，保持传统 ToolMessage 输出
12
    if not any(isinstance(o, Command) for o in flat_outputs):
13
        if input_type == "list":
14
            return flat_outputs
15

16
        return {
17
            self._messages_key: flat_outputs
18
        }
19

20
    # 3. 如果有 Command，需要让 LangGraph runtime 处理
21
    combined_outputs = []
22

23
    for output in flat_outputs:
24
        if isinstance(output, Command):
25
            combined_outputs.append(output)
26
        else:
27
            if input_type == "list":
28
                combined_outputs.append([output])
29
            else:
30
                combined_outputs.append({
31
                    self._messages_key: [output]
32
                })
33

34
    return combined_outputs

解释#

在 create_agent 的常规场景中，最常见返回：

1
{
2
    "messages": [
3
        ToolMessage(...),
4
        ToolMessage(...),
5
    ]
6
}

这会被 LangGraph reducer 追加到 state["messages"]。

如果工具返回 Command，说明工具不仅产生消息，还想控制图状态或流程。

运行时总结、伪代码与源码证据#

以下为保留关键控制流的简化伪代码，不是源码逐字复制：

1
def run_agent(state):
2
    ai_message = call_model(state)
3
    state = append_message(state, ai_message)
4
    if ai_message.tool_calls:
5
        return execute_tools_and_loop(state)
6
    return state

源码证据：

9. 关键分支、异常与边界#

Agent 是否“能调用工具”只是第一层能力，更关键的是它能否可靠收敛。循环控制需要同时处理正常结束、结构化结果、人工中断和框架级熔断，不能只依赖模型自觉停止。

stop condition 源码拆解#

最小判断#

1
def should_continue(state):
2
    last_message = state["messages"][-1]
3

4
    if isinstance(last_message, AIMessage):
5
        if last_message.tool_calls:
6
            return "tools"
7

8
    return END

多情况判断#

1
def should_continue(state):
2
    last = state["messages"][-1]
3

4
    # 1. 不是 AIMessage，不能继续工具循环
5
    if not isinstance(last, AIMessage):
6
        return END
7

8
    # 2. 模型请求业务工具
9
    business_tool_calls = _get_business_tool_calls(last)
10
    if business_tool_calls:
11
        return "tools"
12

13
    # 3. 模型请求结构化响应工具
14
    structured_calls = _get_structured_output_calls(last)
15
    if structured_calls:
16
        parse_result = _parse_structured_output(structured_calls)
17

18
        if parse_result.success:
19
            return END
20

21
        return "model"  # 让模型修复
22

23
    # 4. 没有工具调用，正常结束
24
    return END

解释#

create_agent 的停止不是“循环次数到了就停”，而是：

1
模型不再请求工具

这就是 ReAct 的自然终点。

但为了防止模型一直调用工具，LangGraph 运行时还有：

1
recursion_limit

recursion_limit 与 iteration limit#

业务层停止条件#

1
AIMessage 不再包含 tool_calls

框架层保护条件#

1
LangGraph 执行步数超过 recursion_limit

设置方式#

1
result = agent.invoke(
2
    {
3
        "messages": [
4
            {"role": "user", "content": "..."}
5
        ]
6
    },
7
    config={
8
        "recursion_limit": 20
9
    }
10
)

重要区别#

1
recursion_limit 不是“最多工具调用次数”。
2
recursion_limit 限制的是图执行步数 / super-step。

一次 ReAct 循环可能包括：

1
model node
2
  ↓
3
tools node

所以：

1
一次工具调用循环 ≈ 多个图执行步。

无限循环场景#

1
model 总是输出 tool_calls
2
  ↓
3
tools 返回结果
4
  ↓
5
model 仍然输出同一个 tool_call
6
  ↓
7
tools 再执行
8
  ↓
9
...
10
  ↓
11
达到 recursion_limit

这时需要排查：

1
工具结果是否让模型误解？
2
工具描述是否诱导重复调用？
3
stop condition 是否被结构化响应策略影响？
4
system_prompt 是否要求必须查工具？
5
模型是否缺少“何时停止”的指令？

structured_response 源码链路#

输入#

1
class TravelSummary(BaseModel):
2
    destination: str
3
    days: int
4
    highlights: list[str]
5

6
agent = create_agent(
7
    model="openai:gpt-4.1-mini",
8
    tools=[search_city_info],
9
    response_format=TravelSummary,
10
)

ProviderStrategy 伪代码#

1
def _call_with_provider_strategy(request):
2
    model = request.model
3

4
    model = model.with_structured_output(
5
        request.response_format.schema,
6
        method="provider_native",
7
    )
8

9
    parsed = model.invoke(request.messages)
10

11
    return {
12
        "structured_response": parsed,
13
        "messages": [_make_final_message_if_needed(parsed)],
14
    }

ToolStrategy 伪代码#

1
def _call_with_tool_strategy(request):
2
    schema_tool = _schema_to_virtual_tool(
3
        request.response_format.schema
4
    )
5

6
    model = request.model.bind_tools(
7
        [*request.tools, schema_tool],
8
        tool_choice="any",
9
    )
10

11
    ai_message = model.invoke(request.messages)
12

13
    if _contains_schema_tool_call(ai_message):
14
        parsed = _parse_schema_tool_call(ai_message)
15
        return {
16
            "messages": [ai_message],
17
            "structured_response": parsed,
18
        }
19

20
    return {
21
        "messages": [ai_message]
22
    }

解释#

structured_response 不是从最终自然语言里手写正则解析出来的。

它来自：

1
Provider-native structured output
2
或
3
Tool calling structured output

最后被写入：

1
state["structured_response"]

interrupt_before / interrupt_after 与工具审批#

配置#

1
agent = create_agent(
2
    model=model,
3
    tools=[create_booking],
4
    interrupt_before=["tools"],
5
    checkpointer=checkpointer,
6
)

执行心智模型#

1
model 输出 tool_calls
2
  ↓
3
准备进入 tools node
4
  ↓
5
interrupt_before["tools"] 触发
6
  ↓
7
图状态保存到 checkpoint
8
  ↓
9
等待人工确认
10
  ↓
11
resume
12
  ↓
13
tools node 真正执行

适合场景#

1
下单
2
付款
3
退款
4
发邮件
5
删文件
6
提交工单
7
修改数据库

这说明：

1
高风险工具不应该只靠 prompt 控制，
2
应该在 graph 层或 middleware 层强制中断审批。

异常传播与恢复边界#

异常类别	处理位置	默认行为
模型调用失败	`wrap_model_call` 或模型节点	重试、fallback 或继续抛出
工具参数错误	`ToolNode` / `BaseTool`	按工具错误策略转成 ToolMessage 或抛出
中间件拒绝	middleware hook	跳转、返回状态更新或中断
图不收敛	LangGraph runtime	达到 `recursion_limit` 后抛出错误

分支语义总结、伪代码与源码证据#

以下为保留关键控制流的简化伪代码，不是源码逐字复制：

1
def route_after_model(state):
2
    last_message = state["messages"][-1]
3
    if last_message.tool_calls:
4
        return "tools"
5
    if state.get("structured_response") is not None:
6
        return END
7
    return END

源码证据：

10. 扩展机制与框架协作#

扩展点总览#

扩展点	执行时机	适用能力
`before_agent` / `after_agent`	Agent 生命周期边界	初始化、汇总和清理
`before_model` / `after_model`	每次模型调用前后	上下文处理、输出审查
`wrap_model_call`	包裹模型请求	重试、fallback、动态模型
`wrap_tool_call`	包裹工具执行	审批、缓存、工具错误策略
`response_format`	最终响应生成	structured_response
`interrupt_before` / `interrupt_after`	节点执行边界	Human-in-the-loop

Middleware 包装链伪代码#

以下为保留关键控制流的简化伪代码，不是源码逐字复制：

1
def call_model_with_middleware(request, middleware):
2
    handler = base_model_handler
3
    for item in reversed(middleware):
4
        handler = item.wrap_model_call(handler)
5
    return handler(request)

与 LangGraph 的协作边界#

1
create_agent 负责标准 Harness
2
  ↓
3
CompiledStateGraph 负责图执行
4
  ↓
5
Checkpointer 负责持久化
6
  ↓
7
Store / Runtime 负责运行上下文

何时使用扩展点，何时手写图#

需求	推荐方式
修改模型或工具调用前后行为	middleware
增加标准状态字段	middleware state schema
改变主执行拓扑	手写 LangGraph
多阶段确定性审批和并行汇合	手写 LangGraph

源码证据#

11. 工程决策与适用场景#

适用场景#

create_agent 适合标准的“模型决定是否调用工具，工具结果回填模型，直到生成最终结果”循环。

不适用场景#

复杂业务状态机、强确定性流程、多阶段并行汇合和跨节点审批，应直接使用 LangGraph 构建显式拓扑。

工程决策表#

决策点	推荐
标准 ReAct 工具循环	`create_agent`
仅增加日志、重试或上下文处理	middleware
改变节点和边	手写 `StateGraph`
高风险工具	middleware + interrupt + checkpointer

12. 常见误区与源码纠正#

create_agent 返回的是 Chain#

纠正：

1
create_agent 返回 CompiledStateGraph。

create_agent 就是 while 循环#

纠正：

1
它是 LangGraph 图。
2
循环由 conditional edge + 回边表达。

ReAct 必须显式输出 Thought#

纠正：

1
现代实现更依赖 AIMessage.tool_calls、ToolMessage、trace。

ToolMessage 是普通文本#

纠正：

1
ToolMessage 携带 tool_call_id，必须和 AIMessage.tool_calls 中的 id 对应。

middleware 只能记录日志#

纠正：

1
middleware 可以修改 state、改模型、过滤工具、重试、fallback、跳转 END。

recursion_limit 是工具调用次数限制#

纠正：

1
recursion_limit 是 LangGraph 图执行步数上限，不是单纯工具调用次数。

structured_response 是从自然语言里解析的#

纠正：

1
create_agent 的 structured_response 通常来自 ProviderStrategy 或 ToolStrategy。

create_agent 适合所有业务系统#

纠正：

1
create_agent 适合标准 ReAct 工具循环；
2
复杂业务流程、强状态机、多阶段审批更适合手写 LangGraph。

13. 最终心智模型与掌握检查#

构建期#

1
model + tools + system_prompt + middleware + response_format
2
  ↓
3
create_agent
4
  ↓
5
StateGraph
6
  ↓
7
model node
8
  ↓
9
tools node
10
  ↓
11
conditional edge
12
  ↓
13
CompiledStateGraph

运行期#

1
HumanMessage
2
  ↓
3
model node
4
  ↓
5
AIMessage(tool_calls)
6
  ↓
7
tools node
8
  ↓
9
ToolMessage
10
  ↓
11
model node
12
  ↓
13
AIMessage(final answer)

ToolNode 内部#

1
state["messages"][-1]
2
  ↓
3
AIMessage.tool_calls
4
  ↓
5
_tools_by_name 查找工具
6
  ↓
7
ToolRuntime 注入
8
  ↓
9
wrap_tool_call
10
  ↓
11
BaseTool.invoke(full_tool_call)
12
  ↓
13
ToolMessage
14
  ↓
15
{"messages": [ToolMessage]}

停止条件#

1
最后一条 AIMessage 有 tool_calls
2
  ↓
3
tools node
4

5
最后一条 AIMessage 无 tool_calls
6
  ↓
7
END
8

9
图执行超过 recursion_limit
10
  ↓
11
GraphRecursionError

一句话总结：

create_agent 是 LangChain v1 的高层 Agent Harness。它通过 LangGraph 构建 model node → tools node → model node 的循环，用 AIMessage.tool_calls 表示 Action，用 ToolMessage 表示 Observation，用条件边判断是否继续循环，并用 middleware、structured_response、interrupt、checkpointer 与 recursion_limit 把 ReAct 类行为工程化为可观测、可恢复、可扩展的 Agent Runtime。

14. 参考资料与下一篇衔接#

官方资料#

官方文档#

API Reference#

官方源码#

下一篇衔接#

下一篇建议进入：

1
第 6 篇：LangGraph StateGraph 源码解剖

重点回答：

1
StateGraph 是 builder 还是 Runnable？
2
add_node 如何注册节点？
3
add_edge 如何注册固定边？
4
add_conditional_edges 如何注册条件边？
5
compile 之后生成什么？
6
node 返回 partial state update 如何合并？
7
messages reducer 如何追加消息？
8
CompiledStateGraph.invoke 如何驱动图执行？

LangChain 源码学习路线：第 5 篇 create_agent 与 ReAct Agent 运行机制源码解剖#

0. 本篇在源码学习主线中的位置#

1. 本篇问题、学习目标与能力边界#

2. 核心概念与最小心智模型#

经典 ReAct 范式#

LangChain / LangGraph 现代映射#

为什么生产中更推荐结构化 trace#

3. 完整执行链路#

最小代码#

你应该观察什么#

create_agent 的输出不是字符串#

执行流程#

高层执行流程#

细粒度执行流程#

4. 源码地图、关键文件与阅读顺序#

核心目录#

重点文件#

重点类与函数#

推荐阅读顺序#

5. 对象模型、继承关系与协议边界#

核心对象关系#

协议边界#

6. 源码阅读策略与证据标准#

按执行边界追踪源码#

阅读顺序#

证据标准#

7. 构建期源码解剖#

create_agent 顶层入口伪代码#

伪代码#

逐行解释#

第 1 步：标准化模型#

第 2 步：标准化工具#

第 3 步：标准化结构化输出#

第 4 步：标准化 middleware#

第 5 步：合成 State schema#

第 6—11 步：构图与编译#

模型初始化源码拆解#

伪代码#

解释#

工具归一化源码拆解#

伪代码#

解释#

response_format 归一化源码拆解#

伪代码#

解释#

ProviderStrategy 心智模型#

ToolStrategy 心智模型#

注意#

middleware 归一化源码拆解#

伪代码#

解释#

State schema 合成源码拆解#

最小 AgentState#

加 structured_response 后#

加 middleware state 后#

伪代码#

解释#

构建期总结、伪代码与源码证据#

8. 运行时主链源码解剖#

Model 调用链：从 State 到 AIMessage#

model node 构造源码拆解#

model node 的职责#

伪代码：model node 外壳#

before_model hooks 细拆#

伪代码#

每一步解释#

读取 hook#

执行 hook#

合并 state#

典型 before_model 用途#

system_prompt 注入源码拆解#

伪代码#

解释#

ModelRequest 构造源码拆解#

伪代码#

字段解释#

为什么要有 ModelRequest#

wrap_model_call 链源码拆解#

目标#

伪代码：包装链构造#

`create_agent` 顶层入口伪代码#

`RunnableCallable`#

`_tools_by_name`#

`_injected_args`#

`wrap_tool_call`#

核心细节 1：补 `type="tool_call"`#