多智能体这块,Supervisor Agent是我在LangChain生态里用下来最接近“能解决实际问题”的架构。这个系列写到第二十二集,上一集我们把核心概念和第一个微缩版Demo跑通了,这集不再重复基础演示,专门聊那些让你敢把多智能体往生产环境里放的东西:路由决策的底层机制、子Agent的封装方式、State怎么设计、上下文怎么控制,以及那些文档里不会写、但线上一定会踩的坑。
适合看这集的人,我默认你已经用LangChain写过单Agent,对LangGraph里的StateGraph、node、edge这些有基本概念。如果只是刚听说LangChain,建议先回头翻前面几集。对有一定基础的朋友,这集可以当一份“Supervisor Agent工程化清单”来用。
1. 从单Agent到多Agent:为什么调度比堆工具更重要
1.1 单Agent的天花板在哪里
早期做Agent应用,最自然的思路是给一个Agent塞一大堆工具:查订单的、查物流的、算价格的、推荐商品的、处理退款的……工具列表越接越长。真跑到线上,你会发现模型面对二三十个工具的description时,选错工具的概率会明显上升。这不是模型变笨了,而是信息密度太高。每个工具的description就一行话,模型要在相似描述之间做区分,本身就是很模糊的决策。
除了误选工具,上下文污染是另一大痛。单Agent每轮都要把全部工具的返回结果带进历史,查物流返回一长串轨迹点,算价格返回一堆JSON,这些内容对最终回复很可能是噪音。等到用户问第二个问题时,之前的工具输出已经占了大量上下文窗口,模型回答问题反而更不聚焦。
Supervisor Agent解决的是这两个问题:不是“一个Agent做所有事”,而是“一个主管配上多个专职Agent”。主管只负责一件事——判断这个任务该交给谁。专职Agent只处理自己领域的工具和逻辑,输出经过结构化之后交回去。职责切分清楚了,每个环节的prompt都短了,工具选择准确率自然就上去了。
1.2 两种Supervisor架构形态对照
我见过两种主流实现方式,各有适用场景。
第一种是“工具封装式”。每个子Agent被包装成一个Tool,Supervisor通过工具调用的方式“直接呼叫”子Agent。实现最简单,几行代码就能跑通,适合做Demo和快速验证。但缺点也很明显:子Agent的内部对话状态没法保留,Supervisor把指令发给它、它返回一个字符串,下次再调用又得从头来。而且这种架构下,整个运行链路是一条调用链,不适合在中间插入人工审核或动态分支。
第二种是“图节点路由式”。用LangGraph把每个子Agent作为独立节点嵌进StateGraph,Supervisor也作为一个节点存在。Supervisor通过Tool Calling输出决策,决定下一步走哪个子Agent节点。子Agent执行完后把结果写回State,流程图回到Supervisor,由它决定继续调度还是收尾。这个模式状态可追踪、对话可回放、中间可以接入checkpointer做持久化,工程上明显更稳。
两种方式的对比可以看下面这张表:
| 维度 | 工具封装式 | 图节点路由式 |
|---|---|---|
| 实现复杂度 | 低,几分钟可跑通 | 中,需要理解图与条件边 |
| 子Agent多轮状态 | 无法保留 | 可由State统一维护 |
| 调试与回放 | 弱,只能看日志 | 强,可结合checkpointer回放 |
| 中间插入审核节点 | 困难 | 容易 |
| 生产环境推荐度 | 低 | 高 |
这集的实操部分以图节点路由式为主线。工具封装式适合做概念验证,但真正上了生产、需要稳定性和可观测性的时候,图的方式优势是碾压级的。
1.3 什么样的任务才适合Supervisor Agent
这里必须先泼一盆冷水:不是所有场景都该上多智能体。Supervisor Agent适合的是“有明确职能边界、任务可以拆成相对独立的子任务”的场景。比如客服系统,查订单、处理退款、答商品问题,边界清晰,拆给不同Agent没有问题。
不适合的场景是任务高度耦合、每个子任务都需要共享前后文才能推理的情况。比如让一个Agent写代码、另一个Agent不停重构、再一个Agent做代码审查,它们之间需要频繁交换大量代码片段,这时候来回调度的开销远大于收益,不如直接把问题交个一个强Agent自己处理。
我的经验是:先从单Agent开始,只有当“工具数量超过15个”或者“prompt里职责描述超过三块”的时候,才考虑拆成Supervisor架构。为了多智能体而多智能体,是最容易翻车的一条路。
2. 核心机制拆解:Supervisor到底是怎么“指挥”子Agent的
2.1 State设计:多智能体之间的公共数据面
LangGraph里所有节点都通过State交换数据。State本质上就是一个TypedDict,节点从里面读数据,把处理结果写回去。Supervisor架构里,State是最容易被忽视但最关键的设计点。
我常用的State结构长这样:
from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] session_id: str current_task: str visit_counts: dict这里messages是历史对话数据,通过add_messages这个reducer自动合并各个节点追加的消息。需要注意add_messages是按消息对象来合并的,不是简单往list里append。也就是说,同一个节点如果往State里写了两条内容,reducer会判断消息id,避免重复追加。
session_id和current_task是我为了工程化加的辅助字段。多Agent协作时,不是所有信息都适合塞进messages里。比如当前正在处理的任务编号、用户订单号,这些业务上下文单独存字段,比让每个子Agent从对话历史里自己找要可靠得多。
visit_counts用来统计每个子Agent被调用的次数,后面做循环防护会用到。这个字段不在LangGraph规范里,是我自己加的。多智能体系统刚上线时模型经常在几个Agent之间来回横跳,没这个字段根本没法优雅地跳出循环。
2.2 路由决策:靠Tool Calling实现的“指挥权移交”
Supervisor节点的核心机制非常简单:给Supervisor绑定一个只包含“子Agent名称”的工具列表,让模型每次都做一次工具调用决策。如果模型返回了tool_calls,主图根据工具名路由到对应子Agent;如果模型没有调用工具,说明任务可以收尾。
关键在Supervisor的System Prompt设计。我第一版做的Supervisor总想自己回答问题,后来发现必须显式告诉它:
SUPERVISOR_PROMPT = """ 你是一个任务调度主管。你的唯一职责是根据用户的请求,从可用执行者列表中选择一个合适的执行者来处理问题。 你绝对不要在未调用工具的情况下直接回答用户的实质性问题。 每个执行者都封装了具体的能力和内部逻辑,你需要把任务完整地转交给对应执行者。 执行者返回结果后,你需要整理结果生成最终回复。 如果执行者的结果已明确满足用户需求,不要再次调用工具,直接给出最终汇总答复。 """这段prompt强调了三件事:主管不亲自做事、必须调用工具、拿到结果后收尾。实际跑下来,这样能显著减少Supervisor“幻觉式回答”的概率。
条件路由函数是这里的技术细节。每个工具名需要映射到Graph节点名:
def route_supervisor(state: AgentState) -> str: last_message = state["messages"][-1] if not last_message.tool_calls: return "__end__" tool_name = last_message.tool_calls[0]["name"] mapping = { "order_query_agent": "order_agent", "after_sales_agent": "after_sales_agent", "product_info_agent": "product_agent", } return mapping.get(tool_name, "supervisor")我建议把“工具名”和“节点名”的映射统一收敛到一个地方维护,不要散落在多个函数里改成两套。早期我分散维护,结果某次重构只改了工具名忘了改映射,整条链路静默失效,排查到怀疑人生。
2.3 子Agent的封装要点:命名、描述、返回格式
子Agent的封装是整个架构里最考验细节的部分。命名要直白,description要写清楚触发条件和边界。比如“订单查询Agent”的工具描述可以这样写:
@tool def order_query_agent(task: str) -> str: """ 查询订单状态与物流信息。 仅当用户要求查看订单进度、物流轨迹、发货状态时使用。 输入参数task为完整的中文指令。返回结果为JSON字符串。 """ ...注意description里那句“仅当……时使用”。没这句,Supervisor很容易把“我要退款”这种请求误派给订单查询Agent。边界写清楚了,模型的工具选择准确率才能上来。
子Agent内部实现可以是简单的AgentExecutor,也可以是一个嵌套的LangGraph子图。我更推荐后者,因为子图可以直接复用主图的State管理能力,也能独立编译调试。如果子Agent内部逻辑很简单,直接用函数调用普通工具也行,没必要为简单的数据库查询建一个Agent,那是在浪费token。
子Agent的返回格式我建议统一为JSON字符串,至少包含result和requires_more_info两个字段。这样Supervisor拿到的执行结果是一个结构化数据,很容易判断任务是真的结束了还是需要追加追问。
return json.dumps({ "result": "订单20250110-001已签收,签收时间为昨天下午。", "requires_more_info": False, }, ensure_ascii=False)这个习惯后来帮我省了很多事。多Agent跑久了,只要从返回结果里一眼能看到requires_more_info,Supervisor就能做出“继续追问”还是“收尾”的决策,不需要靠模型自己瞎猜。
3. 实操:从零搭建一个Supervisor Agent项目
3.1 场景设计与Agent清单
用一个电商客服场景来演示。三个子Agent,一个Supervisor:
| Agent名称 | 职责边界 | 依赖能力 | 返回格式 |
|---|---|---|---|
| 订单查询Agent | 只处理订单状态、物流轨迹 | 订单数据库 | JSON字符串 |
| 售后处理Agent | 处理退款、退货、换货申请 | 售后工单系统 | JSON字符串 |
| 商品咨询Agent | 回答商品规格、库存、活动规则 | 商品知识库向量检索 | JSON字符串 |
Supervisor读取用户输入,判断这个请求属于哪个领域,路由到对应子Agent;子Agent执行完返回结构化结果;Supervisor汇总成最终回复。
3.2 搭建Supervisor节点与主图
核心代码分成四段:定义State、定义Supervisor节点、定义条件路由、组装图。
from typing import Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, SystemMessage from langchain_core.runnables import RunnableConfig # 1. 定义State class AgentState(TypedDict): messages: Annotated[list, add_messages] session_id: str current_task: str visit_counts: dict # 2. 定义Supervisor节点 def supervisor_node(state: AgentState, config: RunnableConfig): llm = get_llm(config) # 在config里传入模型实例 tool_names = list(agent_registry.keys()) tools = [agent_registry[name]["tool"] for name in tool_names] llm_with_tools = llm.bind_tools(tools) messages = [SystemMessage(content=SUPERVISOR_PROMPT), *state["messages"]] response = llm_with_tools.invoke(messages) return {"messages": [response]}这里有个容易踩的坑:llm_with_tools.invoke(messages)如果返回了tool_calls,这个AIMessage会作为新消息追加进State;如果没返回任何tool_calls,就是我们说的“收尾信号”。这个逻辑要在条件路由里判断,不能靠节点本身返回值判断。
3.3 子Agent节点接入主图
每个子Agent需要一个包装函数,把Agent内部的逻辑封装成LangGraph节点:
def order_agent_node(state: AgentState, config: RunnableConfig): last_ai = state["messages"][-1] # 从tool_call中取出传递给子Agent的任务指令 if last_ai.tool_calls: task = last_ai.tool_calls[0]["args"].get("task", "") else: task = state["current_task"] result = run_order_query(task, session_id=state["session_id"]) # 把结果以ToolMessage形式追加回State tool_call_id = last_ai.tool_calls[0]["id"] return {"messages": [ToolMessage(content=result, tool_call_id=tool_call_id)]} # 类似地定义售后节点和商品节点注意ToolMessage的tool_call_id必须和Supervisor发出的工具调用id一一对应。LangGraph在编译时会做工具调用的校验,如果AIMessage带了一个tool_call,但状态里没有对应的ToolMessage补上,graph执行到后续节点就会报校验错误。这是我一开始最容易疏忽的地方。
子Agent不一定要用LLM,如果订单查询是确定性逻辑,直接在函数里查数据库就行。真正需要模型推理的子Agent,才在里面挂AgentExecutor或子图。很多开发者误以为子Agent必须是完整的Agent,其实没必要——能用规则解决的子任务,用规则的稳定性和成本都优于走模型。
3.4 编译运行与完整链路验证
组装主图和编译:
graph = StateGraph(AgentState) graph.add_node("supervisor", supervisor_node) graph.add_node("order_agent", order_agent_node) graph.add_node("after_sales_agent", after_sales_node) graph.add_node("product_agent", product_agent_node) graph.set_entry_point("supervisor") graph.add_conditional_edges("supervisor", route_supervisor) graph.add_edge("order_agent", "supervisor") graph.add_edge("after_sales_agent", "supervisor") graph.add_edge("product_agent", "supervisor") app = graph.compile()这里有一个容易忽略的细节:add_conditional_edges的route_supervisor返回值一定要包含"__end__",这样模型在决定不调用任何工具时,执行流程才会走到终点。如果不加,图会在理想流程里卡死。
调用方式:
result = app.invoke( {"messages": [HumanMessage(content="我订单号20250110-001到哪了")]}, config={"configurable": {"thread_id": "session-001"}, "recursion_limit": 40} ) print(result["messages"][-1].content)recursion_limit建议从40起步。太小不够多Agent轮转,太大又容易陷入死循环消耗token。先跑几轮真实对话,观察图正常需要几步,在这个基础上加50%余量是比较合理的设置。
我建议第一次跑这个Demo时,把verbose=True打开或者干脆打印每个节点执行前后的State摘要。看到Supervisor先选了order_agent,order_agent返回ToolMessage,再到Supervisor生成最终答复,整个链路就算通了。
4. 进阶优化:让多智能体系统真正具备生产条件
4.1 上下文窗口与消息裁剪策略
多Agent协作最大的暗坑是上下文膨胀。Supervisor每个决策点都会读取state["messages"]里的全部历史,子Agent节点也一样。跑十几轮之后,消息列表里堆满了系统提示、工具返回、中间结果,token消耗呈指数级增长,而且模型在长上下文里抓关键信息的精度会下降。
我采用的策略是分层裁剪:
- 传给Supervisor的消息列表只保留最近6轮对话和最新工具返回,更早的历史做摘要拼在SystemMessage里
- 子Agent只需要读取自己所属的那段任务描述,不传给它们全量历史
- 业务上下文单独存字段,比如
current_task,不依赖模型从历史里自行提取
实现一个简单的裁剪函数:
def trim_messages(messages, max_rounds=6): user_turns = [m for m in messages if isinstance(m, HumanMessage)] if len(user_turns) <= max_rounds: return messages # 超过轮次时,只保留最近max_rounds轮,同时对前面的内容做摘要 recent = messages[-max_rounds * 3:] # 粗略裁剪 return recent这个函数思路是粗暴的,但它让token消耗从“线性增长”变成“有上限”。真要做得更精细,可以在多Agent协作里接入摘要节点,每几轮把早期对话压成一条SystemMessage,但这会引入额外的摘要延迟,用户多的时候不划算。我的原则是:能用规则截断的,先不用模型摘要。
4.2 循环防护与退出机制
多Agent的死循环问题是每个上线Supervisor架构的人都会遇到的。现象是:Supervisor反复调用同一个子Agent,子Agent每次都返回类似结果,Supervisor又选择继续调用它,直到撞上recursion_limit报错。
防循环不能只靠prompt,要用代码兜底。我在State里放了一个visit_counts字典,路由函数里做检查:
def route_supervisor(state: AgentState) -> str: last_message = state["messages"][-1] if not last_message.tool_calls: return "__end__" tool_name = last_message.tool_calls[0]["name"] mapping = { "order_query_agent": "order_agent", "after_sales_agent": "after_sales_agent", "product_info_agent": "product_agent", } target = mapping.get(tool_name, "supervisor") # 同一分支连续执行次数限制 prev = state["visit_counts"].get(target, 0) if prev >= 2: return "__end__" # 强制收尾 state["visit_counts"][target] = prev + 1 return target这里限制单个子Agent连续触发不超过2次。第3次时强制结束,宁可让最终答复质量差一点,也不能让系统卡死。对用户来说,一个“我暂时无法处理”的明确答复,比一个无限转圈的系统要体面得多。
4.3 调用链追踪与性能观测
多Agent系统出问题时,最让人头疼的是“不知道哪一步出了问题”。因为一个请求可能在Supervisor和两三个子Agent之间转了三四轮,任何一环出错都会导致最终结果异常。
我习惯在每个节点里加回调,把执行信息打到结构化日志:
from langchain_core.callbacks import BaseCallbackHandler class TraceCallback(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): node = kwargs.get("node_name", "unknown") log_info(f"[START] {node} prompts_len={sum(len(p) for p in prompts)}") def on_llm_end(self, response, **kwargs): node = kwargs.get("node_name", "unknown") log_info(f"[END] {node} tokens={response.llm_output.get('token_usage', {})}")这样每个请求可以生成一条完整调用链:哪个节点调了模型、输入多少token、输出多少token、耗时多少。排查性能瓶颈时,只要把日志按耗时倒序排,一眼就能看到是哪个Agent在拖时间。
还有一个容易被忽视的点:每个子Agent应该独立记录trace ID,与主请求关联。否则多个用户同时请求时,所有Agent日志混在一起完全没法区分。我在项目里用session_id加自增序号拼一个trace_id,每次节点调用都带上,排查问题效率翻了不止一倍。
5. 常见问题与排查技巧实录
5.1 典型故障对照表
| 现象 | 可能原因 | 解决动作 |
|---|---|---|
| 主管每次选错子Agent | 工具description边界不清 | 重写description,加“仅当…时使用” |
| 子Agent执行后图报校验错误 | ToolMessage未与tool_call_id配对 | 检查每个tool_call是否有对应ToolMessage |
| 主管反复调用同一子Agent | prompt里没写收尾规则 | 显式告诉主管“拿到结果后就汇总,不再调用工具” |
| 上下文膨胀导致变慢/超限 | 全量历史塞给每个节点 | 裁剪最近N轮,业务上下文单独存字段 |
| 子Agent说“不知道用户问什么” | 子Agent没看到HumanMessage | 检查子Agent读取State的字段路径 |
| 系统在几个Agent间死循环 | 无循环防护 | 用visit_counts限制同一分支连续次数 |
这些是我实际项目里每隔一段时间就会遇到一次的经典问题。大部分不是代码逻辑错误,而是多Agent交互的“软”问题:prompt没写清楚、上下文传递不完整、边界条件没约束住。
5.2 三个具体的排查案例
第一个案例是Supervisor每次把售后问题派给订单查询Agent,用户投诉“我说退款却让我查物流”。排查到最后发现是订单查询Agent的description开头写了“查询所有订单相关信息”,这个“所有”让模型产生了误判。改成“查询订单状态与物流轨迹,仅限物流与进度查询,不处理退款相关请求”之后,误派率立刻降下来。工具description是模型选择的重要依据,措辞一定要精确到“什么时候不该用”。
第二个案例是子Agent跑完返回结果后,图直接报tool_calls校验错误。原因是子Agent节点返回的ToolMessage里的tool_call_id和Supervisor发起的AIMessage里的id不一致。后来我封装了一个通用的agent_node包装函数,自动从最后一条AIMessage里提取tool_call_id,再构造ToolMessage,这个问题再也没出现过。
第三个案例更隐蔽:系统在客服场景里正常,换到售后场景之后频繁卡死。看日志发现路由函数里工具名映射在客服场景配置里存在,但售后场景没有注册这个工具映射,导致返回了"supervisor"这个兜底分支,结果Supervisor又发出同样的工具调用,形成死循环。最终收集所有工具映射到统一的注册表,在任何场景里都从注册表取映射,兜底分支也改成了直接收尾而不是回到Supervisor。
5.3 工程红线与配置建议
- 第一个红线:所有子Agent节点的异常必须被catch,返回错误字符串而不是抛出异常。多Agent系统里一个节点抛异常会让整条链路崩,用户只能得到“系统错误”。
- 第二个红线:不要让Supervisor看到底层子Agent的私有工具。假如订单Agent内部有“更新订单状态”的工具,而Supervisor绑定了这个工具,它可能绕过订单Agent直接调用,造成不可控操作。工具权限边界要严格隔离。
- 第三个红线:模型调用要设超时和重试。多Agent系统的一次用户请求可能触发5-8次模型调用,单次卡住整体全卡。
配置方面,我建议Supervisor和子Agent可以选不同温度参数。Supervisor做路由决策,温度设为0最合适;子Agent如果承担生成类任务,温度可以稍微调高。这个细节很多人忽视——调度的随机性越少越好,内容生成的随机性可以适度放宽。
最后再分享一点个人心得
这个系列写到这一集,Supervisor Agent基本就能在你们自己的项目里落地了。我个人实际操作中的最大体会是:不要把Supervisor当成一个“万能分发器”,从设计上它只是一个状态机里的路由决策节点。它的价值在于让每个子Agent保持简单和专注,如果子Agent本身又臃肿又庞杂,那这套架构的收益会大幅缩水。
第一版落地时,先让Agent数量少一点、职责边界绝对清晰,把链路跑通,再逐步增加新的子Agent。多Agent系统是一个典型的“开始容易、调优较难”的架构,对token消耗和排查功夫的要求都比单Agent高。如果你手头正在调多智能体,先别急着加Agent,把一轮调用的日志和State流转看明白,可能比盲目堆架构更有效果。
调试的时候还有个小技巧:把Supervisor的SystemMessage里加一条“你最后一次行动必须是整合结果,而不是调用工具。”这句话能明显减少结尾阶段的多余工具调用。多Agent的坑大多是调度和上下文层面的,抓住这两点,这套架构离生产可用就不远了。