子图简介

子图（Subgraph）是 LangGraph 里的核心模块化特性：

它允许你把一个完整的工作流图，当作另一个图里的普通节点来使用。
本质是一种「图中套图」的设计，让复杂的 AI 工作流可以像搭积木一样被组合起来，从而提升系统的灵活性和可维护性。

子图的主要优势

构建多智能体系统

每个智能体（比如一个专门负责检索、一个专门负责写文案）都可以被独立设计成一个子图，再组装到主流程里，结构更清晰。

代码复用

通用的工作流（比如「文档解析」「工具调用」）可以做成子图，在多个不同的主流程里重复调用，避免重复写代码。

分布式开发

不同团队可以并行开发不同的子图模块，最后再统一接入主图，大幅提升团队协作效率。

系统解耦

只要子图对外的输入输出接口保持一致，主图（父图）完全不需要关心子图内部是怎么实现的。子图可以独立迭代、优化，不会影响上层逻辑。

模块化价值

子图设计的本质：

将复杂系统分解为可管理的模块，提高代码复用性和团队协作效率。

简单来说，子图就是 LangGraph 里实现「高内聚、低耦合」的关键手段，让大型 AI 工作流的开发、维护和扩展都变得更轻松。

💡 举个通俗的例子：

就像你做一顿大餐，主流程是「备菜→烹饪→装盘」，而「备菜」本身可以是一个子图（里面包含洗菜、切菜、腌制等步骤），「烹饪」也可以是一个子图。你可以随时替换「备菜」的方式，只要它最终输出处理好的食材就行，完全不用改主流程。

LangGraph 中两种实现子图（Subgraph）的核心方式

对比维度	方式一：从节点调用子图	方式二：将子图直接添加为节点
父子图状态	状态结构完全独立（不同的 `TypedDict`）	共享同一套状态结构（同一个 `TypedDict`）
状态转换	需要手动写转换逻辑（父状态 → 子状态 → 父状态）	无需额外转换，直接共享状态
代码复杂度	更高，需要额外封装调用节点	更简洁，直接把子图作为节点添加
适用场景	父子图逻辑差异大、状态结构完全不同	父子图结构相似、共享状态键的场景

从节点调用子图（手动状态转换）

当父图和子图的状态结构完全不同时使用。比如父图的状态是 {"foo": str}，子图的状态是 {"bar": str}，两者没有任何共享字段

# 子图状态（独立定义）
class SubgraphState(TypedDict):
    bar: str

def subgraph_node_1(state: SubgraphState) -> SubgraphState:
    return {"bar": "hi! " + state["bar"]}

# 构建子图
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()

# 父图状态（和子图完全不同）
class State(TypedDict):
    foo: str

# 关键：封装一个调用节点，手动处理状态转换
def call_subgraph(state: State) -> State:
    # 1. 父图状态 → 子图状态
    subgraph_output = subgraph.invoke({"bar": state["foo"]})
    # 2. 子图状态 → 父图状态
    return {"foo": subgraph_output["bar"]}

# 构建父图，把调用节点作为普通节点加入
builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()

子图和父图各有一套独立的 State，完全解耦。
必须写一个中间节点 call_subgraph，负责把父图的状态转换成子图能接收的格式，调用子图后再把结果转换回父图状态。
优点是子图可以完全独立开发，和父图的状态无关；缺点是多了一层封装和转换逻辑。

将子图直接添加为节点（共享状态）

当父图和子图共享同一套状态结构时使用。比如两者都用同一个 TypedDict 定义状态，子图只修改其中的部分字段。

# 父子图共享同一个状态定义
class State(TypedDict):
    foo: str

def subgraph_node_1(state: State) -> State:
    return {"foo": "hi! " + state["foo"]}

# 构建子图
subgraph_builder = StateGraph(State)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()

# 构建父图，直接把子图作为节点添加进去
builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # 关键：直接传入编译好的子图
builder.add_edge(START, "node_1")
graph = builder.compile()

父子图使用同一个 State，状态是共享的，子图的修改会直接体现在父图的状态里。
无需手动转换状态，直接把编译好的子图 subgraph 当作普通节点 add_node 即可。
优点是代码非常简洁，状态流转自然；缺点是子图和父图的状态耦合度较高，修改状态结构时需要同步考虑父子图。

怎么选？给你一个简单判断标准

✅ 选方式一：
- 子图是通用工具，可能被多个不同状态的父图调用；
- 父子图的业务逻辑差异很大，状态字段完全不同；
- 希望子图完全独立，不依赖父图的状态结构。
✅ 选方式二：
- 子图是父图的一个内部流程，和父图共享大部分状态字段；
- 追求代码简洁，不想写额外的状态转换逻辑；
- 父子图由同一个团队开发，状态结构变更可控。

子图持久化（Checkpoint）

LangGraph 为子图提供了自动持久化支持：

你只需要在编译父图的时候，给它配置好检查点存储器（比如 MemorySaver），那么所有被父图直接添加为节点的子图，都会自动继承这个持久化配置，无需额外操作。

这意味着：

父图和子图默认共享同一个检查点、同一个对话历史和状态快照。
子图的执行过程、状态变化，都会被父图的 checkpointer 完整记录下来。

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing_extensions import TypedDict


class State(TypedDict):
    foo: str


# 子图定义
def subgraph_node_1(state: State) -> State:
    return {"foo": state["foo"] + "bar"}


subgraph_builder = StateGraph(State)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()


# 父图配置持久化
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)


# 如果希望子图拥有独立的记忆空间
subgraph = subgraph_builder.compile(checkpointer=True)

代码解读：

基础导入与状态定义
1
2
3
4
5
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver

class State(TypedDict):
foo: str
- MemorySaver 是 LangGraph 提供的内存型检查点存储，用来保存对话 / 状态快照，实现持久化、断点续跑、回滚等功能。
- State 是父子图共享的状态结构，这里只有一个 foo 字段。

定义子图

# 子图节点逻辑
def subgraph_node_1(state: State) -> State:
    return {"foo": state["foo"] + "bar"}

# 构建子图
subgraph_builder = StateGraph(State)
subgraph_builder.add_node("subgraph_node_1", subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()

子图本身没有配置 checkpointer，但它会被父图继承持久化配置。
这个子图的作用是把状态里的 foo 字段拼接上 "bar"。

构建父图并配置持久化（关键部分）

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # 直接把子图作为节点加入父图
builder.add_edge(START, "node_1")

# 配置检查点存储器
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

这里是核心逻辑：
1. 父图通过 builder.add_node("node_1", subgraph) 把子图添加为自己的节点。
2. 父图编译时，传入了 checkpointer=checkpointer。
3. 子图会自动继承父图的 checkpointer，不需要额外配置。
效果：父图和子图的状态变化都会被 MemorySaver 记录下来，支持对话记忆、中断恢复等功能。

进阶：让子图拥有独立的记忆空间
1
2
# 如果希望子图拥有独立的记忆空间
subgraph = subgraph_builder.compile(checkpointer=True)
- 这里的 checkpointer=True 是一个快捷方式，它会为子图创建一个独立的、默认的检查点存储器。
- 效果：
  - 子图有自己独立的记忆空间，不会和父图的检查点混在一起。
  - 父图的检查点不会记录子图内部的状态细节，只会记录子图节点的整体输入 / 输出。
  - 适用于子图需要独立会话历史、或者不想和父图共享记忆的场景。

两种持久化模式对比

模式	配置方式	记忆空间	适用场景
继承父图持久化	子图不配置 `checkpointer`，父图配置后子图自动继承	父子图共享同一个检查点	父子图流程紧密耦合，需要完整记录子图内部状态（比如调试、审计）
子图独立持久化	子图编译时传入 `checkpointer=True`	子图有自己独立的检查点	子图是独立的服务 / 工具，不需要和父图共享会话历史

💡 补充理解：

当你把子图当作节点加入父图时，LangGraph 会把它当成一个 “黑盒” 节点，但持久化机制会自动处理好状态的传递和记录。
共享持久化：就像一个函数调用，函数内部的变量变化也会被记录到同一个日志里。
独立持久化：就像调用一个外部 API，API 自己有日志，父图只记录请求和响应。

状态查看与调试

状态查看

核心作用

当子图执行被中断时，查看子图内部的状态，方便定位问题。

# 获取父图状态
parent_state = graph.get_state(config)

# 获取子图状态（仅在子图被中断时可用）
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state

graph.get_state(config)：默认获取父图的当前状态，看不到子图内部。
subgraphs=True：开启子图状态查看模式，此时返回的状态对象里会包含当前正在执行的子图任务。
.tasks[0].state：取第一个正在执行的任务（也就是子图节点）的内部状态。

关键注意事项

子图状态只能在子图被中断时查看（比如设置了中断节点、或者手动暂停执行）。
一旦恢复执行，子图的内部状态就会被 “压栈”，无法再直接访问，只能看到父图状态。

流式输出调试

核心作用

实时监控子图的执行过程，同时看到父图和子图的状态更新，适合调试复杂的嵌套工作流。

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True,  # 启用子图输出
    stream_mode="updates",
):
    print(chunk)

subgraphs=True：让 stream() 同时输出父图和所有子图的执行更新。
stream_mode="updates"：只输出状态的更新部分，而不是完整状态，让日志更简洁。
效果：执行时，会按顺序打印父图进入子图、子图内部节点执行、子图退出、父图继续执行的每一步更新。

方式	使用场景	优势	限制
`get_state(subgraphs=True)`	中断时查看快照	可以拿到子图当前的完整状态，适合事后排查	只能在中断状态下使用，恢复执行后无法访问
`stream(subgraphs=True)`	实时监控执行过程	能看到完整的执行流，父图 / 子图的步骤都能看到	是执行时的日志输出，不能回溯历史状态

LangGraph 子图的两大核心实际应用场景

多智能体系统和模块化工作流

多智能体系统

核心思路

把每个智能体封装成独立的子图，再由一个主图（路由智能体）来调度分发任务。

# 1. 两个独立的智能体子图
# 客服智能体子图
customer_service_agent = build_customer_service_agent().compile()
# 技术支持智能体子图
tech_support_agent = build_tech_support_agent().compile()

# 2. 路由智能体：根据任务类型分配给对应子图
def router_agent(state: State) -> State:
    if "billing" in state["query"]:
        return {"next_agent": "customer_service"}
    else:
        return {"next_agent": "tech_support"}

# 3. 主协调图：把路由和两个智能体子图组合起来
main_graph = StateGraph(State)
main_graph.add_node("router", router_agent)
main_graph.add_node("customer_service", customer_service_agent)
main_graph.add_node("tech_support", tech_support_agent)

优势

解耦与独立开发：客服和技术支持两个智能体可以由不同团队独立开发、测试、迭代，互不影响。
灵活调度：路由节点可以根据用户的问题类型，动态选择调用哪个子图，实现 “分工协作”。
可扩展性强：后续要新增 “账单智能体”“物流智能体”，只需要新增一个子图和路由规则，不用修改主流程。

模块化工作流

核心思路

把通用的、可复用的处理流程封装成子图模块，再像搭积木一样组合成完整的业务流程。

# 1. 封装可复用的子图模块
# 文本预处理子图
text_preprocessor = build_text_preprocessor().compile()
# 情感分析子图
sentiment_analyzer = build_sentiment_analyzer().compile()
# 报告生成子图
report_generator = build_report_generator().compile()

# 2. 组合成完整的分析流水线
analysis_pipeline = StateGraph(AnalysisState)
analysis_pipeline.add_node("preprocess", text_preprocessor)
analysis_pipeline.add_node("analyze", sentiment_analyzer)
analysis_pipeline.add_node("generate", report_generator)

优势

高代码复用：文本预处理、情感分析这些通用模块，可以在多个不同的业务流程里重复使用。
流程可替换：比如后续想把情感分析换成更精准的模型，只需要替换 sentiment_analyzer 这个子图，不用改动整个流水线。
清晰易维护：复杂的业务流程被拆分成了多个单一职责的模块，调试和维护都更简单。

两种场景的核心共性

这两个场景本质上都在利用子图的模块化、解耦、可复用特性：

单一职责原则：每个子图只做一件事，逻辑更聚焦。
高内聚低耦合：子图内部逻辑对外透明，主图只关心输入输出。
可组合可扩展：可以通过组合子图，快速搭建复杂的 AI 系统。

核心价值总结

LangGraph 的子图功能为构建复杂 AI 系统提供了强大的模块化能力。通过合理使用子图，你可以：

这句话点明了子图的本质：用模块化的方式，解决复杂 AI 系统的构建问题。

优势	说明	通俗理解
提高代码复用性	通用模块一次开发，多处使用	比如文本预处理、情感分析这些通用流程，做成子图后，多个业务流程都能直接调用，不用重复写代码。
简化系统架构	复杂系统分解为可管理的模块	把一个巨大的工作流拆成多个小模块（子图），每个模块只做一件事，结构更清晰，一眼就能看懂流程。
支持团队协作	不同团队可以并行开发不同模块	客服智能体、技术支持智能体可以由不同团队独立开发、测试，最后再像搭积木一样拼到主流程里，互不干扰。
便于调试维护	模块边界清晰，问题定位更容易	出问题时，你可以单独调试某一个子图，不用在几百行的大流程里找 bug；更新功能时，只改对应子图就行，不会影响其他模块。

无论是简单的数据处理流水线还是复杂的多智能体系统，子图都能帮助你构建更加健壮和可维护的 AI 应用。

这句话点出了子图的适用范围：从简单的脚本级任务，到企业级的多智能体系统，都能通过子图的模块化设计，变得更稳定、更好维护。

子图就是 LangGraph 里的「积木」。它让你不用从零开始搭建复杂系统，而是通过复用、组合、分工，像搭乐高一样，高效地构建出可扩展、好维护的 AI 应用。

Jean's Blog

LangGraph之父子图（多工作流）