多智能体(Multi-Agent)一直是LangChain生态里被聊得最多、也最容易翻车的主题。前面二十集把Chain、Memory、RAG、单Agent都过了一遍,到了这一集终于轮到整个系列的重头戏:多智能体编排。很多人做单Agent的时候觉得已经很酷了,但一旦面对真实业务——查外部资料、查内部库、做分析、生成报告,几个能力混在一起——单个Agent就开始力不从心。提示词变得臃肿、工具一多模型就会乱选、排查问题的时候你根本不知道它脑子里在想什么。这一集要拆的Supervisor Agent,就是多智能体编排里最经典、最好落地的一种形态:一个“主管”Agent不干活,专门负责看全局、做调度,把任务分给后面一群专业Agent去干。因为内容量不小,我会分成上下两集。这篇是上篇,主线是把Supervisor的架构逻辑讲透,并跑通一个最小闭环,让你从零把主管-工人的结构搭起来。适合已经玩过LangChain单Agent、想往多智能体方向走的人,也适合那种“一个Agent什么都干不了”的业务场景。
1. 为什么单Agent跑不动真实业务:多智能体的出现逻辑
1.1 单Agent的边界到底在哪
先把话说清楚:单Agent不是不能用,而是它有明确的容量上限。当你给一个Agent挂上五六个甚至十几个工具的时候,模型的选择压力会指数级上升。表面上它还能跑,但实际效果是:路由不准、调用频繁出错、上下文被各种工具结果塞满。我见过很多项目把工具列表从三个加到八个之后,准确率直接从九成掉到六成。这不是模型变笨了,而是你让同一个大脑同时处理太多不同类型的决策,它的注意力被稀释了。
还有一个更隐蔽的问题:单Agent的系统提示词会越来越长。为了让它知道“什么时候查资料、什么时候查库存、什么时候算价格、什么时候写报告”,你得不断往System Prompt里塞规则。规则一多,模型就开始“选择性失明”,排在后面的约束经常被忽略。到最后维护成本高得离谱——改一个业务逻辑,要连带考虑它对其他十几个逻辑的影响。
这时候多智能体就成了一种自然的选择。把一个大而全的系统,拆成几个各司其职的小系统,每个小系统只关心自己那一摊事。每个Agent的工具数量控制在两到三个,System Prompt也能写得更聚焦,模型的选择压力瞬间降下来了。
1.2 Supervisor模式到底解决什么问题
多智能体不是把几个Agent随便拼在一起。拼在一起之后,最核心的问题来了:任务怎么分配?
你可以让每个Agent都去听用户说话,然后抢着干活——这基本会乱成一锅粥。更靠谱的思路是:找一个“大脑”来做集中调度。这就是Supervisor Agent模式,它有一个很形象的英文名字:Centralized Control(集中式控制)。
在这个模式里,Supervisor本身不直接执行任务,它的唯一职责是“看懂用户请求,然后把请求派给最合适的下游Worker”。下游Worker执行完之后,把结果汇报回来,Supervisor再决定是继续派下一个任务,还是把最终答案还给用户。
用个生活化的类比:Supervisor像一个经验丰富的项目经理,Worker是团队里的各个专家。项目经理自己不一定能写代码,但他知道“这个问题该找后端、那个问题该找前端”。专家们干完活之后向项目经理汇报,项目经理整合信息后回复客户。客户不需要跟每个专家直接沟通,他只需要面对一个总接口。
这种架构解决的核心问题有三个:一是职责隔离,每个Worker只做一类事,不需要知道其他领域的东西;二是上下文聚焦,每个Agent看到的有效上下文只跟自己的任务相关;三是可扩展性,要加新能力,只需要新增一个Worker,并在Supervisor的“接线表”里注册一下,不用动其他Worker。
1.3 多智能体常见形态对比:为什么先学Supervisor
多智能体不止Supervisor一种玩法。业内常见的还有环形编排(Round-Robin)、层级编排(Hierarchical)、协议协作(如A2A风格)等。为什么绝大多数教程和项目都从Supervisor讲起?因为它最容易理解、最容易调试,也最适合业务系统的真实形态。
环形编排说白了就是几个Agent轮流发言,适合头脑风暴,但结果不可控;层级编排更像一个多层级的公司,适合超大系统,但复杂度高、出错了不好定位。Supervisor这种“一个主管带几个专工”的模式,跟真实组织架构高度相似,业务方也最容易接受。你自己写代码的时候也直观:图的节点就两类,一个Supervisor节点加若干个Worker节点,边也很清晰,几乎没有理解成本。所以我一直建议,多智能体入门就从Supervisor下手,先把这套跑稳了,再去玩别的花样,才有底气。
2. Supervisor的最小架构:角色、状态、消息流
2.1 架构里都有哪些角色
一套Supervisor模式里,角色拆开看就四样:Supervisor、Worker、Agent内工具、共享状态。
Supervisor是路由大脑,负责看用户请求并决定下一步去哪个Worker,或者结束对话;Worker是干活的,它通常是一个完整的Agent,有自己的System Prompt和专属工具集,能独立完成一小类任务;工具则是Worker执行任务时调用的具体能力,比如查数据库、调接口、扫文档;共享状态则描述的是这次对话里所有节点共同读写的信息。不同节点在同一轮对话里必须能互相看到对方的输出,否则没法协作。
有意思的地方是Worker通常也被实现成一个完整的Agent,而不是一个单纯的LLM调用。原因很简单:Worker收到的任务往往不是一句话就能解决的,它内部需要拆解、查资料、再总结。如果Worker只是个裸的工具调用器,那它就只能处理一步任务,任何需要多步思考的请求都得Supervisor自己去拆,这又退化回单Agent了。所以正确姿势是:Supervisor是个轻量级路由器,Worker是能够自主多步推理的专家。
2.2 共享状态的设计:messages、next、sender
多智能体能协作,靠的是一套大家都能读写的共享状态。LangGraph里这个状态通常用TypedDict定义。我常用的最小状态就三个字段:
from typing import Annotated, TypedDict, Literal import operator class State(TypedDict): messages: Annotated[list, operator.add] next: str sender: str看起来简单,但每个字段都有讲究。
messages用来存整个对话过程中的所有消息,它是一个会用operator.add聚合的列表,意思是每个节点返回的新消息会被自动追加到全局消息里。这样Supervisor和Worker都能看到完整的历史对话。
next字段是Supervisor写的“路由指令”。Supervisor每次决策后把自己的路由结果写进这个字段,图的边会根据这个字段决定下一步跳到哪个节点。
sender字段是最容易被忽略、但又特别关键的字段。它记录的是“当前这组消息是谁产生的”。为什么不直接靠messages里的角色判断?因为在LangGraph里,所有节点往同一个messages列表里追加消息,如果大家都只是追加,后面读上下文时根本分不清哪条消息是哪个Agent说的。加上sender之后,我们既可以判断消息归属,也可以让Supervisor在组装提示词时用sender信息区分“用户说了什么、内部哪个Worker回了什么”。
2.3 权限隔离:为什么Worker不能乱调用别人的工具
架构里一个必须提前定好的规矩是:Worker只能调用自己所属的工具集。很多人一开始图省事,把工具列表全挂在每个Worker身上,想着“反正模型会自己选”。结果就是干活最少的工具永远被冷落,某些工具会被不同Agent抢着调,上下文里工具选择的噪音极大。
Supervisor模式里的权限隔离做起来其实很直接:每个Worker在构建时只注入自己的工具列表,其他Worker的工具它根本看不见。LangGraph的工具调用机制下,模型只会从传入tools参数中选择,没传进去的工具它连名字都不知道,自然就不会乱调。
但你还需要考虑另一个层面:虽然Worker看不到别的Worker的工具,但Supervisor自己知道所有Worker的各自能力。Supervisor在决策时才需要全量信息。所以完整的设计是:Worker看到的工具是局部子集,Supervisor看到的“工具”其实是Worker的名字和它的能力描述。Supervisor不需要知道Worker内部的工具细节,它只需要知道“这个Worker能干什么”。
3. 动手搭建一套Supervisor最小闭环
3.1 场景设定与工具准备
纸上谈兵没用,直接上代码。我们搭一个“智能信息整合助手”,场景不复杂但足够说明问题。整套系统有两个Worker:
- 外部资讯查询员:负责检索公开的外部知识或资讯,对应工具是
search_external_info。 - 内部文档查询员:负责检索企业内部知识库,对应工具是
query_internal_doc。
用户的典型问法有两种,一种是“帮我查一下某技术的最新进展”,一种是“内部文档里关于某个规范是怎么写的”。有时候还有混合问法,比如“先查外部资料,再看内部有没有对应规范”。这就要Supervisor来拆分了。
先准备工具。为了演示,工具内部就用简单函数返回固定文本,你可以理解成真实项目里这些函数内部是接检索服务或者数据库查询。
from langchain_core.tools import tool @tool def search_external_info(keyword: str) -> str: """用于检索外部公开资讯,输入关键词,返回资讯摘要。""" # 真实场景这里是调用搜索API或知识服务 return f"外部资讯结果:关于【{keyword}】的公开信息显示,该方向当前主流方案是A和B。" @tool def query_internal_doc(doc_id: str) -> str: """用于查询企业内部知识库文档,输入文档编号,返回文档要点。""" # 真实场景这里是查询内部文档库 return f"内部文档【{doc_id}】要点:公司内部规范要求使用C方案,并遵循D流程。"工具写完之后,把每个工具的描述认真打磨一下。这步我后面会强调,这里先记住:工具描述直接决定Supervisor和Worker能不能正确理解“这个工具是干嘛的”。
3.2 定义Worker Agent
Worker我直接用LangChain的create_react_agent来构建。它接收模型、专属工具列表和一个System Prompt,返回一个可运行的Agent。这里每个Agent被工厂函数包一层,方便扩展。
from langchain.agents import create_react_agent from langchain_core.messages import SystemMessage from langchain_core.prompts import ChatPromptTemplate def make_worker( name: str, system_prompt: str, tools: list, model ): # 每个worker的system prompt聚焦于自己的业务范围 prompt = ChatPromptTemplate.from_messages([ ("system", system_prompt), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ]) agent = create_react_agent(model, tools, prompt) def node(state: State): result = agent.invoke({"input": state["messages"]}) return { "messages": [result["messages"][-1]], "sender": name, } return node两个Worker的System Prompt可以这样写:
external_worker_prompt = """你是一名外部资讯分析助手。你的职责是检索外部公开资讯并总结给用户。 只使用工具列表里的 search_external_info,不要猜测内部信息。 如果用户询问内部资料,请明确告知该任务应交给内部文档查询员。"""internal_worker_prompt = """你是一名企业内部知识库助手。你的职责是查询内部文档并总结要点。 只使用工具列表里的 query_internal_doc,不要编造外部信息。 如果用户询问外部动态,请明确告知该任务应交给外部资讯查询员。"""这里有个容易忽略的细节:result["messages"][-1]只是把Agent最后一条消息(通常是最终回复,也可能是工具调用请求)追加到共享状态里。如果你把中间过程也全部追加进去,上下文会快速膨胀。所以最小闭环里只取最终消息,这是一个经验选择。
3.3 定义Supervisor路由大脑
Supervisor本质上也是一个LLM节点,只不过它的输出不是自然语言,而是“路由决策”。为了让决策稳定,我用结构化输出,而不是直接让LLM吐自然语言再解析。
from langchain_core.pydantic_v1 import BaseModel, Field from typing import Literal from langchain_core.messages import HumanMessage class Route(BaseModel): """路由决策结果""" next: Literal["external_agent", "internal_agent", "FINISH"] = Field( description="下一个要执行的节点" ) def make_supervisor(model, workers_info: dict): def supervisor_node(state: State): system = SystemMessage(content=( "你是协调多个专业助手的主管。请根据用户最新需求和历史对话," "决定下一步该交给哪个助手处理。可选助手信息:\n" + "\n".join( f"- {name}: {desc}" for name, desc in workers_info.items() ) + "\n如果用户请求已经完成,输出FINISH。" )) # 只把supervisor关心的上下文传给它:最近几轮消息 recent_messages = state["messages"][-20:] result = model.with_structured_output(Route).invoke( [system] + recent_messages ) return {"next": result.next} return supervisor_node注意两点。第一,我把上下文裁成了最近20条消息,避免整个会话无限增长导致超Token。第二,结构化输出是Supervisor模式的关键,它强制模型只能从预定义节点里选一个,从根本上杜绝了“模型说了句废话导致后续没法路由”的尴尬。
3.4 组装StateGraph并编译运行
组装图是整个架构里最直观的一步:注册节点、定义边。
from langgraph.graph import StateGraph, END def assemble_supervisor_graph(supervisor_node, workers): graph = StateGraph(State) # 注册所有节点 graph.add_node("supervisor", supervisor_node) for name, worker_node in workers.items(): graph.add_node(name, worker_node) # supervisor是入口 graph.set_entry_point("supervisor") # supervisor到各个worker的条件边 def route_next(state: State): return state["next"] graph.add_conditional_edges( "supervisor", route_next, { "external_agent": "external_agent", "internal_agent": "internal_agent", "FINISH": END, } ) # worker都回到supervisor for name in workers: graph.add_edge(name, "supervisor") return graph.compile()运行的时候,直接用invoke传初始消息:
initial_state = { "messages": [HumanMessage(content="帮我查一下某项技术的外部最新进展,再看看内部文档里有没有对应规范")], "next": "", "sender": "user", } result = app.invoke(initial_state, {"recursion_limit": 30}) print(result["messages"][-1].content)注意我传了recursion_limit是30。这个参数很关键,下文会专门展开。
整个流程跑起来之后,你会看到它经历:Supervisor先决策 → 跳到外部Agent → 外部Agent返回结果 → 回到Supervisor → Supervisor再次决策 → 跳到内部Agent → 内部Agent返回结果 → 回到Supervisor → Supervisor判断已完成 → FINISH。这就是一个完整的主管-工人闭环。
4. 运行效果与关键参数调优
4.1 跑通一个完整查询流程后的现象观察
如果你跑通了上面的代码,第一次看到输出的时候,建议别急着高兴,先认真观察几个现象。
第一个现象是,每次Supervisor被调用,它都会重新审视整个上下文再做路由。这意味着它具备多步协同能力,能根据上一个Worker的输出决定下一步做什么。我实测这个能力在“先查外部再查内部”这类组合任务上至关重要,因为第二个Worker往往依赖第一个Worker查到的某些关键词。
第二个现象是,消息顺序有时候会让人困惑。因为所有节点都往同一个messages里追加,你可能会在列表里看到一条工具调用消息、一条AI回复消息、一条System消息混在一起。这时候sender字段就显得格外重要。生产环境里,我强烈建议每次调试时打印每轮的关键信息,包括当前节点名和sender。
第三个现象是,模型偶尔会偷懒。比如用户明明问的是内部资料,Supervisor却因为内部Agent的工具描述不够清晰,把任务派给了外部Agent。这种问题一般不是模型能力问题,而是工具描述和路由提示词没写好,我会在下面展开说怎么优化。
4.2 影响路由准确率的几个关键参数
Supervisor模式下,路由准确率几乎决定系统成败。我自己试下来,影响最大的参数有这么几个。
模型选择是第一位。路由这个动作本身不复杂,但对语义理解能力要求很高。能力弱的模型经常把FINISH和继续执行弄混,或者无视候选Agent列表输出一个奇怪结果。建议用支持Function Calling或结构化输出的主流大模型,路由效果才稳定。从工程角度,我倾向于把路由模型和Worker模型分开配置,路由用高能力模型、Worker用成本更低的模型,这样可以在保证准确率的同时控制成本。
Temperature一定要低。路由是确定性决策,不需要创造性。设成0或者很低的数值,可以大幅减少“模型偶尔脑袋一抽跳错分支”的概率。我见过有人用默认temperature跑Supervisor,结果同一个问题十次里有两三次路由不同,这种不确定性在业务系统里是不可接受的。
结构化输出一定要开。如果不用with_structured_output,模型可能返回“我认为应该让内部Agent来处理”这种自然语言文案,你还得用正则或者字符串匹配去解析,极易出错。开了结构化输出之后,模型必须严格按Schema输出,路由结果天然就合法。
还有一个小细节,各Worker的名字和描述要起得让模型一目了然。external_agent就比agent_1好一万倍,描述里要说清楚“这个Agent擅长什么、什么时候该选它、它绝不负责什么”。模型虽然不会像人一样逐字读描述,但描述里的关键语义确实会直接影响它的路由选择。
4.3 防止失控:限制循环与超时
多智能体系统最常见的失控模式就是无限循环。Supervisor在两个Worker之间来回跳,每次都产生新消息,对话永不终止。这既浪费Token,也拖垮响应时间。
LangGraph的recursion_limit是防失控的第一道防线。它限制的是图中节点总执行次数,默认值是25。在Supervisor模式下,每多跳一个Worker,图就多执行至少两个节点(Worker节点+回到Supervisor节点),多轮任务很容易超限。我一般把recursion_limit设到30到50,具体看任务链路多长。但设太高也有隐患,万一真有Bug,服务器会跑很久才报错。
第二种防失控手段是给Worker内部的Agent设置迭代上限。create_react_agent内部有max_iterations参数,控制它在思考-行动循环里最多跑多少步。如果Worker内部一直在调工具却始终不收敛,说明要么它的工具返回了无效信息,要么它的提示词没写清楚“何时停止”。把上限设到5到10能减少很多无谓消耗。
第三种手段是给整个流程设置合理的超时时间。LangGraph里可以在compile()或.invoke()时传timeout字段。生产环境建议按业务响应要求设置,比如30秒或60秒。超时后直接返回错误提示,而不是让用户无限等待。
5. 常见问题与排查实录
5.1 问题速查表
这一节把我在实际跑Supervisor模式时碰到过的高频问题整理成表格,方便你照着排查。
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| Supervisor路由结果不稳定 | 模型能力不足或temperature过高 | 换更强模型;把temperature设为接近0 |
| Worker反复调用同一个工具不收敛 | Worker的System Prompt没有定义停止条件 | 在Prompt里增加“得到结果后直接总结回复” |
| 上下文越来越长最后超Token | 所有节点把全部中间消息都append进state | 每个节点只追加最终消息,或定期裁剪历史 |
| 两个Worker来回跳不停 | 历史消息里缺少语义锚点,Supervisor判断不了进度 | 增加上下文消息数量;排查工具返回内容是否包含可判断的关键信息 |
| 模型无视可选节点输出非法值 | 没有启用结构化输出 | 使用with_structured_output约束路由结果 |
| 某个Worker永远不被选中 | 工具描述或Worker描述里缺少该场景的触发词 | 重写描述,把该Worker负责的业务典型问法写进去 |
调用报str object has no attribute content | 消息列表里混入了纯字符串而非消息对象 | 检查所有节点返回值,确保追加的是Message对象 |
5.2 踩过最深的坑:工具命名空间与上下文污染
两个坑都很有代表性,值得单独说。
第一个坑是工具命名空间冲突。我最早做实验时,给外部Agent定义了一个search工具,给内部Agent也定义了一个search工具。虽然每个Worker只挂了自己的search,但模型日志里偶尔会出现工具名调用混乱的情况,尤其在上下文较长时。后来排查发现,部分模型API在渲染工具列表时会存在同名工具被合并或互相干扰的情况。解决办法很简单:给工具名加业务前缀,比如external_search和internal_search,彻底隔离命名空间。这不只是命名习惯,而是一个必要的安全措施。
第二个坑是上下文污染。所有Agent的输出都会进入共享的messages列表,但如果Worker在推理过程中输出过长的中间结果(比如工具原始返回值有几千字),这些内容全被塞进messages后,下一轮任何Agent读上下文都会看到大量无关的垃圾信息。Token消耗大还是小事,关键是模型会被噪音干扰,路由准确性直线下降。我的经验是,只有在确实需要“跨Agent传递信息”时才保留中间消息;一般情况下,Worker的节点返回值只保留最终回复,工具原始结果留在Worker内部消费完就丢弃。如果你的业务确实需要把上游查询结果传给下游,可以单独设计一个字段来存结构化中间结果,而不是把所有杂文都塞给messages。
5.3 实测之后的三条心得
文章写到这里,分享几条我个人把Supervisor从Demo推到可用状态后的心得。
第一,Supervisor模式的成败很大程度不在框架代码,而在工具描述和Worker提示词。我调试过程中最有效的动作不是改代码,而是反复雕琢每个工具描述里的“什么时候用、什么时候别用”。模型是聪明的,但你需要把边界说清楚,否则它迟早用脚投票。
第二,不要把Supervisor做成神。它的路由能力再强,也不能替Worker完成领域判断。如果某个业务线特别复杂,更合适的做法是把它抽成独立的Worker,而不是把细节塞进Supervisor的提示词里。Supervisor保持轻量,才能保证路由长期稳定。
第三,调试时一定把中间过程打出来。我建议在开发阶段给每个节点加一段辅助打印,输出当前节点名、输入消息来源和路由结果。多智能体系统不像单Agent那样一条路走到黑,你不看节点跳转,根本不知道问题出在哪个环节。图结构可视化是辅助,真正靠得住的还是逐步打日志。搞定这一步,排错效率能提升一个量级。
这一集讲到这里,Supervisor模式的架构逻辑和最小闭环已经跑通了。上篇的核心目标就是让你彻底理解“主管调度、工人执行”这套设计,并具备自己搭一个最小系统的能力。下一集我们继续往下走,我会把Supervisor的进阶玩法——比如人工介入审批、动态路由、多级Supervisor嵌套、以及Worker内部状态隔离这些生产级细节——拆开揉碎讲清楚。