☰
多智能体协作实战:基于Coze API构建AI Agent指挥官与调度官
2026/10/20 9:47:58 网站建设 项目流程

先说个现象:我们团队内部发起的“智能体来了(西南总部)”这个项目,刚开始只是想把手里几个零散的 AI 机器人收拢起来,结果越做越发现,真正难的并不是单个智能体有多聪明,而是当你同时运行十几个 Bot、几十条工作流、外加一堆知识库和插件之后,这些智能体之间怎么协作、怎么排队、怎么不吵架、怎么不重复干同一件事。最后我们把整套体系收敛成两个核心角色:AI Agent 指挥官和 AI调度官。这篇文章就把我们基于 Coze API 构建这套智能体体系的完整思路、落地过程、踩坑记录都摊开讲,包括架构怎么拆、接口怎么调、Prompt 怎么写、并发和上下文怎么管。如果你也在做多智能体编排,或者正准备从“单 Bot 演示”走向“多 Agent 生产化”,这篇内容应该能帮你少走不少弯路。

1. 先搞清楚:指挥官和调度官到底解决什么问题

1.1 智能体一多,乱成一锅粥是必然

我见过很多团队做 Agent 的方式是“来了一个需求就建一个 Bot”。业务部门说要做报销助手,就建一个报销 Bot;说要查合同,又建一个合同 Bot;说要自动回邮件,再建一个邮件 Bot。看起来每个 Bot 都能独立值班,但用起来会发现几个很头疼的问题:用户根本不知道该找谁;同一个问题多个 Bot 都能回,但答案经常互相矛盾;知识库各自维护,一个业务规则改了要同步改 N 个地方。

“智能体来了(西南总部)”这个项目启动时,我们统计了一下,光试点范围内就有 23 个在跑的智能体,但真正被高频使用的不到三分之一。问题不是智能体不够聪明,而是缺少一个统一的“入口”和“调度机制”。用户需要的是一个总入口,进去之后有人帮他判断该找谁、该调谁、该汇总谁的结果。这就是我们决定引入 AI Agent 指挥官和 AI调度官这两个角色的直接原因。

指挥官的定位是“大脑”,负责理解目标、拆解任务、判断上下文、汇总结论;调度官的定位是“枢纽”,负责把指挥官拆出来的任务转成分发给对应执行智能体的一系列具体操作,管理队列、并发、重试、超时和状态上报。一句话总结:指挥官管“想清楚做什么”,调度官管“确保做出来”。

1.2 没有调度层,Agent 之间的协作就是灾难

刚开始我们也想过一个更简单的方案:让指挥官 Agent 直接调用其他 Bot。试了两周就放弃了,原因很典型:第一,指挥官 Agent 的 Prompt 里要塞下所有子系统的调用说明,token 消耗巨大,而且只要某个子 Bot 的入参格式变了,指挥官的 Prompt 就得跟着改;第二,Agent 自己“凭感觉”决定调用顺序和并发策略,结果就是经常同时呼叫好几个 Bot,把上游接口限流打爆;第三,没有统一的超时和重试机制,某个子 Bot 卡住了,指挥官也跟着傻等,整个链路超时。

后来我们把调度逻辑从指挥官里拆出来,做成一个独立的 AI调度官模块,问题一下清晰了。指挥官只输出意图和技术无关的结构化指令,比如“需要查询合同库并汇总风险点”;调度官再根据路由规则,决定调哪个 Bot、带什么参数、超时多久、失败了怎么办。这个拆分本质上就是把“决策”和“执行控制”解耦,跟我们在后端设计中把路由和业务逻辑分开是一个道理。

1.3 为什么底座选 Coze API

选型的时候我们对比过几个方案,最后定了 Coze API,核心是看中它这几点:

Coze API 能力我们用来做什么
Bot 管理 API动态创建、更新、发布执行 Agent,指挥官和调度官的配置走 Git 化
会话/消息 API以编程方式向任意 Bot 发消息、拿响应,这是调度官调用执行体的通道
工作流 API把编排逻辑固化成流程,适合做告警处理、派单、审核这类确定性环节
知识库 API给不同执行 Agent 挂独立的领域知识,互不污染
插件系统对接内部系统、第三方工具,执行体通过插件拿真实数据

Coze API 对我们来说最舒服的一点是:Bot 的界面搭建、Prompt 调试、知识库上传这些偏重体验的操作可以在平台上完成,而 Bot 之间的协作、路由、状态管理这些偏工程的操作可以通过 API 在自己服务端实现。这样产品和开发的边界很清晰,产品同学调 Prompt,开发同学调调度。我们不用从零去训模型,也不用自己维护对话基础设施,等于把“智能体的运行时”外包给平台,把精力集中在智能体治理上。

2. 整体架构:一主一调,多执行体

2.1 三层架构:指挥官层、调度层、执行层

我们最终的架构收敛成三个层级,名字听起来唬人,其实画出来非常朴素:

  • 指挥官层:一个主 Agent,所有用户请求先到它这里。它负责理解意图、判断任务类型、拆解子任务,但自己不做具体查询和操作,只输出一份带路由意图的 JSON 指令。
  • 调度层:一个 AI调度官服务,跑在我们自己的后端容器里,不依赖平台界面。它接收指挥官输出的 JSON,校验参数,查路由表,然后按规则调用对应的执行 Bot,并处理并发、重试、超时、结果聚合。
  • 执行层:N 个专业子 Agent,各自有独立的知识库、插件和工作流。它们只做具体的事,比如查合同、写摘要、发通知、算指标,不感知上层决策逻辑。

为什么不让指挥官直接调子 Agent?除了上一节说的 token 和耦合问题,还有一个很现实的原因:审计。所有调度请求都会经过调度官,意味着每一步都有日志:谁调的、调的谁、参数是什么、结果如何。这个对于企业落地太重要了,不然出了问题根本没法追溯。

2.2 消息流与状态流设计

一次完整的请求生命周期,我们是这么设计的:

  1. 用户把问题发给入口 Bot,实际上就是指挥官 Agent。
  2. 指挥官 Agent 根据内部 Prompt,输出类似{"intent": "contract_risk", "entities": {...}, "subtasks": [...]}的结构化指令。
  3. 调度官收到指令,生成一个全局唯一的request_id,把指令拆解成多个可执行的 job。
  4. 调度官查路由表,按依赖关系提交 job 给对应的执行 Agent。可以并行的并行,必须串行的串行。
  5. 每个执行 Agent 返回结果后,调度官把结果写入状态存储,标记成功或失败。
  6. 所有 job 结束后,调度官把聚合结果返回给指挥官,由指挥官生成最终面向用户的答案。

状态流转我们只定义了五个:pending、running、success、failed、timeout。每个 job 从创建到结束都会完整更新状态和时间戳。这个设计看起来简单,但排查问题的时候帮了大忙,因为每个环节卡住了都能定位。

2.3 服务端目录结构与关键配置

我们的调度官服务是 Python 写的,目录结构大致是这样:

agent-orchestrator/ ├── config/ │ ├── agents.yaml # 执行 Agent 的 Bot ID、能力标签、超时配置 │ └── router.yaml # 意图 -> Agent 的路由规则 ├── orchestrator/ │ ├── commander_client.py # 调用指挥官 Agent 的 Coze API 封装 │ ├── dispatcher.py # 调度核心:队列、并发、重试 │ ├── router.py # 路由解析与参数映射 │ └── state_store.py # 状态存储,我们用的 Redis ├── api/ │ └── webhook.py # 接收前端/IM 回调的入口 └── logs/ └── audit.log # 全链路审计日志

配置里最核心的是router.yaml,它决定了调度的走向。我们初期把路由规则直接写成静态映射,不急着做动态路由,因为稳定比灵活更重要。后续要扩展再慢慢加模型打分路由。

3. 核心实现:用 Coze API 把指挥官和调度官跑起来

3.1 准备阶段:Bot 创建与 API 凭证管理

在 Coze 平台上,先把执行层需要的子 Bot 都建好。这一步没什么捷径,但有几个经验值得分享:每个执行 Bot 的 Prompt 必须限定边界,比如“你只负责从合同文本中提取付款条款,不回答其他任何问题”。这样后续做路由才可靠。

API 凭证方面,Coze 的接口通常需要 API Key 和 Bot ID。我们的要求是凭证只能存在于服务端环境变量或密钥管理服务里,前端直接传用户消息到我们后端,再由后端附带凭证去调 Coze API。如果你把 API Key 写在前端代码里,等于把后台完全暴露了,这个坑我们最初差点踩进去。

一个容易被忽略的点:不同环境的 Bot 要分开。我们有 dev、staging、prod 三套 Bot,配置通过agents.yaml区分。用同一个 Bot 联调和上线,会出现这边改 Prompt 还没验证,那边线上已经在受影响的情况。

3.2 指挥官 Agent 的任务拆解逻辑

指挥官的 Prompt 我们不追求花哨,核心就三块:角色边界、拆解规则、输出格式。

角色边界会明确告诉它:你是任务指挥官,不是执行者,不要直接回答业务问题,你的职责是理解用户需求,然后输出 JSON 指令。这一句话就能挡掉很多“指挥官越权去搜知识库”的情况。

拆解规则会给它一套方法论:判断意图属于哪个域;每个域对应哪些子任务;子任务之间有没有依赖关系;需要提取哪些关键实体。为了防止它自由发挥,我们还加了否定清单,比如“不要生成超过 5 个子任务”“不要自行判断哪个执行 Bot 可用,那是调度官的事”。

输出格式用 JSON 约束,并给一个具体的 few-shot 示例。实测下来,加上 few-shot 后解析成功率从 70% 出头直接拉到 95% 以上。指挥官模型我们选的偏强推理的版本,因为任务拆解质量直接影响下游;执行层 Bot 可以用更轻量的模型,成本能省不少。

3.3 调度官的实现细节:路由表、超时与重试

调度官看起来像个 Agent,其实本质上是一个带路由规则的任务分发服务。我用一段简化代码说明核心逻辑,基于常见实践写的伪代码:

def dispatch(commander_output: dict) -> dict: request_id = generate_request_id() subtasks = commander_output.get("subtasks", []) results = {} for subtask in subtasks: intent = subtask["intent"] target_agent = router.lookup(intent) # 查路由表 if target_agent is None: results[subtask["id"]] = fallback_submit(subtask) continue payload = bind_params(target_agent, subtask.get("entities", {})) try: resp = call_coze_bot(target_agent, payload, timeout=15) results[subtask["id"]] = normalize(resp) except TimeoutError: retry_times = 0 while retry_times < 2: resp = call_coze_bot(target_agent, payload, timeout=15) if resp: results[subtask["id"]] = normalize(resp) break retry_times += 1 else: results[subtask["id"]] = {"error": "timeout", "request_id": request_id} return aggregate(commander_output, results)

几个关键参数可以给参考:没有依赖关系的子任务并行度我们限制在 3,因为很多子 Bot 底层共享同一个限流桶,并行太多会触发 429。单个子任务超时 15 秒,最多重试 2 次,重试间隔按 1 秒、2 秒递增。超过这个极限就直接返回失败给指挥官,让它在回答里如实说“部分任务未完成”,而不是假装全都成功了。

3.4 上下文管理与记忆传递

多智能体协作里最容易被低估的就是上下文。Coze 平台自带的会话记忆是针对单个 Bot 的,但指挥官和多个执行 Agent 之间来回传递内容,不能指望每个 Bot 都记住前因后果。

我们的做法是三层记忆:

  • 全局上下文:用户问题、指挥官拆解结果、最终答案摘要,用request_id存一份到 Redis,保留 7 天。
  • 子任务上下文:每次调执行 Bot,我们会把该子任务相关的内容拼成一段精简的上下文传过去,但有意控制长度。比如只传“用户原始问题摘要 + 本次任务实体”,不传全量对话历史,因为全量传上去 token 成本暴涨。
  • 摘要记忆:如果链路很长,比如用户连续追问,我们会在下一轮把上一轮的结论摘要回填给指挥官,而不是回填完整记录。

这一步做完之后,多轮对话的体验稳定了很多。之前经常出现用户第二轮问“那合同编号是多少”,执行 Agent 完全不记得上一轮提过哪份合同,现在通过摘要回填基本能衔接上。

4. 实操过程:从零搭一套“告警接入 + 自动跟进”的双 Agent 体系

4.1 设定一个具体场景

光讲架构容易飘,我拿我们内部一个真实跑起来的场景来拆解:系统告警处理。这个场景非常适合演示指挥官和调度官,因为它天然多步骤、需要调用多个执行体、并且对时效性有要求。

流程需求是这样的:监控系统推送一条线上告警(比如数据库连接池打满、支付接口超时率上升),指挥官 Agent 收到后先判断告警类型和影响范围;然后调度官根据类型,同时调度三个执行 Agent:一个查知识库找对应应急预案,一个去查指标系统拿最近 5 分钟趋势,一个查值班表找到当前值班人;最后指挥官汇总这些信息,输出一段包括“故障现象、影响面、建议动作、值班人”的完整告警通知,推送到工作群。

这个场景里,指挥官不是医生,它只是把“诊断线索”收集齐;真正翻手册、看指标、找人,都是执行 Agent 干的事。

4.2 在 Coze 平台侧编排子 Agent 与工作流

第一步,在 Coze 平台分别创建三个执行 Bot:

  • ops-doc-bot:挂载运维知识库,Prompt 限定为“只根据知识库内容回答应急预案相关问题”;
  • metrics-bot:通过插件对接内部指标查询接口,入参是服务名和时间范围,出参是趋势摘要;
  • oncall-bot:查值班表插件,返回当前值班人姓名和联系方式。

这三个 Bot 互相不知道对方存在,也不需要知道。这是刻意设计的,执行层越“笨”越好,减少越权行为。

第二步,在入口侧设置指挥官 Agent,把它的能力集中在理解告警文本和结构化输出上。为了让指挥官能读懂各种奇奇怪怪的告警消息,我们在知识库里放了几十条历史告警样本,用来做 few-shot 对齐。

第三步,调度官服务通过 Coze API 的 Bot 消息接口去调用这三个子 Agent。整个编排看起来像是“在 Coze 平台上搭积木,在自己后端搭管道”。

4.3 调度调用的关键代码与返回解析

调度官调用执行 Agent 的代码并不复杂,核心是 Post 一条消息并解析返回结果,基于常见实践的写法参考如下:

import requests, os COZE_API_BASE = os.getenv("COZE_API_BASE") API_KEY = os.getenv("COZE_API_KEY") def call_bot(bot_id: str, user_message: str, conversation_id: str = None) -> dict: headers = { "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", } payload = { "bot_id": bot_id, "user_id": "orchestrator", "stream": False, "auto_save_history": True, } if conversation_id: payload["conversation_id"] = conversation_id if user_message: payload["additional_messages"] = [ {"role": "user", "content": user_message, "content_type": "text"} ] resp = requests.post(f"{COZE_API_BASE}/v1/chat", headers=headers, json=payload, timeout=25) return parse_chat_response(resp.json())

这里的要点有四个:

  • stream先设成False,方便做同步聚合。等要上用户直连场景再改流式,不然调度逻辑会被打散。
  • user_id我们用orchestrator来标记这是系统调用而不是真人,这样在后续审计时能区分来源。
  • 返回结果要自己做 normalize,不同 Bot 返回的文本风格差别很大,调度官要做一层清洗,把纯文本变成结构化字段再传给指挥官。
  • 如果某个子 Bot 需要多轮才能拿到结果,必须用同一个conversation_id,但这种情况在我们场景里很少,因为我们是把上下文拼在一条消息里发过去的。

4.4 成本与性能调优的体会

整套体系跑起来之后,最先受到冲击的不是准确性,而是成本。我们有一次联调测试,一个告警处理链路跑完,光 token 消耗就是单 Bot 单轮对话的几十倍,因为指挥官读一遍告警、写一遍拆解、看三份结果、再写一遍汇总,加上三个子 Agent 的上下文,算下来非常可观。

我们后来做了三个优化:

  • 所有子 Agent 用轻量模型,指挥官用强模型。任务拆解这种“想清楚再做”的活值得花钱,查知识库这种“关键是检索质量”的活不需要太强推理。
  • 限制子 Agent 返回结果的长度。我们会在 Prompt 里明确“回答不超过 200 字,不要输出步骤说明,直接给结论”,能省大量 token。
  • 告警场景加了“相似告警合并”。5 分钟内同一条告警重复触发时,调度官先查 Redis 里有没有未完结的request_id,有就直接复用之前的分析结果,不重新调一套 Agent。这个优化直接让整体成本降了接近 40%。

5. 实际情况中的坑:从测试到上线的排查笔记

5.1 问题速查表

现象可能原因排查思路解决参考
指挥官频繁输出非法 JSONPrompt 里格式约束不够,或少 few-shot看原始输出日志,找规律加 few-shot,并要求“只输出 JSON,不要解释”
子任务全部超时上游 API 限流或共享平台配额检查响应状态码和限流头降低并发度,加指数退避重试
结果张冠李戴上下文被多个子任务共享污染检查会话 ID 是否复用不同子任务强制新会话,只传摘要
用户追问时执行 Agent 失忆全局上下文没有回填查 Redis 里该请求的摘要下一轮把摘要注入指挥官
同一告警被重复分析好几遍调度入口没有做幂等查日志看request_id是否复用加去重窗口,Redis SETNX

这张表是在一次大型联调之后整理出来的,当时我们花了整整两天追一个“结果完全错乱”的问题,最后发现就是会话上下文串了。

5.2 四个我自己踩过的典型坑

第一个坑:指挥官被业务方误当成了知识问答机器人。因为入口 Bot 挂了指挥官的 Prompt,测试人员直接问它业务问题,指挥官为了显得有用,真的会强行回答,结果答错。后来我们给入口 Bot 加了一条铁律:所有与业务相关的问题必须先走调度流程,否则回复“请稍等,正在为你查询”。实际上就是把“不自作主张”写进了角色设定。

第二个坑:并发调度打爆了平台限流。我们的调度官一开始并行度设的是 6,结果一有告警风暴,六个子 Agent 同时发请求,限流直接触发,所有任务全失败。后来把并行度降到 3,并且加了“每 100 毫秒最多发出 2 个请求”的本地令牌桶,再没出现过限流。

第三个坑:重试导致重复派单。第一个子 Bot 超时了,重试机制重新调用,但前一个请求其实已经成功执行,只是响应超时,于是值班人收到两条一模一样的通知。解决办法是在消息内容里带上request_id,执行 Agent 在发通知前先查重;同时把重试语义明确为“查询型任务可重试,通知型任务不重试”。

第四个坑:审计日志太冗余。最开始我们一行一条日志地打,状态变化全打到同一个文件里,结果排查问题时要 grep 半天。后来改成按request_id聚合日志,每条链路打一个 JSON 结构,包含所有子任务的状态和时间戳,排障效率明显提升。

5.3 稳定性兜底:限流、熔断与降级

这套体系毕竟跑在生产上,稳定性必须有兜底。我们在调度官里做了三级保护:

  • 限流:上面提到的令牌桶,按子任务维度限制对 Coze API 的调用频率。
  • 熔断:如果连续 10 个请求中有 50% 以上超时或返回错误,调度官直接熔断 30 秒,期间不调用子 Agent,由指挥官直接返回“当前智能体服务繁忙,请稍后再试”。
  • 降级:核心子任务失败时,调度官可以走一条预先配置的降级路径,比如查知识库的 Bot 挂了,就改为从静态备份文件里匹配关键词。

这些机制听起来复杂,但实现都很轻。我们在dispatcher.py里总共不到两百行就全做了。多智能体系统最怕的不是某个 Agent 出错,而是错误没有兜底一路向上传播,最后用户看到的是一个胡说八道的总结。有兜底之后,最差的情况也是用户看到“部分能力暂不可用”,至少不会看到完全编造的内容。

这套“指挥官 + 调度官”的模式,我们内部已经跑成了基座,后续往更多区域复制时基本不用改架构,只要改路由表和执行 Agent 的知识库就行。我个人最大的体会是:做多智能体编排,最重要的不是让每个 Agent 变得更聪明,而是让它们之间形成清晰的边界和流程。指挥官不要太贪,什么都想自己干;调度官不要太复杂,稳定可靠比花哨重要;执行体要足够专一,只做一件事并且做好。如果你准备动手搭自己的智能体体系,我建议先别追求大而全,挑一个足够典型的场景,用两三个执行 Agent 跑通闭环,再逐步扩张。等你真正经历过一次告警风暴下所有 Agent 协同不出错之后,就会理解这套治理逻辑的价值了。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询