LangChain Supervisor Agent：多智能体编排主管调度模式实战

做Agent开发的人迟早会遇到一个问题任务一复杂单个大模型根本顶不住。工具堆了一大堆提示词越写越长模型反而开始“摆烂”——答非所问、工具调用错乱、上下文被无关信息塞满。我在用langchain做多智能体编排时最常用也最容易被忽略的方案就是Supervisor Agent一个负责“派活”的主管节点带着一群各怀绝技的子Agent干活。这篇是系列第二十一集的上篇我把从原理到第一版能跑的完整过程拆开讲希望能帮你在自己的项目里少走弯路。很多刚接触langchain的人会误以为多智能体就是“多几个Agent串在一起跑”实际上真正的多智能体协作需要一个调度核心。Supervisor Agent恰恰就是这个核心。下面我会从为什么需要它、它的底层机制、我第一版实现踩过的坑以及后续可以怎么扩展这几个方面展开内容偏实战代码是完整的。1. 从单智能体到多智能体Supervisor模式解决的三个根本问题1.1 单智能体的边界工具堆得越多决策反而越乱先说一个直观现象。给单个Agent挂上五六个工具一开始效果还行用户问什么都能接住等工具涨到十几个问题就来了。模型在选择“下一步调用哪个工具”时要从海量工具描述里做判断注意力被稀释经常出现类似“订单查询和物流查询拿不准该用哪个”的尴尬。我在一个客服项目里做过实验工具数量从6个提到15个工具调用的准确率从95%掉到78%而修复手段只有反复修改工具描述。这不是提示词写得不细而是结构性缺陷。单智能体的决策空间随着工具数量线性膨胀而模型能同时关注的信息有限。你当然可以把所有工具合并成一个超级工具但那又回到了“一个函数里塞一大堆if-else”的老路任何新增需求都要改动中心逻辑维护成本爆炸。更麻烦的是工具之间会有隐性的依赖关系。比如“查询订单”和“查询物流”明明是两步但用户一句话抛出来时模型要么只做一步要么把两步揉成一团。这时候就该考虑把能力拆成多个独立的子Agent每个子Agent只对一个小领域负责然后用一个主管来统筹。1.2 三种多智能体编排方式Sequential、Router、Supervisor我见过不少项目自称“多智能体”其实就是三种形态顺序执行、路由器、Supervisor。这三者的区别直接决定了系统能处理复杂度的上限。编排方式调度时机典型实现适用场景Sequential顺序链运行前固定A跑完必跑BLangChain Chain流程固定、步骤顺序明确Router路由器运行开始一次性选择分支提示词分类 if/else输入分类、多个互斥业务线Supervisor主管每轮动态决定可能多次切换LangGraph LLM 路由复杂任务、步骤不可枚举、需交叉调用Sequential最简单但死板。你写了一个“查库存→生成报表→发送邮件”的链那所有请求都必须走完这三步哪怕用户只是想查库存。Router比Sequential灵活一点可以根据输入选择一个分支去执行但它只做一次决策剩下路是固定的。Supervisor完全不同。它让一个主管节点在每一轮都看一眼当前对话状态然后决定“现在该由哪个子Agent上场”甚至可以反复切换。比如用户问“先查一下A仓库的库存再根据库存生成补货报表”Supervisor可以先把控制权交给库存Agent拿回结果后再把控制权交给报表Agent。这是Router那种一次性分流做不到的。1.3 Supervisor和Router的本质区别调度权交给了谁Router把调度逻辑固化在“输入特征→分支”的映射里本质上还是用代码规则或一次模型分类来定方向。Supervisor则把调度权交给一个常驻的LLM决策节点由它根据上下文动态分配任务。你可以把Router理解成火车站的分岔道火车一旦驶入某条轨道就不会回头而Supervisor是机场塔台它可以让同一架飞机在多个停机位之间反复调整直到任务完成。这个区别带来的最大变化是子Agent之间可以进行多轮协作。A把结果给主管主管判断还需要B参与就让B接手B返回后主管可能又觉得需要A补充就再把控制权交回A。这种动态协调能力才是多智能体真正区别于“多个单体Agent串在一条链上”的地方。LangChain生态里能落地这种模式的核心引擎是LangGraph下面我展开讲它怎么支撑起Supervisor的调度。2. Supervisor Agent的底层逻辑LangGraph状态机如何撑起主管调度2.1 LangGraph在LangChain生态中的位置多智能体的地基先回顾一下LangChain传统的Chain结构它把Prompt、Model、OutputParser串成一条管道数据从一头流到另一头。这种设计处理线性任务很顺但要表达“多节点之间有条件的跳转”就捉襟见肘了。LangGraph就是为了拆掉这条线性锁链而出现的。LangGraph把自己定义为“窗口版的LangChain”窗口代表的是状态图执行。它用StateGraph定义一个有向图每个节点是一个函数函数接收当前全局状态返回更新后的字段节点之间用边连接边可以是固定边也可以是条件边——条件边的走向由某一个函数动态决定。多智能体的执行过程天然就是图Supervisor节点、子Agent节点、结束节点以及它们之间的条件跳转。LangGraph把最底层的状态传递、循环、终止条件都处理好了我们只需要写节点函数和定义路由规则。这也是Supervisor Agent能优雅落地的原因主管节点的下一步动作不是写死的而是由条件边根据模型输出决定。2.2 状态与消息传递主管如何知道派谁Supervisor Agent的图里最核心的数据结构是全局状态State。我用的状态通常是一个MessagesState——本质就是保存了对话历史的messages列表。每个节点执行完都往messages里追加自己的输出比如一条AIMessage下一个节点就能从状态里读到目前为止的全部上下文。主管节点要做的事很简单读messages结合系统提示词里对各个子Agent的描述调用一次大模型让模型返回“下一步应该调用哪个Agent”的工具调用结果。这里的关键是我们把每个子Agent都注册成一个工具工具的名字、描述就是子Agent的“名片”。举个例子主管的系统提示里写了三个工具code_agent擅长写代码、调试程序、解释代码片段。data_agent擅长数据分析、SQL查询、生成统计报告。doc_agent擅长写文档、总结内容、润色文本。模型收到用户问题后如果问题涉及“帮我统计这批数据”它输出的tool_calls就会带data_agent。这个输出会被我们的路由函数读到从而把图流转到data_agent节点。子Agent执行完返回结果追加状态图又回到主管节点主管再看看是不是要派下一个活。2.3 把控制流从代码手里抢回来LLM作为决策器有人会问这跟普通function calling有什么区别本质上确实一样但关键是“工具”变成了“子Agent整体”。普通function calling里每个工具是一个函数执行完把返回值给模型Supervisor里每个工具是一个带独立上下文、独立系统的子Agent它可能有自己的记忆、自己的工具列表、自己的循环逻辑。主管只负责“点将”不负责“替子Agent干活”。这种做法把控制流从代码手里抢了回来交给了自然语言。传统方式是写一堆if user_input contains sql then go_to_data_agent这种规则永远写不全Supervisor的方式是让模型根据自然语言意图自己决定路由你只需在提示词里写清楚每个子Agent的职责边界。代价是引入了不确定性。模型可能选错Agent、可能陷入循环、可能因为提示词描述歧义反复横跳。这些坑我后文会重点讲。先来看一个完整可运行的实现。3. 手把手实现一个Supervisor Agent代码准备到第一版跑通3.1 环境准备需要装哪些包我用的是langchainlanggraph的组合Python版本建议3.10以上。安装命令如下pip install langchain langchain-openai langgraph我测试时用的是langgraph 0.2.x如果你用的是更新的版本部分API可能有细微调整但整体流程一致。模型接口我用了langchain_openai.ChatOpenAI你可以换成任何支持函数调用的模型只要它返回tool_calls格式即可。需要在环境变量里配好你的API Key示例里我用占位符替代export OPENAI_API_KEYsk-你的key3.2 定义三个子Agent节点这里的核心思想是每个子Agent也是一个节点函数入参是当前状态内部构建自己的提示词和模型调用最后返回一条消息。先说第一个子Agent代码助手。from typing import Annotated, Literal from typing_extensions import TypedDict from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, MessagesState, END # 基础模型后续子Agent都会复用 llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) def code_agent(state: MessagesState): system_prompt 你是一名资深程序员。只负责解决代码问题包括写代码、调试、解释代码。不要回答与代码无关的问题。 messages [{role: system, content: system_prompt}] state[messages] response llm.invoke(messages) return {messages: [AIMessage(contentresponse.content)]}注意返回结构必须是字典键为messages值是消息列表。这样LangGraph会把这条消息追加到全局状态中。同理定义数据助手和文档助手def data_agent(state: MessagesState): system_prompt 你是数据分析师。只负责数据处理、SQL编写、统计汇总。不要参与代码编写和文档写作。 messages [{role: system, content: system_prompt}] state[messages] response llm.invoke(messages) return {messages: [AIMessage(contentresponse.content)]} def doc_agent(state: MessagesState): system_prompt 你是文档专家。只负责写文档、总结内容、润色文本。不要处理代码或数据分析。 messages [{role: system, content: system_prompt}] state[messages] response llm.invoke(messages) return {messages: [AIMessage(contentresponse.content)]}这样写有个隐患子Agent会把主管发给它的“任务请求”当成用户消息可能会主动问用户而不是直接干活。为了让它直接执行可以在子Agent的提示词里加一句“你收到的上一条消息是主管派给你的任务直接执行不要反问”。我后文调试里会再展开。3.3 定义Supervisor节点和路由逻辑主管节点的难点在于它要输出“接下来该调用谁”的决定。我们的做法是把三个子Agent注册成工具绑定到同一个模型上让模型的tool_calls结果决定图走向。先定义三个工具from langchain_core.tools import tool tool def code_agent_tool(query: str) - str: 处理代码相关任务包括写代码、调试、解释代码。 return query tool def data_agent_tool(query: str) - str: 处理数据分析、SQL、统计相关任务。 return query tool def doc_agent_tool(query: str) - str: 处理文档撰写、内容总结、文本润色相关任务。 return query这三个工具只是占位真正执行时并不调用它们而是利用它们的名称和描述让模型产生tool_calls。接下来定义主管节点from langchain_core.runnables import RunnableLambda model ChatOpenAI(modelgpt-4o-mini, temperature0.1) supervisor_llm model.bind_tools([code_agent_tool, data_agent_tool, doc_agent_tool]) def supervisor(state: MessagesState): system_prompt ( 你是主管。你管理着以下三个Agent\n - code_agent负责写代码、调试、解释代码。\n - data_agent负责数据分析、SQL、统计。\n - doc_agent负责写文档、总结、润色。\n 根据用户请求调用合适的Agent来完成任务。一次只调用一个Agent 如果任务完成或没有合适的Agent直接回复结束。 ) messages [{role: system, content: system_prompt}] state[messages] response supervisor_llm.invoke(messages) return {messages: [response]}注意这里返回的消息直接是模型的原始响应里面可能包含tool_calls。如果模型决定结束它不会带tool_calls而是正常文本回复。然后定义路由函数它从主管节点返回的消息里解析工具调用名def route_to_agent(state: MessagesState) - Literal[code_agent, data_agent, doc_agent, END]: last_message state[messages][-1] if not last_message.tool_calls: return END tool_call last_message.tool_calls[0] # 工具名对应节点名 return tool_call[name].removesuffix(_tool)这里有个细节我之前工具名定义成code_agent_tool所以要剥离_tool后缀才能对应到节点名。你也可以直接让工具名等于节点名省去转换。3.4 建图、编译、运行用StateGraph把上述节点串起来workflow StateGraph(MessagesState) workflow.add_node(supervisor, supervisor) workflow.add_node(code_agent, code_agent) workflow.add_node(data_agent, data_agent) workflow.add_node(doc_agent, doc_agent) workflow.add_edge(__start__, supervisor) workflow.add_conditional_edges(supervisor, route_to_agent, { code_agent: code_agent, data_agent: data_agent, doc_agent: doc_agent, END: END, }) # 子Agent执行完后回到主管由主管决定是否继续 for agent in [code_agent, data_agent, doc_agent]: workflow.add_edge(agent, supervisor) agent_graph workflow.compile()编译好后就可以用invoke或stream运行了from langchain_core.messages import HumanMessage result agent_graph.invoke({ messages: [HumanMessage(content帮我统计用户反馈中提到价格这个词的次数然后用一句话总结)], }) for message in result[messages]: print(message.type, message.content)输出里你会看到supervisor的tool_calls、data_agent的回答、再回到supervisor调doc_agent最后supervisor结束。这个过程我可以拍胸脯说是第一次跑通时最让人爽的时刻。4. 实测效果与参数调试我的第一版Supervisor表现如何4.1 第一轮测试输出主管是怎么把一句话拆成两票活的我实际运行了一段测试输入是“统计用户反馈里提到价格这个词的次数然后用一句话总结大家的态度。”图执行的大致路径是输入 - supervisor supervisor 决定调用 data_agent_tool data_agent 返回“共23条反馈提到价格。” - supervisor supervisor 决定调用 doc_agent_tool doc_agent 返回“用户对价格普遍比较敏感希望更透明的定价。” - supervisor supervisor 返回最终总结“统计显示23条反馈提到价格普遍反映更关注价格透明度。” - END这段路径完美展示了Supervisor的价值它不是一次性把所有活都派出去而是等数据Agent的结果回来再决定是否需要文档Agent总结。这在Router模式里要写死两步都很麻烦因为不是每个问题都需要两步。4.2 实测中翻车的三个典型问题翻车才是开发常态。我把第一版里踩过的三个典型问题列出来每个都能让你卡半天。第一个问题是子Agent“越权”。因为子Agent也接到了完整对话历史它经常会从历史里找到主管的工具调用记录然后自己也试图调用工具甚至假装自己是主管去给另一个子Agent下命令。解决办法有两个一是不要让子Agent绑定任何工具它只做纯文本回复二是在子Agent的系统提示里明确“你仅回答任务不调度其他Agent”。第二个问题是循环卡死。有一次主管反复调度code_agent因为code_agent每轮都输出“我还在检查……”但没给出结论主管觉得任务没完成就继续派它。根源是子Agent缺少“结束任务”的信号。我后来在子Agent提示词里加了一句“无论结果如何必须在第一轮回复中给出明确结论或能交付的最优内容”循环才明显减少。第三个问题是上下文爆炸。每次子Agent执行完结果都会追加到messages里。任务步骤多、Agent数量多时很快会把上下文窗口撑爆。我的临时方案是在每个子Agent返回给主管之前把过长的输出摘要成200字以内的要点正式的方案应该引入记忆压缩或定期裁剪这在下集可以细聊。4.3 参数和提示词的调试心得运行稳定后我总结出几个很实际的经验temperature要压低。主管路由和子Agent执行都用0.1到0.3之间太高会让模型乱选Agent太低又会变得机械。工具描述的措辞非常关键。写“处理代码相关任务”比“代码”更利于模型判断写清“什么情况下不用”比“什么情况下用”更能避免误调。如果模型不支持工具调用可以退而求其次让主管以JSON格式输出{next: agent_name}然后解析。效果略差但也能跑通。尽量让图有“出口”。我给主管画了“如果用户只说谢谢那就直接回复结束”这样的终止路径避免模型礼貌性地派活。这几点调完第一版才算真正可用。5. 再往前一步Supervisor模式的可扩展玩法5.1 从“一个主管”到“主管带主管”Supervisor的图结构天然支持分层。比如一个总主管下面有“代码主管”和“数据主管”而“代码主管”自己又是一个子图下面带着“前端Agent”和“后端Agent”。LangGraph允许你把一个编译好的图作为另一个图的节点这叫做嵌套图。我试过两层结构外层主管只负责判断该让哪个子主管接手内层子主管再细分任务灵活性直接翻倍。分层之后要注意的是状态隔离。子图里的消息可能和外层状态相互干扰最好在子图内用独立的状态字段只把必要的结果回传给外层。5.2 加入人工审批Human-in-the-loop有些动作不适合让模型完全放手。比如主管决定要删数据、发邮件、执行高成本计算应该先暂停一下让真人确认。LangGraph提供了interrupt机制可以在节点执行前中断图的运行等待外部输入后恢复。我在这套图上加了一个“审批节点”主管如果决定调用高风险工具就先走到human_approval节点由人工点击确认才继续。这个机制特别适合金融、运维等场景也是Supervisor Agent从“演示”走向“生产”的关键一步。5.3 我的个人建议回头看我做这个Demo的整个过程最大的体会是Supervisor不是万金油别为了多智能体而上多智能体。如果你的任务流程固定、分支有限Router就能解决如果只是简单的串行处理Sequential都够。只有当任务里确实存在“多步骤、可切换、依赖判断”的空间时Supervisor模式才值得引入。另外提示词的工程量比模型能力还重要。主管的“识人能力”取决于你给他的Agent描述够不够准确子Agent的执行能力取决于你给它的边界够不够清楚。在你怀疑“是不是模型太笨”之前先怀疑自己的描述是不是太模糊。我目前这套实现还在迭代接下来会继续把Supervisor的上下文记忆、子Agent之间的共享状态、以及与外部工具链的更大范围集成补完。这一集先到这里下集我会把更复杂的调度场景和状态共享细节展开来写。如果你也在玩langchain多智能体欢迎对照这份代码跑一遍遇到别的问题我们一起讨论。