基于 StateGraph 的状态图 Agent 设计
基于 StateGraph 的状态图 Agent 设计
学习目标
本章要解决什么问题：
- 如何使用 LangGraph 的
StateGraph声明式构建 Agent 工作流 - 如何设计状态 Schema（TypedDict vs Pydantic）
- 如何定义节点、边、条件边实现动态路由
- 编译后的图如何执行和交互
前置知识
本章涉及 Agent 状态图的通用原理，建议先阅读：
- StateGraph Agent 循环 — 将 Agent 循环建模为有向图的设计动机
- 状态 Channel 语义 — 状态更新语义与 Channel 类型选择
下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。
项目实践
StateGraph 的 Builder 模式
LangGraph 的 StateGraph 使用 Builder 模式逐步构建图：
from langgraph.graph import StateGraph, START, END
class State(TypedDict): messages: Annotated[list, add_messages] current_step: str
graph = StateGraph(state_schema=State)graph.add_node("model", call_model)graph.add_node("tools", execute_tools)graph.add_edge(START, "model")graph.add_conditional_edges("model", route)graph.add_edge("tools", "model")compiled = graph.compile()关键流程：
- 定义状态 Schema（
state_schema） - 可选定义输入/输出 Schema（
input_schema、output_schema） - 可选定义上下文 Schema（
context_schema） - 注册节点（
add_node） - 定义边（
add_edge）和条件边（add_conditional_edges） - 编译（
compile()）
状态 Schema 的三种角色
StateGraph 接受四个 Schema 参数，各承担不同角色：
| Schema | 职责 | 默认值 |
|---|---|---|
state_schema | 运行时完整状态 | 必填 |
input_schema | 调用 invoke() 时的输入格式 | 同 state_schema |
output_schema | 图的最终输出格式 | 同 state_schema |
context_schema | 不可变的运行时上下文（如 user_id、db_conn） | 无 |
为什么需要分离：输入可能只需要部分字段（如只需用户消息），但内部状态需要更多字段（如计数器、中间结果）。分离后，调用者不需要知道内部状态的全部结构。
条件边的动态路由
add_conditional_edges 接受一个路由函数，根据当前状态决定下一步：
def route(state): if state["messages"] and state["messages"][-1].tool_calls: return "tools" return END
graph.add_conditional_edges("model", route)路由函数的返回值必须是图中已定义的节点名或 END。如果返回不存在的节点名，编译时会报错。
Send 原语：动态多目标分发
除了固定的条件边，Send 允许节点在运行时动态决定向哪些目标节点分发任务：
from langgraph.types import Send
def fan_out(state): queries = extract_queries(state["messages"]) return [Send("research_agent", {"query": q}) for q in queries]每个 Send 创建目标节点的一个独立执行实例，有自己的状态副本。
编译与执行
compile() 将 StateGraph（Builder 对象）转换为 CompiledStateGraph（可执行对象）：
app = graph.compile()
# 同步调用result = app.invoke({"messages": ["你好"]})
# 流式输出for chunk in app.stream({"messages": ["你好"]}, stream_mode="updates"): print(chunk)编译后的图支持 invoke()、stream()、ainvoke()、astream() 等方法。
问题与规避
状态 Schema 不匹配导致编译失败
问题：如果 input_schema 或 output_schema 中包含了 state_schema 中没有的 Channel，或者类型不一致，编译时会报错。
规避：确保 input/output schema 是 state schema 的子集或超集，Channel 类型必须一致。
条件边返回不存在的节点名
问题：路由函数返回了未在图中注册的节点名，运行时会抛出 ValueError。
规避：使用类型安全的枚举或常量定义节点名，路由函数的返回值受类型检查约束。
无限循环
问题：如果条件边始终回到前一个节点而没有退出条件，图会无限循环。
规避：
- 设置
recursion_limit参数 - 在状态中使用
RemainingStepsmanaged value 追踪剩余步数 - 确保条件边有明确的退出路径
设计取舍
优势
- 声明式：图结构一目了然，可视化友好
- 类型安全：Schema 驱动，IDE 可推断状态字段
- 灵活：支持条件分支、并行、子图嵌套等复杂模式
代价
- 学习曲线：需要理解 Builder 模式、Schema、Channel 等多个概念
- 冗长：简单线性流程也需要定义 Schema、注册节点、定义边
- 编译时 vs 运行时：编译时无法检测所有运行时错误（如条件边返回无效节点名）
替代方案
- Functional API（
@entrypoint/@task）：用装饰器定义工作流，适合线性/并行流水线，详见 Functional API：以函数式风格定义工作流 - Prebuilt
create_react_agent：一键构建 ReAct Agent，适合标准场景
参考来源
- 源码验证:
libs/langgraph/langgraph/graph/state.py— StateGraph 实现 - 文档: LangGraph StateGraph API, https://docs.langchain.com/oss/python/langgraph/state