实战!LangGraph Multi-Agent Supervisor 模式:手把手构建生产级多智能体系统

📅 2026/6/30 2:14:49
实战!LangGraph Multi-Agent Supervisor 模式:手把手构建生产级多智能体系统
手把手构建生产级多智能体系统 · 代码全部可运行 · 含避坑指南 高频面试题含参考答案Q1LangGraph 中 Supervisor 模式和 Swarm 模式有什么区别Supervisor 模式一个中央 Supervisor Agent 负责任务分发和结果汇总所有 Worker Agent 只和 Supervisor 通信结构清晰易控制适合任务边界明确的场景。Swarm 模式Agent 之间可以直接互相交接控制权handoff更灵活但更难调试适合任务边界模糊、需要多轮协商的场景。面试重点Supervisor 集中控制Swarm 去中心化协作。Q2LangGraph 的 StateGraph 如何实现多 Agent 间的状态共享LangGraph 使用**共享状态Shared State**实现跨 Agent 通信。所有 Agent 节点读写同一个 TypedDict 状态对象通过 Reducer 函数如 operator.add处理并发写入冲突。每个节点只返回自己修改的字段框架自动 merge 到全局状态。Checkpoint 机制SQLite/Redis保证状态持久化支持断点续跑。Q3多 Agent 系统如何防止幻觉传染即一个 Agent 的错误输出影响整个流程核心方案①Critic Agent独立审查节点每个关键输出都经过批评者评分低分触发重做②结构化输出 Schema 验证强制每个 Agent 输出符合 Pydantic 定义的结构不合规直接拒绝③人工审批节点Human-in-LoopLangGraph 的 interrupt_before/after 机制对高风险决策暂停等待人工确认④沙箱执行Code Agent 产出的代码在隔离环境运行避免副作用扩散。Q4生产环境中 Multi-Agent 系统的 Token 消耗如何控制①上下文压缩Agent 间传递摘要而非原文用 LLM 对中间结果做 100 token 摘要②工具调用缓存相同参数的工具调用结果缓存 TTL 1分钟避免重复检索③分级模型Grader/Router 用 GPT-4o-miniGenerator 用 GPT-4o成本降 60%④并行减少往返Fan-out 并发执行多个子任务减少串行次数。一、为什么单 Agent 不够用2026 年的标志性变化Agent 开发进入多智能体时代。单 Agent 面临三大天花板维度单 AgentMulti-Agent上下文窗口受 Token 限制复杂任务截断各 Agent 独立上下文无限扩展专业深度通才但不精专业 Agent 各司其职更精准并行能力串行执行慢Fan-out 并发速度 5-10× 提升可靠性单点故障全盘崩独立 Agent 隔离失败可重试Anthropic 2026年4月数据Claude Managed Agents 已在 Notion、Rakuten 等企业跑真实业务Multi-Agent 系统商业化进程远超预期。二、Supervisor 模式角色分工与通信机制Supervisor 模式的核心是集中调度一个 Supervisor 负责任务拆解和路由多个专业 Worker 负责执行。️Supervisor Agent接收用户意图拆解子任务路由给 Worker汇总最终答案Research Agent网络搜索知识库检索数据收集Code Agent代码生成沙箱执行Bug 修复✍️Writer Agent内容整合格式化输出报告生成Critic Agent质量审查事实核验评分反馈**状态流转**用户输入 → Supervisor 分析 → 并行派发 Research Code Agent → 汇总中间结果 → Writer 整合 → Critic 审查不通过则返回对应 Agent 重做→ 最终输出。关键所有通信通过 LangGraph 共享 State 传递无需直接 API 调用。三、LangGraph Supervisor 核心代码实现以下为生产可用的 Supervisor 骨架代码LangGraph v1.ximport operator from typing import Annotated, TypedDict from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import END, StateGraph from langgraph.prebuilt import create_react_agent # 1. 定义共享状态 class TeamState(TypedDict): messages: Annotated[list, operator.add] # Reducer 并发安全 task: str research_result: str code_result: str final_answer: str next_worker: str # Supervisor 决定下一步 # 2. Supervisor 节点LLM 路由 def supervisor_node(state): supervisor_prompt 你是团队协调者。根据当前任务状态 决定下一步分配给哪个 Worker或直接 FINISH。 可选researcher / coder / writer / critic / FINISH 当前任务: {task} 已完成: {messages} # 注意确保你已经在外部定义并初始化了 llm 实例 response llm.invoke(supervisor_prompt.format(**state)) return {next_worker: response.content.strip()} # 3. Worker 节点使用 create_react_agent # 注意确保 web_search, vector_search, python_repl, code_formatter 等工具已在外部定义 research_agent create_react_agent( llm, tools[web_search, vector_search], system_message你是专业研究员负责信息收集和事实核验, ) code_agent create_react_agent( llm, tools[python_repl, code_formatter], system_message你是资深工程师负责代码生成和调试, ) # 提示原代码中使用了 writer_agent 和 critic_agent请确保它们也像上面一样被定义。 # 4. 条件路由Supervisor 的决策 def route_by_supervisor(state): next_w state[next_worker] if next_w FINISH: return END return next_w # 返回 worker 名称作为下一节点 # 5. 构建 Supervisor 图 builder StateGraph(TeamState) builder.add_node(supervisor, supervisor_node) builder.add_node(researcher, research_agent) builder.add_node(coder, code_agent) builder.add_node(writer, writer_agent) builder.add_node(critic, critic_agent) builder.set_entry_point(supervisor) # 所有 Worker 完成后都回到 Supervisor 决策 for worker in [researcher, coder, writer, critic]: builder.add_edge(worker, supervisor) builder.add_conditional_edges( supervisor, route_by_supervisor, { researcher: researcher, coder: coder, writer: writer, critic: critic, END: END, }, ) # 6. 开启 Checkpointer 持久化状态支持断点续跑 memory SqliteSaver.from_conn_string(:memory:) graph builder.compile(checkpointermemory)四、三种多智能体架构横向对比️ Supervisor 模式集中调度逻辑清晰易于 Debug 和监控Supervisor 是性能瓶颈适合流程固定、任务边界清晰 Swarm 模式去中心化灵活协作Agent 自主 handoff难以调试追踪适合探索性任务、边界模糊️ 层级化团队多层 Supervisor 嵌套扩展性最强延迟叠加明显适合大型复杂项目五、生产环境落地4 个必须踩的关键点1必须加 Human-in-the-Loop 审批节点对高风险操作数据库写入、邮件发送、API 调用使用 LangGraph 的 interrupt_before“node_name” 暂停图执行等待人工审批后再继续。这是生产级 Agent 系统的安全底线即使 AI 判断100% 正确也不能跳过。2为每个 Agent 设置 Token 预算和超时在共享 State 中维护 agent_token_budget 字段每个 Agent 节点消耗 Token 后更新计数。超出预算触发 early_stop 路径返回 Supervisor。同时为每个节点设置 timeout30s防止工具调用阻塞整个图。3Checkpointer 选 Redis别用内存生产环境 Agent 任务可能运行数分钟进程重启会丢失所有状态。使用 RedisCheckpointerlanggraph-checkpoint-redis持久化状态任何时刻重启都能从断点恢复对用户透明。4LangSmith 全链路追踪是必选项Multi-Agent 系统出 Bug 时没有 trace 几乎无法定位问题。LangSmith 会自动记录每个节点的输入输出、工具调用、Token 消耗并支持在 UI 中回放整个 Agent 执行流程。开发环境设置 LANGCHAIN_TRACING_V2true 一行搞定。六、真实项目踩坑记录⚠️ 坑1Supervisor 陷入无限循环现象Supervisor 一直在 researcher 和 coder 之间反复路由永不结束。根因Supervisor prompt 没有清晰的完成条件LLM 不知道什么时候该输出 FINISH。修复在 Supervisor prompt 中明确写“当 research_result 和 code_result 均非空时必须输出 FINISH”并在 State 中记录 iterations 计数器超过 10 次强制终止。⚠️ 坑2Worker Agent 输出不被 Supervisor 理解现象Code Agent 输出了代码但 Supervisor 判断任务未完成继续派发。根因Worker 的输出作为 messages 列表追加Supervisor 上下文过长导致遗忘中间结果。修复每个 Worker 完成后将关键结果存入 State 的专属字段research_result/code_resultSupervisor prompt 直接读取这些字段而非 messages 列表。⚠️ 坑3并行 Fan-out 导致状态写入冲突现象两个并行 Agent 同时写入 messages 字段后写的覆盖了先写的。修复将 messages 字段的类型改为 Annotated[list, operator.add]使用 Reducer 函数合并而非直接覆盖。这是 LangGraph 处理并发写入的标准方案。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】