Subgraph 与多 Agent 协作——三种模式详解——从零开始学 LangGraph（十一）

Agent 协作——三种模式详解写在前面前 10 期我们一直在构建单个Agent。但在真实项目中一个 Agent 往往不够用太复杂一个 Agent 里塞了 8 个工具、5 个路由条件代码爆炸职责不清查天气的逻辑和写邮件的逻辑混在一个 Node 里难以复用写好的审批流程想用到另一个 Agent 里只能复制粘贴Subgraph子图就是来解决这些问题的。它把图中的一部分封装成独立的子图然后作为节点嵌入到主图中。这一期讲三种多 Agent 协作模式路由模式Router主 Agent 分发任务到不同的子 Agent委托模式SubagentAgent 把部分工作委托给子 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 → [传给子图的输入] → 子图执行 → [子图返回输出] → 合并到主图 State2. 最简单的 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, temperature0) 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.54. 模式二委托模式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 个子图选一个执行委托模式主图同时委托多个子图整合结果交接模式多个子图顺序执行流水线作业下期预告第 12 期全栈实战——智能 RAG 知识库 Agent用 LangGraph 构建一个真正能用的 RAG Agent检索 增强 生成 审批多轮对话 记忆系统带来源追溯LangSmith 集成调试参考资料LangGraph Subgraph 概念Graph API overview - Docs by LangChainLangGraph Subgraph 指南Use the graph API - Docs by LangChain多 Agent 模式Multi-agent - Docs by LangChain多 Agent 的核心价值不是多个 Agent而是每个 Agent 干一件事干好。把复杂系统拆成多个单一职责的子图每个子图独立测试、独立部署系统的复杂度和可维护性会大幅改善。