StateGraph.compile:把蓝图编译成 Pregel 运行体

源码版本1.2.9

职责

StateGraph.compile 是构建器与运行时之间的「翻译官」。它吃进去一个 StateGraph(节点 spec、边、分支条件、通道字典),吐出来一个 CompiledStateGraph(CompiledStateGraph:1391)——后者继承自 Pregel,带 invoke / stream / astream / get_state 等所有运行期方法。compile 之前你的图只是配置数据,compile 之后才变成可执行体。

它的核心工作是四件:1) 校验图结构 (validate),2) 决定输出/流式通道,3) 注入运行期依赖 (checkpointer / store / cache / interrupt 配置),4) 把每个 StateNodeSpec / edge / BranchSpec 翻译成 Pregel 的 PregelNode + 通道订阅 (构造 CompiledStateGraph:1333-1357)。翻译这一步是循环调 compiled.attach_node / compiled.attach_edge / compiled.attach_branch 完成的——这些方法把 builder 里的声明式数据结构转换成 PregelNode.writers / PregelNode.triggers 这类运行期字段。

compile 还兼任两个次要职责:1) 给所有未显式指定 policy 的节点应用 set_node_defaults 设的默认值 (retry / cache / error_handler / timeout),2) 在启用了严格 msgpack 序列化时(_serde.STRICT_MSGPACK_ENABLED),build 一个 serde allowlist 让 checkpointer 只序列化 schema 里出现的字段(serde allowlist:1220-1241)。

设计动机

为什么不直接让 StateGraph 自己就是 Pregel,非要分两步?

• 构建期 vs 运行期关注点分离:builder 阶段只关心「图长什么样」,运行期阶段关心「怎么跑」。分开后 builder 可以被多次编译成不同运行体——同一个 StateGraph 配不同的 checkpointer / store / interrupt 节点跑出多个 Pregel 实例。
• 不可变运行体:compile 后的 CompiledStateGraph 字段在 Pregel.__init__ 里基本固定(Pregel.__init__:758-836),auto_validate=False 让 compile 自己控制何时校验,避免被构造函数提前校验。
• 延迟校验:用户 add_node 时只 append 进容器,不立即检查「边的起点是否存在」——这样允许用户用任意顺序注册节点和边;真正校验在 compile 时一次性做 (validate 调用:1247-1254)。
• 中断配置参数化:interrupt_before / interrupt_after 是 compile 参数,不是图本身的属性——同一张图可以编译出「带中断」和「不带中断」两个版本,人机协作场景里很常见 (interrupt 参数:1170-1171)。
• checkpointer 类型规范化:compile(checkpointer=...) 接受 None / True / False / BaseCheckpointSaver 四种,ensure_valid_checkpointer(ensure_valid_checkpointer:107-117)统一校验,避免运行时炸。
• 默认值在编译期才应用(defaults 应用:1299-1325)——这样允许用户在 add_node 之后再调 set_node_defaults,新默认值会覆盖所有没显式指定的节点。

关键文件

• compile 签名:1164-1217 — 接收 checkpointer / store / cache / interrupt_before / interrupt_after / debug / name / transformers。
• ensure_valid_checkpointer:107-117 — 校验 checkpointer 必须是 None / True / False / BaseCheckpointSaver 之一,否则 TypeError。
• ensure_valid_checkpointer 调用:1218 — compile 第一件事就是规范化 checkpointer。
• serde allowlist:1220-1241 — 严格 msgpack 下,build allowlist 并 apply 到 checkpointer,让 checkpoint 只存 schema 字段。
• interrupt 合并 + validate:1243-1254 — "*" 表示 All(所有节点),把 interrupt_before / interrupt_after 合并成一个 list 传给 validate。
• output / stream channels:1256-1273 — 单字段且是 __root__ 时直接用字符串,否则用 list 过滤掉 managed value。
• 默认 error handler 节点:1278-1297 — set_node_defaults 设的全局 error_handler 会被注入成一个名为 __default_error_handler__ 的特殊节点。
• defaults 应用:1299-1325 — 把 set_node_defaults 的 retry / cache / error_handler / timeout 应用到每个未显式指定的 spec;cache 和 error_handler 不应用到 error-handler 节点本身。
• node_error_handler_map:1327-1331 — 生成 node_name -> handler_node_name 映射,运行时根据这个 map 把失败节点的执行路由到对应 handler。
• 构造 CompiledStateGraph:1333-1357 — 把 builder 的 channels / managed 合并,加 START: EphemeralValue(input_schema) 当入口通道,设 stream_mode="updates"、input_channels=START。
• attach 三件套:1360-1388 — 循环调 compiled.attach_node(START, None) + 每个节点、attach_edge 每条边、attach_branch 每个分支;最后 compiled.validate()。
• Pregel.__init__:758-836 — CompiledStateGraph.__init__ 通过 super().__init__(**kwargs) 委托给 Pregel,这里展开所有运行期字段并 auto_validate=True 时调 self.validate()。

数据流

下面是 compile 真正构造运行体的关键段——把 builder 的所有字段塞进 CompiledStateGraph,并加一个 START 入口通道:

compiled = CompiledStateGraph[StateT, ContextT, InputT, OutputT](
    builder=self,
    schema_to_mapper={},
    context_schema=self.context_schema,
    nodes={},
    channels={
        **self.channels,
        **self.managed,
        START: EphemeralValue(self.input_schema),
    },
    input_channels=START,
    stream_mode="updates",
    output_channels=output_channels,
    stream_channels=stream_channels,
    checkpointer=checkpointer,
    interrupt_before_nodes=interrupt_before,
    interrupt_after_nodes=interrupt_after,
    auto_validate=False,
    debug=debug,
    store=store,
    cache=cache,
    node_error_handler_map=node_error_handler_map,
    name=name or "LangGraph",
    stream_transformers=transformers,
)
compiled._serde_allowlist = serde_allowlist

compiled.attach_node(START, None)
for key, node in self.nodes.items():
    compiled.attach_node(key, node)

(构造 + attach_node:1333-1362)

最后把边和分支也翻译成 Pregel 的通道订阅关系,再触发一次 validate:

for start, end in self.edges:
    compiled.attach_edge(start, end)

for starts, end in self.waiting_edges:
    compiled.attach_edge(starts, end)

for start, branches in self.branches.items():
    for name, branch in branches.items():
        compiled.attach_branch(start, name, branch)

return compiled.validate()

(attach edge/branch:1378-1388)

Pregel.__init__ 拿到这些字段后,会把 nodes 里的 NodeBuilder 翻成 PregelNode,给 TASKS 通道安一个 Topic(Send, accumulate=False),然后 auto_validate=True 时调 self.validate()(Pregel init:800-836):

self.nodes = {
    k: v.build() if isinstance(v, NodeBuilder) else v for k, v in nodes.items()
}
self.channels = channels or {}
if TASKS in self.channels and not isinstance(self.channels[TASKS], Topic):
    raise ValueError(
        f"Channel '{TASKS}' is reserved and cannot be used in the graph."
    )
else:
    self.channels[TASKS] = Topic(Send, accumulate=False)

(Pregel init nodes:800-809)

边界与失败

• checkpointer=True 不能用于根图(True 报错:2583-2584)——True 表示「从父图继承」,只有子图才能用;根图必须显式给 saver 或 False。
• interrupt_before="*" 跟 interrupt_after="*" 互斥处理(* 处理:1249-1253)——只有 interrupt_after != "*" 时才把 interrupt_before 列表合进 interrupt;这是为了用 * 表示「所有节点」时的优先级约定。
• TASKS 通道名被保留(TASKS 保留:804-807),用户在 schema 里定义名为 __pregel_tasks 的字段会直接报错——这是引擎内部用来扇出 Send 的 Topic 通道。
• 重复 attach 同名节点:builder 阶段已经挡了重名(重名:792-793)——但 __default_error_handler__ 这种自动生成的节点名也要避免跟用户节点冲突(default handler 冲突:1280-1284)。
• cache 和 error_handler 默认值不应用到 handler 节点本身(cache 不给 handler:1313-1321)——因为「缓存 handler 结果」不安全(失败节点的 state 每次可能不同),「handler 抓自己」会死循环。
• 同一 builder 多次 compile 是允许的:每次 compile 都创建新的 CompiledStateGraph,builder 的 compiled 字段虽然在 validate() 时被设成 True(compiled=True:1161),但只是个标志位,不阻止再 compile 或继续 add_node(只是会 warning)。

小结

compile 是构建器和运行时之间的边界——它把 StateGraph 里 add_node / add_edge / add_conditional_edges 累积的 spec 全部翻译成 Pregel 的运行期结构,绑上 checkpointer / store / interrupt 配置,最后返回一个可直接 invoke 的 Pregel 子类。理解了它就理解了「图声明 → 可执行体」的整条链路。

接下来可以看 Pregel 引擎 里 Pregel.__init__ 之后 invoke / stream 怎么跑这个编译产物,或看 StateSnapshot 怎么从编译产物配 checkpointer 后取状态。

对照官方资料:LangGraph 文档 · README。