LangChain多智能体Supervisor Agent核心实现与踩坑指南

多智能体这个系列写到第二十二集前面聊过Chain、Agent、Memory、工具调用上集我们终于把Supervisor Agent搭起来了。但说实在的能跑通demo和能在真实场景下稳定运行中间差了不止一个级别。这一集继续往下挖专门讲Supervisor Agent在实现过程中的那些关键决策和细节坑worker如何设计、supervisor如何做路由、状态怎么管理、循环怎么防以及怎么把整个系统从“能跑”调优到“能用”。适合正在用LangChain搭多智能体的朋友尤其是那种已经看过不少文档但一上手就发现文档没讲清楚的部分。1. 先想清楚Supervisor模式到底解决了什么问题1.1 为什么需要“老板”这个角色多智能体不是“Agent越多越好”。当你把两个Agent串起来之后很快会发现它们经常抢话、重复干活、或者在一个问题上绕圈子。这就是缺少一个统一调度者的典型症状。Supervisor Agent的思路非常像现实里的项目分工老板不亲自写代码、不做测试但他需要决定“这件事交给谁做、做完没有、下一件事是什么”。在LangChain的多智能体框架里Supervisor Agent承担的就是这个中枢决策角色。它本身也是一个Agent有自己的模型和提示词但它不直接调用业务工具而是调用一组“子Agent工具”。每个子Agent通常叫Worker被包装成一个可供调用的工具对外暴露一个名字和一段功能描述。Supervisor看到任务后根据这些描述决定应该调用哪个Worker并把任务原文传递下去。这样做最直接的好处是职责清晰。Worker只关心自己那摊事不需要理解全局目标Supervisor只关心调度不需要关心每个业务的细枝末节。修改其中一个Worker时其他部分基本不受影响这一点在真实项目里尤其重要。1.2 与“群聊式”多智能体的关键差异有一种很多时候也会被提及的设计叫“群聊式”多智能体所有Agent共享同一个消息队列互相之间都能看到对方的消息谁有话谁就说。听起来很民主但实际跑起来问题不少。消息一多每个Agent都要从大量历史消息里分辨“哪些跟我相关”某个Agent抛出一个问题后很可能没人接或者好几个人同时接聊天一长上下文成本也涨得飞快。Supervisor模式相当于把所有沟通都收敛到老板一个人身上。Worker之间不直接通信只跟Supervisor对话。每个Worker看到的是Supervisor单独发给它的任务而不是整个群聊记录。这样就不存在“这条消息该谁处理”的争执也减少了上下文里的无关信息。当然代价也很明显Supervisor节点容易成为瓶颈所有决策都要过它这一轮模型调用而且Supervisor一旦判断失误整个流程可能跑偏。所以后面第4部分讲的防循环、防误调度都是在这个基础上展开的。2. 核心实现拆解LangChain/LangGraph中的Supervisor Agent结构2.1 从AgentExecutor到LangGraph的演进理解早几年写LangChain多智能体大家常用的是AgentExecutor。它的工作方式是一个Agent内部反复执行“思考-调用工具-观察结果”的循环。这种方式对单个Agent没问题但想在上面做“多个Agent协作”就很别扭要么把另一个Agent塞进工具列表里当普通工具调用要么自己写很多胶水代码去管理循环结束条件。后来LangGraph把这一整套抽象成了“图”。Agent变成了图里的节点节点之间的边可以是条件边下一轮走哪个节点由当前节点的输出动态决定。这个变化对多智能体是本质性的。因为它让“路由”从Agent自己脑内完成变成了一种图级别可以显式观察和控制的东西。Supervisor Agent在LangGraph里的本质就是一个特殊节点它根据当前对话状态决定跳转到哪个Worker节点或者跳转到结束节点。这里要记住一个关键认知在Supervisor模式里Worker并不是“主动发言”的节点而是被Supervisor通过工具调用触发的。当Supervisor调用了代表某个Worker的工具时LangGraph会执行那个Worker并把它的最终输出作为“工具调用结果”返回给Supervisor。Supervisor看到结果后再决定继续还是结束。理解了这一层后面看代码就不会懵。2.2 create_supervisor的骨架和关键参数现在LangChain生态里最省事的实现方式是使用langgraph-supervisor提供的create_supervisor函数。它把“Supervisor节点 Worker工具绑定 条件路由”这些逻辑都封装好了。下面是一个标准骨架from langchain.chat_models import init_chat_model from langgraph.prebuilt import create_react_agent from langgraph_supervisor import create_supervisor # 用你的模型服务初始化 model init_chat_model(your-model-id) # 定义两个worker research_agent create_react_agent( modelmodel, tools[search_knowledge_base], nameresearch_agent, prompt你是资料研究员根据用户的查询检索知识库给出相关段落。, ) write_agent create_react_agent( modelmodel, tools[format_markdown], namewrite_agent, prompt你是摘要作者把调研结果整理成结构化摘要。, ) # 定义supervisor supervisor create_supervisor( agents[research_agent, write_agent], modelmodel, prompt( 你是一名项目负责人。根据用户需求决定调用research_agent还是write_agent。 如果用户需要查资料先调用research_agent 如果资料齐了再调用write_agent产出摘要 当工作完成时直接回复用户最终结果。 ), supervisor_namesupervisor, output_modelast_message, ) app supervisor.compile()这里有几个参数值得认真对待。agents列表就是Supervisor手里能调用的Worker集合每个Worker会被自动转换成工具描述工具名就是Agent的name工具描述是从Prompt和Agent信息里生成的。所以name一定要取得辨识度高别叫什么agent1、agent2否则Supervisor没法准确决策。output_mode控制了最终返回给用户的信息形态。用last_message时Supervisor结束时会直接返回它最后生成的那条回复这对大多数“最终要一个答案”的场景够用。如果你希望把整个对话里的中间过程也暴露出去可以用full_history但这时返回内容会非常大成本也高。model最好支持工具调用这是硬前提。如果模型不支持工具调用Supervisor没法稳定地“决定调用哪个Worker”整个路由就退化成靠文本生成的碰运气。2.3 Worker Agent的设计要点工具与提示词Worker内部也是一个完整的React Agent它有自己独立的工具和提示词。很多人花了很多时间调整Supervisor的Prompt但忽略了Worker的设计结果Supervisor再怎么聪明也没用。Worker的提示词要回答三个问题我是谁我的输出格式是什么我什么情况下该说“做不到”。因为Worker不会主动和别的Worker沟通它只会收到Supervisor传来的任务文本。如果Supervisor对它能力的理解有偏差Worker就可能硬着头皮做一件它不擅长的事。所以Worker名和提示词里要有清晰的边界描述。比如上面的research_agent我写的是“根据查询检索知识库给出相关段落”如果写的是“负责用户研究”Supervisor就可能在用户问另一个领域问题时也把它调度出来。Worker内部工具的描述同样决定路由质量。因为Worker被封装成工具后Supervisor看到的是“这个工具能做什么”而不是“这个Agent内部有什么工具”。如果你的search_knowledge_base工具描述写得太泛Supervisor会高估或者低估它。我习惯在工具描述里把“使用前提、返回内容、最好在什么场景下调用”都写清楚。例如def search_knowledge_base(query: str) - str: 在本地产品知识库中搜索与query相关的片段。仅适合查询产品功能、参数和常见问题。 ...这样描述之后Supervisor看到的是research_agent这个工具并继续从其描述里继承细节。每层描述清晰决策准确率自然就高。3. 实操手写一个“查资料写摘要”的Supervisor系统3.1 准备环境与模型在动手之前先把依赖装好pip install -U langchain langgraph langgraph-supervisor模型部分我强烈建议选一个支持function calling的模型。具体是哪个厂商的不重要关键是“工具调用”能力要稳。初始化时用init_chat_model这样可以方便切换模型服务。如果你跑在本地也可以直接切成Ollama或者其他本地模型。不过本地小模型在Supervisor这种需要复杂决策的场景里表现差距会很明显建议先用一个中大型模型跑通链路然后再做降级优化。本文场景很简单我有一个模拟的产品知识库里面存了几段产品资料另一个Worker负责把资料整理成固定结构的Markdown摘要。整个系统就是“先查资料再写摘要”的两步流程。3.2 定义两个Worker为了完全脱离外部API依赖我直接写一个内存里的知识库检索函数这样任何人都能复现。实际项目里把这个函数换成向量数据库检索即可。knowledge_base { 价格: 产品X基础版定价299元/月专业版599元/月。, 功能: 产品X支持多租户、权限管理、报表导出和API接口。, 部署: 产品X提供公有云SaaS和私有化部署两种模式。, } def search_knowledge_base(query: str) - str: 在本地产品知识库中搜索与query相关的片段。仅适合查询产品功能、参数和价格信息。 results [v for k, v in knowledge_base.items() if k in query or query in k] return \n.join(results) if results else 知识库中未找到相关信息。然后是写摘要的Worker。它不需要额外外部工具但为了演示Worker也有自己的工具调用能力我给它一个format_markdown工具。当然你也可以不给工具让它直接用文本输出。def format_markdown(text: str) - str: 把纯文本整理成带标题和列表的Markdown格式。 lines text.strip().splitlines() sections [f- {line} for line in lines if line.strip()] return ## 摘要\n \n.join(sections) research_agent create_react_agent( modelmodel, tools[search_knowledge_base], nameresearch_agent, prompt( 你是产品研究员。用户需要产品信息时用search_knowledge_base检索。 如果检索结果为空如实说明不要编造。 ), ) write_agent create_react_agent( modelmodel, tools[format_markdown], namewrite_agent, prompt( 你是摘要作者。收到调研材料后用format_markdown把它整理成Markdown摘要。 保留关键数据不要新增信息。 ), )两个Worker之间的协作关系完全由Supervisor来调度。research_agent不会自己调用write_agentwrite_agent也不会反过来调用research_agent。3.3 定义Supervisor并组装状态图现在用create_supervisor把两个Worker捆到一起supervisor_agent create_supervisor( agents[research_agent, write_agent], modelmodel, supervisor_namesupervisor, prompt( 你是一个产品咨询团队负责人。用户会问产品相关问题。\n 流程如下\n 1. 需要查询产品资料时调用research_agent。\n 2. 如果已经拿到资料需要整理成Markdown摘要时调用write_agent。\n 3. 如果工作已完成直接给用户最终答案不要再调用任何工具。 ), output_modelast_message, ) app supervisor_agent.compile()如果你不用create_supervisor想理解底层原理可以用LangGraph手写等价逻辑。核心就三点把Worker工具绑定到Supervisor模型上解析模型输出的tool_calls根据tool_calls决定下一跳节点。关键代码大致长这样from langgraph.graph import StateGraph, END def supervisor_node(state): messages state[messages] response model_with_worker_tools.invoke(messages) return {messages: [response]} # 根据最后的tool_call路由到对应worker或结束 def route_after_supervisor(state): last state[messages][-1] if not last.tool_calls: return END tool_name last.tool_calls[0][name] if tool_name research_agent: return research_worker elif tool_name write_agent: return write_worker return supervisor这样写虽然要处理不少边界细节但对理解整个调度机制非常有用。封装函数能让你快速起步手写能让你排障时不慌。3.4 运行与结果分析直接调用result app.invoke({ messages: [{role: user, content: 请查一下产品X的价格和功能并整理成摘要}] }) print(result[messages][-1].content)正常情况下Supervisor会先看到用户请求然后调用research_agent收到知识库内容后再调用write_agent最后生成摘要。运行过程里的中间消息会包含多次模型调用但从外部看就是一个普通的invoke调用。最终输出大概是## 摘要 - 产品X基础版定价299元/月专业版599元/月。 - 产品X支持多租户、权限管理、报表导出和API接口。出现这个结果说明Supervisor的两步路由都成功了。如果直接一次性生成摘要没有先调用research_agent通常有两种原因一是模型自带知识里也有相关内容没有触发工具调用二是Prompt里流程描述不够强硬。要让流程真正“去查”可以在用户请求里加一句“必须先查知识库再写摘要”或者在Supervisor Prompt里强调“禁止跳过research_agent”。4. 踩坑实录Supervisor Agent的6个高频问题与排查方法4.1 supervisor反复把任务分给同一个worker死循环现象调用invoke后日志里显示同一个Worker被连续调用很多次最后要么超时要么把上下文撑爆。原因最常见的是Worker返回内容里含有“我不确定需要再查一次”之类的话Supervisor看完后觉得“还没完成”于是再次调用原Worker。也可能是Supervisor把“调用工具后的观察结果”误判成了“任务还没开始”。排查方法先看最后一次tool_calls和对应的ToolMessage。如果Worker已经返回了信息但Supervisor继续调度它那就是提示词问题。我在Supervisor Prompt里通常会加一句“当工具返回了内容时认为该步骤已经完成除非内容为空或明确表示失败否则不要重复调用同一个工具。”还有一种情况是Worker本身也在内部循环它可以在自己的React循环里反复调用同一个工具。这种问题要单独调Worker的工具描述和Supervisor无关。4.2 worker返回了“最终答复”supervisor却还要再调度现象Worker明明已经给出了最终答案Supervisor却不管不顾地调用下一个Worker结果把内容搞串了。原因Worker被包装成工具后Supervisor看到的是工具返回值它并不清楚这个返回值到底是“中间结果”还是“最终结果”。如果Worker的返回文本没有明确的“完成”信号Supervisor会倾向于继续走流程。解决思路在每个Worker的Prompt里要求它返回时加上一个结束标记比如“【研究完成】”或者“【摘要已完成】”。Supervisor看到这个标记就停止调度。这个法子虽然土但在模型能力不够强时非常有效。4.3 上下文窗口被对话历史塞满现象对话轮次稍微多一点state[messages]里积攒了所有Worker的执行记录、工具返回值、中间思考还没跑完就报超长错误。原因LangGraph的State默认会把消息全量追加。Supervisor模式里每次调用Worker都会产生好几条消息用户和系统之间的上下文膨胀速度比单Agent快得多。解法一是在State里用last_k之类的函数对消息列表做截断只保留最近几轮二是把Worker的中间过程收拢成一条汇总消息而不是把所有原始消息都放回Supervisor上下文。我习惯在Worker执行后只把它的最终输出追加到supervisor_messages内部琐碎步骤不进入总状态。4.4 状态冲突多个worker修改同一份state现象当多个Worker并行跑或者顺序跑时如果都试图往同一个自定义字段里写东西后写的会覆盖先写的数据就丢了。原因LangGraph里默认的state reducer对列表是追加对普通字段是覆盖。如果你自定义了summary、research_results这些字段就要小心。解决方式给这些字段配置合适的reducer比如用operator.add把新值追加到列表或者写一个自定义的合并函数。最简单的是把每次结果都放在消息列表里不要单独维护平铺字段。4.5 worker内部异常导致整个图崩溃现象某个Worker内部工具报错比如查不到数据、JSON解析失败整个invoke直接抛出异常前面的结果全部白费。原因LangGraph里默认情况下节点异常会向外传播。你以为Supervisor模式能“包容”子Agent的错误但实际上并不会自动消化。处理办法在Worker的Prompt里明确要求“工具出错时把错误信息转换为普通文本返回而不是抛异常”同时可以用LangGraph的invoke(..., config{recursion_limit: ...})控制最坏情况。更稳妥的是在工具函数内部做try/except把异常信息包成字符串返回。这样Supervisor至少还有机会基于错误信息做下一步决策。4.6 使用追踪与日志定位supervisor决策链路多智能体最让人头疼的是“结果不对但不知道是哪一步导致的”。我强烈建议从第一版就打开LangGraph的追踪/日志记录功能。这样每次invoke之后你能看到完整的节点跳转序列、每一步的输入输出以及耗时和token消耗。实际排查时我一般先看节点跳转序列是不是符合预期。如果预期是supervisor - research_worker - supervisor - write_worker - supervisor但实际变成了supervisor - write_worker - supervisor说明Supervisor跳过了查询步骤这时候去改Prompt比靠猜有效得多。追踪日志里的token消耗还能帮你发现上下文膨胀的元凶——往往是某个Worker内部循环太多。5. 从“能跑”到“好用”的进阶经验5.1 给Supervisor加记忆和人工确认默认的Supervisor只处理当前这轮对话。如果用户今天问了价格明天问“昨天你查的价格里包含部署费用吗”Supervisor并没有“昨天”的概念。解决方式是在State里增加长期记忆字段把历史关键信息比如用户反复提到的产品、偏好格式在每轮开始前注入Prompt。如果你做的是高价值场景比如生成合同、给出医嘱千万别让Supervisor全自动跑完。可以在流程中间加一个人工确认节点Supervisor在调用write_agent之前先把research_agent的结果展示给用户确认。这个用Interrupt或自定义工具都能实现。模型决策再稳关键节点上人来拍板能挡住大多数低级错误。5.2 多级Supervisor让老板只管方向组长管执行当Worker数量超过5个时让一个Supervisor直接面对所有Worker决策质量会明显下降。原因很简单工具列表太长模型在众多描述中挑一个最优项越来越难。我建议做两级Supervisor。顶层Supervisor只管“这个任务交给哪个小组”比如“内容组”还是“数据组”底层每个小组再有自己的Supervisor负责调度组内的具体Worker。这样每个Supervisor手里的工具数量都控制在3到4个决策准确率会高很多。代价是多一轮模型调用但换来的是稳定性和可维护性。在实际项目里这个取舍是值得的。5.3 成本与速度优化不要什么都走LLM最后聊一个容易被忽视的点Supervisor的本质是“用模型做路由”但路由不一定都要模型来决策。比如用户输入里含有关键词“价格”可以直接用规则先判断应该走research_agent输入里出现“总结”“摘要”可以直接走write_agent。你可以把这种规则放在Supervisor之前作为一个快速路由节点命中规则就直走没命中再交给Supervisor。我在一个真实项目里做过测试引入前置规则后整个系统的响应时间下降了约40%token消耗也明显减少。原因是大量固定问法根本不需要“智能决策”。这个经验放在任何多智能体架构里都成立能用规则解决的不麻烦模型能用一个Worker解决的不启动两个Worker。结构越简单越好维护。写到这Supervisor Agent从概念到封装、到手写、到排障、再到进阶优化基本都覆盖了。如果让我说一条最值得记住的经验那就是多智能体系统的瓶颈通常不在模型而在“职责边界”和“状态流转”设计。Worker边界清晰一点Supervisor手里的工具描述准确一点状态里别塞无关消息系统稳定性就会有质的提升。这一集你按3.2那套代码跑通之后再回头加Agent、加工具应该会顺畅很多。下一集我打算聊聊如何评估这种多智能体系统的回答质量顺便给出一个可落地的测试集设计思路。