尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

多Agent协作实战指南:从架构设计到代码实现

多Agent协作实战指南:从架构设计到代码实现 最近我被反复问到同一个问题一个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”针对同一个方案来回交换意见最后能产出一份同时包含机会和风险的综合建议。这种结构也更接近我们理想中的A2AAgent-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_factorylambda: str(uuid4())) msg_type: str text # text | result | error content: str created_at: datetime Field(default_factorydatetime.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( modelself.model, system_promptself.system_prompt, user_contentmessage.content ) return Message( senderself.role, receiverorchestrator, task_idmessage.task_id, msg_typeresult, contentresponse_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), timeoutself.timeout ) except asyncio.TimeoutError: retry 1 print(f[orchestrator] {role} timeout, retry {retry}) return Message( senderorchestrator, receivermain, task_idtask.task_id, msg_typeerror, contentf{role} failed after 3 attempts )真正的编排器会比这个复杂但核心骨架就是上面这几行。三个Agent并行的调度逻辑可以简化为这样async def run_team(self, task_desc: str): task Message(senderuser, receiverorchestrator, contenttask_desc) # 第一步研究员收集资料 research_result await self.assign(researcher, task) # 第二步分析师 审查员可以并行执行 analysis_task Message( senderresearcher, receiveranalyst, task_idtask.task_id, contentresearch_result.content ) [analysis_result, critique_result] await asyncio.gather( self.assign(analyst, analysis_task), self.assign(critic, Message( senderresearcher, receivercritic, task_idtask.task_id, contentresearch_result.content )) ) # 第三步写作者汇总 final await self.assign(writer, Message( senderanalyst, receiverwriter, task_idtask.task_id, contentf{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也只是三十场无休止的低效会议。如果你正准备动手我建议从最小的三人小组开始让它们在真实任务里先跑起来再根据暴露出的问题一点点加规则——这比纸上谈兵式的架构设计靠谱得多。
返回列表