子图组合与多 Agent 编排

子图组合与多 Agent 编排

学习目标

本章要解决什么问题：

• 如何将一个编译后的图嵌套为另一个图的子节点
• 如何管理父子图之间的状态传递与命名空间隔离
• 如何使用 Send 创建多个子图实例并行执行
• 子图与多 Agent 分工的映射关系

前置知识

本章涉及子图组合的通用原理，建议先阅读：

• 子图组合模式 — 子图嵌套、状态隔离、Send 分发
• 多 Agent 路由 — 多 Agent 协作的架构模式

下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。

---

项目实践

将图编译为子图节点

# 定义子图 1：研究 Agent
research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("analyze", analyze_node)
research_graph.add_edge(START, "search")
research_graph.add_edge("search", "analyze")
research_graph.add_edge("analyze", END)

# 定义子图 2：写作 Agent
writing_graph = StateGraph(WritingState)
writing_graph.add_node("outline", outline_node)
writing_graph.add_node("write", write_node)
writing_graph.add_edge(START, "outline")
writing_graph.add_edge("outline", "write")
writing_graph.add_edge("write", END)

# 父图：将子图作为节点注册
parent_graph = StateGraph(ParentState)
parent_graph.add_node("research", research_graph.compile())
parent_graph.add_node("write", writing_graph.compile())
parent_graph.add_edge(START, "research")
parent_graph.add_edge("research", "write")
parent_graph.add_edge("write", END)

app = parent_graph.compile()

关键：子图必须 先编译 再作为节点注册到父图。

状态映射：父子图之间的数据流

父图传递到子图的状态字段必须匹配子图的 input_schema。子图返回的状态字段会合并回父图的状态：

class ParentState(TypedDict):
    query: str
    research_result: str
    final_output: str

class ResearchState(TypedDict):
    query: str        # 从父图接收
    findings: str     # 子图输出，合并回父图

# 父图的 "research_result" 接收子图的 "findings"
# 需要在节点函数中做映射

Send 创建多个子图实例

from langgraph.types import Send

def fan_out(state):
    queries = ["AI 安全", "量子计算", "区块链"]
    return [Send("research", {"query": q}) for q in queries]

parent_graph.add_conditional_edges(START, fan_out)

每个 Send 创建 research 子图的一个独立实例，各实例有隔离的状态和执行轨迹。

子图的命名空间隔离

子图使用命名空间隔离状态：

• 子图的 checkpoint 使用父图 thread_id + 子图路径前缀
• 同一父图的不同子图实例共享 thread_id 但命名空间不同
• 嵌套子图时，命名空间层级递增

这意味着子图内部的状态字段不会与父图或其他子图冲突。

问题与规避

命名空间冲突

问题：子图与父图使用相同 Channel 名称时，虽然在命名空间中隔离，但在状态映射时需要小心处理。

规避：在父图节点函数中做显式的字段映射，避免依赖同名自动匹配。

子图错误传播不透明

问题：子图内部的异常可能不会直接暴露给父图。

规避：在子图出口设置错误处理，将错误信息编码为状态字段返回父图，父图根据状态字段决定后续流程。

子图 checkpoint 与父图版本不一致

问题：时间旅行恢复父图的某个 checkpoint 时，子图的状态也需要回滚到对应版本。

规避：fork 时显式指定子图的 checkpoint_id，或在子图中使用独立的 thread_id。

设计取舍

优势

• 模块化：每个子图独立开发、测试、演进
• 可复用：同一子图可在多个父图中复用
• 多 Agent 分工：每个子图代表一个 Agent 的完整工作流

代价

• 调试复杂度：多层嵌套时追踪执行路径困难
• 状态映射开销：父子图之间的状态转换需要额外设计
• 性能开销：每层子图有独立的 checkpoint 和上下文切换

替代方案

• 扁平图：所有节点在同一个图中定义，适合简单工作流
• 条件分支：通过条件边实现多路径，而非子图嵌套

参考来源

• 源码验证: libs/langgraph/langgraph/pregel/main.py — 子图执行逻辑
• 文档: LangGraph Subgraphs, https://docs.langchain.com/oss/python/langgraph/subgraphs