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

写在前面前面的文章里我们学过流程控制顺序、并行、路由、循环。那都是单个图内的节点编排。但实际应用中我们经常需要多个独立的 Agent协作每个 Agent 有自己的职责分析、决策、执行每个 Agent 有自己的 State独立的数据结构多个 Agent 需要共享信息、协同工作这就是多 Agent 协作要解决的问题。多 Agent vs 多节点特性多节点多 Agent独立性节点共享 State每个 Agent 有独立的 State复用性节点难以复用Agent 可以独立复用复杂度适合简单流程适合复杂系统Subgraph 是实现多 Agent 协作的核心机制。这篇文章深入讲解 Subgraph 的工作原理和使用方法。一、为什么需要 Subgraph1.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 只包含必要字段父图协调流程不关心具体逻辑高级用法子图嵌套分层设计条件调用根据情况选择循环调用迭代改进