Day 49 / 共 60 天 · 阶段 8 函数式 API 与子图

子图基础:一张图,能当另一张图的节点

前面的图都是"一层的":节点是函数。今天我们让节点本身也是一张图——这就是子图(subgraph)。你会惊讶地发现:LangGraph 几乎没为此写特殊代码。因为编译后的图本来就是一个 Runnable,而节点接受任何 Runnable。今天看两件事:编译好的子图怎么被当节点接上,以及父子图如何"共享同名字段"的状态。

📍 阶段 8 · 函数式 API 与子图(6 天)你在这里
D47 @entrypoint/@task D48 func→pregel D49 子图基础 D50 子图隔离/stream D51 CachePolicy D52 重试容错
🤔 痛点:图越写越大,一个文件几十个节点 一个复杂 Agent 系统:检索模块 5 个节点、推理模块 4 个节点、审阅模块 3 个节点……全塞进一张 StateGraph,节点名冲突、无法复用、无法单独测试。你想像搭积木一样:把"检索模块"打包成一个可复用组件,在多个大图里当一个节点用。
💡 本质:子图 = 把一张编译好的图当节点嵌进另一张图 就像编程里"函数调用函数"。父图跑到某个节点,这个节点其实是一整张子图,子图跑完把结果交回父图继续。类比:公司的组织架构——总公司(父图)下面有一个"财务部"(子图),对总公司来说财务部是"一个节点",但点进去它内部又是一整套流程。关键问题只有一个:父图和子图怎么传递/共享状态?答案是"看字段名对不对得上"。
L01

两种嵌法:直接当节点 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 上。
L02

为什么图能直接当节点:它就是个 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 循环。
控制流:父图跑到 "sub" 节点 → 调子图 invoke → 结果回父图 父图节点 p1 改 foo 节点 "sub" = 子图 subgraph.invoke(state) 内部跑完整一轮 Pregel 写回父图 同名 foo 通道 END 对父图而言 "sub" 就是普通一步;点进去是一整张子图(独立命名空间存档)
图注:子图执行像"函数调函数"——调 invoke、跑完、结果按同名字段合并回父图。
💡 设计取舍①:为什么不专门设计一个"SubgraphNode"类型?朴素框架会为"嵌套"引入一等公民概念(如 add_subgraph)。LangGraph 选择复用 Runnable 抽象:既然节点吃 Runnable、编译图又是 Runnable,那"图当节点"就免费成立,零特殊代码。好处是极其统一——子图、普通函数、LCEL 链、甚至远程图(RemoteGraph)都能当节点,因为它们都是 Runnable。代价是"这个节点是不是子图"这件事变得隐式,需要靠运行时探测(L03)才能知道,用于流式和状态展开等特殊处理。这是"用统一抽象换特殊能力"的经典权衡。
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 就行"。探测是"想看清内部时才需要的地图"。
L04

状态共享的真相:按字段名对齐

为什么 L01 例子里父图改 foo、子图也改 foo,二者能接上?因为父图把整个 state 传给子图当输入,子图读它认识的字段、写回同名字段,父图再用同名通道的 reducer 合并。共享的前提就是字段名相同

数据结构:共享状态靠同名字段(foo)对齐 父图 State foo bar(父独有) 节点 "sub" = 子图 收到整个 State 子图 State foo 子图读 foo、改 foo 写回父图同名通道 同名 → 打通
图注:只有同名字段(foo)在父子图之间流动;父图独有的 bar 子图看不见也动不了。
父传整个 state父图执行子图节点时,把当前 state 作为输入调 subgraph.invoke(state)。子图按自己的 input schema 取用其中认识的键。
子图写回同名键子图返回的更新(如 {"foo": ...})被当作父图的节点写,走父图 foo 通道的 reducer 合并(Day 07 reducer)。
不同名 = 不共享父图的 bar 子图 schema 里没有,子图既读不到也改不了。想让子图用一个不同名字段,就得走 D50 的"包一层转换"。
⚠️ 边界:共享字段的 reducer 以父图为准如果 foo 在父图用 add 累加、在子图用覆盖写,最终写回父图时按父图 foo 通道的 reducer 处理。父子图对同名字段的 reducer 语义不一致时容易踩坑(比如你以为覆盖,父图却累加)。规避:共享字段在父子图保持一致的类型与 reducer 约定;不确定时用 D50 的显式转换隔离,别依赖隐式同名共享。
L05

子图有自己的命名空间(存档隔离)

状态字段可以共享,但存档(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。井水不犯河水。
💡 设计取舍②:字段共享 + 命名空间隔离,为什么要"半共享半隔离"?两种极端都不好:全共享(父子图连存档都混在一起)会让子图的中间状态污染父图的档、时间旅行时无法区分层级;全隔离(父子图状态完全不通)又回到 D50 要手写一堆转换的麻烦。LangGraph 选了中间路线——运行时的值按同名字段流动(协作方便),持久化的坐标按命名空间分层(互不干扰)。这样父子图既能自然传数据,又能各自独立地断点、恢复、时间旅行。代价是使用者要理解"字段共享但存档隔离"这个看似矛盾的模型,但它精确对应了"协作"与"隔离"两种不同需求的分离。
💡 本质:状态"字段"共享,但存档"空间"隔离这看起来矛盾,其实分工明确:字段共享让父子图能协作传数据(运行时的值流动);命名空间隔离让父子图的存档、断点、时间旅行互不干扰(持久化的坐标分层)。一个子图中断(interrupt)时,恢复能精确定位到"父图的哪个 task 下的子图的哪一步",正是靠这条 父ns|子ns 的嵌套路径。Day 41 讲的 interrupt 在子图里也能用,就是这个原因。
L06

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
明天预告 · Day 50:如果父子图连一个同名字段都没有怎么办?如何显式转换输入/输出让 schema 完全解耦?以及流式时怎么"下钻"看到子图内部的每一步(stream(subgraphs=True))?明天讲子图的状态隔离与转换。
← Day 48 func→pregel Day 50 · 子图状态隔离与转换 →