子图基础:一张图,能当另一张图的节点
前面的图都是"一层的":节点是函数。今天我们让节点本身也是一张图——这就是子图(subgraph)。你会惊讶地发现:LangGraph 几乎没为此写特殊代码。因为编译后的图本来就是一个 Runnable,而节点接受任何 Runnable。今天看两件事:编译好的子图怎么被当节点接上,以及父子图如何"共享同名字段"的状态。
两种嵌法:直接当节点 vs 包一层函数
子图有两种接法,今天讲第一种(共享状态),明天讲第二种(隔离转换):
| 方式 | 写法 | 状态关系 |
|---|---|---|
| 直接当节点 | parent.add_node("sub", subgraph) | 共享:父子图同名字段自动互通(今天) |
| 包一层函数 | parent.add_node("sub", lambda s: subgraph.invoke({...})) | 隔离:手动转换输入/输出,schema 可完全不同(D50) |
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
foo: str # 父子图都有 foo → 共享
# 子图
sub = StateGraph(State)
sub.add_node("s1", lambda s: {"foo": s["foo"] + "|sub"})
sub.add_edge(START, "s1"); sub.add_edge("s1", END)
subgraph = sub.compile()
# 父图:把编译好的子图直接当节点
parent = StateGraph(State)
parent.add_node("p1", lambda s: {"foo": s["foo"] + "|p1"})
parent.add_node("sub", subgraph) # ← 子图当节点!
parent.add_edge(START, "p1"); parent.add_edge("p1", "sub"); parent.add_edge("sub", END)
app = parent.compile()
app.invoke({"foo": "start"}) # {'foo': 'start|p1|sub'}
注意 add_node("sub", subgraph) 第二个参数是编译好的图对象,不是函数。父图把整个 State 传给子图,子图对 foo 的修改直接体现在父图的 foo 上。为什么图能直接当节点:它就是个 Runnable
秘密在类型继承链。编译后的图是 Pregel,而 Pregel 实现了 PregelProtocol,后者直接继承 Runnable pregel/protocol.py:25:
# pregel/protocol.py:25
class PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, OutputT]):
...
# pregel/main.py:450
class Pregel(PregelProtocol[StateT, ContextT, InputT, OutputT], ...):
...
而 add_node 对 action 的处理,就是一句 coerce_to_runnable graph/state.py:874:
# graph/state.py:872
self.nodes[node] = StateNodeSpec(
coerce_to_runnable(action, name=node, trace=False), # 把 action 归一成 Runnable
metadata, input_schema=..., retry_policy=..., cache_policy=..., ...
)
Pregel is-a Runnable因为编译好的子图本身就是 Runnable,coerce_to_runnable 见到它原样返回(无需包装)。所以子图能像普通函数节点一样被存进 nodes。子图有 .invoke父图跑到这个节点时,就调子图的 .invoke(state, config)——和调一个普通函数节点没区别。子图内部自己跑完整一轮 Pregel 循环。add_subgraph)。LangGraph 选择复用 Runnable 抽象:既然节点吃 Runnable、编译图又是 Runnable,那"图当节点"就免费成立,零特殊代码。好处是极其统一——子图、普通函数、LCEL 链、甚至远程图(RemoteGraph)都能当节点,因为它们都是 Runnable。代价是"这个节点是不是子图"这件事变得隐式,需要靠运行时探测(L03)才能知道,用于流式和状态展开等特殊处理。这是"用统一抽象换特殊能力"的经典权衡。引擎怎么知道"这个节点是子图":自动探测
虽然子图当节点靠的是 Runnable 抽象,但引擎有时需要知道"这里面藏着一张图"(比如流式要下钻、get_state 要展开子图状态)。PregelNode 构造时会自动探测 pregel/_read.py:182:
# pregel/_read.py:182
if subgraphs is not None:
self.subgraphs = subgraphs
elif self.bound is not DEFAULT_BOUND:
try:
subgraph = find_subgraph_pregel(self.bound) # 从 bound 里挖有没有 Pregel
except Exception:
subgraph = None
if subgraph:
self.subgraphs = [subgraph]
else:
self.subgraphs = []
else:
self.subgraphs = []
find_subgraph_pregel 会顺着 Runnable 的结构往里挖 pregel/_utils.py:47:
# pregel/_utils.py:47
def find_subgraph_pregel(candidate: Runnable) -> PregelProtocol | None:
from langgraph.pregel import Pregel
candidates: list[Runnable] = [candidate]
for c in candidates:
if (isinstance(c, PregelProtocol)
and (not isinstance(c, Pregel) or c.checkpointer is not False)):
return c # 找到一张图!
elif isinstance(c, RunnableSequence) or isinstance(c, RunnableSeq):
candidates.extend(c.steps) # 是链 → 挖每一步
elif isinstance(c, RunnableLambda):
candidates.extend(c.deps) # 是 lambda → 挖它依赖的对象
elif isinstance(c, RunnableCallable):
if c.func is not None:
candidates.extend( # 是函数 → 挖闭包里引用的对象
nl.__self__ if hasattr(nl, "__self__") else nl
for nl in get_function_nonlocals(c.func))
...
return None
直接是 Pregel → 返回L01 第一种写法:子图直接当节点,bound 就是 Pregel,一眼探测到。是链/lambda → 继续挖就算你把子图包在函数里(L01 第二种写法),它也会顺着闭包的 get_function_nonlocals 挖出被引用的子图。所以"包一层"也能被识别为含子图。checkpointer is not False细节:显式关掉存档(checkpointer=False)的子图不算——因为它不参与父图的持久化协调(D50 讲 ns)。node.subgraphs 这个列表被 get_subgraphs()(L06)、流式下钻(D50)、时间旅行展开子图状态用到。平时执行子图并不需要它——执行只靠"它是 Runnable,调 invoke 就行"。探测是"想看清内部时才需要的地图"。状态共享的真相:按字段名对齐
为什么 L01 例子里父图改 foo、子图也改 foo,二者能接上?因为父图把整个 state 传给子图当输入,子图读它认识的字段、写回同名字段,父图再用同名通道的 reducer 合并。共享的前提就是字段名相同。
父传整个 state父图执行子图节点时,把当前 state 作为输入调 subgraph.invoke(state)。子图按自己的 input schema 取用其中认识的键。子图写回同名键子图返回的更新(如 {"foo": ...})被当作父图的节点写,走父图 foo 通道的 reducer 合并(Day 07 reducer)。不同名 = 不共享父图的 bar 子图 schema 里没有,子图既读不到也改不了。想让子图用一个不同名字段,就得走 D50 的"包一层转换"。foo 在父图用 add 累加、在子图用覆盖写,最终写回父图时按父图 foo 通道的 reducer 处理。父子图对同名字段的 reducer 语义不一致时容易踩坑(比如你以为覆盖,父图却累加)。规避:共享字段在父子图保持一致的类型与 reducer 约定;不确定时用 D50 的显式转换隔离,别依赖隐式同名共享。子图有自己的命名空间(存档隔离)
状态字段可以共享,但存档(checkpoint)是分层的。父图执行子图节点时,给子图分配一个嵌套的 checkpoint 命名空间 pregel/_algo.py:624:
# pregel/_algo.py:624
task_checkpoint_ns = f"{checkpoint_ns}{NS_END}{task_id}"
# ...
# 传给子图执行的 config 里(同函数体 :746)
CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,
task_checkpoint_ns = 父ns + task_id子图这次执行拿到一个独立的命名空间字符串,形如 sub:<task_id>。它挂在父图的 ns 下面。CONFIG_KEY_CHECKPOINT_NS 传给子图子图内部所有存档都写在这个 ns 下。回想 Day 35 InMemorySaver:存储键第一层是 thread、第二层就是 checkpoint_ns。父图存在 ""(根 ns),子图存在 sub:xxx。井水不犯河水。父ns|子ns 的嵌套路径。Day 41 讲的 interrupt 在子图里也能用,就是这个原因。get_subgraphs:把嵌套结构列出来 + 小结
L03 探测出的 node.subgraphs 在这里派上用场——get_subgraphs 遍历所有节点,把子图(可递归)列出来 pregel/main.py:1076:
# pregel/main.py:1089
for name, node in self.nodes.items():
...
graph = node.subgraphs[0] if node.subgraphs else None # 用探测结果
if graph:
if namespace is None:
yield name, graph # 产出 (节点名, 子图)
if recurse and isinstance(graph, Pregel): # recurse=True 继续下钻
yield from ((f"{name}{NS_SEP}{n}", s)
for n, s in graph.get_subgraphs(namespace=namespace, recurse=recurse))
遍历 nodes 找 subgraphs哪些节点其实是子图,一目了然——这就是 L03 探测的价值兑现。recurse 递归下钻子图里还有子图?recurse=True 会一路挖到底,命名空间用 | 拼成完整路径,如 outer|inner。👶 子图必须和父图用同一个 State schema 吗?
👨🏫 不必完全相同,但想共享的字段名必须一致。父图 State 有 {a, b, c},子图 State 有 {b, d},那么只有 b 共享:父图把 b 传进去、子图改了 b 传回来;子图的 d 是它的私有字段(父图看不见);父图的 a、c 子图碰不到。完全不同 schema(一个字段都不重名)就没法直接当节点共享,得走 D50 的转换方式。
🧠 今天你应该能回答
- 为什么编译好的图能直接当节点?(Pregel 实现 PregelProtocol,后者继承 Runnable,而节点吃 Runnable)
- add_node 怎么处理子图?(coerce_to_runnable 见到 Runnable 原样返回,存进 nodes)
- 引擎怎么知道一个节点是子图?(PregelNode 构造时 find_subgraph_pregel 探测,存进 node.subgraphs)
- 父子图状态怎么共享?(按同名字段对齐:父传整个 state,子图读写同名键,用父图通道 reducer 合并)
- 父图独有字段子图能改吗?(不能,schema 里没有就看不见)
- 子图的存档在哪?(独立的嵌套命名空间
父ns|sub:task_id,与父图隔离)
✋ 10 分钟动手
# 1. 读三段
sed -n '47,75p' libs/langgraph/langgraph/pregel/_utils.py # find_subgraph_pregel
sed -n '150,195p' libs/langgraph/langgraph/pregel/_read.py # PregelNode 探测
sed -n '1076,1114p' libs/langgraph/langgraph/pregel/main.py # get_subgraphs
# 2. 亲手验证共享 + 列子图
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict):
foo: str
sub = StateGraph(S); sub.add_node("s1", lambda s: {"foo": s["foo"]+"|sub"})
sub.add_edge(START,"s1"); sub.add_edge("s1",END)
subgraph = sub.compile()
p = StateGraph(S)
p.add_node("p1", lambda s: {"foo": s["foo"]+"|p1"}); p.add_node("sub", subgraph)
p.add_edge(START,"p1"); p.add_edge("p1","sub"); p.add_edge("sub",END)
app = p.compile()
print(app.invoke({"foo":"start"})) # {'foo': 'start|p1|sub'}
print([name for name,_ in app.get_subgraphs()]) # ['sub']
PY
stream(subgraphs=True))?明天讲子图的状态隔离与转换。