☰
多智能体协作实战:用“虚拟机构”编排Agent团队
2026/10/10 4:28:49 网站建设 项目流程

在Agent开发这个圈子里,“多智能体协作”喊了一整年,真正落地时你会发现困难不在单个Agent本身,而是怎么把一堆各怀绝技的Agent组织成一支能打仗的团队。我目前在做的一个项目代号就叫agency-agents,核心思路很简单:把所有Agent塞进一个“虚拟机构”,像公司一样划分角色、指定流程、定义上下级汇报关系,让它们各司其职地完成一条复杂业务链路。

这个项目解决的典型问题你肯定遇到过:单个Agent面对一个综合性任务时,要么上下文越来越长导致性能崩坏,要么模型能力被“既要又要”的需求拉扯得两头不靠。拆成多个Agent又面临新的头痛——谁来拆解任务?谁来汇总结果?A Agent说的话B Agent能听懂吗?跑着跑着某条链路断了怎么恢复?

这篇文章就把我在agency-agents项目里踩过的坑、验证过的方案、沉淀下来的代码结构完整拆开讲。适合已经在用LLM API开发应用、想从“单Agent”跨到“多Agent协作”的工程师,也适合准备设计企业内部自动化流程的技术决策者。你会发现,把几个人类社会的管理智慧搬进Agent系统里,很多看似玄学的问题突然就有了答案。

1. 项目概览与设计思路

1.1 为什么单个智能体搞不定复杂任务

我先说两个很具体的技术瓶颈。第一个是上下文窗口的物理限制。一个综合型任务,比如“调研竞品并生成投放策略”,如果让单个Agent完成,它需要同时记住市场背景、产品资料、竞品分析、投放平台规则、文案风格指南……这些资料加在一起,即使没超过窗口上限,模型在超长上下文下的召回精度也会明显下降。实测超过一定长度后,模型会开始“选择性失忆”,早期塞进去的关键约束直接被遗忘。

第二个瓶颈是角色冲突。同一个模型要同时扮演“冷静的分析师”和“激进的创意人员”,这不是提示词能彻底解决的。即便你在System Prompt里写了“先分析再创作”,模型依然容易在分析阶段就带出创作腔,或者在创作阶段放不开手脚。让一个Agent干所有事,本质上是在和模型的固有人格倾向做对抗。

所以我的基本判断是:复杂任务必须拆,拆完必须由不同角色分别处理。这就是agency-agents项目的起点。

1.2 “虚拟机构”的协作模型是什么

agency-agents的核心隐喻是把Agent系统组织成一家虚拟公司。每个Agent对应机构里的一个职能部门或岗位,有明确的职责边界、能力说明和其他Agent协作的规则。家不是一个Agent,而是一套完整的组织架构。

在我这个项目里,最基本的机构配置包括四类角色:

  • 规划器(Planner):相当于项目经理,接收原始需求,负责拆解任务、分派给执行者、汇总中间结果。
  • 分析师(Analyst):负责信息调研、数据处理、竞品分析,产出结构化的分析报告。
  • 创作者(Creator):负责文案产出、内容生成、创意表达,基于分析报告产出面向用户的最终物料。
  • 审核员(Reviewer):负责检查最终输出质量,对照需求清单逐项核验,有问题就打回重做。

这套设计的第一个优势是职责清晰。每个Agent只需要精通一件事,提示词可以写得更聚焦,例如分析师的System Prompt里不需要任何文案创作技巧。第二个优势是流程可审计。每个环节的输入输出都作为消息留存,出了问题能定位到具体某一步。第三个优势是容错隔离——创作者即使输出质量不稳定,也不会污染分析师的前期数据。

1.3 三种主流协作模式与选型对比

多智能体协作不是只有“机构”这一种玩法,我在设计前对比过三种主流模式。

协作模式工作方式优点缺点适用场景
管道式(Pipeline)按固定顺序串联,A输出作为B输入实现简单、流程可控链路僵硬、中间结果不可复用流程高度固定的批处理任务
编排器-工作者(Orchestrator-Workers)中央调度器动态分派任务给多个工作者灵活、适合复杂任务分解编排器容易成为瓶颈综合性任务、需求多变
群体协商(Debate)多个Agent平行讨论、交叉质询结论质量高、覆盖角度全Token消耗巨大、收敛慢高风险决策、需要多角度验证

agency-agents最终选择的是“编排器-工作者”的变体,外加半自动化的管道衔接。规划器作为中央编排器动态决定任务怎么拆、分给谁、需要几轮;但每个子任务内部,又按照固定流程(比如分析师先产出,创作者再基于产出创作)走管道逻辑。这种混合架构的好处是:宏观路径由AI动态规划,微观步骤由人工预设兜底,既灵活又不至于完全失控。

注意:纯“管道式”最容易上手,但一旦业务逻辑复杂到需要条件分支,代码里会堆满if-else,维护成本陡增。而纯“群体协商”的Token成本在真实业务里很难被接受。混合架构是个值得优先考虑的中间态。

2. 角色定义与消息协议

2.1 角色设计的关键参数

每个Agent在系统里不只是一条提示词,而是一个包含完整配置的实体。我在代码里用一个数据类来统一承载这些元信息。在设计角色时,下面几个参数是我反复调整的重点:

首先是角色说明。这部分用来让“其他Agent”理解这个Agent能干什么、擅长什么。因为规划器要靠这段文字来分派任务,写得太模糊会导致路由错误。

其次是输出格式约束。每个Agent的回复必须使用严格的结构化格式(通常是JSON),否则下游Agent无法解析。自由文本聊天的时代在Agent系统里已经结束了——机器之间交换信息,格式契约是第一位的。

第三是协作边界。需要明确这个Agent能调用哪些工具、访问哪些数据源、允许它等待几轮回复。例如审核员只能读不能改,创作者只能基于分析师提供的资料动笔,不允许自己重新查资料。

2.2 LLM选型与参数调优

不是所有Agent都应该用同一个模型。我在项目里的经验是:规划器用推理能力强的大模型,执行者用性价比高的中档模型。规划器要处理的是全局调度决策,每轮推理质量影响整条链路;而分析师和创作者做的事情相对机械,用中档模型完全够用,成本能省下60%以上。

有一个小细节特别值得注意:同一个Agent在不同阶段可能需要不同的推理温度。创作类的任务温度设置在0.7到1.0之间比较合适,出稿更有随机性和灵性;但分析、审核这类任务我会把温度压到0.1到0.3,宁可保守也不要自由发挥。如果你平台不支持在请求级别调整温度,那就需要给创作者和分析师分别配置不同模型实例,让参数绑定在Agent配置上,而不是请求调用的现场。

2.3 结构化输出:让Agent“说人话”转成“机器话”

这是整个项目里最值得提前做对的一件事。两个Agent之间交互如果没有严格结构,你会在消息路由和参数提取上浪费大量时间。我最终采用的是“JSON Schema + 有限枚举值”双重约束。

response_format层面的JSON模式,能强制模型输出合法JSON;但这还不够,字段内部的值也需要约束。比如创作者的状态字段只能取"success"或"needs_revision",任务类型字段只能在预设枚举里选,所有自由文本必须放进固定的字段名里。这样下游Agent在做分支判断时,直接用字典取值就行,不用做任何模糊匹配。

提示:第一次尝试时一次加太多限制,模型输出会频繁校验失败。更好的做法是分两步走:先用宽松模式(只约束JSON整体结构)跑通流程,再逐步增加枚举值和必填字段。如果你跳过这一步直接上强约束,排查问题的成本是前者的三倍不止。

2.4 消息协议与事件总线

多Agent之间的消息不能是“说一句话”那么简单。我在项目里定义了一组核心字段作为统一消息协议:

  • sender_id和receiver_id:标明来源和目的地。
  • run_id:属于哪一轮任务编排。
  • message_type:如任务分派、任务结果、审核反馈。
  • payload:实际数据体。
  • correlation_id:关联上下文,用于追踪一组消息的完整生命周期。

有了这套协议,整个系统在逻辑上就是一个消息总线。每个Agent从总线上订阅属于自己的消息,处理完把结果重新发布回去。这个设计为后来了不得的两个能力铺了路:一是审计追踪,可以把任意一条业务链路按run_id完整回放;二是横向扩展,某个Agent压力大了就起多实例订阅同一类消息,天然支持并发。

3. 编排器核心实现

3.1 整体模块结构

先看下我这个项目里目录的基本划分,把职责边界拆清楚会省掉大量后期返工。

agency-agents/ ├── agents/ # 各角色Agent定义 │ ├── base.py # Agent基类:消息收发、LLM调用、重试 │ ├── planner.py # 规划器:任务拆分、结果汇总 │ ├── analyst.py # 分析师:信息处理 │ ├── creator.py # 创作者:内容产出 │ └── reviewer.py # 审核员:质量检查 ├── core/ # 核心框架 │ ├── message.py # 消息协议定义 │ ├── router.py # 消息路由:根据类型分发 │ ├── memory.py # 共享黑板:跨Agent数据存储 │ └── orchestrator.py # 编排器:调度整体流程 ├── tools/ # Agent可调用的外部工具 ├── configs/ # 各Agent的System Prompt和模型参数 └── examples/ # 示例项目与启动入口

实际项目里还会多一个tests/目录,但后面我会专门讲测试策略,这里先不展开。

3.2 核心代码:消息定义与工具函数

先展示最基础的部分:消息体定义。这是所有模块交互的契约,改动它要极其谨慎。

import uuid import time from typing import Any class Message: def __init__( self, sender_id: str, receiver_id: str, message_type: str, payload: dict[str, Any], run_id: str, correlation_id: str = None, ): self.message_id = uuid.uuid4().hex self.sender_id = sender_id self.receiver_id = receiver_id self.message_type = message_type self.payload = payload self.run_id = run_id self.correlation_id = correlation_id or self.message_id self.timestamp = time.time() def to_dict(self) -> dict: """序列化为字典,方便存储和日志输出""" return { "message_id": self.message_id, "sender_id": self.sender_id, "receiver_id": self.receiver_id, "message_type": self.message_type, "payload": self.payload, "run_id": self.run_id, "correlation_id": self.correlation_id, "timestamp": self.timestamp, }

这里有个容易忽略的细节:correlation_id默认等于当前消息ID,但后续的消息回复会沿用第一次的correlation_id,这样整条业务链路的所有消息都能串起来。没有这个字段,出问题时你会面对几十条毫无关联的消息日志,追踪难度指数级上升。

3.3 编排器调度策略

编排器是整台机器的“心脏”,它负责接收用户请求、调用规划器拆分任务、把子任务依次分发出去。

class Orchestrator: def __init__(self, planner: Agent, workers: dict[str, Agent]): self.planner = planner self.workers = workers def run(self, user_request: str, max_rounds: int = 5) -> str: # 1. 规划器拆解任务 plan = self.planner.plan(user_request) # 2. 逐项分派给对应的worker处理 results = {} for task in plan["tasks"]: worker_id = task["assignee"] worker = self.workers[worker_id] reply = worker.execute(task) results[task["task_id"]] = reply # 3. 汇总结果返回 return self.planner.aggregate(results)

这段代码逻辑很直白,但实际项目里需要在每个环节补充三类处理:超时重试、异常分支、结果校验。规划器返回了不存在的assignee怎么办?worker调用LLM时超时要不要重试?某个关键任务连续失败三次,是跳过还是终止整个流程?这些凡是没写清楚的位置,迟早会在生产环境集中爆发。

3.4 完整调用流程演示

我用一个“竞品调研+投放文案生成”的例子,把完整调用流程串一遍:

  1. 用户输入原始需求:“调研某品类三款竞品的定价和卖点,基于分析结果写一篇小红书风格的产品种草文案。”
  2. 规划器接受任务,拆解出三个子任务:
    • 任务A:检索三款指定竞品的公开定价数据与核心卖点,产出结构化对比表。
    • 任务B:基于任务A的对比表,提炼该品类的差异化关键词。
    • 任务C:基于前两步产出,写一篇包含体验感描述、产品卖点、行动号召的种草文案。
  3. 分派关系是:任务A和任务B分给分析师,任务C分给创作者。审核员作为兜底节点,在不满足输出质量时执行“打回重做”操作。
  4. 编排器按依赖关系依次执行:A完成后触发B,B完成后触发C。这里我用的是一个简化策略,直接把C的任务依赖列表里的前序任务结果作为附加上下文传递给创作者。
  5. 最终产出由审核员校验,通过后返回给用户。

这套流程跑通之后,你再往里面加新任务类型,只需要在规划器的提示词里补充新任务模板,再注册好对应worker,大部分情况下不用改动编排器代码本身。

4. 任务分解与上下文管理实操

4.1 任务分解策略与常见误区

任务分解是整个系统质量的“天花板”,规划器的拆解质量决定了后续每一步的上限。最常见的误区是“拆得不够细”。例如把“写一篇产品文案”这个任务原封不动丢给创作者,这不叫拆解,这叫甩锅。合理的拆法至少要分成:信息收集(输入素材)、受众洞察、核心卖点提炼、初稿产出、审校修改。

我总结出一套实用的拆解公式:目标——背景——约束——交付物。规划器拆分出的每个子任务,都必须同时包含这四个要素。

{ "task_id": "task_C", "type": "copywriting", "goal": "基于竞品对比数据,产出一篇种草文案", "context": { "competitor_table": "来自task_A的结构化输出", "keywords": "来自task_B的关键词列表" }, "constraints": [ "字数控制在300字以内", "语气亲切口语化", "必须包含目标产品", "不超过三个核心卖点" ], "deliverable": { "type": "markdown_doc", "format": "正文内容+结尾行动号召" } }

这套结构看起来繁琐,但换来的是下游Agent不需要猜测上游意图。很多Agent系统跑起来后“看起来在聊,实际上全是空对空”,根子就在于任务定义里缺了约束和交付物。

4.2 上下文隔离与共享黑板

Agent系统里最容易被低估的风险是上下文污染:分析阶段的原始数据被传给了文案阶段的Agent,导致创作者在写文案时被过多细节牵着走。我的解法是引入“隔离三级制”:

  • 一级隔离:每个Agent的系统提示词保持独立,不互相包含。
  • 二级隔离:任务执行时上下文只包含本任务相关的前置结果,不加载全部历史。
  • 三级隔离:可选的共享存储(我称它为“黑板”)用于存放大范围公用的中间数据,比如竞品分析表。

为了避免一个短期任务把不属于它的上下文全塞进来,我实现了一个轻量级内存模块,支持按task_id和context_type两个维度取值。这样创作者拿到的永远是“精加工后的结论”,而不是几百行原始数据。

4.3 成本控制与Token预算实战

多Agent系统的成本是单Agent的数倍,不管理Token预算,一次综合任务的费用会轻松突破单一调用链路的10倍。我上线第一个版本时就没做预算,实测某个月的消耗比预估高了3倍多。

现在我给每个Agent限制最大上下文长度。一个标准策略是:所有源来自上游的消息,在进入当前Agent的上下文之前,先经过一层压缩处理。以分析师为例,它的输出结果如果过长,在作为输入交给创作者之前,会先由规划器调用一次“摘要工具”把核心结论提炼成300字以内的要点,再传给创作者。这一步牺牲了一点信息完整性,换来的却是Token费用和推理延迟的显著下降。

这里有一个再强调不过的原则:体系内应该流动“提炼后的信息”,而不是“原始的全文”。所有Agent都应该被要求输出精简结论,把详细内容写入共享黑板,需要时按需存取。这个原则贯彻到位,成本问题能控制住一大半。

5. 常见问题与排查技巧实录

5.1 典型问题速查表

症状可能原因解决方向
任务A完成后任务B一直不启动依赖关系没有设置或状态没有正确写入检查编排器的依赖标记逻辑,确保A完成时显式触发下游
某个Agent的回复反复解析失败输出格式约束偏弱或提示词引导不足切换到严格的JSON模式,并在System Prompt里附一个完整输出示例
两台Agent在同样的输入下产出不同结果检查是否设置了随机参数或模型版本不一致决策类Agent温度降到0.2以下,并固定模型版本
流程跑到一半就“聊偏了”角色代入失败,Agent开始行使权限之外的职责给每个Agent加严格的行为禁令,明确“不允许”做什么
Token消耗异常偏高消息内容没有做摘要,全量堆积到下游上下文实现信息压缩层,只向下游传递提炼结果

5.2 死循环与路由混乱:亲历的两个故障

我实际调试过程中,印象最深的一次死循环出现在规划器和审核员之间:规划器拆出的方案被审核员打回,规划器在收到反馈后重新调整了方案,但调整后的方案仍然不满足审核员的核心约束,于是再次被打回……两者来回拉锯了14轮,消耗了巨量Token才触顶中断。

排查后发现根因是审核员的“打回理由”表述太模糊。它只说“卖点不够突出”,没有具体指出哪三个卖点不突出、期望的呈现方式是什么。规划器接到的是一条语义含糊的反馈,自然无从修正。解决方案是给审核员添加了结构化审核反馈模板:所有不通过项必须指出具体条款、违反的原因、修正的建议。从那以后,打回轮次稳定控制在两轮以内。

5.3 追踪调试与日志设计

多Agent系统的调试,没法靠“打断点单步跟踪”,因为每个Agent内部是黑盒调用。唯一的办法就是:把全部消息流落盘,并在每一步记录结构化日志。

我在项目里接了一个简单的日志格式规范,核心字段包括时间戳、run_id、发送方、接收方、消息类型、关键动作、Token消耗。只要run_id一致,一条完整链路就能被无缝重组出来。建议你无论用自建存储还是第三方追踪平台,至少保证能做到“按run_id查询全部关联消息”这个能力。

再强调一个调试技巧:准备一套固化输入的回归场景。我用三组固定的业务需求作为每日冒烟测试,每次修改提示词或Agent参数后,先跑一遍这三组,对比关键步骤的输出与上次的差异。很多看似随机的“灵异问题”,实际上是上一轮参数调整的连带效应,没有回归测试你根本抓不住。

6. 扩展方向与个人经验

6.1 引入人工审核节点的必要位置

目前的版本里,我把人工介入收缩为三个必要节点:任务开始前的需求确认、产生异常时的断点续跑、最终交付前的质量验收。不需要在每一步都设人工节点,那样就把多Agent的自动化优势完全抵消了。对业务方来说,他们关心的永远不是“中间过程谁干的”,而是“最终结果能不能放心用”。

6.2 评估与回归测试双轨并行

我强烈建议任何做Agent系统的团队,从第一天就维护一组“黄金测试集”。不需要多,20条左右覆盖典型业务场景即可。每次调优后跑一遍,记录每条任务的完成率、平均轮次、Token消耗和失败原因。这组数据能让你在“感觉新方案更好”和“实测更差”之间抓住真实的因果逻辑。

6.3 我强烈的体会

这个项目做到中后期,我的心态从“让模型做得更多”转成了“让模型做得更准”。多数失败不是模型能力不够,而是我的“组织架构”没有给模型创造正常发挥的环境。Agent调优和带团队极其相似:成员能力固然重要,职责边界、信息流转、反馈机制才是体系运转的地基。把精力花在定义清晰的角色边界和消息协议上,比花在逐个Agent的提示词调教上带来更稳定的整体回报。

最后分享一个我一直在使用的模板细节:在每个Agent的System Prompt最底部,加一句“如果你不确定当前任务是否属于你的职责范围,请直接回答:职责范围内无法处理,并说明原因”。这个简单的护栏,帮助我避免了很多次“好心办坏事”导致的链路偏移。你也可以试一试,成本几乎为零,效果却很立竿见影。

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

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

立即咨询