Subgraph 与多 Agent 协作——三种模式详解——从零开始学 LangGraph（十一）-平芜编程栈

Agent 协作——三种模式详解

写在前面

前 10 期我们一直在构建单个Agent。但在真实项目中，一个 Agent 往往不够用：

太复杂：一个 Agent 里塞了 8 个工具、5 个路由条件，代码爆炸
职责不清：查天气的逻辑和写邮件的逻辑混在一个 Node 里
难以复用：写好的"审批流程"想用到另一个 Agent 里，只能复制粘贴

Subgraph（子图）就是来解决这些问题的。它把图中的一部分封装成独立的子图，然后作为节点嵌入到主图中。

这一期讲三种多 Agent 协作模式：

路由模式（Router）：主 Agent 分发任务到不同的子 Agent
委托模式（Subagent）：Agent 把部分工作委托给子 Agent
交接模式（Handoff）：多个 Agent 顺序交接完成任务

1. Subgraph 是什么

Subgraph 就是一个独立的、可编译的图，它可以作为节点嵌入到另一个图里。

# ── 先定义一个子图 ── sub_builder = StateGraph(SubState) sub_builder.add_node("step1", sub_func1) sub_builder.add_node("step2", sub_func2) sub_builder.add_edge(START, "step1") sub_builder.add_edge("step1", "step2") sub_builder.add_edge("step2", END) subgraph = sub_builder.compile() # ── 作为节点嵌入主图 ── main_builder = StateGraph(MainState) main_builder.add_node("entry", entry_node) main_builder.add_node("sub_agent", subgraph) # ← 子图作为节点 main_builder.add_edge(START, "entry") main_builder.add_edge("entry", "sub_agent") main_builder.add_edge("sub_agent", END) main_graph = main_builder.compile()

1.1 Subgraph 的优势

优势	说明
状态隔离	子图有自己独立的状态命名空间，不污染主图
可复用	同一个子图可以被多个主图引用
可测试	子图可以独立编译和测试
职责清晰	每个子图负责一个明确的功能

1.2 状态传递

主图和子图之间通过输入/输出来通信：

主图 State → [传给子图的输入] → 子图执行 → [子图返回输出] → 合并到主图 State

2. 最简单的 Subgraph 例子

from typing import TypedDict from langgraph.graph import StateGraph, START, END # ─── 子图的状态 ─── class SubState(TypedDict): input_text: str processed: str def sub_node_a(state: SubState) -> SubState: return {"processed": f"[A处理] {state['input_text']}"} def sub_node_b(state: SubState) -> SubState: return {"processed": f"{state['processed']} → [B处理]"} # 构建子图 sub_builder = StateGraph(SubState) sub_builder.add_node("a", sub_node_a) sub_builder.add_node("b", sub_node_b) sub_builder.add_edge(START, "a") sub_builder.add_edge("a", "b") sub_builder.add_edge("b", END) subgraph = sub_builder.compile() # 子图可以独立测试 print(subgraph.invoke({"input_text": "测试", "processed": ""})) # {'input_text': '测试', 'processed': '[A处理] 测试 → [B处理]'} # ─── 主图 ─── class MainState(TypedDict): user_input: str result: str def entry_node(state: MainState) -> MainState: return {"result": "开始处理..."} def sub_wrapper(state: MainState) -> MainState: """把主图状态映射到子图，再映射回来""" sub_result = subgraph.invoke({ "input_text": state["user_input"], "processed": "", }) return {"result": sub_result["processed"]} main_builder = StateGraph(MainState) main_builder.add_node("entry", entry_node) main_builder.add_node("sub_process", sub_wrapper) main_builder.add_edge(START, "entry") main_builder.add_edge("entry", "sub_process") main_builder.add_edge("sub_process", END) main_graph = main_builder.compile() print(main_graph.invoke({"user_input": "LangGraph", "result": ""})) # {'user_input': 'LangGraph', 'result': '[A处理] LangGraph → [B处理]'}

3. 模式一：路由模式（Router）

一个主 Agent 做"路由器"，根据用户意图分发给不同的子 Agent。

┌─────────────┐ │ 路由 Agent │ └──────┬──────┘ │ ┌─────────┼─────────┐ ▼ ▼ ▼ ┌────────┐┌────────┐┌────────┐ │ 天气 ││ 计算 ││ 搜索 │ │ Agent ││ Agent ││ Agent │ └────────┘└────────┘└────────┘

from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END from langchain.chat_models import init_chat_model # ─── 子图 1：天气 Agent ─── class WeatherState(TypedDict): city: str result: str def weather_work(state: WeatherState) -> WeatherState: data = {"北京": "晴 25°C", "上海": "多云 28°C", "深圳": "雷阵雨 30°C"} return {"result": f"{state['city']}天气：{data.get(state['city'], '无数据')}"} weather_builder = StateGraph(WeatherState) weather_builder.add_node("work", weather_work) weather_builder.add_edge(START, "work") weather_builder.add_edge("work", END) weather_agent = weather_builder.compile() # ─── 子图 2：计算 Agent ─── class CalState(TypedDict): expression: str result: str def calc_work(state: CalState) -> CalState: try: r = eval(state["expression"], {"__builtins__": {}}) return {"result": f"{state['expression']} = {r}"} except Exception as e: return {"result": f"计算错误：{e}"} cal_builder = StateGraph(CalState) cal_builder.add_node("work", calc_work) cal_builder.add_edge(START, "work") cal_builder.add_edge("work", END) calc_agent = cal_builder.compile() # ─── 主图：路由 Agent ─── class RouterState(TypedDict): user_input: str intent: str final_result: str model = init_chat_model("gpt-4o-mini", temperature=0) def classify_intent(state: RouterState) -> RouterState: """用 LLM 判断用户意图""" response = model.invoke( f"分析以下用户输入属于哪个类别：weather（天气）、calc（计算）。" f"只返回类别名。\n用户输入：{state['user_input']}" ) intent = response.content.strip().lower() if "weather" in intent: intent = "weather" elif "calc" in intent: intent = "calc" else: intent = "unknown" return {"intent": intent} def route_to_agent(state: RouterState) -> RouterState: if state["intent"] == "weather": # 提取城市名（更复杂的提取可以用 LLM） cities = ["北京", "上海", "深圳", "广州", "成都", "杭州"] city = next((c for c in cities if c in state["user_input"]), "北京") result = weather_agent.invoke({"city": city, "result": ""}) return {"final_result": result["result"]} elif state["intent"] == "calc": # 提取表达式（简化处理） result = calc_agent.invoke({"expression": state["user_input"], "result": ""}) return {"final_result": result["result"]} return {"final_result": "无法处理该请求，请使用天气或计算功能。"} def router(state: RouterState) -> Literal["agent", END]: return "agent" builder = StateGraph(RouterState) builder.add_node("classify", classify_intent) builder.add_node("agent", route_to_agent) builder.add_edge(START, "classify") builder.add_edge("classify", "agent") builder.add_edge("agent", END) router_agent = builder.compile() # ─── 测试 ─── print(router_agent.invoke({"user_input": "北京天气怎么样？", "intent": "", "final_result": ""})["final_result"]) # 北京天气：晴 25°C print(router_agent.invoke({"user_input": "计算 3.14 * 25", "intent": "", "final_result": ""})["final_result"]) # 3.14 * 25 = 78.5

4. 模式二：委托模式（Subagent）

主 Agent 把部分工作"委托"给子 Agent，然后等待子 Agent 返回结果，继续自己的工作。

主 Agent（项目管理） │ ├── 分析需求 → 得到需求文档 ├── 委托：搜索 Agent 查资料 → 得到搜索结果 ├── 委托：摘要 Agent 总结 → 得到摘要 └── 整合结果 → 最终报告

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver # ─── 子图：搜索 Agent ─── class SearchState(TypedDict): query: str results: str def search_worker(state: SearchState) -> SearchState: # 模拟搜索 return {"results": f"关于「{state['query']}」的搜索结果：\n1. 技术文档\n2. 社区讨论\n3. 最佳实践"} search_builder = StateGraph(SearchState) search_builder.add_node("work", search_worker) search_builder.add_edge(START, "work") search_builder.add_edge("work", END) search_agent = search_builder.compile() # ─── 子图：摘要 Agent ─── class SummaryState(TypedDict): text: str summary: str def summary_worker(state: SummaryState) -> SummaryState: return {"summary": f"[摘要] {state['text'][:30]}..."} summary_builder = StateGraph(SummaryState) summary_builder.add_node("work", summary_worker) summary_builder.add_edge(START, "work") summary_builder.add_edge("work", END) summary_agent = summary_builder.compile() # ─── 主图 ─── class ResearchState(TypedDict): topic: str search_results: str summary: str final_report: str def analyze_topic(state: ResearchState) -> ResearchState: return {"final_report": f"研究主题：{state['topic']}\n"} def delegate_search(state: ResearchState) -> ResearchState: search_result = search_agent.invoke({"query": state["topic"], "results": ""}) return {"search_results": search_result["results"]} def delegate_summary(state: ResearchState) -> ResearchState: summary_result = summary_agent.invoke({"text": state["search_results"], "summary": ""}) return {"summary": summary_result["summary"]} def compile_report(state: ResearchState) -> ResearchState: report = ( f"研究主题：{state['topic']}\n\n" f"搜索资料：\n{state['search_results']}\n\n" f"核心摘要：\n{state['summary']}\n\n" f"--- 报告完成 ---" ) return {"final_report": report} builder = StateGraph(ResearchState) builder.add_node("analyze", analyze_topic) builder.add_node("search", delegate_search) builder.add_node("summarize", delegate_summary) builder.add_node("compile", compile_report) builder.add_edge(START, "analyze") builder.add_edge("analyze", "search") builder.add_edge("search", "summarize") builder.add_edge("summarize", "compile") builder.add_edge("compile", END) research_agent = builder.compile() # ─── 测试 ─── result = research_agent.invoke( {"topic": "LangGraph Subgraph 用法", "search_results": "", "summary": "", "final_report": ""} ) print(result["final_report"])

5. 模式三：交接模式（Handoff）

多个 Agent 顺序执行，像生产线一样——上一个 Agent 处理完传给下一个。

用户 → 输入处理 Agent → 业务逻辑 Agent → 输出格式化 Agent → 用户 ↓ ↓ ↓ 验证输入 处理核心业务 格式化输出

from typing import TypedDict from langgraph.graph import StateGraph, START, END # ─── Agent 1：输入验证 ─── class InputState(TypedDict): raw_input: str validated: bool cleaned: str def validate_node(state: InputState) -> InputState: text = state["raw_input"].strip() if len(text) < 2: return {"validated": False, "cleaned": ""} return {"validated": True, "cleaned": text} input_builder = StateGraph(InputState) input_builder.add_node("validate", validate_node) input_builder.add_edge(START, "validate") input_builder.add_edge("validate", END) input_agent = input_builder.compile() # ─── Agent 2：业务处理 ─── class BizState(TypedDict): cleaned_input: str processed: str status: str def process_node(state: BizState) -> BizState: return { "processed": f"[处理] {state['cleaned_input'].upper()}", "status": "done", } biz_builder = StateGraph(BizState) biz_builder.add_node("process", process_node) biz_builder.add_edge(START, "process") biz_builder.add_edge("process", END) biz_agent = biz_builder.compile() # ─── Agent 3：输出格式化 ─── class OutputState(TypedDict): processed: str formatted: str def format_node(state: OutputState) -> OutputState: return {"formatted": f"📋 结果：{state['processed']}"} output_builder = StateGraph(OutputState) output_builder.add_node("format", format_node) output_builder.add_edge(START, "format") output_builder.add_edge("format", END) output_agent = output_builder.compile() # ─── 交接编排 ─── class PipelineState(TypedDict): raw_input: str final_output: str def handoff_pipeline(state: PipelineState) -> PipelineState: # Agent 1 → Agent 2 → Agent 3 r1 = input_agent.invoke({"raw_input": state["raw_input"], "validated": False, "cleaned": ""}) if not r1["validated"]: return {"final_output": "输入无效，请输入至少 2 个字符。"} r2 = biz_agent.invoke({"cleaned_input": r1["cleaned"], "processed": "", "status": ""}) r3 = output_agent.invoke({"processed": r2["processed"], "formatted": ""}) return {"final_output": r3["formatted"]} pipeline_builder = StateGraph(PipelineState) pipeline_builder.add_node("pipeline", handoff_pipeline) pipeline_builder.add_edge(START, "pipeline") pipeline_builder.add_edge("pipeline", END) pipeline = pipeline_builder.compile() # ─── 测试 ─── print(pipeline.invoke({"raw_input": "hello world", "final_output": ""})["final_output"]) # 📋 结果：[处理] HELLO WORLD print(pipeline.invoke({"raw_input": "x", "final_output": ""})["final_output"]) # 输入无效，请输入至少 2 个字符。

6. Subgraph 与主图共享状态

有时候子图需要访问主图的 State，或者把结果写回主图的特定字段。方案是映射函数：

from typing import TypedDict from langgraph.graph import StateGraph, START, END class SharedState(TypedDict): user_query: str sub_result: str final: str class SubSharedState(TypedDict): query: str result: str def sub_work(state: SubSharedState) -> SubSharedState: return {"result": f"子图处理：{state['query']}"} def main_node(state: SharedState) -> SharedState: # 把主图的 user_query 映射到子图的 query sub_result = sub_agent.invoke({"query": state["user_query"], "result": ""}) # 把子图的 result 映射回主图的 sub_result return {"sub_result": sub_result["result"]} sub_builder = StateGraph(SubSharedState) sub_builder.add_node("work", sub_work) sub_builder.add_edge(START, "work") sub_builder.add_edge("work", END) sub_agent = sub_builder.compile() main_builder = StateGraph(SharedState) main_builder.add_node("main", main_node) main_builder.add_edge(START, "main") main_builder.add_edge("main", END) main_graph = main_builder.compile()

关键原则：子图的输入和输出要有明确的映射逻辑，不要假设字段名能自动匹配。

7. 三种模式对比

维度	路由模式	委托模式	交接模式
组织结构	1 个分发器 + N 个执行器	1 个管理者 + 子任务	流水线
子图关系	互斥（选一个执行）	协作（都执行，主图整合）	顺序执行
典型场景	多意图聊天机器人	研究报告生成	数据处理管道
复杂度	低	中	低
灵活性	高（可动态选择）	最高	固定
代码复用	子图独立，可互换	子图独立，可复用	每步独立，可抽换

8. 常见坑点

坑 1：子图的状态字段名冲突

# 主图有 city 字段，天气子图也有 city 字段 # 调用子图时可能不小心传错了值

解决：子图使用独立的命名空间前缀，或在映射函数中显式转换。

坑 2：忘记传子图需要的所有字段

# ❌ 子图 State 需要 2 个字段，只传了 1 个 result = sub_agent.invoke({"query": "test"}) # 如果 processed 有默认值还好，否则报错 # ✅ 传所有必需字段 result = sub_agent.invoke({"query": "test", "processed": ""})

坑 3：子图内部的错误没有被主图捕获

# 子图抛异常，主图的节点没有 try/except def main_wrapper(state): result = sub_agent.invoke(state) # 如果这里崩了... return result # 主图也跟着崩 # ✅ 包装错误处理 def safe_main_wrapper(state): try: result = sub_agent.invoke(state) return result except Exception as e: return {"error": str(e)}

坑 4：过度使用 Subgraph

不要为了用 Subgraph 而用 Subgraph。如果你的主图只有 3 个节点、没有复用的需求，直接写在一个图里更简单。

9. 本期核心总结

概念	一句话
Subgraph	独立编译的图，作为节点嵌入主图
状态隔离	子图有自己的 State，不污染主图
状态映射	主图和子图之间通过明确映射来传递数据
路由模式	一个分发器 + N 个子图（选一个执行）
委托模式	主图同时委托多个子图，整合结果
交接模式	多个子图顺序执行，流水线作业