☰
LangChain Supervisor Agent:多智能体编排主管调度模式实战
2026/10/18 6:09:23 网站建设 项目流程

做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 环境准备:需要装哪些包

我用的是langchain+langgraph的组合,Python版本建议3.10以上。安装命令如下:

pip install langchain langchain-openai langgraph

我测试时用的是langgraph 0.2.x,如果你用的是更新的版本,部分API可能有细微调整,但整体流程一致。模型接口我用了langchain_openai.ChatOpenAI,你可以换成任何支持函数调用的模型,只要它返回tool_calls格式即可。

需要在环境变量里配好你的API Key,示例里我用占位符替代:

export OPENAI_API_KEY="sk-你的key"

3.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(model="gpt-4o-mini", temperature=0.1) def code_agent(state: MessagesState): system_prompt = "你是一名资深程序员。只负责解决代码问题,包括写代码、调试、解释代码。不要回答与代码无关的问题。" messages = [{"role": "system", "content": system_prompt}] + state["messages"] response = llm.invoke(messages) return {"messages": [AIMessage(content=response.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(content=response.content)]} def doc_agent(state: MessagesState): system_prompt = "你是文档专家。只负责写文档、总结内容、润色文本。不要处理代码或数据分析。" messages = [{"role": "system", "content": system_prompt}] + state["messages"] response = llm.invoke(messages) return {"messages": [AIMessage(content=response.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(model="gpt-4o-mini", temperature=0.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多智能体,欢迎对照这份代码跑一遍,遇到别的问题我们一起讨论。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询