子图 Subgraph

当工作流越来越复杂，把所有节点塞进一张图会变得难维护。子图（Subgraph）就是把一块功能封装成一个独立的小图，再作为"一个节点"挂进父图。本篇讲怎么做，以及父子状态怎么对齐。

一、为什么需要子图

动机	说明
模块化	把"研究"和"写作"拆开，各自独立开发测试
复用	同一个子图可在多个父图里用
可读性	父图只看高层流程，细节藏在子图里
团队协作	不同人各管一个子图，互不干扰

类比 Java/LangGraph4j：相当于把一组节点打包成一个可复用的子 StateGraph。

二、子图就是一个编译好的图

子图没什么神秘——它本身就是一个用 StateGraph 定义、compile() 后的可运行对象。把它当节点 add_node 进父图即可。

from langgraph.graph import StateGraph, START, END

# 子图：一个简单的两步流程
class SubState(TypedDict):
    query: str
    result: str

def sub_step1(state):
    return {"result": f"处理了 {state['query']}"}

sub_builder = StateGraph(SubState)
sub_builder.add_node("step1", sub_step1)
sub_builder.add_edge(START, "step1")
sub_builder.add_edge("step1", END)
subgraph_app = sub_builder.compile()   # 编译好的子图

# 父图：把子图当一个节点
parent = StateGraph(ParentState)
parent.add_node("research", subgraph_app)   # 直接把编译好的图当节点

三、状态映射：父子状态怎么对齐

这是子图最关键也最容易踩坑的点。子图被调用时，LangGraph 会从父图状态里取子图需要的字段（按字段名匹配），子图返回时也按字段名写回父图。

情况 1：子图状态是父图状态的子集（字段同名）

最简单的情况：子图状态字段是父图字段的子集，名字一致。

class ParentState(TypedDict):
    query: str
    research_result: str   # 子图会写这个
    final: str

class SubState(TypedDict):
    query: str             # 父图也有 query
    research_result: str  # 父图也有 research_result

# 子图节点读 query、写 research_result，框架自动从父图取 query、回写 research_result

情况 2：字段不同名或需要转换

如果子图状态字段和父图对不上，需要做一层映射。常见做法是在子图外加一层"适配节点"做字段搬运，或让父图节点先把字段重命名好再传给子图。

def adapt_in(state):
    # 把父图的 topic 映射成子图要的 query
    return {"query": state["topic"]}

parent.add_node("adapt_in", adapt_in)
parent.add_edge("adapt_in", "research")  # research 是子图

简单原则：尽量让父子状态字段同名，能省掉大量适配代码。

四、完整示例：父图调度两个子图

下面搭一个"研究子图 + 写作子图"的父图。每个子图内部都有自己的多步流程。

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END

# ============ 子图1：研究 ============
class ResearchState(TypedDict):
    topic: str
    findings: list[str]

def search(state):
    return {"findings": [f"{state['topic']}的资料A", f"{state['topic']}的资料B"]}

def filter(state):
    # 假装过滤，只保留第一条
    return {"findings": state["findings"][:1]}

research_builder = StateGraph(ResearchState)
research_builder.add_node("search", search)
research_builder.add_node("filter", filter)
research_builder.add_edge(START, "search")
research_builder.add_edge("search", "filter")
research_builder.add_edge("filter", END)
research_app = research_builder.compile()

# ============ 子图2：写作 ============
class WritingState(TypedDict):
    findings: list[str]
    article: str

def outline(state):
    return {"article": f"大纲：基于{len(state['findings'])}条资料"}

def draft(state):
    return {"article": state["article"] + " → 写成初稿"}

writing_builder = StateGraph(WritingState)
writing_builder.add_node("outline", outline)
writing_builder.add_node("draft", draft)
writing_builder.add_edge(START, "outline")
writing_builder.add_edge("outline", "draft")
writing_builder.add_edge("draft", END)
writing_app = writing_builder.compile()

# ============ 父图 ============
class ParentState(TypedDict):
    topic: str
    findings: Annotated[list[str], add]   # 用 reducer，两个子图都要写
    article: str

parent_builder = StateGraph(ParentState)
parent_builder.add_node("research", research_app)
parent_builder.add_node("writing", writing_app)
parent_builder.add_edge(START, "research")
parent_builder.add_edge("research", "writing")
parent_builder.add_edge("writing", END)
parent_app = parent_builder.compile()

# 运行
result = parent_app.invoke({"topic": "LangGraph"})
print("资料:", result["findings"])
print("文章:", result["article"])

要点：

• 父图 findings 用 Annotated[list[str], add] 是因为两个子图都往里写，需要 reducer 汇聚，否则会冲突。如果只有一个子图写，直接覆盖即可。
• 子图的入口节点会从父图取 topic/findings，出口字段写回父图。

五、嵌套图的可视化

把父图导出 mermaid，能看到子图作为一个节点出现：

print(parent_app.get_graph().draw_mermaid())

如果想看子图内部细节，单独导出子图：

print(research_app.get_graph().draw_mermaid())

flowchart LR
    subgraph 父图
        R[research 子图] --> W[writing 子图]
    end
    subgraph research子图内部
        S[search] --> F[filter]
    end
    subgraph writing子图内部
        O[outline] --> D[draft]
    end

六、子图与检查点

子图也可以有自己的 checkpointer。父图带 checkpointer 时，子图执行的状态默认也会被父图的 checkpointer 覆盖（嵌套保存）。如果子图单独配了 checkpointer，注意：

• 子图用独立 checkpointer 时，父子是两套检查点体系，恢复时要分别处理。
• 大多数场景子图不单独配 checkpointer，复用父图的即可，避免管理混乱。

HITL（见 人机交互 HITL）配合子图时，interrupt_before 写的是子图内部节点名，行为一致。

七、常见踩坑

踩坑 1：父子状态字段对不上

子图要 findings 但父图字段叫 docs，子图拿到的是空。字段同名是子图工作的前提，设计状态时就要规划好共享字段名。

踩坑 2：多个子图写同一字段没挂 reducer

research 和 writing 都返回 findings，父图该字段没挂 reducer（Annotated[list, add]），第二个子图会覆盖第一个的结果，甚至报 InvalidUpdateError。多写必须 reducer。

踩坑 3：把子图 builder 当节点

parent.add_node("research", research_builder)  # ❌ 传了 builder 而不是编译后的 app

要传 compile() 之后的产品，不是 StateGraph builder。

踩坑 4：子图入口/出口字段缺失

子图第一个节点要读 topic，但父图状态进来时没有 topic（拼写错了或没传），子图拿到 None。调试时先确认进入子图时父图状态确实有这些字段。

踩坑 5：子图里有循环但没终止条件

子图内部的回路同样需要终止条件（计数器/阈值），否则会无限循环，详见 构建完整工作流 的踩坑汇总。

八、小结

• 子图 = 编译好的小图，作为节点挂进父图，实现模块化、复用。
• 父子状态按字段同名自动传递，子集关系最简单。
• 多个子图写同一字段要挂 reducer 汇聚。
• 子图一般复用父图 checkpointer，不单独配，避免管理混乱。
• 调试用分别导出父子图的 mermaid 看拓扑。

下一篇 中断与恢复 讲怎么把被打断的长流程接续跑完。