如果你自己动手搭过带工具调用的单 Agent,大概很快会撞上一堵墙:模型既要理解用户意图,又要拆解步骤,还要决定什么时候调用哪个工具,最后再把一堆中间结果整理成答案。任务一复杂,思路就开始打结,上下文窗口被塞得满满当当,改了这一步忘了那一步。于是自然就想到,能不能让几个专职 Agent 各管一摊,一个负责查资料,一个负责算数据,一个负责写报告?
分工之后,新的问题立刻冒出来:谁来决定这些 Agent 什么时候上场?谁把用户那句“帮我分析一下这个季度的情况”翻译成“先让数据 Agent 跑指标,再让文案 Agent 写摘要”?这个负责拆任务、派活、验收结果的角色,就是 Supervisor(主管)Agent。
这文章我从零把这个角色推导给你看,不讲花哨理论,就讲它到底解决了什么问题、内部机制怎么设计、代码怎么落地、实际跑起来会踩哪些坑。适合正在搞多 Agent 系统、被“Agent 满天飞没人统筹”困扰的开发者,也适合想搞懂 Agent 协作架构的好奇派。
1. 项目概述:Supervisor Agent 到底干了什么事
1.1 从单 Agent 到多 Agent,麻烦出在“没人统筹”
先说个熟悉场景。一个人接客服工单,问“我上个月买的套餐怎么扣了两次钱,能不能退”,传统做法是 Agent 自己理解意图、查订单系统、翻计费记录、想退费规则、组织回复话术。任务链稍微长一点,模型就容易在中间丢信息,或者在一个无关工具上反复打转,最后输出一条“我帮你查了”但什么都没查的废话。
后来大家把流程拆开:意图识别 Agent 专门判断用户想干嘛,订单查询 Agent 只查订单,退款 Agent 只走退费规则,回复生成 Agent 只管写话术。每个 Agent 的任务都变得特别纯粹,单点准确率上去了。但这套架构跑起来马上发现,Agent 之间互相不知道对方的状态,订单查完了没人告诉退款 Agent“订单已查好,可以走退款了”,回复 Agent 拿到一堆字段也不知道该挑哪些写进话术。
问题不在某个 Agent 不够强,而在缺一个“项目组长”。
1.2 Supervisor 模式的精确定义
Supervisor Agent 的本质是一个编排中枢,它不是一个全能的干活者,而是其他 Agent 的统筹者。它的职责是四件事:
- 理解用户的整体目标,把大任务拆成多个可执行的子任务;
- 为每个子任务选定合适的子 Agent,并把手头掌握的必要信息传递过去;
- 回收子 Agent 的产出,做质量校验,不达标就安排返工或换人;
- 维护整个任务过程的全局状态,保证信息不丢、不串、不无限循环。
类比到现实团队里,Supervisor 不是那个亲手写代码的人,而是分配任务、跟进进度、检查交付质量的技术组长。具体干活的是下面的开发、测试、文档专员,组长负责让这些人朝同一个方向使劲。
这个“项目组长”的价值,恰恰在团队成员变多之后体现得最明显。没有 Supervisor,协调成本随 Agent 数量指数增长;有了 Supervisor,沟通路径全部收敛到中心节点,虽然这个节点可能成为瓶颈,但总体复杂度大大降低。
2. 为什么必须“推导”出 Supervisor:方案选型与架构推演
2.1 三种协作方案横向对比
不是一开始就该上 Supervisor,而是推演之后发现它最合适。我先把常见的多 Agent 协作方案摆出来比一比。
| 方案 | 协作方式 | 优点 | 明显缺点 | 适用场景 |
|---|---|---|---|---|
| 无中枢点对点 | Agent 之间直接传递消息,A 完成后手动指定传给 B | 链路透明,没有额外中心开销 | 职责边界模糊,消息乱飞,循环难控制 | 固定三步流水线,结构千年不变 |
| 单一 Supervisor | 一个主管 Agent 统一调度一组专家 Agent | 职责清晰,扩展新 Agent 简单,任务追踪方便 | 主管 Agent 本身可能成为瓶颈,单点故障 | 中小规模任务,5~10 个以内的子 Agent 很合适 |
| 分层 Supervisor | 上层主管只管下一级主管,下下级继续分派 | 可扩展性强,决策边界清晰 | 实现复杂度高,调试链路长 | 大型系统,子 Agent 数量多且领域差异巨大 |
三种方案都有人在实际项目里用过。纯点对点的方案适合一条流水线写死、每个环节固定不变的场景,比如“抓数据 -> 清洗 -> 生成报表”,Agent 之间传什么字段、传给谁都是写死的,不需要一个智能调度者。
但只要任务一开放,用户今天问“投诉”,明天问“退款”,后天问“定制方案”,写死的流水线就崩了。这时候要么用规则引擎硬拼,要么引入一个能根据任务类型动态选择路径的中央调度者,后者就是 Supervisor 的雏形。
2.2 从组织管理视角推导 Supervisor 模式的合理性
我们从管理学的老道理来想这件事:一个人直接管理的下属数量是有限的,超过一定数量,信息同步和执行校准就会出问题。多 Agent 协作也是同理。
假设某个系统里有 7 个专家 Agent,分别管订单、库存、物流、支付、发票、投诉、售后。如果用点对点方式,任何两个 Agent 之间都可能产生消息传递,潜在交互路径是 42 条;出了问题,你根本不知道是哪个环节传错了。但如果这 7 个 Agent 都只向一个 Supervisor 汇报,交互路径收敛到 7 条,每一条由谁负责、传了什么、结果如何,都在 Supervisor 的日志里有迹可循。
这个推导得出一个重要结论:Supervisor 的价值不止是“智能调度”,更是给系统加了一层可观测的控制点。所有决策、派活、校验都在这个节点留下记录,调试的时候能看到全貌,这是点对点模式很难做到的。
3. 核心细节解析:Supervisor 的内部工作拆解
3.1 路由决策:怎么知道该派哪个 Agent 上场
路由是 Supervisor 最核心的决策点。它拿到用户的一条任务后,需要产出两个东西:一是“这是什么类型的任务”,二是“这类任务由谁来执行”。
我见过很多第一版实现,直接让 Supervisor 用关键词匹配做路由,规则是“包含‘退款’就去退款 Agent,包含‘物流’就去物流 Agent”。短期能跑,但用户的话术千奇百怪,“钱什么时候退给我”“我的货到哪了”“这个东西我不想要了”,每一句话里都没有规则里的关键词。所以可靠的做法是让 Supervisor 先做一层“意图分类”,再根据分类结果映射到对应 Agent。
以客服系统为例,可以先定义几个清晰的任务类型和对应 Agent:
- 订单状态查询 -> 订单 Agent
- 退款与售后 -> 售后 Agent
- 商品咨询 -> 商品 Agent
- 投诉与建议 -> 投诉 Agent
Supervisor 每次路由时输出一个结构化结果,包含任务类型、目标 Agent、需要保留的关键信息、执行优先级。这个结构化结果最好让模型以 JSON 形式返回,方便代码直接解析,而不是让模型说一大段自然语言再去猜。
有几个路由经验是从实际测试里得来的。一是路由条件要“互斥且穷尽”,每个任务只可能落在一个类型里,不能既像退款又像投诉;二是任务边界模糊时,给 Supervisor 一个“默认兜底 Agent”,拿不准就先走通用处理通道,别硬塞给某个专家;三是如果用户一句话包含多个意图,Supervisor 要能拆出多个任务并按顺序排队执行,这就引出下面的任务状态管理。
3.2 状态管理:Supervisor 是如何不“失忆”的
多 Agent 协作里翻车最多的地方,不是 Agent 能力不够,而是状态丢了。用户一开始说“我要退订单 12345 的钱”,Supervisor 把订单号传给售后 Agent,售后 Agent 回复了一堆退款细节,结果回复生成 Agent 接到的上下文里已经没有“订单 12345”了,于是写出来的回复没有订单号,用户一脸懵。
为了避免这个问题,Supervisor 需要维护两张表。一张是“任务状态表”,记录当前任务 ID、任务类型、执行状态(待执行 / 执行中 / 已完成 / 已失败)、指派给哪个 Agent、重试了几次。另一张是“全局上下文表”,记录用户目标、已收集到的关键字段、各子 Agent 返回的关键结果摘要。
这个全局上下文不应该是把所有对话历史原样堆在一起。我自己的习惯是维护一个结构化的“信息清单”,比如一个 JSON 对象,里面有 user_intent、keywords、order_id、agent_reports 等字段,随着任务推进不断更新。子 Agent 干活时只需要拿到与它相关的字段,而不是把整个历史都塞给每个 Agent,这样既能省 token,也能降低模型被无关信息干扰的概率。
3.3 反馈循环:返工机制与闭环校验
Supervisor 把任务派下去,不代表事情就结束了。它必须回收子 Agent 的产出,做一轮“质量验收”。通常我让 Supervisor 在子 Agent 返回结果后问自己三个问题:
- 产出的内容是否直接回答了用户原始问题?
- 是否缺少关键字段?
- 有没有明显的逻辑错误或未解决的冲突?
如果不符合要求,就把结果连同“返工原因”一起打回给原子 Agent,让它补充或重做。这里要注意,返工不能无限进行,否则一个差劲的子 Agent 会带着 Supervisor 一起原地空转。我的做法是给每个子任务设置最大重试次数,通常 2 到 3 次,超过次数就降级处理,比如切换到备用 Agent 或者直接转人工。
这里分享一个真实遇到的教训。第一次做返工机制时,我把打回条件写得太宽松,售后 Agent 返回一句“退款金额需要核实”也被判定为合格,结果用户追问细节时,Supervisor 又发起一轮查询,整个对话绕了十几轮还没解决同一个问题。后来改成“必须同时满足字段完整 + 无待确认项才验收通过”,返工才真正起到了作用。
3.4 异常处理与降级策略
Supervisor 不仅要管“正常流程”,更要管“流程断了怎么办”。常见异常有以下几类:
- 子 Agent 调用外部工具超时,比如查物流接口 10 秒没响应;
- 子 Agent 返回了空结果或直接用话术敷衍;
- 子 Agent 自己进入了死循环,反复调用同一个工具;
- 路由结果不明确,Supervisor 不知道该派谁。
每类异常都要有兜底方案。超时就设置超时上限,比如等待 15 秒直接标记失败并走备用逻辑;空结果就让子 Agent 重试一次,再空就换 Agent;死循环就设调用次数上限,触发后终止任务并给用户一个保守但诚实的答复;路由不明确就向用户追问澄清,或者走默认 Agent 处理。
另一个容易忽略的点是“人工介入开关”。对于退款、改地址这类高风险操作,Supervisor 应该只做“方案准备”,不执行最终变更,而是把待确认动作发给用户确认,甚至转人工处理。这一层保护,在真实业务里比任何调优都重要。
4. 从零实现一个轻量版 Supervisor Agent
4.1 开工前的准备与工具选型
从零实现不需要一开始就上重型框架,我的建议是先写一个不依赖具体模型厂商的“最小闭环”,把编排逻辑跑通,再逐步替换组件。
环境方面有三件套就够了:Python 3.10 以上,一个支持结构化输出的模型接口,以及一个能保存任务状态的内存字典(任务量大了再换成 Redis 之类的持久化方案)。
模型选型有个硬性要求:Supervisor 自身的输出必须是可靠的 JSON。因为路由、任务分解、验收判断都要靠结构化输出驱动,如果模型经常返回半截 JSON,后续解析代码会痛苦不堪。尽量选支持工具调用或具备强指令跟随能力的模型,并在提示词里明确“只输出 JSON,不要多余解释”。
4.2 核心代码骨架:一个最小可运行的编排器
我先用一个抽象基类把 Agent 的通用接口定下来。每个子 Agent 只需要实现一个方法:接收任务输入,返回任务输出。
from abc import ABC, abstractmethod class BaseAgent(ABC): agent_name: str = "base" @abstractmethod def execute(self, task_input: dict) -> dict: """执行具体任务,返回结构化结果""" pass下面设计一个模拟的子 Agent,比如售后 Agent,它一般只负责处理退款相关动作:
class AfterSalesAgent(BaseAgent): agent_name = "after_sales" def execute(self, task_input: dict) -> dict: order_id = task_input.get("order_id", "未知") reason = task_input.get("reason", "无") # 实际项目里这里会调用退款系统接口 return { "agent": self.agent_name, "status": "completed", "message": f"订单 {order_id} 已登记退款,原因:{reason}" }重点在 Supervisor 这个类。它要做三件事:认识有哪些子 Agent 可用;把用户任务转化成路由决策;根据路由决策调用对应 Agent 并结果验收。
from typing import Dict, List, Type class SupervisorAgent: def __init__(self, agents: Dict[str, BaseAgent]): self.agents = agents self.task_history = [] # 记录每个任务的执行过程 self.MAX_RETRY = 2 def route(self, user_request: str) -> str: """路由决策。 实际项目中这里应该让模型做意图分类并输出 JSON。 这里为了展示逻辑,用一段模拟规则代替,生产环境请替换。 """ # 伪代码示意:调用模型得到结构化路由结果 route_result = model_route({ "user_request": user_request, "available_agents": list(self.agents.keys()) }) return route_result["target_agent"] def execute_task(self, user_request: str) -> dict: target = self.route(user_request) task_record = { "request": user_request, "target_agent": target, "status": "pending", } self.task_history.append(task_record) retry_count = 0 while retry_count <= self.MAX_RETRY: try: result = self.agents[target].execute({ "user_request": user_request }) if self._validate_result(result): task_record["status"] = "completed" return result else: task_record["status"] = "retry" retry_count += 1 except Exception as exc: task_record["status"] = "failed" retry_count += 1 task_record["status"] = "exhausted" return { "agent": "supervisor", "status": "failed", "message": "该任务多次执行未通过校验,已转人工处理" } def _validate_result(self, result: dict) -> bool: """验收标准:必须有结果文本,且状态为 completed。""" return ( result.get("status") == "completed" and result.get("message") and len(result["message"]) > 5 )这个骨架虽然模型路由部分还是伪代码,但已经把 Supervisor 的完整流程立起来了:路由、执行、验收、重试、降级。后面扩展新 Agent 只需要往 agents 字典里注册,Supervisor 的代码一行都不用动。
4.3 让路由从“关键词拼接”进化到“意图分类”
第一版的路由如果用关键词匹配,很快就会发现用户表达方式的多样性能把你逼疯。比如用户说“我不想要这个了”,你到底应该路由到售后去退款,还是路由到商品去退货?关键词根本给不了答案。
我的进化路线是这样:先把任务类型枚举固定下来,例如 ORDER_QUERY、REFUND_REQUEST、COMPLAINT、PRODUCT_CONSULT。然后给 Supervisor 一个结构化的意图分类指令,让它输出类似下面的 JSON:
{ "task_type": "REFUND_REQUEST", "confidence": 0.92, "key_entities": { "order_id": "20240601", "product": "无线鼠标" } }接着再用 30 到 50 条有代表性的用户话术去测试,把分类不准的样本收集起来,调整提示词里的任务类型描述,让边界更清晰。比如“退货”和“退款”在客户眼里可能是一个意思,但在业务上分属不同 Agent,就需要在描述里把差异写清楚。
这一步做好了,Supervisor 的路由才真正从“规则驱动”变成了“模型驱动”,泛化能力大幅提升。
5. 工程落地中的常见问题与排查实录
5.1 上下文污染:子 Agent 之间互相“串味”
多 Agent 系统最常见的问题就是上下文污染。具体表现是:上一个 Agent 的输出内容大量出现在下一个 Agent 的输入里,导致下一个 Agent 被无关信息干扰。
我遇到过这样的情况:订单 Agent 返回了 500 字详细物流记录,Supervisor 原封不动把这些记录传给投诉 Agent,投诉 Agent 从一堆物流字段中挑了一段“包裹已签收”,就开始写“你应该去查一下快递柜”,完全忘了用户投诉的其实是“客服响应太慢”。
解决思路是 Supervisor 在传递信息时只做摘要转发,而不是原文转发。子 Agent 返回的原始结果进入全局上下文后,先被压缩成三行以内的结构化要点,再把要点传给下一个需要它的 Agent。可以把这条当作硬性规则:任何中间结果都不允许大于 200 字地原样传递。
5.2 子 Agent 的输出不稳定
子 Agent 同样是用模型驱动的,所以它也可能不按套路出牌。有的子 Agent 会在结果里加一段“作为一个 AI 助手……”的废话,有的会返回格式破损的 JSON,还有的直接输出“我无法处理这个任务”就结束。
对策是双管齐下。一个是在子 Agent 的提示词里给出严格的输出模板,要求它只输出 JSON 且字段必须齐全;另一个是在代码里做“解析容错”,第一次解析失败就尝试让模型修复,修复失败才判定任务失败,不要因为一次格式错误就让整个流程崩掉。
还有一招很实用:给子 Agent 一个可以被工具调用替代的“内部工具输出”选项。比如对于查询类的 Agent,可以优先让它调用真实接口,把接口返回值作为结果,而不是让模型自己“编造”查询结果。模型只负责把接口返回整理成自然语言,不负责生成事实内容,能大幅降低幻觉概率。
5.3 死循环与资源失控
资源失控是多 Agent 系统上线后才会暴露的问题。子 Agent 自己疯狂调用工具,Supervisor 疯狂重试,两个环节叠加,token 消耗像开了水龙头一样刷刷流。
我踩过最狠的一个坑是:某个子 Agent 调用一个聚合接口,接口本身每次要处理很多文档,模型在结果里看到了“建议进一步查看详情”这句话,就误以为还需要再调用一次,于是一遍遍调同一个接口,最后 token 消耗是预期的 20 倍。
后来在每个 Agent 的工具调用层统一加了一个计数器,单个任务内调用次数达到 5 次就强制停止,并且把“触发次数上限”本身作为一个重要信号反馈给 Supervisor,让 Supervisor 决定是换工具还是换 Agent。同时对 Supervisor 的返工也设了绝对值上限,防止返工逻辑叠加导致总轮数失控。
5.4 测试与可观测性:让问题“看得见”
Supervisor 系统最容易翻车的地方是“黑盒运行”。你只看到用户问了一句话,系统内部走了 7 个 Agent、调了 11 次模型,但你不知道问题出在哪一环。所以可观测性必须从第一天就做起来。
我推荐的日志标准是每一次模型调用和工具调用都记录如下字段:trace_id、agent 名称、输入摘要、输出摘要、耗时、token 消耗、重试次数。排查问题时,拿着 trace_id 一查就能还原完整执行链路。
下面是我常用的问题排查速查表:
| 现象 | 可能原因 | 排查方法 |
|---|---|---|
| 用户问题答非所问 | 路由分错 Agent | 检查路由日志里的 task_type 是否准确 |
| 回复信息不全,缺订单号等关键字段 | 全局上下文里关键字段被丢弃 | 检查传递到最终回复 Agent 的输入里有哪些字段 |
| 整个流程耗时极长 | 子 Agent 陷入工具循环 | 查工具调用次数和单次耗时日志 |
| token 消耗异常上涨 | 某个子 Agent 反复重试或上下文内容被重复传递 | 查各 Agent 输入文本大小与重试计数 |
| 子 Agent 之间互相矛盾 | 各自基于不同版本的上下文做判断 | 检查全局上下文更新的时机和内容摘要 |
测试方面,除了对每个子 Agent 做单点测试,还必须做“带任务类型的集成测试”:给 Supervisor 一组覆盖所有任务类型的真实请求,跑完后逐条检查路由是否准确、结果是否通过验收、有没有多余的返工轮次。这些用例最好沉淀成回归测试集,每次改完提示词都跑一遍。
6. 写在最后的实操心得
第一版 Supervisor 不需要设计得特别宏大,先用两个子 Agent 跑通闭环,再在真实请求里持续观察路由准确率和返工率。我自己的体会是,Supervisor 的“智能”并不体现在模型有多强的推理能力,而体现在它能不能稳定地控制住流程:该路由到谁就路由到谁,该验收就验收,该停就停。
最容易被低估的是状态管理,信息清单比模型提示词更值得花时间设计。把全局上下文维护好,子 Agent 的表现会跟着稳定。反过来,上下文一团乱麻,再强的大模型也救不回来。另外,返工逻辑一定要有上限,否则 Supervisor 会把一个小问题的处理成本放大到难以接受。
如果你已经有一个跑得不错的单 Agent,想迁移到 Supervisor 模式,最省力的路径是把当前 Agent 的核心能力拆成两个专职 Agent,再套一个最轻的路由壳子,而不是一上来就搞三五个专家 Agent。架构是慢慢长出来的,不是一次性设计出来的。