基于图的状态管理：StateGraph 驱动的 Agent 循环

基于图的状态管理：StateGraph 驱动的 Agent 循环

学习目标

本章要解决什么问题：create_agent() 是如何将 Agent 的核心执行循环（模型调用 → 工具执行 → 判断是否继续）实现为一个状态图的？

读者将学到：

• create_agent() 的图构建流程
• 模型节点与工具节点的条件路由
• 中间件节点在图中的插入方式
• jump_to 状态字段如何控制流

前置知识

本章涉及 StateGraph Agent 循环的通用原理，建议先阅读：

• Agent 核心循环设计
• StateGraph Agent 循环

下文假设你已理解上述概念，直接聚焦 LangChain 的具体实现。

项目实践

create_agent() 的图构建流程

create_agent() 的核心实现是一个 1800+ 行的工厂函数，负责将 Agent 的各个方面编译为 StateGraph：

关键步骤：

1. 模型绑定：根据 response_format 类型选择 ProviderStrategy 或 ToolStrategy，调用 model.bind_tools() 绑定工具列表
2. 工具节点：ToolNode 由 LangGraph 提供，负责执行模型请求的工具调用并将结果写入状态
3. 中间件节点：每个中间件的钩子（before_agent、before_model 等）被编译为独立的图节点，按注册顺序串联
4. 条件边：model_to_tools 和 tools_to_model 两个条件边函数实现 ReAct 循环

条件边的路由逻辑

模型 → 工具（model_to_tools）：

# 伪代码
def model_to_tools(state) -> str | list[Send] | None:
    # 1. 优先检查显式跳转
    if state.get("jump_to"):
        return _resolve_jump(state["jump_to"])

    # 2. 没有 AIMessage → 退出循环
    if no_ai_message(state["messages"]):
        return END

    # 3. 没有工具调用 → 退出循环（经典退出条件）
    if no_tool_calls(last_ai_message):
        return END

    # 4. 有待执行的工具调用 → 分发到 tools 节点
    if pending_tool_calls:
        return [Send("tools", [tool_call]) for tool_call in pending_tool_calls]

    # 5. 已有结构化响应 → 退出循环
    if "structured_response" in state:
        return END

    # 6. 有工具调用但都已处理 → 回到模型（处理人工注入的消息）
    return model_destination

工具 → 模型（tools_to_model）：

# 伪代码
def tools_to_model(state) -> str | None:
    # 1. 所有工具都是 return_direct → 直接结束
    if all(tool.return_direct for tool in executed_tools):
        return END

    # 2. 执行了结构化输出工具 → 结束
    if executed_structured_tool:
        return END

    # 3. 默认：回到模型继续循环
    return "before_model"  # 或 "model"（无中间件时）

中间件节点在图中的插入

中间件的每个钩子都被编译为独立的图节点：

无中间件:  START → model → tools → model → END

有中间件:
START → m.before_agent → m.before_model → model → m.after_model → tools → m.before_model → ...
                                                                    ↓
                                                                  END (via m.after_agent)

边构建策略：

• before_agent 中间件按注册顺序串联，最后一个连接到 loop_entry_node
• before_model 中间件按注册顺序串联，最后一个直接连接到 model 节点
• after_model 中间件按注册逆序串联（model → last.after_model → first.after_model）
• after_agent 中间件按注册逆序串联，最后一个连接到 END

jump_to 状态字段

jump_to 是 AgentState 中的一个临时字段（使用 EphemeralValue 注解），用于中间件显式控制流：

# 伪代码：中间件设置跳转
@before_model(can_jump_to=["end"])
def my_hook(state, runtime) -> dict | None:
    if some_condition(state):
        return {"jump_to": "end"}  # 直接结束，不调用模型
    return None

jump_to 的值在条件边函数中被 _resolve_jump() 解析并映射为实际的节点名称。

问题与规避

递归深度限制

问题：Agent 循环超过默认限制（9999 次）。

对策：create_agent() 在编译时设置 recursion_limit=9999，这通常足够覆盖正常任务。如果业务需要更长循环，可通过 LangGraph 的 config 覆盖。

消息状态竞争

问题：多个中间件同时修改 messages 字段可能导致覆盖。

对策：messages 字段使用 add_messages reducer（LangGraph 的 add_messages 函数），将多个节点的更新合并为追加操作，而非覆盖。

工具节点的可选性

问题：不传入工具时，Agent 退化为简单的模型调用器，但代码路径仍需正确处理。

对策：create_agent() 在没有工具时跳过 ToolNode 的创建，直接连接 model → exit_node。如果存在 after_model 中间件，仍会执行。

设计取舍

条件边 vs 动态路由

方案	优势	代价
条件边（LangGraph）	声明式、图结构可验证、支持可视化	需要理解图概念
动态路由（代码 if/else）	直观、无需额外概念	控制流分散在代码中、难以验证

LangChain 选择条件边，因为 Agent 的循环逻辑是核心基础设施，声明式定义使得图结构在编译时可被验证和分析。

Send vs 直接返回

问题：工具节点返回时，使用 Send("tools", [tool_call]) 而非简单字符串路由。

原因：Send 支持并行工具调用——当模型在一次响应中请求多个独立工具时，LangGraph 会并行分发到多个工具执行实例。

参考来源

• LangGraph 文档
• LangChain Agents 文档