第11章子图与嵌套

11.1 引言

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

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

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

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

本章要点

图作为节点：理解 add_node(name, compiled_graph) 的内部机制和 Pregel 作为 Runnable 的协议
命名空间隔离：深入 NS_SEP/NS_END 分隔符构成的层级命名空间体系
Checkpoint 命名空间：掌握 "parent|child" 格式的检查点命名空间和 checkpoint_map 的跨层级映射
ParentCommand 跨图通信：理解子图如何通过异常冒泡向父图发送控制指令
状态映射：掌握父图与子图之间的状态传递和转换机制
嵌套 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) 可以精确知道哪些任务包含子图，并按需加载它们的状态。

11.3 命名空间隔离

11.3.1 NS_SEP 与 NS_END

LangGraph 使用两个分隔符构建层级命名空间：

python

# langgraph/_internal/_constants.py
NS_SEP = "|"   # 层级分隔符，分隔图的嵌套层级
NS_END = ":"   # 命名空间与任务ID的分隔符

一个完整的检查点命名空间看起来像这样：

"agent:task-id-abc|tool_executor:task-id-def|sub_tool:task-id-ghi"
  ^       ^          ^           ^            ^        ^
  节点名  任务ID      节点名     任务ID        节点名   任务ID

每一层由 node_name:task_id 组成，层与层之间用 | 分隔。这种编码方式不仅标识了子图的嵌套层级，还通过 task_id 区分了同一个节点的不同执行实例。这在并行 map 场景中至关重要：当 Send 创建了多个同类子图的并行实例时，每个实例通过不同的 task_id 获得独立的命名空间，从而拥有互不干扰的检查点和状态。

命名空间的编码规则是完全确定性的：给定相同的图结构和执行路径，生成的命名空间字符串总是相同的。这个特性使得从中断恢复时可以精确定位到子图的检查点——不需要额外的映射表，命名空间本身就包含了足够的信息来重建图的嵌套结构。

11.3.2 命名空间的构建

在 PregelLoop.__init__ 中，命名空间从配置中提取并解析：

python

# langgraph/pregel/_loop.py

class PregelLoop:
    def __init__(self, ...):
        # 检测是否是嵌套图
        self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})

        # 构建命名空间
        scratchpad = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
        if isinstance(scratchpad, PregelScratchpad):
            if cnt := scratchpad.subgraph_counter():
                # 追加子图计数器到命名空间
                self.config = patch_configurable(
                    self.config,
                    {
                        CONFIG_KEY_CHECKPOINT_NS: NS_SEP.join((
                            config[CONF][CONFIG_KEY_CHECKPOINT_NS],
                            str(cnt),
                        ))
                    },
                )

        # 解析命名空间为元组
        self.checkpoint_ns = (
            tuple(
                cast(str, self.config[CONF][CONFIG_KEY_CHECKPOINT_NS]).split(NS_SEP)
            )
            if self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS)
            else ()
        )

subgraph_counter 是 PregelScratchpad 中的一个闭包计数器。当同一个节点的同一次执行中启动多个子图时（例如通过 Send 的并行 map），每个子图获得不同的命名空间后缀。

11.3.3 根图的特殊处理

当一个非嵌套图的配置中已经存在 checkpoint_ns 时（例如通过外部配置传入），根图会清理它：

python

if not self.is_nested and config[CONF].get(CONFIG_KEY_CHECKPOINT_NS):
    self.config = patch_configurable(
        self.config,
        {CONFIG_KEY_CHECKPOINT_NS: "", CONFIG_KEY_CHECKPOINT_ID: None},
    )

这确保了根图总是在空命名空间 "" 下运行，避免外部配置污染。这个清理逻辑看似简单，但对于系统的健壮性至关重要：在 LangGraph Platform 等部署环境中，外部系统可能在配置中设置了 checkpoint_ns 来追踪请求来源，如果不清理这些值，根图可能会错误地将自己当作子图运行，导致中断抑制等行为异常。

11.4 Checkpoint 命名空间

11.4.1 checkpoint_map：跨层级的检查点映射

checkpoint_map 是一个从命名空间到检查点 ID 的映射，它在父图和子图之间传递检查点关联信息：

python

# langgraph/_internal/_constants.py
CONFIG_KEY_CHECKPOINT_MAP = "checkpoint_map"

当父图执行子图时，它将自己的命名空间和检查点 ID 加入到 checkpoint_map 中传递给子图：

python

# 在 PregelLoop._first() 中
metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})

子图在初始化时，使用 checkpoint_map 来定位自己的起始检查点：

python

# PregelLoop.__init__
if (
    CONFIG_KEY_CHECKPOINT_MAP in self.config[CONF]
    and self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS)
    in self.config[CONF][CONFIG_KEY_CHECKPOINT_MAP]
):
    self.checkpoint_config = patch_configurable(
        self.config,
        {
            CONFIG_KEY_CHECKPOINT_ID: self.config[CONF][
                CONFIG_KEY_CHECKPOINT_MAP
            ][self.config[CONF][CONFIG_KEY_CHECKPOINT_NS]]
        },
    )

11.4.2 子图检查点的存储

子图的检查点与父图存储在同一个 CheckpointSaver 中，但通过不同的 checkpoint_ns 隔离。以 InMemorySaver 为例：

python

# InMemorySaver 的存储结构
# storage[thread_id][checkpoint_ns][checkpoint_id] = (checkpoint, metadata, parent_id)

# 父图的检查点
storage["thread-1"][""]["ck-1"] = (parent_checkpoint, ...)

# 子图的检查点
storage["thread-1"]["agent:task-abc"]["ck-2"] = (child_checkpoint, ...)

# 子子图的检查点
storage["thread-1"]["agent:task-abc|tool:task-def"]["ck-3"] = (grandchild_checkpoint, ...)

第11章 子图与嵌套 ​

11.1 引言 ​

11.2 图作为节点 ​

11.2.1 Pregel 的 Runnable 协议 ​

11.2.2 子图的 Checkpointer 继承 ​

11.2.3 子图探测与 subgraphs 属性 ​

11.3 命名空间隔离 ​

11.3.1 NS_SEP 与 NS_END ​

11.3.2 命名空间的构建 ​

11.3.3 根图的特殊处理 ​

11.4 Checkpoint 命名空间 ​

11.4.1 checkpoint_map：跨层级的检查点映射 ​

11.4.2 子图检查点的存储 ​

第11章子图与嵌套

11.1 引言

11.2 图作为节点

11.2.1 Pregel 的 Runnable 协议

11.2.2 子图的 Checkpointer 继承

11.2.3 子图探测与 subgraphs 属性

11.3 命名空间隔离

11.3.1 NS_SEP 与 NS_END

11.3.2 命名空间的构建

11.3.3 根图的特殊处理

11.4 Checkpoint 命名空间

11.4.1 checkpoint_map：跨层级的检查点映射

11.4.2 子图检查点的存储