1. 为什么多智能体一多,第一个崩的总是协作
最近做某跨平台系统的手上项目,最初只用了三个智能体:一个负责意图识别,一个负责工具调用,一个负责结果整理。效果出奇地稳定,主流程跑得顺,偶尔出错也能靠单点重试救回来。后来场景扩展,智能体加到八个:新增了记忆管理、文档审校、外部API对接、代码生成、异常兜底……结果第一个崩掉的不是某个智能体本身,而是它们之间的协作——A把结果递给B,B觉得格式不对,C又在等D的上游输出,D却在重复消费E的结果,整个系统陷入空转。
这个场景应该不少人熟悉。多智能体系统的痛点从来不在单点能力,而在“谁说了算”这件事上。智能体一多,天然会出现三类问题:
- 职责边界模糊:两个智能体都觉得自己该处理某个请求,结果重复执行;或者都觉得该对方处理,结果没人响应。
- 上下文各自为政:每个智能体维护自己的会话记忆,全局状态没人统一维护,到了关键步骤,某个智能体问你要一个只有另一个智能体才知道的中间结果。
- 无意义的环状依赖:智能体A调B,B调C,C又回到A,在分布式场景里还会出现多个节点同时抢占同一个外部资源的情况。
靠提示词约束不是办法。提示词只能约束单个智能体的行为倾向,管不了多个智能体之间的信息流和调用时序。靠人工介入也不现实,智能体一旦跑起来,外部请求是并发的,人工根本追不上节奏。这时候就需要一个总调度层——你可以叫它编排器、协调器、中枢控制器,叫什么名字无所谓,做的事情本质一样:把智能体之间的自由对话,改造成受控的、带清晰边界的数据流。
我个人的结论很直接:多智能体系统设计的第一原则,不是把单个智能体调教得多聪明,而是先想清楚调度层怎么把任务拆下去、把结果收回来、把状态同步给该知道的人。
2. 总调度层到底在调度什么
2.1 四个核心职责,缺一个都会出问题
总调度不是简单做请求转发,它有四个核心职责,每一个都会直接影响系统能不能稳定跑起来。
任务分解。收到一个用户请求后,调度层需要判断这个请求是单一意图还是复合意图,切成多个子任务,并确定子任务之间的依赖关系(串行、并行、条件分支)。这一步做得不好,后面全乱。比如用户说“帮我写一份季度复盘报告,并提取上季度的销售数据”,这就是典型的复合任务——先取数、再生成。调度层要识别出先后依赖,而不是把两个子任务同时丢出去。
智能体路由。子任务确定后,要分给合适的智能体。路由策略决定了是固定映射(某类任务永远走某个智能体)、基于能力描述匹配,还是基于历史效果动态选择。大多数生产系统会用“固定映射+兜底策略”的组合,而不是完全依赖动态匹配,因为动态匹配的试错成本太高,一旦路由错了,下游全错。
上下文管理。这一步是最容易在处理时被忽略的。多个智能体协作,全局状态、中间结果、用户约束条件都需要在调度层统一维护。关键问题在于:上下文不能无脑全传。每个智能体只需要拿到自己负责的那一小块上下文,而不是整个对话历史。信息过载会直接拉低输出质量,还会推高token消耗。我在实际项目中见过一个典型case:调度层把整份五百行文档传给负责文案润色的智能体,结果这个智能体把别人的业务字段也“润色”了。
状态跟踪与异常兜底。调度层要实时掌握每个子任务的执行状态,超时要重试,失败要降级,死循环要中断。没有这层管控,一个跑飞的智能体会消耗掉整个系统的预算和算力。
2.2 调度模型怎么选
调度模型的选型没有标准答案,跟业务场景强相关。我按复杂度从小到大列一下常见的几种模式。
单层直连模式。智能体数量少(4个以内)、任务链路明确时,可以在代码里直接编排调用顺序。优点是简单直接,调试方便;缺点是扩展性差,加一个智能体就要改流程代码,链路稍微复杂点就维护不住。
中心化编排模式。这是目前生产环境的主流选择。所有智能体通过统一调度层接入,不直接互调。调度层内置任务本体、路由表、状态机,对智能体做统一的读写管理和状态同步。优点是可控性强,智能体之间保持解耦,谁挂了调度层能立刻感知并做降级。缺点是调度层本身会变成瓶颈,需要做好水平扩展。
混合编排模式。调度层负责主链路,但允许某些高频场景走局部的直连调用,降低调度层的转发压力。比如主流程必须走调度层,但同一批数据在某个子任务内部多次流转时,可以让子任务内部自行协调。这种模式控制得好效率很高,控制不好会让整个系统的可观测性大打折扣——因为不是所有调用都会经过统一日志。
从我接触过的项目来看,80%以上需要多智能体协作的场景,中心化编排模式都是最稳妥的起点。先别急着玩混合编排,先把中心化做扎实。
3. 实操要点:总调度落地时要盯住的五个环节
3.1 定义智能体的“能力描述”而不是“人设描述”
这是第一个容易踩坑的地方。很多人在定义智能体身份时会写“你是一个负责用户运营的助手”“你是一个严谨的数据分析师”——这等于在写人设,而不是在写能力。
调度层做路由时,需要的是机器可解析的能力清单。我更推荐用结构化的能力描述字段来定义智能体:
{ "agent_id": "data_analyst", "capabilities": ["sql_query", "data_visualization", "trend_analysis"], "input_schema": {"type": "query", "required_fields": ["metric", "time_range"]}, "output_schema": {"type": "analysis_result", "fields": ["summary", "charts"]} }路由时,调度层拿到子任务,按能力标签匹配,而不是靠大模型自己理解这段任务描述。能力描述越精确,路由越稳定。我对能力描述的唯一建议是:要覆盖面,但不要交叉覆盖——两个智能体都声明了同样的能力,就会制造路由歧义。
3.2 设计标准化的任务协议
智能体之间不直接对话,但调度层转发的内容必须遵循统一协议。协议怎么定是工程能力的体现。实际操作中我会把所有传给智能体的payload统一成三段式结构:
{ "task_id": "task_20250117_001", "task_type": "report_generation", "input_data": {...}, "constraints": { "max_tokens": 2000, "timeout": 60, "output_format": "markdown" }, "context_ref": "context_id_8899" }task_id用于全链路追踪,出了问题能精准定位。input_data是执行这个任务真正需要的数据,不要塞历史记录。constraints约定了token上限、超时时间、输出格式等执行约束。这一步很重要,没约束的智能体可能一次输出两万字,直接把下游显存或内存打爆。context_ref是一个上下文引用,调度层用引用传递而不是内容传递,按需拉取,避免上下文被无谓地复制传播。
智能体完成处理后,按同样的协议把结果回传给调度层:
{ "task_id": "task_20250117_001", "status": "success", "source_agent": "data_analyst", "output_data": {...} }这套格式看着简单,但它是整个调度层最关键的骨架。先定协议,再谈实现,顺序不要反。
3.3 上下文仓库怎么设计
带上下文的智能体被各个调度任务复用,上下文仓库的设计直接决定了系统的稳定性。不要把上下文存在某个智能体内部,那样调度层就失去控制了。上下文仓库本身应该是一个独立的存储组件,至少支持以下能力:
- 按
context_id存取任意阶段的上下文快照。 - 支持局部读取,能拿某个键对应的值就好。
- 记录上下文的更新时间与访问记录,时间太久没用的上下文要能自动淘汰。
- 分布式部署时要支持锁机制,多个智能体并发访问同一上下文时能做冲突规避。
举个例子,用户问“把昨天讨论的方案修改一下”,这个请求涉及昨天讨论的记录、方案文档、修改指令。调度层会先通过记忆管理智能体拉取相关上下文,写入仓库,生成一个context_ref,再把修改任务连同上下文引用发给方案生成智能体。整个过程中,方案生成智能体手里只有引用,不会直接把历史记录全塞进去。
3.4 兜底策略必须前置设计
这部分我吃过亏。最初我以为智能体失败了大不了重试一次,结果重试三次全失败,用户等了几分钟只等来一句“系统繁忙”。多智能体系统的兜底设计必须前置,不能在线上出问题时才临时写逻辑。
我在生产环境里主要做四层兜底:
- 重试:对瞬时故障(超时、网络抖动)做有限次重试,一般两次上限,单次重试间隔按指数退避设置,比如第一次等2秒,第二次等4秒。
- 降级:主智能体不可用时,路由到备用智能体;如果备用也没有,就退化为规则引擎处理,比如预设的模板回复、查本地缓存。
- 熔断:某个智能体连续失败达到阈值,比如5次,调度层熔断该智能体一段时间,避免后续请求继续打到故障节点。
- 中断:检测到环状调用或死循环时,调度层直接强杀任务,并返回截止当前累积到的中间结果,而不是让用户全程空等。
3.5 可观测性比智能体本身更早建设
多智能体系统是很难调试的。单智能体出错了,看它的输入输出基本可以定位;多智能体出错了,调用链路长得像盘山路,可能根本不知道哪一段出了问题。所以可观测性一定要在建系统第一天就搭起来,不要等出了问题再回头补。
常规做法是:调度层对每个智能体的调用记录日志,包含task_id全链路串联、耗时统计、token消耗、返回状态码。更进一步的,调度层可以记录路由决策依据——为什么会把任务派给这个智能体,当时的能力匹配分数是多少。这个在排查路由误判的时候能省很多时间。
至少把下面三个指标监控起来:
- 各智能体的成功率与平均响应时间:看单点健康度。
- 全链路任务完成率:看整体系统健康状况。
- token消耗分布:看是不是某个智能体在异常消耗资源,或者上下文传重了。
4. 从零搭建一个带总调度的多智能体系统
4.1 功能拆分示例
假设我们要做一个某跨平台系统的智能客服Demo。把它拆成四个子任务:
- 意图识别,判断用户问的是售后、售前还是闲聊。
- 售后问题处理,涉及订单查询、退款政策匹配。
- 售前咨询处理,涉及产品推荐、参数对比。
- 结果整理与话术生成,负责把业务结果转成自然流畅的客服回复。
四个智能体各司其职。但请求进来时不能直接发给某个智能体,而是先发给调度层。
4.2 调度逻辑伪代码
下面给出核心调度代码,展示了“请求接入→意图识别→路由分发→结果回传”的主流程,注意这段代码只是骨架,生产环境还需要补上鉴权、日志、异常拦截等逻辑。
# orchestrator.py import json from dataclasses import dataclass, field @dataclass class Task: task_id: str task_type: str input_data: dict constraints: dict = field(default_factory=dict) context_ref: str = "" class Orchestrator: def __init__(self, agents: dict): """ agents: {能力标签: 智能体实例} """ self.agents = agents self.task_counter = 0 def handle_request(self, user_input: str): # 1. 意图识别,交给意图识别智能体 intent_result = self._dispatch_task( "intent_classification", {"user_input": user_input} ) intent = intent_result["intent"] if intent == "售后": return self._route_to_after_sales(user_input) elif intent == "售前": return self._route_to_pre_sales(user_input) else: return {"status": "fallback", "message": "暂时无法处理该请求"} def _route_to_after_sales(self, user_input: str): # 2. 先查订单 order_data = self._dispatch_task( "order_query", {"user_input": user_input} ) # 3. 再匹配退款政策 policy_match = self._dispatch_task( "policy_matching", {"user_input": user_input, "order": order_data} ) # 4. 最后生成话术 reply = self._dispatch_task( "reply_generation", {"user_input": user_input, "order": order_data, "policy": policy_match} ) return {"status": "success", "reply": reply} def _dispatch_task(self, capability: str, payload: dict): agent = self.agents.get(capability) if not agent: return {"status": "error", "error": "no_agent_available"} self.task_counter += 1 task = Task( task_id=f"task_{self.task_counter:04d}", task_type=capability, input_data=payload, constraints={"timeout": 30, "max_retries": 1} ) # 调度层统一发起调用,智能体不直接互相调用 return agent.execute(task) # 调用入口(示意) # orchestrator = Orchestrator(agents=agent_registry) # result = orchestrator.handle_request("我想申请退款")这段代码关键点就两个:
- 所有任务都经
_dispatch_task统一路径执行,日志、限流、重试可以在这一层统一做,不用每个流程各写一遍。 - 智能体之间没有直接引用关系,它们只与调度层交互,后天想新增智能体,只需要注册新的能力映射。
4.3 任务依赖的配置化
上面是代码编排。稍复杂一点的系统,我建议把依赖关系配置化,而不是硬编码在函数里。比如下面的YAML配置定义了一个简单流程,调度层读取配置后执行对应的智能体组:
workflows: - name: after_sales_query steps: - id: order_query agent: order_agent - id: policy_match agent: policy_agent depends_on: order_query - id: reply_generate agent: reply_agent depends_on: policy_match这样做的好处是调整流程顺序不用改代码,运营也能参与编辑,只需修改配置文件。生产系统里,很多业务排障修复的时效要求是分钟级,这种配置化能大幅降低改动成本。
4.4 系统治理
最后提一嘴系统治理。总调度层在业务上整合了各个智能体,它在系统治理上的潜力也经常被忽略。调度层能暴露唯一的、统一的系统入口,做权限管控、灰度发布、版本回滚会方便很多。比如,一个新智能体上线,可以先把流量切一小部分过去,对比效果后再放量。这个操作在调度层做,粒度可以非常细——按任务类型、按用户比例、按地域维度都能切。没有调度层,你想对单个智能体做灰度几乎无从下手,只能整系统一起上。
5. 我踩过的四个坑与对应的排查方法
5.1 上下文污染:智能体拿到了不该有的信息
现象是:某个智能体的输出突然不稳定,有时准确,有时凭空捏造字段。排查后发现,调度层为了方便把全局上下文统一注入给了所有子任务,导致了污染。
应对措施:
- 按双模式重新执行上下文注入逻辑:只下发任务相关上下文摘要,完整上下文详情保留在仓库内按需索引。
- 在task payload里增加一条人工审计字段,标记当前子任务实际注入了哪些context,方便事后回查。
5.2 循环调用:智能体陷入了无限对话
现象:系统长时间无响应,日志显示多个智能体在互相接口。某个智能体返回结果,调度层觉得缺字段,就再用另一个智能体补,补完又回传给上一个重新理解,形成了循环。
排查方式:调度层记录每条任务的调用栈和最大深度,超过阈值及时中断。复盘后我把系统规则改成了“同一链路内对同一智能体最多调用三次”,从源头杜绝循环。另外在接口治理上增加调用频率限制,单智能体每秒超过某阈值自动拒绝请求,这也能兜住一部分异常流量。
5.3 token消耗飙升,成本失控
现象不是某一次出问题,而是连续监测时发现token消耗成倍增长。排查后发现根因是历史对话每次都被整体传入,没有做裁剪和摘要化。
调整方案:
- 调度层在写入上下文仓库时,统一做一次“压缩”:超出窗口的历史消息保留摘要或关键片段,完整消息本地存储。
- 在写库前做数据元数据筛选,只保留本轮业务字段,其他无关信息不写入。
5.4 智能体接口偶发超时,导致全链路级联失败
现象:某个智能体偶发响应慢,但主流程依赖它的结果,整个任务跟着失败。
排查思路:先看是单个节点超时,还是调度层同步等待引发阻塞。我把调度层改成了异步任务队列,主流程不等返回,而是订阅结果事件。超时的任务单独补跑,不再拖垮整个链路。
经验总结就一句:总调度层要做成“流式管道”而不是“同步联调现场”。一切设计往解耦、异步、可观测方向靠,系统稳定性会好很多。
6. 最后聊两句我的实操感受
多智能体的总调度,本质上是在给一群“非常聪明但缺少统一意志”的执行者做流程治理。做调度层设计时,我越做越有一种感受:真正难的不是某个智能体能力不够,而是你如何让不同能力、不同模型、不同数据敏感度的智能体,在同一套规则下有序协作。
我个人建议:如果你刚开始做多智能体项目,第一版调度层宁可保守、宁可简单,也不要为了炫技做太多花样。先把“任务切分、路由分发、上下文管理、异常兜底”这四件事做到七十分,系统跑起来稳定了,再考虑动态决策、混合编排这些进阶方向。调度层是典型的“基础不牢,地动山摇”,前期多花一倍时间在设计上,后期能省十倍精力在排查上。
另外还想再分享一个小技巧:在做多智能体评估时,不要只看单任务成功率,要看“端到端任务成功率”。我做过一组对比测试,单个智能体准确率都超过90%,但串联三个智能体后,端到端成功率只剩75%左右。这个衰减曲线,就是调度层优化空间的最直观度量。把这条曲线监控起来,你就能知道什么时候该优化智能体本身,什么时候该改调度策略了。