多智能体这块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被包装成一个ToolSupervisor通过工具调用的方式“直接呼叫”子Agent。实现最简单几行代码就能跑通适合做Demo和快速验证。但缺点也很明显子Agent的内部对话状态没法保留Supervisor把指令发给它、它返回一个字符串下次再调用又得从头来。而且这种架构下整个运行链路是一条调用链不适合在中间插入人工审核或动态分支。第二种是“图节点路由式”。用LangGraph把每个子Agent作为独立节点嵌进StateGraphSupervisor也作为一个节点存在。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_asciiFalse)这个习惯后来帮我省了很多事。多Agent跑久了只要从返回结果里一眼能看到requires_more_infoSupervisor就能做出“继续追问”还是“收尾”的决策不需要靠模型自己瞎猜。3. 实操从零搭建一个Supervisor Agent项目3.1 场景设计与Agent清单用一个电商客服场景来演示。三个子Agent一个SupervisorAgent名称职责边界依赖能力返回格式订单查询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(contentSUPERVISOR_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_idstate[session_id]) # 把结果以ToolMessage形式追加回State tool_call_id last_ai.tool_calls[0][id] return {messages: [ToolMessage(contentresult, tool_call_idtool_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时把verboseTrue打开或者干脆打印每个节点执行前后的State摘要。看到Supervisor先选了order_agentorder_agent返回ToolMessage再到Supervisor生成最终答复整个链路就算通了。4. 进阶优化让多智能体系统真正具备生产条件4.1 上下文窗口与消息裁剪策略多Agent协作最大的暗坑是上下文膨胀。Supervisor每个决策点都会读取state[messages]里的全部历史子Agent节点也一样。跑十几轮之后消息列表里堆满了系统提示、工具返回、中间结果token消耗呈指数级增长而且模型在长上下文里抓关键信息的精度会下降。我采用的策略是分层裁剪传给Supervisor的消息列表只保留最近6轮对话和最新工具返回更早的历史做摘要拼在SystemMessage里子Agent只需要读取自己所属的那段任务描述不传给它们全量历史业务上下文单独存字段比如current_task不依赖模型从历史里自行提取实现一个简单的裁剪函数def trim_messages(messages, max_rounds6): 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主管反复调用同一子Agentprompt里没写收尾规则显式告诉主管“拿到结果后就汇总不再调用工具”上下文膨胀导致变慢/超限全量历史塞给每个节点裁剪最近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的坑大多是调度和上下文层面的抓住这两点这套架构离生产可用就不远了。 
长河君 · 2026/10/19 12:50:24 · 阅读 1.2k Supervisor Agent 工程化指南：多智能体路由与状态管理实战
多智能体这块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被包装成一个ToolSupervisor通过工具调用的方式“直接呼叫”子Agent。实现最简单几行代码就能跑通适合做Demo和快速验证。但缺点也很明显子Agent的内部对话状态没法保留Supervisor把指令发给它、它返回一个字符串下次再调用又得从头来。而且这种架构下整个运行链路是一条调用链不适合在中间插入人工审核或动态分支。第二种是“图节点路由式”。用LangGraph把每个子Agent作为独立节点嵌进StateGraphSupervisor也作为一个节点存在。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_asciiFalse)这个习惯后来帮我省了很多事。多Agent跑久了只要从返回结果里一眼能看到requires_more_infoSupervisor就能做出“继续追问”还是“收尾”的决策不需要靠模型自己瞎猜。3. 实操从零搭建一个Supervisor Agent项目3.1 场景设计与Agent清单用一个电商客服场景来演示。三个子Agent一个SupervisorAgent名称职责边界依赖能力返回格式订单查询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(contentSUPERVISOR_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_idstate[session_id]) # 把结果以ToolMessage形式追加回State tool_call_id last_ai.tool_calls[0][id] return {messages: [ToolMessage(contentresult, tool_call_idtool_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时把verboseTrue打开或者干脆打印每个节点执行前后的State摘要。看到Supervisor先选了order_agentorder_agent返回ToolMessage再到Supervisor生成最终答复整个链路就算通了。4. 进阶优化让多智能体系统真正具备生产条件4.1 上下文窗口与消息裁剪策略多Agent协作最大的暗坑是上下文膨胀。Supervisor每个决策点都会读取state[messages]里的全部历史子Agent节点也一样。跑十几轮之后消息列表里堆满了系统提示、工具返回、中间结果token消耗呈指数级增长而且模型在长上下文里抓关键信息的精度会下降。我采用的策略是分层裁剪传给Supervisor的消息列表只保留最近6轮对话和最新工具返回更早的历史做摘要拼在SystemMessage里子Agent只需要读取自己所属的那段任务描述不传给它们全量历史业务上下文单独存字段比如current_task不依赖模型从历史里自行提取实现一个简单的裁剪函数def trim_messages(messages, max_rounds6): 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主管反复调用同一子Agentprompt里没写收尾规则显式告诉主管“拿到结果后就汇总不再调用工具”上下文膨胀导致变慢/超限全量历史塞给每个节点裁剪最近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的坑大多是调度和上下文层面的抓住这两点这套架构离生产可用就不远了。