LangGraph 设计与实现
第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 接口。这意味着一个编译后的图可以像普通函数一样被调用，也可以作为另一个图的节点：
# 创建子图
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 会原样返回它（编译后的图本来就是 Runnable），连同元数据一起存进 StateNodeSpec；到 compile() 时，attach_node 才把它包装成 PregelNode（libs/langgraph/langgraph/graph/state.py:1323 起）：
# libs/langgraph/langgraph/graph/state.py - add_node（节选）
if not isinstance(node, str):
action = node
if isinstance(action, Runnable):
node = action.get_name() # 编译后的图会给出自己的名字
else:
node = getattr(action, "__name__", action.__class__.__name__)
# ...
# input_schema 显式给出时用它，其次用从类型注解推断出的，最后回落到整图的 state_schema
self.nodes[node] = StateNodeSpec(
coerce_to_runnable(action, name=node, trace=False),
metadata,
input_schema=input_schema, # 或 inferred_input_schema / self.state_schema
retry_policy=retry_policy,
cache_policy=cache_policy,
ends=ends,
defer=defer,
)graph TB
subgraph "类型层级"
R[Runnable] --> P[Pregel]
P --> CSG[CompiledStateGraph]
end
subgraph "作为节点时的包装"
CSG -->|"compile() 时 attach_node"| PN[PregelNode]
PN -->|"invoke(state, config)"| CSG2[子图执行]
end
subgraph "节点执行流程"
PARENT_LOOP[父图 PregelLoop] -->|执行任务| TASK[PregelExecutableTask]
TASK -->|"task.proc.invoke()"| PN
PN -->|"返回结果"| PARENT_LOOP
end
11.2.2 子图的 Checkpointer 继承
子图的 checkpointer 行为通过 Checkpointer 类型控制（libs/langgraph/langgraph/types.py:96）：
Checkpointer = None | bool | BaseCheckpointSaver子图的 compile(checkpointer=...) 有四种取值，具体行为由 Pregel._defaults 里的分支决定（libs/langgraph/langgraph/pregel/main.py:2411-2418：先看是否为 False，再看 config 里有没有父图传下来的 CONFIG_KEY_CHECKPOINTER，最后才用子图自己的值）：
None（默认）：沿用父图的 checkpointer。子图的检查点存在父图的存储后端里，命名空间是带任务 ID 的节点名:task_id；每次调用子图的任务 ID 都不同，所以子图每次被调用都从空检查点起步，只在同一次调用之内（例如中断后恢复）延续自己的检查点True：同样沿用父图的 checkpointer，但子图会用recast_checkpoint_ns去掉命名空间里的任务 ID（连同数字段），检查点固定写在节点名这个命名空间下（pregel/main.py:2613-2615），于是子图状态会在同一线程的多次调用之间累积。根图不能用True，否则抛RuntimeError: checkpointer=True cannot be used for root graphs.False：禁用检查点。即使父图有 checkpointer，子图也不继承、不保存检查点- 具体的
BaseCheckpointSaver实例：只在它作为根图运行时生效。嵌套运行时，父图总会在任务 config 里放入CONFIG_KEY_CHECKPOINTER（父图没有 checkpointer 时值为None），这个分支排在self.checkpointer之前，所以子图自带的 saver 会被忽略，行为与None相同
# 子图沿用父图的 checkpointer，每次调用独立
sub_graph = sub_builder.compile() # checkpointer=None
# 子图沿用父图的 checkpointer，状态跨调用累积
sub_graph = sub_builder.compile(checkpointer=True)
# 子图显式禁用 checkpointer
sub_graph = sub_builder.compile(checkpointer=False)None 和 True 的差别可以直接跑出来。下面的子图每次往带 operator.add 的 items 里追加一个 "x"，父图在同一线程上连续调用两次：
import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class S(TypedDict):
items: Annotated[list[str], operator.add]
def run(sub_checkpointer):
sb = StateGraph(S)
sb.add_node("add", lambda s: {"items": ["x"]})
sb.add_edge(START, "add")
sb.add_edge("add", END)
pb = StateGraph(S)
pb.add_node("child", sb.compile(checkpointer=sub_checkpointer))
pb.add_edge(START, "child")
pb.add_edge("child", END)
saver = InMemorySaver()
graph = pb.compile(checkpointer=saver)
config = {"configurable": {"thread_id": "t"}}
graph.invoke({"items": []}, config)
second = graph.invoke({"items": []}, config)["items"]
namespaces = sorted(ns for tid in saver.storage for ns in saver.storage[tid])
print(sub_checkpointer, second, namespaces)
for cp in (None, True, False, InMemorySaver()):
run(cp)实跑结果（pin 版本，任务 ID 每次运行都不同，这里缩写为 <id1>、<id2>）：
| 子图 checkpointer | 第二次 invoke 的 items |
父图 saver 里的命名空间 |
|---|---|---|
None |
['x', 'x', 'x'] |
''、'child:<id1>'、'child:<id2>' |
True |
['x', 'x', 'x', 'x'] |
''、'child' |
False |
['x', 'x', 'x'] |
'' |
自带 InMemorySaver() |
['x', 'x', 'x'] |
''、'child:<id1>'、'child:<id2>'（子图自带的 saver 里什么也没写） |
第二次调用时父图的 items 是 ['x']，它作为输入传给子图。None 时子图从空检查点起步，得到 ['x', 'x'] 交回父图，经父图的 reducer 追加后是三个；True 时子图先加载上一次留在 'child' 命名空间里的 ['x']，再叠加输入和本次写入，交回 ['x', 'x', 'x']，父图于是得到四个。父图本身没有 checkpointer 时，None 和 True 都不会持久化，也不报错（实跑）。False 是唯一能主动阻断继承的取值——当你确定某个子图不需要持久化（例如一个无状态的数据转换子图），使用 False 可以避免不必要的检查点写入开销。
在 Pregel 的执行过程中，checkpointer 通过 CONFIG_KEY_CHECKPOINTER 从父图传递到子图：
# libs/langgraph/langgraph/_internal/_constants.py:36
CONFIG_KEY_CHECKPOINTER = sys.intern("__pregel_checkpointer")11.2.3 子图探测与 subgraphs 属性
PregelExecutableTask 中的 subgraphs 字段记录了当前任务包含的子图：
# libs/langgraph/langgraph/types.py:537
@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] = ()这个字段抄自 PregelNode.subgraphs（pregel/_algo.py:735、1043），而后者在构造节点时就由 find_subgraph_pregel 静态确定（pregel/_read.py:162-173、pregel/_utils.py:36）：它检查节点绑定的 Runnable，拆开 RunnableSequence，还会用 langchain_core 的 get_function_nonlocals 顺着节点函数引用的全局与闭包变量去找——所以在节点函数里调用的子图也能被识别；checkpointer=False 的子图不计入。stream_mode="debug" 的 checkpoint 事件据 task.subgraphs 给含子图的任务附上子图配置（pregel/debug.py:137）；get_state(subgraphs=True) 则通过 get_subgraphs() 读同一份 PregelNode.subgraphs，据此知道哪些待执行任务包含子图、去加载它们的状态。
11.3 命名空间隔离
11.3.1 NS_SEP 与 NS_END
LangGraph 使用两个分隔符构建层级命名空间：
# libs/langgraph/langgraph/_internal/_constants.py:75,77
NS_SEP = sys.intern("|") # 层级分隔符，分隔图的嵌套层级
NS_END = sys.intern(":") # 节点名与任务 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 获得独立的命名空间，从而拥有互不干扰的检查点和状态。
命名空间的编码规则是确定性的：任务 ID 由父图当前检查点的 ID、命名空间、步数、节点名等算出（pregel/_algo.py:594、775），所以从同一个检查点恢复时，同一个任务会重新得到同一个 task_id、同一个命名空间。这个特性使得从中断恢复时可以精确定位到子图的检查点——不需要额外的映射表，命名空间本身就包含了足够的信息来重建图的嵌套结构。
11.3.2 命名空间的构建
在 PregelLoop.__init__ 中，命名空间从配置中提取并解析：
# libs/langgraph/langgraph/pregel/_loop.py:248-303（节选）
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 中的一个 LazyAtomicCounter（pregel/_algo.py:1142、1224 起），每个任务的 scratchpad 各有一个，从 0 开始计数。同一个任务里第一次启动的子图拿到 0，命名空间不变；同一个任务再启动第二、第三个子图时拿到 1、2，命名空间后面追加 |1、|2。典型场景是一个节点函数里先后调用多个子图——实跑一个先后调用两个子图的节点 caller，父图 saver 里出现 caller:<task_id> 和 caller:<task_id>|1 两个命名空间。Send 的并行 map 则不同：每个 Send 本身就是一个独立任务，各有自己的 task_id，靠 task_id 区分，用不到这个计数器。
11.3.3 根图的特殊处理
当一个非嵌套图的配置中已经存在 checkpoint_ns 时（例如通过外部配置传入），根图会清理它：
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},
)这确保了根图总是在空命名空间 "" 下运行，避免外部配置污染。注意"是不是子图"只看 config 里有没有 CONFIG_KEY_TASK_ID（is_nested），与 checkpoint_ns 无关；如果不做这一步清理，外部传入的 checkpoint_ns 会让根图的检查点读写落到别的命名空间里。
graph TB
subgraph "命名空间层级"
ROOT["根图<br/>checkpoint_ns = ''"]
ROOT --> AGENT["agent:task-abc<br/>checkpoint_ns = 'agent:task-abc'"]
AGENT --> TOOL["tool:task-def<br/>checkpoint_ns = 'agent:task-abc|tool:task-def'"]
TOOL --> SUB["sub:task-ghi<br/>checkpoint_ns = 'agent:task-abc|tool:task-def|sub:task-ghi'"]
ROOT --> REVIEWER["reviewer:task-xyz<br/>checkpoint_ns = 'reviewer:task-xyz'"]
end
style ROOT fill:#e8f5e9
style AGENT fill:#e3f2fd
style TOOL fill:#fff3e0
style SUB fill:#fce4ec
style REVIEWER fill:#e3f2fd
11.4 Checkpoint 命名空间
11.4.1 checkpoint_map：跨层级的检查点映射
checkpoint_map 是一个从命名空间到检查点 ID 的映射，它在父图和子图之间传递检查点关联信息：
# libs/langgraph/langgraph/_internal/_constants.py:51
CONFIG_KEY_CHECKPOINT_MAP = sys.intern("checkpoint_map")父图为任务准备 config 时，把自己的命名空间和当前检查点 ID 并入 checkpoint_map 传给子图；子图写检查点时，再把这份映射记进元数据的 parents 字段：
# libs/langgraph/langgraph/pregel/_loop.py:798 - PregelLoop._put_checkpoint()
metadata["parents"] = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})（checkpoint_map 的组装在 pregel/_algo.py:719-722 与 858-861。）
子图在初始化时，使用 checkpoint_map 来定位自己的起始检查点：
# libs/langgraph/langgraph/pregel/_loop.py:278 起
# 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 为例：
# 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, ...)graph LR
subgraph "CheckpointSaver 存储"
direction TB
subgraph "thread_id = 'thread-1'"
NS0["checkpoint_ns = ''<br/>(根图)"]
NS1["checkpoint_ns = 'agent:task-abc'<br/>(子图)"]
NS2["checkpoint_ns = 'agent:task-abc|tool:task-def'<br/>(子子图)"]
end
NS0 --> CK0["ck-001: {channel_values: {...}}"]
NS0 --> CK1["ck-002: {channel_values: {...}}"]
NS1 --> CK2["ck-003: {channel_values: {...}}"]
NS1 --> CK3["ck-004: {channel_values: {...}}"]
NS2 --> CK4["ck-005: {channel_values: {...}}"]
end
11.4.3 恢复时的层级协调
从中断恢复或做时间旅行时，根图要把"这是一次恢复""这是一次重放"告诉子图。CONFIG_KEY_RESUMING 让子图在已有检查点上继续，而不是把父图传来的输入当作新输入重新应用（pregel/_loop.py:633-643）；普通恢复时子图按自己的 checkpoint_ns 取最新检查点（pregel/_loop.py:1159-1162）。CONFIG_KEY_REPLAY_STATE 只在重放时出现，帮子图找到重放点之前的那个检查点：
# libs/langgraph/langgraph/pregel/_loop.py:762
# PregelLoop._first() - 将恢复和重放标志传递给子图
if not self.is_nested:
replay_state: ReplayState | None = None
if self.is_replaying:
replay_checkpoint_id = self.checkpoint["id"]
if (
self.checkpoint_metadata.get("source") == "update"
and self.prev_checkpoint_config
):
replay_checkpoint_id = self.prev_checkpoint_config[CONF].get(
CONFIG_KEY_CHECKPOINT_ID, replay_checkpoint_id
)
replay_state = ReplayState(replay_checkpoint_id)
self.config = patch_configurable(
self.config,
{
CONFIG_KEY_RESUMING: is_resuming,
CONFIG_KEY_REPLAY_STATE: replay_state,
},
)ReplayState（libs/langgraph/langgraph/_internal/_replay.py）持有父图的重放检查点 ID，并用一个 _visited_ns 集合记录哪些子图命名空间已经加载过——同一个逻辑子图第一次被访问时，它用 checkpointer.list(..., before={"configurable": {"checkpoint_id": ...}}, limit=1) 取回重放点之前的那个检查点；同一子图在后续循环迭代中再被访问时，则退回普通的 "取最新检查点" 逻辑，以免读到过期状态。判断"是否同一个逻辑子图"时会先用 rsplit(NS_END, 1) 剥掉任务 ID 后缀，这样循环里每轮不同 task_id 的同名子图仍被认作同一个。
11.4.4 exit 模式下的子图处理
在 durability="exit" 模式下，检查点只在图退出时保存。对于嵌套图，需要特殊处理：
# libs/langgraph/langgraph/pregel/_loop.py:870-886（节选，省略类型注解与之后的中断抑制逻辑）
def _suppress_interrupt(self, exc_type, exc_value, traceback):
if self.durability == "exit" and (
not self.is_nested # 顶层图
or exc_value is not None # 子图有错误或中断
or all(NS_END not in part for part in self.checkpoint_ns) # 或子图自带 checkpointer=True
):
self._put_checkpoint(self.checkpoint_metadata)
self._put_pending_writes()子图在正常完成时不保存检查点（exit 模式），但在中断或错误时会保存，确保可以从中断点恢复。这种条件性保存的逻辑体现了 LangGraph 在性能与可靠性之间的务实平衡：对于正常完成的子图，其结果已经被父图的检查点捕获，不需要额外的子图级检查点；但对于中断或出错的子图，如果不保存检查点，恢复时就需要从头重新执行子图的所有步骤，这可能涉及昂贵的 LLM 调用或外部 API 请求。条件 all(NS_END not in part for part in self.checkpoint_ns)（源码注释写作 "or a nested graph with checkpointer=True"）用于识别子图自带 checkpointer 的情形——此时命名空间的各段都不含 NS_END（即没有 node:task_id 形式的任务后缀），子图的检查点由它自己负责落盘。
11.5 ParentCommand 机制
11.5.1 从子图到父图的控制流
在第10章中我们已经介绍了 ParentCommand 的基本概念。在子图的上下文中，这个机制变得更加重要：
# 子图节点返回指向父图的 Command
def child_node(state):
result = process(state)
return Command(
graph=Command.PARENT,
update={"child_result": result},
goto="parent_handler"
)11.5.2 异常冒泡的完整链路
sequenceDiagram
participant CN as 子图节点
participant CB as _control_branch
participant CT as 子图任务 run_with_retry
participant CL as 子图 PregelLoop
participant PT as 父图任务 run_with_retry
participant PL as 父图 PregelLoop
CN->>CB: return Command(graph=PARENT, ...)
CB->>CB: 检测 graph == PARENT
CB->>CT: raise ParentCommand(cmd)
CT->>CT: 把 cmd.graph 改写为父图任务的命名空间
CT->>CL: 重新抛出
Note over CL: ParentCommand 是 GraphBubbleUp<br/>子图不抑制此异常
CL->>PT: 异常冒泡到父图任务
PT->>PT: 捕获 ParentCommand，cmd.graph 与自身命名空间相符
PT->>PL: 交给该节点的 writers，转换为父图写入
Note over PL: cmd.update -> 父图状态更新<br/>cmd.goto -> 父图路由写入<br/>子图节点的静态出边照常触发
PL->>PL: apply_writes()
PL->>PL: prepare_next_tasks()
11.5.3 多层嵌套的传递
Command.PARENT 只向上传递一层：子图任务的 run_with_retry 把它改写成父图中那个任务的命名空间（pregel/_retry.py:127-138），父图里对应的任务一接住就处理掉，不会继续往上走。如果确实需要跨越多层传递，中间层的子图必须显式处理和转发：
# 孙图节点
def grandchild_node(state):
return Command(graph=Command.PARENT, update={"msg": "from grandchild"})
# 这只到达子图，不会自动到达祖父图
# 子图节点（需要显式转发）
def child_node(state):
# 如果需要进一步向上传递
return Command(
graph=Command.PARENT,
update={"msg": state.get("msg", "from child")}
)11.6 状态映射
11.6.1 输入映射
把编译后的子图直接作为节点加入父图时，节点的输入模式默认就是父图的 state_schema（编译后的图不是函数，add_node 推断不出输入类型），所以子图收到的是父图的整份状态；子图再按自己的输入模式只取其中同名的键。不需要额外声明映射：
# 父图的状态
class ParentState(TypedDict):
query: str
context: list[str]
result: str
# 子图的状态
class ChildState(TypedDict):
query: str
intermediate: list[str]
# 直接把子图加为节点，只有同名字段 query 会进出子图
parent_builder.add_node("child", child_graph)实跑中子图节点看到的输入只有 {'query': ...}。注意不要在这里给 add_node 传 input_schema=ChildState：input_schema 里的键会被注册成父图的通道，intermediate 随之变成父图通道，子图交回的 intermediate 也会写进父图（实跑 stream_mode="updates" 可以看到 {'child': {'query': ..., 'intermediate': [...]}}），恰好破坏了下文所说的隔离。
输出方向，只有父图认识的字段会被写回，这是通过父图 attach_node 里 _get_updates 的 output_keys 过滤实现的：
# libs/langgraph/langgraph/graph/state.py:1236 起，attach_node 中（节选）
if key == START:
output_keys = [k for k, v in self.builder.schemas[self.builder.input_schema].items()
if not is_managed_value(v)]
else:
output_keys = list(self.builder.channels) + [...]
def _get_updates(input):
if isinstance(input, dict):
return [(k, v) for k, v in input.items() if k in output_keys]如果父子图没有共享的键（例如父图用 question / answer，子图用 query / result），就换另一种方式：在普通节点函数里调用子图，自己完成两个方向的转换：
def call_child(state: ParentState) -> dict:
out = child_graph.invoke({"query": state["question"]}) # 父 -> 子：显式转换
return {"answer": out["result"]} # 子 -> 父：显式转换
parent_builder.add_node("child", call_child)这种写法里子图同样运行在 child:<task_id> 命名空间下，也会沿用父图的 checkpointer（§11.2.2），find_subgraph_pregel 顺着函数引用的变量也能把它识别为该节点的子图（§11.2.3，实跑 graph.nodes["child"].subgraphs 里就是这个子图）。
11.6.2 输出映射
子图的输出同样通过 Schema 匹配映射回父图。子图的最终输出（由 output_schema 定义的通道值）被作为节点的返回值，然后经过 _get_updates 过滤，只有与父图 Schema 匹配的字段才会被写入父图的 Channel。
graph TB
subgraph "状态映射"
PS["父图状态<br/>{query, context, result}"]
CS["子图状态<br/>{query, intermediate}"]
PS -->|"输入映射<br/>同名字段: query"| CS
CS -->|"输出映射<br/>同名字段: query"| PS
NOTE1["context, result<br/>对子图不可见"]
NOTE2["intermediate<br/>对父图不可见"]
end
style NOTE1 fill:#fff3e0,stroke:#e65100
style NOTE2 fill:#fff3e0,stroke:#e65100
11.6.3 完全隔离的子图状态
子图拥有完全独立的 Channel 空间。即使父图和子图有同名的 Channel，它们也是物理隔离的——子图的写入不会直接影响父图的 Channel，只有在子图完成后，通过输出映射才会将结果传递回父图。
这种隔离通过命名空间实现：子图的 Channel 值存储在不同的 checkpoint_ns 下，与父图的值物理分离。这意味着即使父图和子图都有一个叫做 messages 的 Channel，它们也是完全独立的——子图向 messages 写入的值不会出现在父图的 messages 中，反之亦然。只有在子图完成执行后，通过输出映射机制，子图的最终状态中与父图 Schema 匹配的字段才会被写回父图的相应 Channel。
这种设计相比共享状态空间的方案有明显优势：子图的内部实现细节不会泄漏到父图的命名空间中，减少了意外的状态冲突；子图可以自由地使用临时的内部 Channel 而不担心与父图或其他子图产生冲突；子图的 Schema 可以完全独立于父图定义，只需要在输入/输出接口处保持兼容。
11.7 嵌套 Agent 架构
11.7.1 基本的主管-工人模式
最常见的嵌套模式是主管（Supervisor）协调多个工人（Worker）子图：
import operator
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, Send
from langgraph.checkpoint.memory import InMemorySaver
class SupervisorState(TypedDict):
query: str
results: Annotated[list[str], operator.add]
final_answer: str
class WorkerState(TypedDict):
query: str
results: list[str]
class WorkerOutput(TypedDict): # 只把 results 交回父图，避免两个工人并行回写 query 冲突
results: list[str]
# 创建工人子图
def create_worker(name: str):
builder = StateGraph(WorkerState, output_schema=WorkerOutput)
builder.add_node("process", lambda state: {"results": [f"{name}: {state['query']}"]})
builder.add_edge(START, "process")
builder.add_edge("process", END)
return builder.compile()
research_worker = create_worker("研究员")
analysis_worker = create_worker("分析师")
# 主管图
supervisor_builder = StateGraph(SupervisorState)
def supervisor(state: SupervisorState) -> Command:
"""决定分派给哪些工人"""
return Command(
goto=[
Send("research", {"query": state["query"]}),
Send("analysis", {"query": state["query"]}),
]
)
def synthesize(state: SupervisorState) -> dict:
return {"final_answer": " + ".join(state["results"])}
supervisor_builder.add_node("supervisor", supervisor)
supervisor_builder.add_node("research", research_worker)
supervisor_builder.add_node("analysis", analysis_worker)
supervisor_builder.add_node("synthesize", synthesize)
supervisor_builder.add_edge(START, "supervisor")
supervisor_builder.add_edge("research", "synthesize")
supervisor_builder.add_edge("analysis", "synthesize")
supervisor_builder.add_edge("synthesize", END)
graph = supervisor_builder.compile(checkpointer=InMemorySaver())
print(graph.invoke({"query": "Q"}, {"configurable": {"thread_id": "1"}}))
# {'query': 'Q', 'results': ['研究员: Q', '分析师: Q'], 'final_answer': '研究员: Q + 分析师: Q'}两个工人在同一超步并行运行，各自交回的 results 由父图的 operator.add 合并。如果工人子图把整份状态（含 query）交回，两个并行任务同时写父图的 LastValue 通道 query，会抛 InvalidUpdateError: At key 'query': Can receive only one value per step.（实跑）——这正是 output_schema 在这里的作用。
graph TB
subgraph "主管-工人架构"
S((START)) --> SUP[Supervisor]
SUP -->|"Send('research', ...)"| RES[Research Worker]
SUP -->|"Send('analysis', ...)"| ANA[Analysis Worker]
subgraph "Research Worker (子图)"
R_START((S)) --> R_PROC[Process]
R_PROC --> R_END((E))
end
subgraph "Analysis Worker (子图)"
A_START((S)) --> A_PROC[Process]
A_PROC --> A_END((E))
end
RES --> SYN[Synthesize]
ANA --> SYN
SYN --> E((END))
end
11.7.2 递归嵌套
子图可以再次包含子图，形成任意深度的嵌套。每一层都有独立的命名空间和检查点：
# Level 3: 最内层工具
tool_graph = create_tool_graph()
# Level 2: Agent 使用工具
agent_builder = StateGraph(AgentState)
agent_builder.add_node("llm", llm_node)
agent_builder.add_node("tools", tool_graph)
agent_builder.add_edge(START, "llm")
agent_builder.add_conditional_edges("llm", should_use_tool, {"yes": "tools", "no": END})
agent_builder.add_edge("tools", "llm")
agent_graph = agent_builder.compile()
# Level 1: 主管协调多个 Agent
supervisor_builder = StateGraph(SupervisorState)
supervisor_builder.add_node("agent_a", agent_graph)
supervisor_builder.add_node("agent_b", agent_graph)
supervisor_builder.add_node("coordinator", coordinator_fn)
# ...这种多层嵌套的架构在企业级 Agent 系统中很常见：顶层的协调器负责任务分配和结果汇总，中层的 Agent 负责特定领域的推理和决策，底层的工具子图负责与外部系统的交互。每一层都可以独立开发、测试和迭代，只要保持层间接口（即输入/输出 Schema）的稳定即可。
此时的命名空间示例：
根图: ""
agent_a: "agent_a:task-1"
agent_a/tools: "agent_a:task-1|tools:task-2"
agent_b: "agent_b:task-3"
agent_b/tools: "agent_b:task-3|tools:task-4"11.7.3 带中断的嵌套图
子图中的中断会向上冒泡到根图，形成一个完整的暂停/恢复链。子图的 PregelLoop 不吞掉 GraphInterrupt，异常一路冒泡，到根图才被吞掉，根图的调用照常返回带 __interrupt__ 的结果——判断条件在 _suppress_interrupt 里，只在 not self.is_nested 时抑制（pregel/_loop.py:888）：
# 子图中的中断
def child_review(state):
decision = interrupt({
"context": state["proposal"],
"question": "是否批准子任务？"
})
return {"decision": decision}
# 创建子图
child_builder = StateGraph(ChildState)
child_builder.add_node("review", child_review)
child_builder.add_edge(START, "review")
child_builder.add_edge("review", END)
child_graph = child_builder.compile()
# 父图
parent_builder = StateGraph(ParentState)
parent_builder.add_node("prepare", prepare_fn)
parent_builder.add_node("child", child_graph)
parent_builder.add_node("finalize", finalize_fn)
parent_builder.add_edge(START, "prepare")
parent_builder.add_edge("prepare", "child")
parent_builder.add_edge("child", "finalize")
parent_builder.add_edge("finalize", END)
graph = parent_builder.compile(checkpointer=InMemorySaver())执行和恢复：
config = {"configurable": {"thread_id": "t1"}}
# 第一次执行：子图中断
result = graph.invoke({"proposal": "..."}, config)
# result 包含中断信息
# 恢复：值传递到子图的 interrupt()
result = graph.invoke(Command(resume="approved"), config)
# 子图恢复执行，完成后返回父图继续sequenceDiagram
participant U as 用户
participant PG as 父图
participant CG as 子图
participant CP as Checkpoint
U->>PG: invoke(input, config)
PG->>PG: prepare 节点执行
PG->>CG: 调用 child 子图
CG->>CP: 子图逐步保存 checkpoint (ns="child:task-x")
CG->>CG: review 节点
CG->>CG: interrupt("是否批准？")
CG-->>CG: raise GraphInterrupt
Note over CG,PG: GraphInterrupt 冒泡<br/>子图不抑制
CG->>CP: 子图写入 INTERRUPT pending write (ns="child:task-x")
CG-->>PG: GraphInterrupt 传播
PG->>CP: 父图任务写入 INTERRUPT pending write (ns="")
PG-->>U: {__interrupt__: [...]}
U->>PG: invoke(Command(resume="approved"), config)
PG->>CP: 加载父图 checkpoint
PG->>CG: 恢复 child 子图
CG->>CP: 加载子图 checkpoint
CG->>CG: 重新执行 review 节点
CG->>CG: interrupt() 返回 "approved"
CG-->>PG: 子图完成，返回结果
PG->>PG: finalize 节点执行
PG-->>U: 最终结果
11.8 子图与流式输出
11.8.1 CONFIG_KEY_STREAM 传递
父图的流式输出协议通过 CONFIG_KEY_STREAM 传递给子图，使得子图的事件可以出现在父图的输出流中：
# libs/langgraph/langgraph/_internal/_constants.py:38
CONFIG_KEY_STREAM = sys.intern("__pregel_stream")
# libs/langgraph/langgraph/pregel/main.py:2702 - 只有 subgraphs=True 时才把父图的 stream 放进 config
if subgraphs:
loop.config[CONF][CONFIG_KEY_STREAM] = loop.stream
# libs/langgraph/langgraph/pregel/_loop.py:255 - PregelLoop.__init__ 合并流
if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
# libs/langgraph/langgraph/pregel/_loop.py:133 - DuplexStream 本体
def DuplexStream(*streams: StreamProtocol) -> StreamProtocol:
def __call__(value: StreamChunk) -> None:
for stream in streams:
if value[1] in stream.modes:
stream(value)
return StreamProtocol(__call__, {mode for s in streams for mode in s.modes})DuplexStream 将本地流和父图传递的流合并，使得子图的事件同时出现在两个流中。这种双工流的设计确保了无论从父图还是直接从子图观察，都能看到完整的事件序列。合并的实现非常简洁——DuplexStream 的回调函数遍历所有注册的流，根据每个流关注的 stream_mode 选择性地转发事件。这意味着如果父图只订阅了 "values" 模式的输出，子图的 "debug" 事件不会被不必要地传递给父图。
流式输出在嵌套图中的传播需要特别注意事件的归属问题。客户端如何区分一个 "updates" 事件是来自父图的某个节点还是来自子图的内部节点？答案就在事件的命名空间前缀中——每个事件都带有产生它的图的完整命名空间路径。通过解析这个路径，客户端可以构建出嵌套图执行的完整拓扑视图，精确地将每个事件定位到其来源层级和节点。
11.8.2 命名空间前缀
子图的流式事件带有命名空间前缀，客户端可以通过命名空间区分事件来源：
# 流式输出示例（v2 格式的每个 part 都是带 type/ns/data 的 TypedDict）
async for part in graph.astream(input, subgraphs=True, version="v2"):
print(f"ns={part['ns']}, type={part['type']}")
# 输出（updates 在节点结束时产出，所以子图内部的事件先于父图的 child 节点事件）:
# ns=('child:task-abc',), type=updates <- 子图事件
# ns=(), type=updates <- 父图事件注意 subgraphs=True 是必须的：Pregel.stream/astream 只有在这个参数为真时才会把父图的 StreamProtocol 写入 config[CONF][CONFIG_KEY_STREAM]，子图才有可能把事件汇入父图的流。v1 格式下，单一 stream_mode 时以 (namespace, chunk) 二元组返回，多个 stream_mode 时为 (namespace, mode, chunk) 三元组（见第 13 章）；v2 格式（version="v2"）则统一为 ValuesStreamPart / UpdatesStreamPart 等 TypedDict，ns 是一个字符串元组（定义见 libs/langgraph/langgraph/types.py）。
11.9 子图的 checkpoint_ns 与调试
11.9.1 get_state 与子图状态
LangGraph 提供了查看子图状态的能力：
# 获取父图状态
state = graph.get_state(config)
# 获取子图状态
state = graph.get_state(config, subgraphs=True)
# state.tasks 中的 task.state 包含子图的 StateSnapshot11.9.2 子图的时间旅行
子图的检查点可以独立进行时间旅行：
# 列出子图的检查点历史
child_config = {
"configurable": {
"thread_id": "t1",
"checkpoint_ns": "child:task-abc",
}
}
history = list(graph.get_state_history(child_config))这得益于命名空间隔离——每个命名空间有独立的检查点链，可以独立浏览和回退。在调试复杂的嵌套 Agent 系统时，这种能力极为有用：你可以深入到某个特定子图的执行历史中，逐步检查它在每个步骤的状态变化和决策过程，而不需要关心父图或其他并行子图的状态。这种层级化的状态审计能力对于理解和诊断多 Agent 系统的行为至关重要——当最终结果不符合预期时，你可以沿着嵌套层级逐层深入，精确定位问题发生在哪个子图的哪个步骤。结合 stream_mode="debug" 的流式输出，开发者可以获得嵌套图执行的完整可观测性。
需要注意的是，StateSnapshot.tasks 中每个 PregelTask 的 state 字段有两种形态，取决于 subgraphs 参数（pregel/main.py:1069-1100）：get_state(config) 默认只给出一个 RunnableConfig，里面是 thread_id 和子图的 checkpoint_ns，表示"这里有子图检查点"，可以拿它再查询；get_state(config, subgraphs=True) 才递归调用子图的 get_state，给出完整的 StateSnapshot。以 §11.11 的审批示例中断后为例，前者是 {'configurable': {'thread_id': ..., 'checkpoint_ns': 'dept:<task_id>'}}，后者的 next 是 ('review',)（实跑）。这种按需加载的设计避免了在查看父图状态时无条件加载所有子图的状态数据。
graph TB
subgraph "Checkpoint 时间线"
direction LR
subgraph "根图 (ns='')"
P1["step -1<br/>input"] --> P2["step 0<br/>prepare"] --> P3["step 1<br/>child开始"]
P3 -.->|中断| P4["step 1<br/>(中断)"]
P4 -->|恢复| P5["step 2<br/>finalize"]
end
subgraph "子图 (ns='child:task-abc')"
C1["step -1<br/>input"] --> C2["step 0<br/>review"]
C2 -.->|中断| C3["step 0<br/>(中断)"]
C3 -->|恢复| C4["step 1<br/>完成"]
end
end
P3 -.->|启动子图| C1
C4 -.->|返回父图| P5
style P4 fill:#f9f,stroke:#333
style C3 fill:#f9f,stroke:#333
11.10 设计决策分析
11.10.1 为什么选择命名空间隔离而非独立存储？
子图的检查点与父图存储在同一个 CheckpointSaver 中，通过 checkpoint_ns 隔离，而不是为每个子图创建独立的存储实例。这带来了几个优势：
- 查询便利性：通过同一个
thread_id就可以查询到所有嵌套层级的检查点，支持从根图到任意子图的深度状态审查 - 资源效率：避免为每个子图层级创建独立的数据库连接或存储实例，这在深度嵌套的场景下可以节省系统资源
- 统一的生命周期管理：
delete_thread可以一次性清理所有嵌套层级的检查点和写入数据，不会留下孤立的子图数据
11.10.2 为什么 ParentCommand 使用异常而非返回值？
子图向父图传递控制使用异常冒泡机制。另一种可能的设计是将 Command 作为子图的特殊返回值。选择异常的原因：
- 即时性：异常一抛出，发出它的节点立即结束，子图也不再进入下一个超步。不过同一超步里并行的其他任务不会被取消：
_should_stop_others不把GraphBubbleUp当作失败，要等它们跑完，_panic_or_proceed才把异常重新抛出（pregel/_runner.py:462-520，实跑：并行的慢节点照样执行完） - 与中断的统一性：
GraphBubbleUp基类统一了中断和跨图通信的传播机制 - 可跨越多层：虽然
Command.PARENT只上传一层，但异常机制本身支持多层冒泡
11.10.3 为什么子图重新执行而非从缓存恢复？
当父图从中断恢复时，子图是重新执行的（从子图的最新检查点恢复），而不是直接返回缓存的结果。这保证了：
- 中断恢复的一致性：如果中断发生在子图内部，必须重新进入子图才能正确恢复
- 检查点的粒度：子图有自己的检查点，恢复时只需重新执行从最后一个子图检查点开始的步骤
11.10.4 subgraph_counter 的作用
PregelScratchpad.subgraph_counter 是一个递增计数器，用于区分同一个任务在一次执行中先后启动的多个子图。例如：
# 一个节点函数里先后调用两个子图
def caller(state):
sub_a.invoke(state) # 命名空间 caller:<task_id>
sub_b.invoke(state) # 命名空间 caller:<task_id>|1
return {}两次调用来自同一个任务，task_id 相同，subgraph_counter 让第二次调用的命名空间多出 |1 后缀（§11.3.2 实跑），两个子图的检查点才不会混在一起。Send 并行启动同一个子图的多个实例则不需要它：每个 Send 本身是一个独立任务，task_id 各不相同（实跑两个 Send("worker", ...) 得到两个不同的 worker:<task_id>，没有数字后缀）。
11.11 完整示例：层级审批系统
下面是一个将本章所有概念结合在一起的完整示例：
from typing import Annotated, Literal, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
# --- 子图：部门审批 ---
class DeptState(TypedDict):
request: str
dept_decision: str
def dept_review(state: DeptState) -> dict:
decision = interrupt({
"request": state["request"],
"question": "部门是否批准此请求？"
})
return {"dept_decision": decision}
dept_builder = StateGraph(DeptState)
dept_builder.add_node("review", dept_review)
dept_builder.add_edge(START, "review")
dept_builder.add_edge("review", END)
dept_graph = dept_builder.compile()
# --- 父图：审批流程 ---
class ApprovalState(TypedDict):
request: str
dept_decision: str
final_decision: str
log: Annotated[list[str], operator.add]
def prepare(state: ApprovalState) -> dict:
return {"log": [f"收到请求: {state['request']}"]}
def final_review(state: ApprovalState) -> Command[Literal["approved", "rejected"]]:
if state["dept_decision"] == "approved":
return Command(
update={"final_decision": "approved", "log": ["最终批准"]},
goto="approved"
)
else:
return Command(
update={"final_decision": "rejected", "log": ["最终拒绝"]},
goto="rejected"
)
def on_approved(state: ApprovalState) -> dict:
return {"log": ["执行审批通过流程"]}
def on_rejected(state: ApprovalState) -> dict:
return {"log": ["通知申请人被拒绝"]}
parent_builder = StateGraph(ApprovalState)
parent_builder.add_node("prepare", prepare)
parent_builder.add_node("dept", dept_graph) # 子图作为节点
parent_builder.add_node("final_review", final_review)
parent_builder.add_node("approved", on_approved)
parent_builder.add_node("rejected", on_rejected)
parent_builder.add_edge(START, "prepare")
parent_builder.add_edge("prepare", "dept")
parent_builder.add_edge("dept", "final_review")
parent_builder.add_edge("approved", END)
parent_builder.add_edge("rejected", END)
graph = parent_builder.compile(checkpointer=InMemorySaver())graph TB
S((START)) --> PREP[prepare]
PREP --> DEPT["dept (子图)"]
subgraph "部门审批子图"
DS((Start)) --> DR["review<br/>(interrupt)"]
DR --> DE((End))
end
DEPT --> FR[final_review]
FR -->|"Command(goto='approved')"| APP[approved]
FR -->|"Command(goto='rejected')"| REJ[rejected]
APP --> E((END))
REJ --> E
style DR fill:#f9f,stroke:#333
style FR fill:#9cf,stroke:#333
使用流程：
config = {"configurable": {"thread_id": "approval-1"}}
# 1. 启动审批流程（子图中断）
result = graph.invoke({"request": "采购10台服务器"}, config)
# -> 中断: {"request": "采购10台服务器", "question": "部门是否批准此请求？"}
# 2. 部门批准（恢复子图，继续到 final_review）
result = graph.invoke(Command(resume="approved"), config)
# -> 最终结果: {final_decision: "approved", log: [...]}
# 3. 查看完整执行历史
for state in graph.get_state_history(config):
print(f"Step {state.metadata['step']}: next={state.next}")11.12 常见陷阱与最佳实践
11.12.1 状态 Schema 不匹配
最常见的子图问题是父图和子图的 Schema 不匹配导致数据丢失。当父图的状态包含子图不需要的字段时，这些字段在子图执行期间保持不变，子图完成后它们的值被保留在父图中。但如果子图的输出中包含父图不认识的字段，这些字段会被静默丢弃。
一个常见的误解是：如果父图和子图都定义了同名的 Channel 但使用了不同的 reducer，数据会被"转换"。实际上不会——状态映射只基于字段名匹配，不关心 reducer 的差异。如果父图的 messages 使用 operator.add（追加），子图的 messages 使用 LastValue（覆盖），子图的最终 messages 值会被传回父图，然后通过父图的 operator.add reducer 追加到已有的消息列表中。这种行为可能不是你期望的——你可能期望子图的消息替换父图的消息。解决方案是在子图的输出 Schema 中使用不同的字段名，然后在父图的后续节点中进行显式的数据整合。
11.12.2 子图的性能考量
每次调用子图都会创建一个新的 PregelLoop 实例，涉及到配置解析、命名空间构建、Channel 初始化和检查点加载等开销。对于频繁调用的轻量级子图，这些固定开销可能比子图本身的业务逻辑更大。在这种情况下，考虑将子图的逻辑内联为普通节点函数可能更高效。
子图的检查点写入也是一个性能考量点。在默认的 async 持久化模式下，每个子图的每个步骤都会触发一次检查点写入。对于深度嵌套的图（三层或更多），检查点写入的频率可能非常高。如果子图的执行是确定性的且不需要中断恢复，可以通过 checkpointer=False 禁用子图的检查点来提升性能。
11.12.3 调试嵌套图的技巧
调试嵌套图的困难在于错误消息和日志可能不包含足够的上下文信息。以下几个技巧可以帮助诊断问题：
首先，使用 stream_mode="debug" 并传入 subgraphs=True，可以获得最详细的执行日志，包括每个子图的每个步骤的输入、输出和中间状态（不传 subgraphs=True 时看不到子图内部的事件，见 §11.8.2）。其次，get_state(config, subgraphs=True) 在发生中断或错误后非常有用——它可以显示整个嵌套图的完整状态树，帮助定位问题发生在哪一层。最后，在开发阶段可以给每个子图添加 interrupt_before="*" 配置，实现单步执行来逐步检查行为。
11.12.4 实测：§11.3-11.5 涉及的常量与异常类的真实位置
§11.3 命名空间 / §11.4 Checkpoint / §11.5 ParentCommand 三节涉及多个常量与异常类——把它们在源码里精确定位——
libs/langgraph/langgraph/_internal/_constants.py 116 行 + 36 个 sys.intern 常量——
# 实测 _constants.py:75,77
NS_SEP = sys.intern("|") # ← §11.3 命名空间分隔符
NS_END = sys.intern(":") # ← §11.3 节点名与任务 ID 的分隔符libs/langgraph/langgraph/errors.py 127 行 + 8 个异常类 + ErrorCode 枚举——
| 类 | 起始行 | 角色 |
|---|---|---|
ErrorCode |
29 | Enum 错误码 |
GraphRecursionError(RecursionError) |
45 | recursion_limit 超限 |
InvalidUpdateError(Exception) |
68 | 状态更新非法 |
GraphBubbleUp(Exception) |
80 | §11.5 异常冒泡基类 |
GraphInterrupt(GraphBubbleUp) |
84 | 第 9 章的主角——interrupt() 的暂停/恢复靠它冒泡 |
NodeInterrupt(GraphInterrupt) |
96 | 节点级 interrupt（已弃用，源码提示改用 interrupt()） |
ParentCommand(GraphBubbleUp) |
111 | §11.5 子图→父图通信主角 |
EmptyInputError |
118 | 空输入 |
TaskNotFound |
124 | task 找不到 |
（__all__ 里还有第 9 个异常名 EmptyChannelError，但它不是本文件定义的——第 9 行从 langgraph.checkpoint.base import 进来再 re-export，真身在 checkpoint 包里，见第 4 章 §4.2。）
两条值得记住的物理事实——
ParentCommand和GraphInterrupt都继承自GraphBubbleUp——异常冒泡基础设施是共用的——印证 §11.10.2 的 "与中断的统一性"——子图→父图的控制流复用 interrupt 的异常路径——一套机制承担两个语义场景。第 9 章 §9.10.2 讨论的"恢复时重新执行整个节点"在子图层面同样成立：节点里在interrupt()之前写下的副作用，在 resume 时会再跑一遍_constants.py116 行 / 36 个sys.intern常量——其中 19 个CONFIG_KEY_*（SEND / READ / CALL / CHECKPOINTER / STREAM / CACHE / RESUMING / REPLAY_STATE / TASK_ID / THREAD_ID / CHECKPOINT_MAP / CHECKPOINT_ID / CHECKPOINT_NS / NODE_FINISHED / SCRATCHPAD / RUNNER_SUBMIT / DURABILITY / RUNTIME / RESUME_MAP），其余是 INPUT / INTERRUPT / RESUME / ERROR / TASKS / PREVIOUS / PUSH / PULL / NS_SEP / NS_END / CONF / NULL_TASK_ID 等——跨子系统共享的字符串集中在这一个文件里；另一部分协议字面量（如SCHEDULED = "__scheduled__"）则住在langgraph-checkpoint包的langgraph/checkpoint/serde/types.py，因为它们要被 checkpointer 实现共享
11.13 小结
本章深入剖析了 LangGraph 的子图与嵌套机制。我们看到，子图不是简单的"图中图"，而是一个涉及多个子系统协调的复杂工程：
- 图作为节点：编译后的图通过
Runnable协议无缝嵌入父图，PregelExecutableTask的subgraphs字段支持运行时的子图感知 - 命名空间隔离：
NS_SEP（|）和NS_END（:）构建的层级命名空间确保每个子图拥有完全独立的通道空间和检查点历史，即使父图和子图有同名通道也不会产生冲突 - Checkpoint 嵌套：
checkpoint_map在父子图之间传递检查点关联，使得恢复时可以正确定位子图的检查点。同一个 CheckpointSaver 通过checkpoint_ns实现物理隔离 - 跨图通信：
ParentCommand通过GraphBubbleUp异常机制实现子图到父图的控制流传递，与中断机制共享冒泡基础设施 - 状态映射：父图和子图之间的状态传递通过输入输出模式的字段名匹配自动完成，只有在两个模式中同时存在的同名字段会被传递，子图的内部状态对父图完全不可见
- 流式输出整合：
DuplexStream合并父图和子图的输出流，命名空间前缀帮助客户端精确区分每个事件的来源层级和节点
子图机制是构建复杂 Agent 系统的关键能力。通过合理的分层和组合，开发者可以将大型系统分解为可管理、可复用、可独立测试的子单元，同时保持完整的状态持久化和人机协作能力。