【LangGraph】六.多 Agent 协作：Subgraph 机制

写在前面

前面的文章里，我们学过流程控制：顺序、并行、路由、循环。那都是单个图内的节点编排。

但实际应用中，我们经常需要多个独立的 Agent协作：

每个 Agent 有自己的职责（分析、决策、执行）
每个 Agent 有自己的 State（独立的数据结构）
多个 Agent 需要共享信息、协同工作

这就是多 Agent 协作要解决的问题。

多 Agent vs 多节点：

特性	多节点	多 Agent
独立性	节点共享 State	每个 Agent 有独立的 State
复用性	节点难以复用	Agent 可以独立复用
复杂度	适合简单流程	适合复杂系统

Subgraph 是实现多 Agent 协作的核心机制。 这篇文章深入讲解 Subgraph 的工作原理和使用方法。

一、为什么需要 Subgraph

1.1 问题：复杂系统的状态爆炸

假设我们要构建一个代码审查系统：

质量分析 Agent
安全分析 Agent
性能分析 Agent
报告生成 Agent

如果所有 Agent 都在一个图里，State 会是什么样？

# ❌ 问题：State 字段爆炸
class State(TypedDict):
    code: str
    quality_analysis: str
    security_analysis: str
    performance_analysis: str
    quality_score: int
    security_score: int
    performance_score: int
    suggestions: list
    report: str
    # ... 还有更多中间字段

问题：

State 字段越来越多，难以维护
不同 Agent 的字段混在一起，职责不清
想复用一个 Agent，但 State 耦合太紧

1.2 解决：用 Subgraph 隔离上下文

Subgraph 的核心价值：每个子图有独立的 State，父图和子图通过明确的接口通信。

# 子图 1：质量分析
class QualityState(TypedDict):
    code: str
    analysis: str
    score: int

# 子图 2：安全分析
class SecurityState(TypedDict):
    code: str
    analysis: str
    issues: list

# 父图：协调多个子图
class ParentState(TypedDict):
    code: str
    quality_result: dict  # 子图的结果
    security_result: dict
    final_report: str

好处：

每个子图的 State 职责单一
子图可以独立开发、测试、复用
父图只关心结果，不关心中间过程

二、Subgraph 核心语法

2.1 创建子图

子图就是一个普通的 StateGraph，可以独立编译和运行。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


# 1. 定义子图的 State
class SubState(TypedDict):
    input: str
    output: str


# 2. 定义子图的节点
def process_node(state: SubState) -> dict:
    result = f"处理：{state['input']}"
    return {"output": result}


# 3. 创建并编译子图
sub_graph = StateGraph(SubState)
sub_graph.add_node("process", process_node)
sub_graph.set_entry_point("process")
sub_graph.add_edge("process", END)

sub_app = sub_graph.compile()

2.2 在父图中使用子图

子图编译后，可以作为节点添加到父图中。

# 1. 定义父图的 State
class ParentState(TypedDict):
    data: str
    result: str


# 2. 定义父图的节点（准备数据）
def prepare_node(state: ParentState) -> dict:
    # 将父图的数据转换为子图的输入
    return {"input": state["data"]}


# 3. 定义父图的节点（处理结果）
def finalize_node(state: ParentState) -> dict:
    # 从子图的结果中提取需要的数据
    return {"result": state["output"]}


# 4. 创建父图
parent_graph = StateGraph(ParentState)
parent_graph.add_node("prepare", prepare_node)
parent_graph.add_node("subgraph", sub_app)  # 子图作为节点
parent_graph.add_node("finalize", finalize_node)

# 5. 连接边
parent_graph.set_entry_point("prepare")
parent_graph.add_edge("prepare", "subgraph")
parent_graph.add_edge("subgraph", "finalize")
parent_graph.add_edge("finalize", END)

parent_app = parent_graph.compile()

2.3 关键问题：状态如何传递？

子图的输入：

子图从父图的 State 中读取输入。具体来说：

子图执行时，会读取父图 State 中与子图 State 字段匹配的部分
这些字段会被传递给子图作为输入

子图的输出：

子图执行完后，输出会合并到父图的 State中：

子图返回的字段会添加到父图 State
如果父图已有相同字段，会被覆盖（除非使用追加更新）

例子：

# 父图 State
class ParentState(TypedDict):
    data: str
    output: str  # 注意：这个字段名与子图的输出字段相同

# 子图 State
class SubState(TypedDict):
    input: str
    output: str

# 子图节点
def process_node(state: SubState) -> dict:
    return {"output": f"处理：{state['input']}"}

# 执行流程：
# 1. 父图执行 prepare_node，设置 input 字段
# 2. 子图执行，读取 input，返回 output
# 3. 子图的 output 合并到父图 State
# 4. 父图执行 finalize_node，读取 output

四、高级用法

4.1 子图嵌套

子图可以嵌套：父图包含子图，子图包含孙图。

# 孙图：错误处理
error_app = create_error_handler()

# 子图：质量分析（包含错误处理）
quality_graph = StateGraph(QualityState)
quality_graph.add_node("error_handler", error_app)
quality_graph.add_node("analyze", analyze_quality)
# ...

# 父图：审查系统（包含质量分析子图）
review_graph = StateGraph(ReviewState)
review_graph.add_node("quality", quality_app)
# ...

适用场景： 复杂系统分层设计。

4.2 条件调用子图

根据条件决定是否执行子图。

def should_run_security(state: ReviewState) -> str:
    # 如果代码包含敏感操作，需要安全分析
    if "eval" in state["code"] or "exec" in state["code"]:
        return "security"
    else:
        return "skip_security"


review_graph.add_conditional_edges(
    "quality",
    should_run_security,
    {
        "security": "security",
        "skip_security": "report",
    }
)

适用场景： 根据情况选择分析维度。

4.3 循环调用子图

迭代改进结果。

def should_improve(state: ReviewState) -> str:
    if state["quality_result"]["score"] < 80:
        return "improve"
    else:
        return "done"


review_graph.add_conditional_edges(
    "quality",
    should_improve,
    {
        "improve": "refactor",  # 重构子图
        "done": "security",
    }
)

适用场景： 多轮迭代改进。

五、常见错误

错误 1：子图 State 设计不合理

# ❌ 错误：子图 State 包含父图专有字段
class SubState(TypedDict):
    code: str
    user_id: str      # 这是父图的字段
    session_token: str  # 这也是父图的字段

# ✅ 正确：子图 State 只包含需要的字段
class SubState(TypedDict):
    code: str
    analysis: str
    score: int

错误 2：忘记子图需要独立编译

# ❌ 错误
parent_graph.add_node("subgraph", sub_graph)  # 直接添加未编译的图

# ✅ 正确
sub_app = sub_graph.compile()  # 先编译
parent_graph.add_node("subgraph", sub_app)  # 再添加

错误 3：状态字段名不匹配

# 父图 State
class ParentState(TypedDict):
    code: str
    result: dict

# 子图 State
class SubState(TypedDict):
    input: str  # ❌ 父图没有 input 字段
    output: str

# ✅ 正确：字段名匹配
class ParentState(TypedDict):
    input: str  # 父图有这个字段
    output: dict

class SubState(TypedDict):
    input: str
    output: str

小结

为什么需要 Subgraph：

隔离不同 Agent 的 State
提高代码复用性
简化复杂系统设计

核心语法：

创建子图：定义 State、节点、编译
添加到父图：parent_graph.add_node("name", sub_app)
状态传递：字段名匹配自动传递

设计原则：

每个子图职责单一
子图 State 只包含必要字段
父图协调流程，不关心具体逻辑

高级用法：

子图嵌套：分层设计
条件调用：根据情况选择
循环调用：迭代改进

转载自CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/hrh1234h/article/details/160691559