第 07 讲 · StateGraph 构建与状态 schema 映射

学习目标

• 理解 StateGraph 的四类构建 API：add_node / add_edge / add_conditional_edges。
• 看懂 状态 schema（TypedDict + Annotated）如何映射成 channel。
• 理解 add_messages 这类 reducer 的本质。

StateGraph 是个"图蓝图"

libs/langgraph/langgraph/graph/state.py，类 StateGraph（130 行起）。 它本身不执行任何东西，只是收集：节点、边、条件分支、状态 schema。 真正变成可执行对象是在 compile()（下一讲）。

class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):

四个泛型：StateT（状态）、ContextT（运行时上下文）、InputT/OutputT（输入/输出 schema）。 最常见用法只关心 StateT：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 有 reducer
    step: int                                 # 无 reducer（默认 LastValue）

g = StateGraph(State)

状态 schema → channel 的映射

这是本讲的核心。当你把 schema 传给 StateGraph(State) 时， _add_schema（342 行）会遍历每个字段，根据有没有 Annotated[T, reducer] 决定用哪种 channel：

• 字段是 Annotated[T, reducer] → 该字段的 channel 用对应 reducer：
  • add_messages → 消息合并通道
  • operator.add → BinaryOperatorAggregate
  • 自定义 BaseChannel 子类 → 直接用它（回顾第 3 讲的自定义通道）
• 字段是裸类型 T（无 Annotated）→ 默认 LastValue（回顾第 4 讲：单写者、并发写报错）。

换句话说：你在 TypedDict 里写 Annotated，就是在为这个字段挑 channel/reducer。 这把第 4-5 讲学的通道，和你日常写的状态定义连在了一起。

这也解释了第 4 讲那个报错：字段没加 reducer = LastValue，两个节点同步写就 InvalidUpdateError。

add_node：注册节点

add_node（375 行起，有多个重载）。最常见形态：

g.add_node("agent", call_model)        # 名字 + 可调用对象
g.add_node(call_model)                 # 用函数名当节点名

节点函数签名约定：接收 state（可能还有 runtime），返回"状态更新 dict"。 返回的 dict 的每个 key 会被写到对应 channel（走该字段的 reducer 合并）。 节点也可以返回 Command（第 18 讲）做"更新 + 跳转"。

add_node 还能配置 per-node 的 retry_policy / cache_policy / defer（延迟执行）/ 错误处理节点等，这些会在 compile 时落到 PregelNode（下一讲）。

add_edge：静态边

add_edge（915 行）。三种形态：

g.add_edge(START, "agent")     # 入口
g.add_edge("tools", "agent")   # 普通边
g.add_edge(["a", "b"], "c")    # 多对一：等 a 和 b 都完成才触发 c（join）

• 单起点边：start 节点完成后触发 end。
• 多起点边（list）：这就是第 5 讲讲的 NamedBarrierValue 屏障的来源—— 编译时会生成一个 join 通道，等所有起点到齐才触发 end。

add_conditional_edges：动态路由

add_conditional_edges（969 行）。这是实现"循环/分支"的关键。

def route(state) -> str:
    return "tools" if state["needs_tool"] else END

g.add_conditional_edges("agent", route)
# 或带映射表：
g.add_conditional_edges("agent", route, {"tools": "tools", END: END})

• route 函数返回下一个节点名（或名字列表、或 Send 列表）。
• 返回 Send(node, arg) 列表 → 动态 fan-out（map-reduce，第 19 讲）。
• 路由函数在运行时由"条件分支"机制执行，产出写到目标节点的触发通道（下一讲拆解）。

add_messages：reducer 的范例

graph/message.py。add_messages 是处理对话消息列表的 reducer，比 operator.add 智能得多：

• 按消息 id 去重 / 更新（同 id 的新消息覆盖旧的）。
• 支持 RemoveMessage 删除特定消息、REMOVE_ALL_MESSAGES 清空。
• 自动把 dict/字符串等"类消息"对象规整成 BaseMessage。

Messages = list[MessageLikeRepresentation] | MessageLikeRepresentation

REMOVE_ALL_MESSAGES = "__remove_all__"

它就是普通的 (left, right) -> merged 函数 reducer，挂在 Annotated[list, add_messages] 上。 理解它，你就能写出自己的高级 reducer（比如带优先级的任务队列合并）。

构建 API 总览图

flowchart TD
    SG[StateGraph State] --> AN[add_node 注册 actor]
    SG --> SCH[_add_schema: 字段→channel]
    AN --> AE[add_edge 静态边/join]
    AN --> ACE[add_conditional_edges 动态路由]
    SCH -.Annotated reducer.-> CH[BinaryOperatorAggregate / add_messages / 自定义]
    SCH -.裸类型.-> LV[LastValue]
    AE --> CP[compile 见第8讲]
    ACE --> CP

使用场景：设计可维护的状态结构

实战经验：

• 能不加 reducer 就不加：单写者字段用默认 LastValue，意图最清晰，并发写还能帮你抓 bug。
• 多写者/聚合字段必须加 reducer：否则运行时 InvalidUpdateError。
• 消息历史一律 add_messages：免去手动去重/更新的麻烦。
• 大状态考虑 DeltaChannel（第 6 讲）：但要确认 reducer 满足 batching-invariant。
• 把"临时中间量"和"长期状态"分开：临时量用 EphemeralValue/不放进对外 schema，避免污染 checkpoint。

动手实验

1. 定义含三种字段的 State：add_messages 的、operator.add 的、裸类型的。 各写一个节点更新它们，打印每步 state，观察三种合并行为差异。
2. 用 add_conditional_edges 做一个"最多循环 3 次"的图，对照第 1 讲实验。
3. 用 g.compile().get_graph().draw_mermaid() 打印图结构，确认边/分支符合预期。

阅读作业

• 精读 graph/state.py 的 _add_schema（342 行）与 add_node/add_edge/add_conditional_edges。
• 精读 graph/message.py 的 add_messages 实现，理解去重/删除逻辑。

小结

• StateGraph 是蓝图：收集节点、边、分支、状态 schema，不执行。
• Annotated[T, reducer] 决定字段用哪种 channel——这是状态定义与通道家族的连接点。
• 多起点边 → join 屏障；条件边 → 动态路由（含 Send fan-out）。

下一讲：compile() 如何把这张蓝图变成会跑的 Pregel。