所谓子图，就是：

一个状态图
  ↓
在另一个状态图中被调用
  ↓
形成父图 + 子图结构

子图最大的价值，是把复杂工作流拆成多个可以独立开发、调试和复用的小图。

1. LangGraph 中有哪两种子图用法？

LangGraph 中常见的子图嵌入方式有两种：

方式	适用场景	父子图通信
节点函数中 subgraph.invoke()	父子图 State 完全隔离	手动映射输入输出
直接把编译后的子图作为节点	父子图共享状态字段	共享字段自动通信

第一种方式：

def call_subgraph(state):
    input_text = state["input_text"]

    result = subgraph.invoke(
        {
            "raw_text": input_text
        }
    )

    return {
        "cleaned_text":
            result["punctuated_text"]
    }

这里父图使用：

input_text
cleaned_text

子图使用：

raw_text
stripped_text
punctuated_text

两套 State 不一样，因此需要节点函数负责转换：

父图 State
    ↓
手动转换
    ↓
子图 State
    ↓
子图运行
    ↓
结果再次转换
    ↓
写回父图 State

这种方式更适合：

子图是一个独立模块
父图不需要知道子图内部状态结构

第二种方式，是直接把编译好的子图作为父图节点：

subgraph = builder.compile()

builder = StateGraph(
    state_schema=OverallState
)

builder.add_node(
    "subgraph_node",
    subgraph
)

这种方式一般要求父图和子图之间存在共享状态字段。

例如：

class OverallState(TypedDict):
    raw_text: str
    cleaned_text: str

父图把 raw_text 传给子图，子图执行后更新 cleaned_text，父图可以直接继续读取。

所以可以简单记：

State 不同
→ 节点函数里 invoke 子图

State 有共享字段
→ 直接把子图作为节点

另外，可以通过：

parent_graph.get_subgraphs()

查看父图中包含的子图。

如果要把子图内部拓扑也展开出来，则可以：

parent_graph
    .get_graph(xray=True)

2. 子图和 Checkpoint 是什么关系？

子图并不会脱离 LangGraph 的持久化体系。

当父图启用 Checkpointer 后，子图也可以产生自己的检查点状态。

例如父图：

checkpoint_ns = ""

而某个子图可能是：

call_subgraph:<task_id>

也就是说：

thread_id
负责区分整条运行线程

checkpoint_ns
负责区分父图和不同子图

checkpoint_id
负责区分具体检查点

所以在嵌套图中，之前学习的：

thread_id
checkpoint_ns
checkpoint_id

就真正发挥作用了。

父图的某个任务快照中，还可以记录当前子图对应的：

thread_id
checkpoint_ns

从而把父图任务和子图检查点关联起来。

如果继续嵌套：

父图
 └── 子图 A
       └── 子图 B

检查点命名空间还会继续向下组合，用于区分不同层级的执行状态。

3. 子图的三种持久化策略

在父图已经启用 Checkpointer 的前提下，子图可以有三种策略：

策略	子图编译参数	多轮记忆	中断恢复
Per-invocation	checkpointer=None	否	是
Per-thread	checkpointer=True	是	是
Stateless	checkpointer=False	否	否

Per-invocation

这是默认方式。

subgraph = builder.compile()

或：

subgraph = builder.compile(
    checkpointer=None
)

每次调用使用独立的子图状态。

适合：

子图只负责完成一次任务
不需要跨多次调用保存自身历史

Per-thread

subgraph = builder.compile(
    checkpointer=True
)

同一个 thread_id 下，子图状态可以在多次调用之间延续。

比较适合：

多轮对话子图
需要独立记忆的子 Agent

Stateless

subgraph = builder.compile(
    checkpointer=False
)

子图完全不保存检查点。

因此：

不能中断恢复
不能保存多轮历史

适合纯计算、无状态处理。

所以选择时可以简单看需求：

只执行一次
→ Per-invocation

需要跨调用连续状态
→ Per-thread

完全不需要持久化
→ Stateless

4. 如何流式观察子图？

上一篇我们已经学习：

graph.stream(...)

默认情况下主要观察父图。

如果想把子图内部的流数据也一起返回，只需要：

for chunk in parent_graph.stream(
    input_data,
    subgraphs=True,
    stream_mode=["updates"]
):
    print(chunk)

这里最关键的是：

subgraphs=True

它的含义不是：

只返回子图数据

而是：

父图流数据
+
子图流数据
一起返回

为了区分数据来自哪里，流式 chunk 会增加一个 namespace。

在 v1 格式下，可以理解为：

(
    namespace,
    stream_mode,
    data
)

例如子图：

(
    (
        "call_subgraph:<task_id>",
    ),
    "updates",
    {
        "subgraph_strip_node": {
            "stripped_text":
                "LangGraph 真有意思"
        }
    }
)

父图的 namespace 则是：

()

所以：

namespace = ()
→ 父图

namespace 有值
→ 某个子图

需要注意，这里的流式 namespace 和检查点里的 checkpoint_ns 用途不同。

前者主要用于：

区分流数据来源

后者主要用于：

区分父图和子图的持久化检查点