多Agent协作架构选型与落地实践:从消息协议到状态机设计
2026/9/7 3:50:39 网站建设 项目流程

多agent协作这个词,近半年热度一直没下来。我身边已经有团队把需求分析、代码生成、评审、测试拆成四个agent,跑了一个季度的内部项目;也有人被多agent的token消耗吓到,转头回去用单个长上下文模型。我属于中间派:多agent确实有用,但前提是你要想清楚它解决的到底是哪个问题。这篇文章不聊宏大的“Agent宇宙”,就聊一个务实的idea——把多个AI agent组织起来协作时,架构怎么选、消息怎么传、状态怎么管、坑在哪里,以及一个可以直接照着改的框架设计。

如果你正在评估多agent方案,或者已经在为多agent协作里的乱象头疼,这篇文章应该能帮你省下不少试错时间。我会尽量用做过项目的人才说得出来的方式去讲,该给代码给代码,该给表给表,不绕弯子。

1. 多agent协作解决的到底是什么问题

1.1 一个从需求到测试的真实场景

我去年在做一个内部工具时,试过让一个通用agent一口气完成“读懂需求→输出技术方案→写代码→写测试用例”四步。结果代码能跑,测试用例却写得很水,几个关键边界条件全漏了。后来换成四个agent各管一段,质量明显上来了,但同时也冒出一堆协作问题:有人改需求没人同步、评审结论和开发实现对不上、某个agent卡住导致整条流水线停摆。

多agent协作最朴素的价值,不是“人多力量大”,而是把一个复杂任务拆分给多个有明确边界的执行者,每个执行者用自己独立的上下文处理一小块,再通过接口把结果喂给下一个环节。为什么单agent搞不定?三个原因最明显:上下文打架、角色冲突、验证失效。

单个agent不是不能做,而是它在做需求分析时脑子里要装着产品背景,写代码时还得记得需求细节,等写到测试用例时,前面的推理链早被长上下文稀释了。更麻烦的是角色冲突——让同一个模型自己写自己审,相当于学生自己批改自己的卷子,它对自己的错误天然不敏感。多agent把“提出方案”和“挑刺方案”分成两个独立上下文,等于引入了真正的对抗性验证。

1.2 核心价值拆解

我用一个表说清楚单agent和多agent的区别,这也是判断要不要上多agent的决策依据。

维度单agent多agent
上下文负担全流程共享,长任务容易稀释每个角色隔离,上下文更聚焦
角色冲突一个人包揽审查,自查盲区大写与审分离,盲区更小
并行度只能串行可并行处理多个环节
可验证性结果难拆开验证每个环节有明确交付物
成本相对低token消耗成倍增长

我的判断标准很简单:如果你的任务具备“上下文隔离、角色专业化、可并行、可验证交付物”中的任意一条,多agent就值得试;如果一条都不沾,硬上多agent只是增加系统复杂度。比如翻译一段话、整理一封邮件摘要,单agent又快又省,没必要硬拆。

多agent不是银弹,它更像生产线。只有当流程能被明确切分成工序时,分工才有意义。你没法把一瓶矿泉水包装成一条生产线,也没必要把一句话的需求拆给五个agent去协作。

2. 协作架构怎么选:三种模式对比与我的取舍

2.1 中心化编排:可控但容易变成瓶颈

最常见也最稳妥的是中心化编排模式。一个主控agent(orchestrator)负责拆任务、派给worker、收集结果、决定下一步。我最早跑通的多agent demo就是这个结构,可靠性最高。

主控agent的prompt骨架大概长这样:

你是一个任务编排者。你的职责是: 1. 将用户目标拆解为不超过5个子任务 2. 为每个子任务指定一个执行角色:产品/架构/开发/测试 3. 等所有子任务返回后,汇总并检查依赖 4. 如果有子任务失败,指派修复角色重试,最多重试2次

这段prompt写得好不好,直接决定整个系统的天花板。你不需要给每个worker写非常复杂的prompt,但主控的拆解逻辑一定要清楚,因为它决定了任务到角色的映射质量。子任务拆得越明确,worker执行就越不容易跑偏。

缺点也很明显:主控是单点。主控一旦对某个子任务理解错,下游全部跟着错;子任务一多,主控的token消耗和延迟都会变成瓶颈。我的处理方式是让主控只做“拆、派、收、汇”,不参与具体业务内容,把业务判断下沉到各worker。

2.2 去中心化对等:灵活但容易失控

去中心化对等模式下,worker之间可以直接互相丢消息,各自判断谁该接下一棒。适合探索型任务,比如让几个agent头脑风暴一个方案:A抛一个想法,B看到缺口就补,C负责质疑,最后得到一个互相碰撞的结论。

这个模式在原型阶段很好玩,但放进生产环境就要小心。没有固定流程,两个agent可能为了一个措辞来回争十几轮;一个agent的处理结果没有统一归口,最后得你自己拼图。我用P2P只做两件事:创意发散和问题诊断。真正要交付结果的任务,我不会让所有agent随便互聊,而是给他们一张“能找谁”的路由表,谁负责什么、谁能回答什么都写清楚。

2.3 共享黑板:适合知识密集协作

共享黑板模式是让所有agent读写同一块共享上下文,互动都发生在黑板层,agent之间不直接对话。比如做行业研究报告:检索agent把资料写入黑板,分析agent从黑板取数据写结论,写作agent再基于结论出稿。

这个模式的优势是上下文一致、方便追溯,特别适合知识密集、多方贡献的任务。隐患同样不小:并发写入冲突、脏数据污染。两个agent同时改同一段内容,后写的覆盖先写的,信息直接蒸发。解决思路是分区写权限——每个agent只能写自己负责的区块,其他区块只能读,改动走消息申请。

2.4 我自己的选型逻辑

模式优点缺点适用场景
中心化编排可控、易调试主控瓶颈、单点风险流程明确、工序固定
去中心化对等灵活、可探索难调试、易死锁头脑风暴、开放探索
共享黑板上下文一致、可追溯并发冲突、上下文污染知识密集、多方贡献

我的原则是:流程确定用编排,问题开放用对等,知识密集用黑板。真实生产系统通常不是单一模式,而是编排为主,局部引入黑板或对等。比如主控编排全局流程,但某个子任务内部让三个agent用黑板方式并行研究,最后再把结论汇总回主控。

3. agent之间怎么说话:消息协议、上下文与工具调用

3.1 消息格式设计

agent之间通信,本质是消息传递。消息格式就是你们的接口契约,定义得脏,后面调试全是泪。我常用的消息格式长这样:

{ "msg_id": "msg_20250201_001", "trace_id": "trace_8f3a", "from": "architect", "to": "developer", "msg_type": "task", "payload": { "task_id": "t_001", "requirement": "用户登录支持验证码", "acceptance": "验证码5分钟内有效,错误3次锁定" }, "created_at": "2025-02-01T10:00:00Z" }

字段看起来多,实际每个都有用。msg_id是唯一消息ID,用于去重和回执;trace_id贯穿整条协作链路,排错时靠它在日志里把一串消息串起来;from/to决定路由;msg_type区分task、result、error、request_info;payload放业务数据。

回传结果的消息也要规范:

{ "msg_id": "msg_20250201_010", "trace_id": "trace_8f3a", "from": "developer", "to": "architect", "msg_type": "result", "payload": { "task_id": "t_001", "status": "success", "artifacts": ["/code/login.ts"], "summary": "已完成验证码登录功能" } }

要注意的是,错误消息不能只写“失败了”,要带上error_code和可执行的下一步建议,否则下游agent收到错误也不知道该干嘛。

3.2 A2A与MCP:两个容易混淆的协议

最近AgentScope 2.0这类框架开始支持A2A模式,很多同学一上来就把它和MCP搞混。一句话区分:A2A管agent与agent之间怎么互相发现、派活、回传结果;MCP管agent怎么调用外部的文件、数据库、API。A2A更像你和同事之间的工作对接规范,MCP更像你操作电脑的键盘鼠标。

协作系统里两个都会用到:agent之间用A2A互相派活,干活时用MCP工具读写外部资源。如果你自己写协议,可以参考A2A的message语义(task、result、error),但没必要照搬全部字段,够用就行。协议这东西,简单可靠永远比大而全实用。

3.3 上下文共享与冲突处理

消息协议解决的是“怎么传”,上下文共享解决的是“怎么共”。我坚持一个原则:每个agent的私有上下文独立,全局目标共享。共享上下文用事件驱动更新,而不是各写各的。

举个例子就明白了。两个agent都在写同一份架构文档,agent A把第3节改成MySQL,agent B同时把第3节改成PostgreSQL,后写入的覆盖先写入的,之前的信息直接蒸发。解决方式:共享上下文不能给所有人写权限,要按模块owner授权。谁负责数据库选型,谁才有那段内容的写权限,其他人只能通过消息请求修改。

如果你做过团队协作平台,会发现这和多人实时文档同步是同一套逻辑。react + nestjs + socket.io做多人协作时,文档同步靠的是操作转换和版本号,不能靠最后写入直接覆盖。agent协作的共享上下文,就是同一逻辑的AI版本。

4. 落地时最容易翻车的三个环节:状态、验证、死锁

4.1 状态一致性:每个状态必须有唯一owner

多个agent共用一个任务状态字段,最容易互相覆盖。任务状态从PENDING变成IN_PROGRESS,到底由执行agent改还是由编排者改?必须有一个唯一owner。我的习惯是把状态机放在编排层,agent只上报结果,不改状态。

如果你碰到“任务被两个agent同时标记为完成,一个报告成功一个报告失败”,那基本就是状态owner不明确导致的。要么加锁,要么用版本号,要么干脆把状态修改权限收拢到单一节点。多人协作平台里一条工单不能两个人同时改,agent协作同理。

4.2 信任与验证:agent的输出必须被验证

大模型会一本正经地胡说八道,这是所有多agent系统必须正视的底色。多agent还引入了一个额外风险:错误会沿着消息链被放大。A产生一份错误的需求理解传给B,B基于错误理解生成设计,C再在错误设计上写代码,最后整条链都是错的,而且每个agent都觉得自己做对了。

所以系统设计里必须加验证层。验证可以是规则校验(JSON schema校验、正则、编译、单测),也可以是另一个agent(代码review agent),还可以是人。我的建议是:能用确定算法验证的,绝不用LLM验证LLM;LLM验证只能作为补充。

以自动化测试为例:开发agent生成代码后,先跑编译和静态检查,再跑核心单测,这些是确定性的;如果都过了,再让测试agent设计边界用例做补充验证。顺序不能反,先机器后模型。不验证的协作就是事故放大器。

4.3 死锁与循环:协作系统也会卡死

多agent协作也会出现类似并发系统的死锁。A在等B提供接口文档,B在等C确认字段格式,C又在等A给最终需求。三个agent各自都认为自己blocked,整个流程就卡住了。

还有一种更烧钱的情况:争论循环。评审agent认为代码要重构,开发agent认为不用,两边来回掰扯,每轮都在消耗token,谁也说服不了谁。我管这类问题叫“协作湍流”。

问题现象常用解法
互相等待A等B、B等C、C等A设置依赖超时、路由表明确“找谁”
争论循环多个agent反复互怼最大轮次上限、语义相似度检测打断
失败重试风暴同一任务无限重试重试次数硬上限,超过转人工

解决思路就两条:限制自由度和保留终止条件。没有终止条件的多agent协作,就是一台停不下来的烧钱机器。

5. 并发与成本:多agent不是免费午餐

5.1 并发配置:别把所有agent一起拉满

很多团队一上来就追求高并发,恨不得10个agent同时跑。厂商API有速率限制,超了就是限流报错,重试逻辑没写好,编排系统先雪崩。用信号量限制并发是最直接的办法:

import asyncio semaphore = asyncio.Semaphore(4) async def run_agent_with_limit(agent_name, task): async with semaphore: try: return await call_agent(agent_name, task) except RateLimitError: await asyncio.sleep(2) # 简单退避,生产环境建议指数退避 return await call_agent(agent_name, task)

同时要根据角色分优先级:关键路径上的agent优先拿并发额度,辅助性质的agent(比如润色、摘要)排在后面。并发不是越高越好,我实测里4个并发对多数业务系统已经够用,再高只是把压力转移到下游。

5.2 Token成本与预算控制

多agent的第一个错觉是“多个agent并行肯定比单agent快”,但token消耗是乘法不是加法。假设4个agent各跑3轮,每轮平均2000 token,一个任务就是24000 token。如果中间反复重试,轻松翻倍。

我建议做三层预算:任务预算(单个任务最多跑几轮)、角色预算(单个角色最多出多少token)、总预算(一天跑多少任务)。超了直接转人工,而不是继续循环。

还有一个性价比做法:把任务四象限化。横轴是复杂度,纵轴是风险。低复杂度低风险的任务,一个便宜模型单agent直接做;低复杂度高风险,单agent做但要加验证;高复杂度低风险,可以多agent但不强求重活;高复杂度高风险,才上完整的多agent协作流程。这就是我理解的“四象限协作提示词”——它本质不是提示词技巧,而是任务分级策略。别让写周报的agent和写核心代码的agent用同一档次模型,低价值任务换便宜模型,能省一大笔。

5.3 可观测性:没有日志的协作系统没法排错

多agent系统出问题,最怕的就是不知道哪一环出了错。trace_id贯穿始终,日志带上角色、任务、状态。我常打的日志字段长这样:

字段含义
trace_id整条协作链路ID
agent_role哪个角色
task_id哪个子任务
event开始/完成/失败/重试
token_usage消耗量
latency_ms耗时

如果你用了消息队列,记得把消息内容也持久化一份,方便事后回放。我踩过最深的坑是:线上协作出问题,日志只有结果没有过程,根本没法定位。多agent系统排错靠的是日志,不是猜。

6. 一个从零可搭的协作框架:角色定义、状态机与人工介入点

6.1 用四要素定义每个agent角色

角色定义的模板我固定用四要素:职责、输入输出、验收标准、退出条件。职责说要干什么;输入输出写清楚格式;验收标准要可量化;退出条件决定它卡住时找谁。

角色:测试工程师 职责:根据需求文档和接口文档编写测试用例 输入:需求文档(markdown)、接口文档(OpenAPI) 输出:测试用例集合(JSON) 验收标准: - 用例覆盖核心场景和至少3个边界条件 - 每条用例包含前置条件、步骤、期望结果 - 输出JSON通过schema校验 退出条件: - 缺少接口字段信息时,向开发者角色请求补充,最多2次 - 超过2次仍缺少,标记NEEDS_HUMAN

很多团队写角色提示词只写职责,不写退出条件和验收标准,结果就是agent卡住后无限重试,或者输出一堆没法用的东西还自我感觉良好。

6.2 任务状态机:让每一步都有迹可循

我把任务生命周期定成7个状态:PENDING、ASSIGNED、IN_PROGRESS、BLOCKED、DONE、FAILED、NEEDS_HUMAN。

状态触发条件下一步
PENDING任务创建,等待派发编排者派发 -> ASSIGNED
ASSIGNED已指定执行角色角色开始 -> IN_PROGRESS
IN_PROGRESS执行中完成 -> DONE / 缺信息 -> BLOCKED / 失败 -> FAILED
BLOCKED缺少依赖信息请求信息补充 -> IN_PROGRESS / 超时 -> NEEDS_HUMAN
DONE输出通过验收流转到下游任务
FAILED连续重试失败转人工或结束
NEEDS_HUMAN需要人工决策人处理后重新派发

状态机不是画给领导看的,是让系统在任意时刻都能回答:这个任务卡在哪一步,该谁动。

6.3 人工介入点:别太早,也别太晚

多agent系统再有自动化,也必须有闸门。我固定的介入点有四类:高危操作前,比如删除数据、发版本、对外发消息,这类操作必须过人工确认,哪怕agent已经输出完整方案;同一子任务连续失败两次,大概率是目标本身有问题,再跑就是烧钱;多agent意见分歧,评审与开发争论超过2轮还没收敛,直接人工拍板;成本超阈值,单个任务token消耗超过预算上限,立刻转人工。

人工介入不是兜底,是控制节点。介入太早,你成了人肉agent;介入太晚,事故已经发生。

6.4 一个最小可参考的实现骨架

如果不依赖重型框架,用Python写一个最小闭环大概长这样:

from dataclasses import dataclass from enum import Enum class TaskState(Enum): PENDING = "PENDING" ASSIGNED = "ASSIGNED" IN_PROGRESS = "IN_PROGRESS" BLOCKED = "BLOCKED" DONE = "DONE" FAILED = "FAILED" NEEDS_HUMAN = "NEEDS_HUMAN" @dataclass class Task: task_id: str role: str payload: dict state: TaskState = TaskState.PENDING retry_count: int = 0 result: dict = None def run_pipeline(tasks, executor, max_retry=2): for task in tasks: if task.state != TaskState.PENDING: continue task.state = TaskState.ASSIGNED while task.state in (TaskState.ASSIGNED, TaskState.IN_PROGRESS): result = executor.run(task.role, task.payload) if executor.validate(task, result): task.result = result task.state = TaskState.DONE else: task.retry_count += 1 if task.retry_count > max_retry: task.state = TaskState.NEEDS_HUMAN else: task.state = TaskState.ASSIGNED return tasks

这段代码在生产环境肯定不够用,真正的系统要加消息队列、持久化、trace、并发控制。但核心逻辑就是这个循环:派发、执行、验证、重试、转人工。

如果你对团队协作平台比较熟,可能会发现这套东西和你在react + nestjs + socket.io里做的多人实时协作没有本质区别:都是参与者在不同状态之间流转、通过消息互相通知、在共享资源上做协同。区别只是参与者从人变成了agent,而agent比人更需要明确的契约和更强的校验。

最后说一点我个人体会。多agent协作最难的从来不是技术选型,而是你愿不愿意把“这个流程到底该怎么走”想清楚。人协作时可以靠默契补位,agent协作没有任何默契,你定义不清楚的地方,它就用token帮你试错。所以我的建议是:先用最小角色集跑通一个真实任务,再加复杂度。你不需要第一次就做出一个十全十美的agent宇宙,能把一个平时要花两小时的任务,用多agent可靠地缩短到二十分钟,就已经赢了。

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

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

立即咨询