最近我被反复问到同一个问题:一个Agent搞不定的任务,多加几个Agent是不是就搞定了?老实说,这个想法既对也危险。对是因为很多复杂任务本身就适合分工协作,危险是因为多Agent协作带来的系统复杂度会成倍上升,如果不提前设计好边界和通信规则,最后只会得到一个互相甩锅的消息黑洞。
我最近做了一个多Agent协作的实验性项目,核心场景是让几个角色完全不同的LLM Agent组成一个小团队,一起完成一份行业调研报告。今天这篇文章想把这套方案背后的设计思路、踩过的坑、可以直接抄走的工程代码一起整理出来,希望对正打算把“多Agent协作”从概念推向工程实现的人有帮助。
如果你还在犹豫要不要做多Agent,或者已经开始折腾但总感觉不稳定,这篇文章应该能帮你省掉不少试错时间。
1. 先想清楚:多Agent协作到底在解决什么问题
1.1 单Agent的天花板在哪里
很多人一开始都会有一个惯性思维:单Agent能力不够,是不是把Prompt写得再长一点、把工具接得再多一点就行了?我一开始也是这么想的,但实际跑下来会发现,单Agent模型再怎么优化,都会撞上几堵墙。
第一堵墙是上下文窗口。一个Agent处理长任务时,历史记录会越滚越大,早期关键结论很容易被挤掉,或者直接超出上下文限制。哪怕模型支持很长的窗口,token成本也会随着轮数肉眼可见地涨,最后变成“任务没完成多少,钱先烧了不少”。
第二堵墙是工具链和角色冲突。让同一个Agent既当资料收集员、又当数据分析师、还要当批判性审查者,它在切换角色时很容易串味。提示词里塞进七八个角色要求,最终结果往往就是每个角色都没做到位,表现会被平均掉。
第三堵墙是调试困难。单Agent逻辑一旦复杂起来,出了问题你根本分不清是哪一步判断出了问题。你是该改Prompt,还是该换模型,还是该修工具?这种“黑盒式”排查体验,折腾过的人都懂。
多Agent协作解决的就是这三堵墙:用专门的Agent做专门的事,每个Agent上下文更聚焦,角色更清晰,出问题的时候也能按角色定位到具体环节。
1.2 多Agent协作能落地的场景长什么样
多Agent不是万金油,它最适合的是那些“任务可以拆解、子任务相对独立、需要多视角交叉验证”的场景。我用一个比较典型的例子来说明:生成一份完整的行业调研报告,拆开来看是这样的。
资料收集Agent负责检索和筛选信息;数据分析Agent负责整理数据、算增长率、做对比;写作Agent负责把分析结果组织成通顺的正文;最后还可以安排一个审查Agent,专门挑毛病,检查有没有数据来源不明、逻辑跳跃、结论和正文不一致的地方。你看,这个流程本质上就是一个微型公司的运作方式。
其他适合多Agent协作的场景还有不少。代码仓库维护场景里,可以让写码Agent、测试Agent、Review Agent各司其职;运营决策场景里,可以让用户反馈汇总Agent、竞品监控Agent、策略建议Agent协同产出报告。本质上都是同一种思路:把一个不可控的大任务,拆成多个相对可控的小任务,再由一个协调者或联席机制汇合。
我特别想强调一点,工业领域里的协作机器人其实也是同一个系统思维——单臂灵活,多臂能干的活翻倍,但协同控制才是真正的难点。Agent领域的多智能体协作面对的也是同样的核心矛盾:各个单元都有能力,但怎么让它们步调一致、结果不互相冲突,这才是工程上真正要花心思的地方。
2. 架构选型:把Agent组织起来的三种主流模式
2.1 集中编排:先搭一个Leader
最直观、也是最容易上手的多Agent架构,就是中央编排模式,通常叫Orchestrator-Worker。这个模式的核心很清晰:有一个总控Agent负责接收用户需求,把任务拆解成多个子任务,再派发给不同的执行Agent,最后由总控Agent收集结果、汇总输出。
这种做法的好处非常实在。第一,可控性强,所有任务分发路径都经过编排器,出问题的时候查日志就能看出是哪一步卡住了。第二,调试方便,每个Worker只负责自己的那一段,你可以单独给某个Worker输入测试数据,看它的输出是否正常。第三,对新人友好,代码结构几乎就是“一个调度中心 + 一群Worker”,心智负担小。
缺点也很明显:编排器会成为整个系统的单点。如果它拆不清楚任务,后面所有Agent再强也白搭;如果编排器调用的LLM偶尔抽风,整个流程就会跟着一起抽风。所以集中编排模式更适合任务链路相对固定、子任务边界清晰的业务场景,比如报表生成、标准问答流程、固定模板的文案生产。
2.2 对等协商:Agent之间直接对话
与集中编排相对的,是对等协商模式,也就是Peer-to-Peer。这种架构里的Agent没有绝对的上下级关系,它们各自身怀绝技,可以直接向其他Agent发消息、索要结果、抛出问题,共同推进一个任务往前走。
这种模式很灵活,特别适合开放式讨论、头脑风暴、策略辩论这类型任务。比如你让“市场分析Agent”和“风险控制Agent”针对同一个方案来回交换意见,最后能产出一份同时包含机会和风险的综合建议。这种结构也更接近我们理想中的A2A(Agent-to-Agent)互操作——不同Agent之间通过某种协议自主发现能力、协同完成工作。
不过,对等协商最大的坑就是容易变成唠嗑。两个Agent互相反驳,如果Prompt设计得又特别对抗,它们可以无限循环下去,直到把上下文塞满或者成本爆炸。所以走这条路必须有强约束:约定最大对话轮数,要求每个Agent在输出结尾显式给出结论标记,或者加一个独立的中立Agent负责终止对话并汇总结论。
我在实际测试里发现,没有“收敛机制”的对等协商只适合作为实验玩具,真正要进生产环境,至少需要一个“会议主持人Agent”来控场,哪怕主持人不做具体分析,只负责喊停和总结,整个系统也会稳定很多。
2.3 自组织群:让Agent自己认领任务
最近社区里冒出来不少Swarm形态的多Agent项目,名字各不相同,但核心思想很一致:把调度动作从中心化变成分布式,让Agent们自己认领任务、互相补位,而不是由一个Leader强制安排。
这种模式很像真实的开源社区或者创业团队:任务贴出来,谁有能力谁上,遇到阻塞大家商量,必要时再临时选一个协调者出来。它的优势是扩展性特别好,Agent可以动态加入退出,某条链路挂了也不至于拖垮全局,适合任务池密集、边界模糊的动态场景。
但它的工程代价也最重。分布式系统里的经典难题——全局状态不可见、消息时序不确定、结果难以复现——在这里一个都跑不掉。你自己测试时能跑通,换个任务可能就完全乱套。所以我个人认为,除非你的业务确实需要非常高的动态性和弹性,否则一开始不建议直接上这种架构;可以先从集中编排起步,等流程跑顺了再逐步去中心化。
三种模式不是一个替代另一个的关系,它们可以混合使用。比如现实里常用的是“总体集中编排 + 局部对等协商”:Leader拆任务,某些复杂子任务内部让两个Agent互相辩论几轮,再回到主流程汇总。我用表格简单对比一下:
| 架构模式 | 控制力 | 灵活性 | 工程复杂度 | 适用场景 |
|---|---|---|---|---|
| 集中编排 | 强 | 中 | 低 | 链路固定、职责清晰 |
| 对等协商 | 弱 | 高 | 中 | 头脑风暴、多视角讨论 |
| 自组织群 | 弱 | 很高 | 高 | 动态任务池、弹性团队 |
| 混合架构 | 中 | 中高 | 中高 | 大多数真实业务 |
3. 关键细节:协议、上下文和状态管理
3.1 Agent之间用什么消息通信
不管选哪种架构,Agent之间要协作,就必须有共同语言。工程实现上我强烈建议:不要传输自然语言长文本,直接用结构化的消息对象。
我的做法是定义一个标准Message结构,核心字段包括sender、receiver、task_id、msg_type、content、created_at。sender和receiver用于路由,task_id用于追踪整条任务链路,msg_type标记这条消息是普通文本、结果返回、请求补充还是异常信息,content是核心内容。
用JSON作为消息载体是成本最低、兼容性最好的方案,几乎所有语言和LLM框架都能直接解析。传输层的选择取决于你的部署形态:如果所有Agent在同一个进程里跑,直接函数调用或者asyncio队列就够了;如果拆成了微服务,可以考虑HTTP/WebSocket;如果任务是异步重型的,建议上消息队列,比如Redis Stream或RabbitMQ,这样能天然获得重试和削峰能力。
通信协议上还有一个趋势值得关注。MCP已经逐渐成为Agent调用工具的通用标准,相当于让Agent和各种外部工具之间有一个“USB-C口”;而Agent与Agent之间的互操作协议,业界也已经出现了A2A这样的方向,目的就是让不同厂商、不同技术栈的Agent能像服务之间调用API一样直接协作。虽然这些协议还在快速演进中,但对我们做系统设计是很好的信号——消息层的抽象做得越通用,未来接入外部Agent的成本就越低。
3.2 上下文和记忆怎么共享
多Agent协作里最容易出错的地方,就是对上下文的理解。很多人想当然地认为,既然大家是协作关系,就应该把完整历史对话复制给每个Agent。这样做三个Agent以内还能勉强跑,Agent一多,每个Agent都在处理重复信息,token成本和推理延迟双双爆炸。
正确的思路是做分层记忆。每个Agent保留自己的“短期工作记忆”,也就是当前正在处理的这一小段上下文;团队层面共享一个“长期记忆”,通常是经过整理的结构化数据或者摘要;更长期的跨任务知识则可以放到向量数据库里,需要时按需检索,而不是塞进Prompt里。
我常用的一种做法是“阶段摘要”:每个Agent在完成自己的环节后,不是把原始对话记录传给下一个Agent,而是生成一段几百字的结构化摘要,包含关键结论、数据来源、未决疑问。下一个Agent拿着这份摘要继续工作,既保证了信息传递,又不会背着巨大的历史包袱。这就像项目团队不会让所有人都参加所有会议,而是用会议纪要和项目文档来同步进度。
另外一个容易被忽略的点是共享画布。在复杂任务里,多个Agent可能需要读写同一份成果文档,比如研究员不断补充参考资料,分析师在表格里填数据,写作者在工作区里更新章节。这个共享工作区可以用数据库实现,也可以用版本化的文件存储,关键在于要有锁或者版本号,避免两个人同时改同一段内容导致互相覆盖。
3.3 任务如何拆分、结果如何汇合
任务拆分有两种路线。自顶向下是由编排Agent先输出一个结构化的任务清单,明确每个子任务的依赖关系,然后再派发;自底向上则是先让各Agent自由产出局部结果,最后由一个汇合Agent去整合和再加工。实际项目中,我几乎都是两者结合:先由编排器给出大致框架,执行过程中允许Agent反馈“这个子任务还需要前置信息”,动态调整依赖关系。
汇合是整个系统里最考验设计的地方。如果只是简单地把所有结果拼在一起,那不叫协作,叫复制粘贴。真正有效的汇合需要做三件事:结构对齐、冲突消解、质量校验。
结构对齐是要求所有子结果都输出成统一的Schema,比如统一日期格式、统一指标口径、统一章节结构。冲突消解是指当两个Agent给出的数值或结论不一致时,必须有仲裁机制,是取平均、找原始数据核实、还是让一个裁判Agent决定,都要提前定义好。质量校验则是在最终交付前,让一个“挑剔的读者Agent”检查整份产出,专门挑逻辑漏洞和事实错误。
任务执行层面,并行度也很关键。我把没有依赖关系的子任务标记为可以并行,用协程并发调度,几个Agent同时开跑;有依赖的则等前置任务完成后再触发。这样能把端到端耗时从“所有任务串行总和”降低到“关键路径长度”。如果你的场景是几十个甚至上百个Agent并发协作,那就要考虑并发配置的合理设置,避免同时请求过多导致服务端限流。
3.4 冲突和异常怎么兜底
多Agent系统比单Agent更容易出乱子,所以兜底机制一开始就要设计进去,而不是等跑挂了再补。
超时控制是第一道防线。每个Agent调用都要有timeout和max_retries,不能无限等。LLM服务偶尔会变慢或者返回异常,重试一两次是合理的,但超过三次基本就是提示词或参数有问题,这时候应该立刻失败退出,把控制权交回编排器,而不是默默重试烧钱。
第二道防线是“评审Agent”。我在跑调研场景时,最终输出一定会经过一个专门做质检的Agent,它的System Prompt非常简单:“你是一个严格的审稿人,你的任务是指出报告中的事实错误、逻辑漏洞和不完整之处,不要给出赞美。”实践证明,一个立场偏负面的评审Agent能把整个系统的产出质量拉高一个档次。
第三道防线是失败信息的结构化。不要让Agent在出错时只返回一句“我失败了”,要让它返回失败类型、失败阶段、部分结果和期望协助。多Agent系统的调试核心不是看结果对不对,而是看消息流转链路上每一步的状态。我后来习惯了在日志里记录每个Agent收到的输入、产出的输出、这一步用了多少token、耗时多久,排查问题时效率高出好几倍。
4. 实操记录:从零搭一个最小可用系统
4.1 技术选型与工程结构
理论聊再多,不如动手写一个demo。我这次选的技术栈很简单:Python + Pydantic + 一个兼容OpenAI接口的大模型SDK。没有上重量级框架,因为就这个规模来说,自己搭更能理解每一层的职责,后面要接CrewAI或LangGraph之类的框架也更容易迁移。
工程结构我拆成了四个文件,核心思想是让模型定义、Agent逻辑、编排逻辑、入口脚本互相独立:
agent_team/ ├── models.py // 消息与任务数据结构 ├── agents.py // Agent基类和具体角色 ├── orchestrator.py // 编排器:派发、重试、汇总 └── main.py // 演示入口这个结构看起来简单,但足够支撑起后续扩展。等哪天角色多了,每个Agent单独一个文件也不会乱;编排逻辑复杂了,可以再抽出task_queue模块。
4.2 定义Agent骨架和消息协议
数据模型是一切的基础,我先定义Message类。它承担两个职责:既是一次会话中的消息单元,也是不同Agent之间传递结果的载体。
from pydantic import BaseModel, Field from datetime import datetime from uuid import uuid4 class Message(BaseModel): sender: str receiver: str = "any" task_id: str = Field(default_factory=lambda: str(uuid4())) msg_type: str = "text" # text | result | error content: str created_at: datetime = Field(default_factory=datetime.utcnow)然后定义Agent基类。每个Agent只需要做三件事:接收一条Message,读取自己的角色设定,调用LLM生成回复,再返回一条Message。这样设计的最大好处是Agent之间完全解耦,后续替换模型、调整Prompt都不会影响全局。
class BaseAgent: def __init__(self, role: str, system_prompt: str, model: str = "gpt-4o-mini"): self.role = role self.system_prompt = system_prompt self.model = model async def run(self, message: Message) -> Message: # 这里实际调用LLM response_text = await call_llm( model=self.model, system_prompt=self.system_prompt, user_content=message.content ) return Message( sender=self.role, receiver="orchestrator", task_id=message.task_id, msg_type="result", content=response_text )你可能会问,为什么不用现成的Agent框架?我的回答是:框架解决的是复杂调度问题,但如果你连最基础的消息结构都还没想清楚,被框架牵着走反而更容易翻车。自己搭的最小骨架,每一步都在掌控之中。
4.3 实现编排器:派单、聚合、重试
编排器是这个系统的核心,也是最花心思的部分。它要解决的问题有三个:把任务派给谁、等多久算超时、结果回来后怎么处理。
import asyncio from typing import Dict, List class Orchestrator: def __init__(self): self.agents: Dict[str, BaseAgent] = {} self.timeout = 30 def register(self, agent: BaseAgent): self.agents[agent.role] = agent async def assign(self, role: str, task: Message) -> Message: retry = 0 while retry < 3: try: return await asyncio.wait_for( self.agents[role].run(task), timeout=self.timeout ) except asyncio.TimeoutError: retry += 1 print(f"[orchestrator] {role} timeout, retry {retry}") return Message( sender="orchestrator", receiver="main", task_id=task.task_id, msg_type="error", content=f"{role} failed after 3 attempts" )真正的编排器会比这个复杂,但核心骨架就是上面这几行。三个Agent并行的调度逻辑,可以简化为这样:
async def run_team(self, task_desc: str): task = Message(sender="user", receiver="orchestrator", content=task_desc) # 第一步:研究员收集资料 research_result = await self.assign("researcher", task) # 第二步:分析师 + 审查员可以并行执行 analysis_task = Message( sender="researcher", receiver="analyst", task_id=task.task_id, content=research_result.content ) [analysis_result, critique_result] = await asyncio.gather( self.assign("analyst", analysis_task), self.assign("critic", Message( sender="researcher", receiver="critic", task_id=task.task_id, content=research_result.content )) ) # 第三步:写作者汇总 final = await self.assign("writer", Message( sender="analyst", receiver="writer", task_id=task.task_id, content=f"{analysis_result.content}\n---审查意见---\n{critique_result.content}" )) return final.content这一步明确展示了一个真实的多Agent并行协作流程:先串行触发资料收集,再并行执行数据分析和质量审查,最后交给写作者统一汇总。并行节点的增加能显著压缩端到端耗时,这是多Agent团队相比单Agent处理方式的一大优势。
4.4 用竞品调研场景做一次完整演示
把一个具体的场景跑通,比抽象讲十遍都有用。我这次选择的任务是:“生成一份关于智能客服机器人2025年市场趋势的简短调研报告”。
团队配置了三个Agent:
- 研究员Agent:负责从网络搜索接口获取市场数据,返回关键事实和数字。
- 分析师Agent:负责解读这些数字,计算同比增速,提炼出3条核心趋势。
- 写作Agent:负责把分析和数据组织成一段结构清晰的报告正文。
演示里我故意没有加审查Agent,方便展示基础版的输出效果。研究员初步返回的关键信息是一组真实感很强的片段:“2025年智能客服市场规模预计在50亿美元左右,年增长率约17%;头部厂商正在从规则引擎转向大模型驱动;客户最关心的是首响时间和回答准确率”。
分析师拿到这些资料后,输出的是结构化分析:“市场仍处于高速成长期,17%的增速意味着年增量约8.5亿美元;大模型替换规则引擎是确定性方向;准确率是用户留存的核心指标,预计会成为产品差异化重点。”
写作Agent再组织成正式段落,输出一段约三百字的短文。整个流程跑下来耗时约15秒,调用LLM三次,合计消耗token不到一万。这个效率放在单Agent场景里几乎不可能实现——单Agent要在一次上下文里完成检索、分析、写作,质量很容易稀释。
代码里真正需要调的核心参数我也列一下:temperature建议设为0.2到0.4之间,任务型Agent不适合太高;max_tokens要给足,尤其是写作Agent,太小会截断;超时建议按任务复杂度动态设置,简单问答10秒,研究报告30秒以上。这些参数没有绝对标准,但按我的经验,这样设置起步不会太差。
5. 常见问题与排查技巧实录
5.1 循环调用停不下来
多Agent系统最典型的问题就是循环。两个Agent讨论一个问题,你来我往,谁也不服谁,直到把上下文塞满或者预算耗尽。
我遇到得最多的是“对抗式角色循环”。比如一个Agent负责挑毛病,另一个负责辩护,如果没有终止机制,它能永远互相反驳下去。
解决思路不复杂:第一,从对话轮数上限做硬限制;第二,给每个Agent的System Prompt里明确写出退出条件,比如“当你认为信息已经充分,请输出你的最终结论,并在结尾添加 标记”;第三,在编排器里解析这个标记,一旦检测到就强制收束本环节。
我后来甚至把这种“评审-修订”打包成一个固定步骤:批评Agent先输出意见,写作Agent根据意见修订一次,然后直接进入下一环节,不搞无限循环。多Agent协作的本质是合作,不是辩论赛,别让角色设定带偏了系统目标。
5.2 上下文和token预算爆炸
第二个高频问题就是token成本失控。我见过有人把完整对话历史发给每个Agent,结果跑一次任务烧掉几万token,产出质量还不咋地。
这里有三条实用策略。第一,设置记忆级别,短期上下文只保存最近3到5轮,更早的信息通过摘要保留,彻底不要传原文。第二,限制工具返回的内容长度,搜索接口返回的结果先做截断或提取,只把关键事实传给Agent,而不是整篇网页文本。第三,在每轮Agent调用前用token计数库估算Prompt长度,超过预设阈值就触发压缩流程。
我还习惯给每个任务设置token预算,比如“整个调研任务最多消耗15000 token”,到达预算上限就直接走降级方案:用更小的模型、更短的输出、跳过非关键步骤。预算意识如果不在开发期养成,上线后成本一定会给你上一课。
5.3 多个Agent的结论互相打架
当研究员说“市场增长率约17%”,分析师说“我认为实际增速不会超过8%”,你信谁?这种结论冲突在多Agent系统里非常常见,本质原因是不同Agent拿到的资料口径不同,或者它们在用各自的经验值做补全。
我的处理手段是:数据类任务必须加事实核对步骤。用一个专门Agent去检查每个关键数字是否能追溯到来源,并强制要求它给出来源链接或者检索关键词。如果数字对不上,就以前置原始数据为准,后置解读只能引用不能篡改。换句话说,定义好“谁产生事实,谁产生观点,观点不能覆盖事实”。
另一个偏工程的手段是给Agent的输出定义JSON Schema。让重要结论必须是一个固定结构的数据对象,比如字段名、数值类型、来源、置信度。这样做不仅方便下游程序解析,还能从结构上逼着Agent减少含糊表达。
5.4 成本和延迟怎么控制
多Agent系统最容易被人吐槽的就是慢和贵。我测过最简单的三Agent协作任务,端到端也要10秒以上,复杂任务到30秒都很正常。
延迟优化有几个有效抓手。第一是并行化,把没有依赖关系的Agent并发运行,这基本能把耗时砍掉一半以上。第二是模型分级,简单任务比如格式整理、摘要压缩用小尺寸模型,复杂推理才用大模型,实践下来总成本能降六成。第三是局部流式返回,如果下游不需要等全部结果,把已经完成的内容先吐给用户,体验提升非常明显。
成本控制还有一个容易被忽略的点:合理设置Agent的最大输出长度。很多模型默认输出很长,导致token消耗大,但实际任务答案往往一段话就够。在Agent基类里把max_tokens根据任务类型调低,节省的成本你看账单就能体会到了。
6. 个人经验与后续扩展
6.1 我的三个建议
这个项目跑下来,我最想分享的三个建议,都是从实际踩坑得来的。
第一条,先从两三个Agent起步,不要一上来就搭十几个角色的“豪华团队”。我最初设计了六个角色,结果大部分冲突都发生在角色边界模糊的地方,后来砍到三个,整体稳定性肉眼可见地上了一个台阶。Agent越少,出问题的时候越容易定位。
第二条,每个Agent都要能单独“单测”。我给每个角色准备了一份固定输入样本,每次改完Prompt就跑一遍,确保它的输出结构没有偏离。Agent单测的断言重点不是内容准不准,而是结构对不对、关键字段是否存在,这能防止你说好A逻辑结果模型返回了B格式。
第三条,日志里尽量记录“意图”。除了Prompt和Response,我还让Agent输出一句“这一步我打算做什么”,这行日志在排查问题时价值极高。多Agent系统是个黑盒,任何能让你更快定位问题的信息都值得花一点token去换。
6.2 后续可以往哪个方向扩展
这个最小系统的下一步,我正在验证几个方向。
一个方向是接RAG,让Agent不仅能靠模型本身的知识工作,还能从私有知识库检索背景资料,这样行业调研报告的质量会明显提升。另一个方向是给Agent接更多真实工具,包括数据库查询、代码执行容器、定时触发器和外部搜索接口,让它们从“聊天者”变成“执行者”。还有一个方向是引入用户反馈闭环,把每次任务的最终结果和用户评分回灌到一个经验库,让Agent团队在后续任务中主动参考历史经验,这其实就是一种轻量级的持续学习机制。
另外,如果打算对接更开放的生态,消息层用MCP、Agent间走A2A类协议,可以让自己的Agent团队和外部系统互通。想象一下,你的写作Agent直接调用别人做好的行业数据Agent作为数据源,这种事情正在从设想变成现实。
我最大的体感是:多Agent协作不是堆数量,而是设计边界。边界清楚,三个Agent就能撑起一条完整的业务流水线;边界模糊,三十个Agent也只是三十场无休止的低效会议。如果你正准备动手,我建议从最小的三人小组开始,让它们在真实任务里先跑起来,再根据暴露出的问题一点点加规则——这比纸上谈兵式的架构设计靠谱得多。