杨艺韬讲堂

Appearance

第11章 子图与嵌套 ​

11.1 引言 ​

在前面的章节中，我们已经深入了解了单个图的全部运行机制：Channel 如何承载状态，Pregel 循环如何调度任务，Checkpoint 如何持久化状态，Command 如何控制流程。然而，当系统复杂度增长到一定程度时，单个扁平的图将变得难以维护。正如软件工程中函数调用和模块化的必要性，LangGraph 的子图（Subgraph）机制允许将一个复杂的图分解为可组合、可复用的子单元。

子图在 LangGraph 中不是简单的"函数调用"。它涉及到命名空间隔离、检查点嵌套、状态映射、跨图通信等一系列精密的工程机制。一个子图拥有自己独立的 Channel 空间、自己的 Checkpoint 历史、自己的执行循环——同时又通过精心设计的接口与父图保持协调。

理解子图机制的关键在于认识到它解决的根本问题：复杂性管理。一个完整的客服系统可能包含意图识别、知识检索、工具调用、人工审批、对话管理等多个功能模块。如果将所有这些逻辑放在一个扁平的图中，节点数量会迅速膨胀到难以维护的地步，而且不同模块之间的状态容易产生意外的耦合。子图提供了自然的封装边界：每个子图定义自己的状态 Schema、自己的执行逻辑和自己的检查点历史，通过明确定义的输入/输出接口与外部交互。这种封装不仅提升了代码的可维护性，还支持了模块的独立开发和测试——一个子图可以单独编译和运行，只有在嵌入到父图中时才需要关心状态映射的问题。

本章将从源码层面剖析 LangGraph 的子图体系：从如何将一个编译后的图作为节点添加到另一个图中，到命名空间如何隔离子图的状态，再到 Checkpoint 如何在嵌套层级中工作，以及 ParentCommand 如何实现跨图的控制流传递。在实际应用中，子图模式是构建多 Agent 系统的核心范式——每个 Agent 可以被封装为一个独立的子图，由协调器（主管）图统一编排和管理。理解子图的内部运作机制，是从"能用 LangGraph"进阶到"深度掌握 LangGraph"的关键一步。

本章要点

  1. 图作为节点：理解 add_node(name, compiled_graph) 的内部机制和 Pregel 作为 Runnable 的协议
  2. 命名空间隔离：深入 NS_SEP/NS_END 分隔符构成的层级命名空间体系
  3. Checkpoint 命名空间：掌握 "parent|child" 格式的检查点命名空间和 checkpoint_map 的跨层级映射
  4. ParentCommand 跨图通信：理解子图如何通过异常冒泡向父图发送控制指令
  5. 状态映射：掌握父图与子图之间的状态传递和转换机制
  6. 嵌套 Agent 架构：通过实际案例了解多层嵌套图的设计模式

11.2 图作为节点 ​

11.2.1 Pregel 的 Runnable 协议 ​

LangGraph 中的编译后的图（CompiledStateGraph）是 Pregel 的子类，而 Pregel 实现了 LangChain 的 Runnable 接口。这意味着一个编译后的图可以像普通函数一样被调用，也可以作为另一个图的节点：

python
# 创建子图
sub_builder = StateGraph(SubState)
sub_builder.add_node("process", process_fn)
sub_builder.add_edge(START, "process")
sub_builder.add_edge("process", END)
sub_graph = sub_builder.compile()

# 将子图作为父图的节点
parent_builder = StateGraph(ParentState)
parent_builder.add_node("sub_agent", sub_graph)
parent_builder.add_edge(START, "sub_agent")
parent_builder.add_edge("sub_agent", END)
parent_graph = parent_builder.compile(checkpointer=InMemorySaver())

当 add_node 接收到一个 Runnable（包括编译后的图）时，它通过 coerce_to_runnable 将其包装为 PregelNode：

python
# langgraph/graph/state.py - add_node

if isinstance(action, Runnable):
    node = action.get_name()
# ...
self.nodes[node] = StateNodeSpec(
    coerce_to_runnable(action, name=node, trace=False),
    metadata,
    input_schema=input_schema or self.state_schema,
    # ...
)

11.2.2 子图的 Checkpointer 继承 ​

子图的 checkpointer 行为通过 Checkpointer 类型控制：

python
Checkpointer = None | bool | BaseCheckpointSaver

三种取值对应三种行为：

  • None（默认）：继承父图的 checkpointer。子图的检查点存储在与父图相同的存储后端中，使用独立的命名空间
  • True：显式启用检查点。效果与 None 相同但语义更明确
  • False：禁用检查点。即使父图有 checkpointer，子图也不保存检查点
python
# 子图继承父图的 checkpointer
sub_graph = sub_builder.compile()  # checkpointer=None

# 子图显式禁用 checkpointer
sub_graph = sub_builder.compile(checkpointer=False)

值得注意的是，checkpointer=None（继承）和 checkpointer=True（显式启用）在当前版本中的效果几乎相同——两者都会让子图使用父图的 checkpointer。区别在于语义明确性：True 明确表示开发者期望子图有持久化能力，而 None 则表示"跟随父图的决策"。当父图也没有 checkpointer 时，None 会导致子图同样没有持久化，而 True 在这种情况下也不会创建一个 checkpointer（因为没有可继承的存储后端）。False 是唯一可以主动阻断继承链的选项——当你确定某个子图不需要持久化（例如一个无状态的数据转换子图），使用 False 可以避免不必要的检查点写入开销。

在 Pregel 的执行过程中，checkpointer 通过 CONFIG_KEY_CHECKPOINTER 从父图传递到子图：

python
# langgraph/_internal/_constants.py
CONFIG_KEY_CHECKPOINTER = "__pregel_checkpointer"

11.2.3 子图探测与 subgraphs 属性 ​

PregelExecutableTask 中的 subgraphs 字段记录了当前任务包含的子图：

python
@dataclass(**_T_DC_KWARGS)
class PregelExecutableTask:
    name: str
    input: Any
    proc: Runnable
    writes: deque[tuple[str, Any]]
    config: RunnableConfig
    triggers: Sequence[str]
    retry_policy: Sequence[RetryPolicy]
    cache_key: CacheKey | None
    id: str
    path: tuple[str | int | tuple, ...]
    writers: Sequence[Runnable] = ()
    subgraphs: Sequence[PregelProtocol] = ()

这使得父图的执行引擎可以感知子图的存在，从而在流式输出、调试信息生成和状态快照查询时正确地处理子图的事件和状态。这种显式的子图追踪（而非在运行时动态探测）使得 get_state(subgraphs=True) 可以精确知道哪些任务包含子图，并按需加载它们的状态。

基于 VitePress 构建

Copyright © 2024-present 杨艺韬