第 16 章子图（Subgraph）与模块化

当 Agent 系统规模增长到十数节点以上、或多个团队协作开发同一应用时，单张图会出现状态字段膨胀（State Schema 动辄二十字段以上）、节点名冲突（多人都想给节点起名 agent/tool_node）、上下文难隔离（一个 Agent 的中间思考污染其他 Agent 的 messages）、无法跨项目复用（写好的研究流程无法直接被另一个产品复用）等问题。子图（Subgraph）是 LangGraph 的模块化原语：一个已经 compile 的 CompiledStateGraph 可以作为节点加入另一张图，像普通节点一样被调度、被边连接、被条件路由选中、被 Send API 派发。子图拥有独立的 State Schema、节点集合、边配置、检查点命名空间与 config_schema，对外只暴露父图关心的输入输出字段，内部实现完全封装。该机制支持复杂系统按领域拆分、团队并行开发、跨项目复用，并天然作为多智能体系统中单个 Agent 的封装单元。

16.1 子图的添加与 State 映射

添加子图通过 builder.add_node("subgraph_name", compiled_subgraph) 完成，传入的必须是已经 compile 完成的 CompiledStateGraph 实例；传入未编译的 StateGraph 会抛 TypeError: Expected a compiled graph, got <StateGraph>。子图在父图中就是一个普通节点：可以被 add_edge 指向、被条件路由返回、被 Send API 派发、可以设置 retry/timeout 等节点级参数，可以有自己的 debug 开关、cache 配置（通常由父图统一传入）。compile 子图时建议传入 name="xxx" 参数，该 name 会作为子图的命名空间前缀出现在 checkpoint_ns、流式 ns、LangSmith trace 名中，便于调试与观测。

State 映射是子图集成的核心契约。父图调用子图节点时，运行时自动完成两件事：(1) 输入映射：父图 State 中与子图 State Schema 同名字段被自动提取出来作为子图的 invoke 输入；未在子图 Schema 中声明的父图字段不会传递给子图（子图读不到）。(2) 输出映射：子图执行到 END 后，其输出 State 中与父图 State Schema 同名字段会被合并回父图，合并规则遵循父图对应字段的 reducer（LastValue 覆盖、list 追加、add_messages 合并等）；子图私有字段（父图 Schema 中没有的字段）不会出现在父图 State 中。这意味着"共享字段名"即是子图与父图最直接、最低成本的契约。

当字段名不一致、或需要做结构转换（父图用 user_query、子图用 topic；父图用完整 messages、子图只要最新一条）时，必须显式提供转换器，有三种等价写法：

包装节点函数（推荐，最灵活）：写一个普通 Python 节点，函数内手动构造子图输入 dict，调用 result = await subgraph.ainvoke(input_dict)，再把 result 映射为父图增量 dict 返回。这种方式能做任意复杂转换、错误处理、日志记录。
input_transform / output_transform（LangGraph 1.2.x 在 add_node 层提供的轻量转换钩子）：通过 add_node(name, sub, input_transform=fn_in, output_transform=fn_out) 传入两个函数，分别负责父→子、子→父的 dict 转换。
子图自定义 input_schema/output_schema：在子图侧用 Pydantic BaseModel 声明严格的输入输出结构，父图必须提供匹配字段，不匹配在调用时校验失败并抛 ValidationError，适合子图作为独立产品发布时强制接口契约。

按字段可见性，父子图 State 交互可归纳为三种模式：

共享字段模式：父图与子图 State Schema 中保留一组同名字段（如 messages、topic、notes），这些字段自动双向传递；简单直接，适合父子图语义高度一致的场景（如研究子图与父图共用 topic 字段）。
输入/输出转换器模式：父图与子图各自用自己的术语表（父图叫 user_query、子图叫 topic），通过包装函数或 transform 显式做键名映射与结构裁剪；父图不污染子图命名空间，子图可独立复用。
私有 State 模式：子图 State Schema 中独有的字段（如子图内部的 planner_output、retrieval_docs、scratchpad）不会暴露给父图，父图也读不到；这些字段只在子图节点间流转并在子图 checkpoint 命名空间内持久化，是模块封装与信息隔离的关键。多智能体场景下每个 Agent 子图私有的 messages、plan、tool_results 字段都应是私有 State。

16.2 子图与 Checkpoint、流式、中断、跨图通信

Checkpoint 命名空间。父图配置 checkpointer 后，子图的 checkpoint 自动在父 thread_id 之下通过 checkpoint_ns（checkpoint namespace，检查点命名空间）隔离。命名空间路径按层级为 子节点名（单层子图）或 父::子::孙（多层嵌套），get_state_history 返回的历史条目带有 ns 字段，按命名空间还原完整调用栈。子图不需要单独配置 checkpointer——如果子图自己 compile 时也传了 checkpointer，运行时会使用父图的 checkpointer 实例覆盖，避免父子各存一份造成数据不一致、resume 时找不到断点的问题。子图独立测试时可以单独传 MemorySaver，但在父图中集成时应让父图统一配置。

流式透传。调用父图 graph.stream(..., subgraphs=True) 时，子图内部所有 stream_mode 事件（messages token、custom 进度、tasks 生命周期等）透传到父图迭代器，事件的 ns 字段标明来源路径（如 research::retrieve 表示 research 子图中的 retrieve 节点）。未开启 subgraphs 时（默认 False），子图作为黑盒只在其入口（节点被调度）与出口（节点返回）产出一次父图层 events 事件，内部 token 流与进度事件对子图外不可见。前端可按 ns.split("::") 渲染层级进度条（"研究子图 → 检索节点 → 60%"）。

中断传播。子图内部节点调用 interrupt() 或子图编译时设置了 interrupt_before/after，中断会冒泡到父图调用方：父图的 invoke/stream 调用在子图中断点处返回中断事件，调用方通过 get_state(config) 读取 interrupt payload（此时返回的 state snapshot 的 ns 字段指向子图命名空间，tasks 列表中的任务路径包含子图节点）。resume 时必须在同一 config（同一 thread_id）下调用 Command(resume=value)，运行时根据断点快照中的 ns 自动定位恢复到子图内断点，调用方无需手动计算路径；若需要在 resume 同时跳转到父图其他节点（如"审批拒绝后直接结束对话，不走子图后续"），可用 Command(resume=value, goto="other_node", graph=Command.PARENT) 显式指定跨图跳转。

子图向父图通信。子图节点可以返回 Command(graph=Command.PARENT, update={...}) 直接更新父图 State（而不仅是子图 State），或 Command(goto="父图节点名", graph=Command.PARENT) 跨图跳转。这是子图主动"上报"结果、提前结束、触发父图路由的主要通道。正常情况下子图只需执行到 END，通过同名字段合并即可把结果传回父图；但当子图需要在中途（还没到 END）就通知父图（如"发现致命错误，提前终止并上报"、"已经找到答案提前返回"），就需要用 Command.PARENT 跨图通信。子图也可以通过 Command(graph=子图名, goto=...) 跳转到兄弟子图，但这会引入子图间的强耦合，不推荐。

多层嵌套子图。命名空间以 ::（流式 ns 字段）或元组路径（tasks 模式 path 字段）分层，层级数没有硬限制，但工程上建议不超过 3 层；过深嵌套会让 checkpoint_ns 过长、debug 信息难以阅读、跨图 Command 跳转变成脆弱的字符串路径、resume 时定位断点逻辑复杂。每一层子图应保持独立可运行性（自己有 START/END、单独有单元测试、能脱离父图独立 invoke），父图只做组合与路由。

子图作为多智能体封装单元。Supervisor 模式下每个 Worker Agent 都应封装为独立子图——拥有自己的 State Schema（含私有 messages、私有工具调用历史）、自己的 LLM 节点、自己的 ToolNode、自己的重试策略。Supervisor 图通过 add_node 把这些子图当节点调度，子图执行完毕通过同名字段（如 worker_result）把结构化摘要合并回父图，或通过 Command.PARENT 上报状态。这种封装带来三个好处：(1) 每个 Agent 的 messages 列表是私有 State，不污染其他 Agent 与 Supervisor，避免单一 messages 无限膨胀；(2) 每个 Agent 子图可独立开发、独立测试、独立复用（同一个研究子图可被多个产品的 Supervisor 使用）；(3) Agent 失败隔离——单个子图抛异常不会破坏其他子图的 State，可按子图做错误处理。

完整示例：父图调度研究子图与写作子图。

python

# langgraph==1.2.x
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ---------- 研究子图：私有 messages + 输出 notes ----------
class ResearchState(TypedDict):
    topic: str
    notes: list[str]
    messages: Annotated[list[BaseMessage], add_messages]  # 私有

def do_search(r: ResearchState) -> dict:
    out = llm.invoke([("user", f"请给出关于「{r['topic']}」的 3 条关键要点")])
    return {"notes": [out.content], "messages": [out]}

r_builder = StateGraph(ResearchState)
r_builder.add_node("search", do_search)
r_builder.add_edge(START, "search").add_edge("search", END)
research_subgraph = r_builder.compile(name="research")

# ---------- 写作子图：私有 messages + 输入 notes + 输出 article ----------
class WriteState(TypedDict):
    notes: list[str]
    article: str
    messages: Annotated[list[BaseMessage], add_messages]  # 私有

def do_write(w: WriteState) -> dict:
    ctx = "\n".join(w.get("notes", []))
    out = llm.invoke([("user", f"基于以下要点写一篇 300 字短文：\n{ctx}")])
    return {"article": out.content, "messages": [out]}

w_builder = StateGraph(WriteState)
w_builder.add_node("write", do_write)
w_builder.add_edge(START, "write").add_edge("write", END)
write_subgraph = w_builder.compile(name="writer")

# ---------- 父图：共享字段 topic/notes/article ----------
class ParentState(TypedDict):
    topic: str
    notes: list[str]
    article: str

def schedule_write(s: ParentState) -> Command:
    return Command(goto="writer")

parent = StateGraph(ParentState)
parent.add_node("research", research_subgraph)  # 共享字段自动传递
parent.add_node("writer", write_subgraph)       # notes 自动传入
parent.add_edge(START, "research")
parent.add_edge("research", schedule_write)
parent.add_edge("writer", END)
graph = parent.compile()

r = graph.invoke({"topic": "LangGraph 流式输出的价值"})
print(r["article"][:200])

本章小结

子图是 LangGraph 的模块化单元，编译后的 CompiledStateGraph 通过 add_node("name", compiled_sub) 加入父图，像普通节点一样被调度。父/子图 State 同名字段自动双向传递（输入映射+输出映射），字段名不一致时通过包装函数/transform/Pydantic schema 显式映射；子图独有字段为私有 State，不暴露给父图，实现信息隔离。Checkpoint 通过 checkpoint_ns 按命名空间 父::子::孙 隔离，子图不需单独配置 checkpointer（父图统一传入即可）。流式通过 subgraphs=True 透传子图内部事件，ns 字段标明来源路径。子图内部 interrupt 冒泡到父图调用方，resume 按命名空间自动定位断点；子图通过 Command(graph=Command.PARENT, update=...) 主动向父图通信或跨图 goto。多层嵌套建议不超过 3 层，每层子图保持独立可测试。多智能体系统中每个 Agent 封装为独立子图（私有 messages）是推荐架构，可避免上下文污染、支持团队并行开发与跨项目复用。

常见错误

父/子图 State 字段名不一致导致数据丢失。现象：父图传入了数据，子图读不到对应字段；或子图产出了字段，父图 State 没有更新。原因：字段名不一致时同名自动传递机制不生效，又未提供显式转换器。修复：共享字段保持同名并在团队规范中固化；不同名字段必须写包装节点做显式映射，不要依赖隐式猜测。
子图未 compile 就 add_node 报错。现象：TypeError: Expected a compiled graph, got <StateGraph>。原因：传入了 StateGraph builder 而非 compile 后的实例。修复：先 sub = builder.compile(name="xxx") 再 add_node；compile 时传 name 便于调试。
子图单独配置 checkpointer 与父图冲突。现象：同一 thread_id 下 resume 找不到断点、历史快照错乱、fork 后分支丢失。原因：父子各存一份 checkpointer，命名空间隔离不彻底，写入与读取发生在不同实例。修复：子图 compile 时不主动传 checkpointer，只在最顶层父图 compile 时统一配置；子图独立测试时单独传 MemorySaver。
误以为子图节点名与父图冲突会被覆盖。现象：父子都有叫 "agent" 的节点，担心冲突或调试时混淆。原因：子图节点在 checkpoint_ns 与流式 ns 下使用 子图名::节点名 路径，运行时不会冲突；但调试时同名可能让日志难读。修复：不担心运行时冲突；为了可读性，子图内节点命名带上子模块前缀或保持语义清晰（research 子图里叫 search/summarize，不叫 agent）。
过度嵌套导致调试与 Command 跳转困难。现象：报错栈显示 ns 路径达五六层；get_state_history 难以定位；Command(goto=..., graph=Command.PARENT) 想跳到祖父节点时路径写错。原因：滥用多层嵌套。修复：层级控制在 3 层内；跨层跳转由各层父图中转（子→父→祖父），不要在深层子图直接 Command.PARENT 跳多层；每一层都有明确的单一职责。

生产建议

每个子图独立目录/包维护，自带单元测试与最小可运行示例（不依赖父图即可 invoke 验证输入输出），父图通过包引入，CI 中单独对每个子图跑测试；
子图 compile 时统一传入 name 参数，保证 ns 路径稳定可读，便于前端渲染进度层级与 LangSmith trace 聚合；
团队约定共享字段名规范（messages、topic、notes、user_id、request_id 等常见字段统一命名），转换器代码集中在父图组合层节点，子图内部保持自己的术语体系不妥协；
多 Agent 架构下每个 Agent 子图独立维护自己的 messages 字段（子图私有 State），Supervisor 父图只维护轻量的 task/next_agent/worker_result 字段，避免单一消息列表无限膨胀造成上下文污染与 token 浪费；
子图入口节点统一做输入校验（Pydantic BaseModel 或手动 dict 检查），非法输入在子图边界快速失败（抛 ValueError 返回清晰错误），不污染父图执行链；
禁止在子图内硬编码父图节点名做跨图 goto（破坏子图可复用性）；子图通过 Command.PARENT 只上报结构化结果（result/error 字段），由父图条件路由决定下一步；
子图抛出的异常统一由父图节点级 RetryPolicy 或 catch 节点处理；不要在子图内部 try/except 吞掉所有异常，否则父图感知不到失败、无法做全局兜底；
大项目维护一份"子图注册表"（Python 字典或 YAML），列出所有子图名、输入输出字段契约、维护团队、版本号，新子图接入通过 PR 更新注册表。

C 语言

语言点

深度

新标准

面试整理

第 16 章 子图（Subgraph）与模块化 ​

16.1 子图的添加与 State 映射 ​

16.2 子图与 Checkpoint、流式、中断、跨图通信 ​

本章小结 ​

常见错误 ​