这年头聊AI代理的人不少,但真正把AI代理放到“协同作战”场景里、还要让多个人的多个AI代理互相配合去办成一件事的项目,其实还是少数。我最近就在折腾这样一套系统架构:让多个不同角色、不同归属的AI代理部署在同一套体系下,代替各自的人类去发起交互、协商、认领任务、执行动作,而人只需要在关键节点给一个明确授权。说白了,就是AI代理替人跑腿,人多、AI也多、活还得协同着干。这篇文章就是我对“基于AI代理代为交互的多人多AI协同系统架构”的一份完整研究笔记,包含需求拆解、架构取舍、核心模块设计、可复现的最小原型步骤,以及我踩过的几个典型的坑。
这套东西能解决什么问题?最直接的一个场景:以前一个跨部门任务,需要几个人拉群、开会、对齐、催进度,耗时短则半天,长则一周。现在每个人的AI代理可以直接对接收发消息,自动拆分工作项、认领任务、汇总结果,再把待确认的决策项抛回给人。它解决的是“多人在场但执行链路太长”的效率问题,同时又把“AI自治”约束在人类授权的边界内。适合正在做多Agent系统、考虑Agent编排、企业智能助理或自动化运营的朋友参考,也适合那些带团队做系统架构的技术负责人拿来当需求评审的清单。
1. 先拆清楚需求:多人多AI协同到底要解决什么问题
1.1 从“单人单Agent”到“多人多Agent”的跃迁
大多数人对AI代理的认知还停在第一阶段:一个人面对一个Agent,问答也好、写代码也好、整理文档也好,交互对象始终是人和AI之间的单一对话。这个阶段里Agent再强,本质还是个“副驾”,大部分动作需要人逐步驱动,Agent不会主动找别人、也不会跟别的Agent沟通。
第二阶段是“单人多Agent”,一个用户同时调度多个分工不同的Agent,比如一个负责查资料、一个负责做分析、一个负责写报告。这时候Agent之间还是靠人来转发信息,协调者是人,Agent只是工具。我做了几个项目之后发现,这个模式很快会碰到天花板:人成了整个系统的瓶颈,所有信息的转发、任务的派发、结果的核对都堆在人身上,Agent虽然多但并没能真正分担协同工作。
真正值得做的是第三阶段:“多人多Agent”。在同一个全局架构下,不同的人拥有各自的AI代理,这些代理可以直接对话、交换中间结果、协商任务归属。人类从“每一步都要管”退到“只定目标、划边界、审结果”。这个跃迁带来了三个质变。第一,交互模式从人机对话变成了Agent对话,你发出目标之后,跑腿的是Agent;第二,系统从单机逻辑变成了分布式系统,涉及消息路由、状态一致性、超时重试、故障恢复,复杂度完全是另一个量级;第三,责任边界变了,没有人类明确授权的Agent动作是危险的,你必须从架构层面让人全程可介入、可追溯、可回滚。
我举一个实际设想过的场景:一个产品团队里,产品经理、后端开发、测试三个人各挂一个代理。产品经理的Agent在群里发出一条任务:“下周二前完成订单模块的压测方案。”后端Agent自动拆解出需要提供的接口文档,测试Agent自动认领压测环境准备工作,三个Agent在后台交换了几条消息之后,把一份分工明确的执行计划推回到三人的审批流里。人只需要各自点“同意”或“调整”。这套流程如果用传统方式走,光是“拉齐三人的时间”就得花半天。
1.2 核心需求与边界梳理
动手设计架构之前,我先列了一份需求清单,这份清单后来帮我避开了很多设计返工。总共五条核心需求:
- 多人并行:多个人的多个Agent可以同时工作,谁也不用排队等谁。
- 多AI协商:Agent之间要有标准语言和协商流程,能讨论、能讨价还价、能拒绝。
- 人工可控:人在关键节点可以随时介入、中止、回滚Agent的动作。
- 可追踪:任何一次协作都能回放,结果能被审计。
- 数据隔离:不同人或部门的数据不能串,同一个任务内部的数据则要共享。
第一和第二条决定了系统必须采用异步消息机制,不能用简单的函数调用来串联Agent。第三和第五条决定了必须有严格的权限模型和审批闸门。第四条看起来最不起眼,恰恰是生产环境上线后最重要的能力,没有审计日志,出了问题就只能靠猜。
与此同时,我还明确了四个边界条件,这四个词后来成了整个架构设计的骨架:身份边界(Agent代表谁)、目标边界(Agent要达成什么)、权限边界(Agent能做什么)、记忆边界(Agent能知道什么)。这四个边界一定要先写清楚再动手,否则后面每加一个Agent,都会出现一次权限和数据的混乱。
我特别建议做这类项目的人,第一份交付物不是架构图,而是“同步协议”文档。这份文档明确三件事:谁跟谁说话、说话用什么格式、什么时候必须停下来等人类。文档比代码先写,能省掉后面大量的重构。
2. 架构设计的四个关键决策,以及我为什么这么选
2.1 分层架构:接入、代理、协同、数据四层拆分
我最终采用的分层架构可以概括为四个层次:接入层、代理层、协同层、数据层。接入层面对所有人和外部系统,负责身份的认证、消息的收发;代理层负责Agent本身的运行,包括接收消息、调用模型做决策、执行工具调用、维护短期记忆;协同层是整个系统的心脏,负责消息路由、任务分发、仲裁、人工闸门和状态管理;数据层则承载所有结构化数据、事件日志和向量知识库。
这个分层思路我很大程度上是借鉴了“分布式交换机系统架构”的设计理念。交换机把控制面和数据面分开,数据面只管转发,控制面集中管理策略。对应到多AI协同里,代理层就是数据面,只管“干活”,协同层就是控制面,管“谁在什么条件下允许干什么”。控制面与数据面分离之后,任何一个Agent更换模型、升级工具,都不需要动到其他Agent内部的东西,只要它对外发消息的协议不变就行。
层与层之间的交互,我坚持用消息而不是函数调用。函数调用看着直接,但会把各层耦合成一个巨型单体;消息则天然解耦,每层只关心自己订阅的数据。数据流的典型路径是这样的:人类A在接入层发出一条任务指令,代理层A收到目标后,向协同层创建一个任务事件;协同层的路由器根据能力标签把子任务分给代理B、代理C;代理B、C各自执行完,把产出用消息发回协同层;协同层聚合出结果,发起人工审批请求;人类审批通过后,任务状态更新,事件落库。整个链路里没有任何一层直接调用另一层的内部方法,全部通过消息传递。
2.2 通信模型选型:事件总线和点对点直连的权衡
最直觉的通信方式是点对点直连,Agent A直接给Agent B发一条HTTP请求。好处是好理解、延迟低、实现起来快,但坏处很快会在系统变大之后集中爆发。首先是耦合问题,多个Agent互相调用会形成网状依赖,改一个接口就要通知所有调用方;其次是观测问题,点对点调用没有一个统一的地方做监控和全链路重放;再就是容错问题,只要有一个Agent挂了,所有依赖它的调用方都会跟着失败,故障面会像多米诺骨牌一样扩散。
所以我的选择是“事件总线为主、点对点为辅”。所有任务类、状态类、审批类的消息都走事件总线,保证可追溯、可重放、可广播;协商过程中一次性的请求回复则可以用点对点方式,降低延迟。通信层我用的是NATS,它在原型阶段非常合适:部署就是一个二进制文件,自带JetStream可以做消息持久化,还内置了请求-回复模式,既支持总线订阅也支持点对点,一鱼两吃。
这里有一个特别容易踩的设计错误:事件总线不等于所有消息都广播给所有人。总线只是通信管道,路由键必须严格设计。我见很多项目把所有Agent都订阅到同一个通配主题上,结果Agent A读到了Agent B的任务,Agent B看到了Agent C的审批内容,数据直接串了。正确的做法是至少用两级路由:一级面向任务域,比如task.created、task.assigned;一级面向具体实例,比如task.{id}.update、agent.{id}.message,让每个Agent只处理跟自己相关的那一部分。
2.3 协调策略:编排、协商和人在回路怎么配合
多人多AI系统里的“协调”是决策密度最高的一块,我从实践里总结了三种策略:编排、协商、人在回路。编排适合流程明确的任务,由一个编排器或者总控Agent负责拆任务、派活、收结果。它的优点是结构清晰、状态推进可控,缺点是总控会变成单点瓶颈,而且总控Agent的决策能力直接决定任务质量。协商则适合开放性问题,没有总控,多个Agent平等地交换提议、反馈和修订,直到收敛。这个模式像一群人开会,能处理编排解决不了的模糊需求,但容易陷入死循环,两个Agent都有礼貌地坚持自己的观点,谁也不让步。
我的经验是:先用编排把流程跑起来,再把确定性强的环节全部自动化,最后把不确定性环节交给协商。不要开局就上全自动协商,那基本会在细节里卡死。举个例子,一个“写跨部门周报”的任务,数据收集环节是确定的,直接用编排路由给数据Agent;而“这周的亮点怎么写”是开放性的,可以让两个内容Agent协商出一版初稿。
不管用编排还是协商,系统里必须有一条贯穿始终的安全线——人在回路。我把人工介入设计成三个层次:审批是“做完之前等确认”,适合高风险动作;监控是“正在做但随时可打断”,适合长时间任务;回滚是“做完了还能撤销”,适合带了痕迹系统的操作。在原型阶段,我建议把审批做全,所有敏感动作都必须经过一个统一的人工审批中心,系统上线初期宁可多审几次,也不要为了省事漏掉一个关键闸门。
3. 核心模块落地的细节与避坑要点
3.1 Agent身份建模和代理关系的处理
我见过不少多Agent项目的失败,本质就是身份模型没理顺。一个Agent绝对不能只是一个“能聊天的程序”,它在系统里必须有完整的身份档案。我给每个Agent设置了四个核心字段:identity(ID、名称、归属人)、capability(能力标签)、status(空闲、忙碌、等待审批)、memory(引用短期会话和长期知识库的ID)。
最关键的是“代理关系”的设计。每个Agent都是某个人的数字代表,因此它发出的任何消息都必须携带代表身份。实际实现时,我要求消息体里必须同时包含agent_id和principal_id,前者是代理的身份,后者是人类主人的身份。这样任何一个收到消息的Agent都能立刻判断这条消息到底是谁在代表谁发声,系统也才能在权限校验时判断“这个Agent有没有权力做这件事”。
有人问过我可不可以让一个Agent代表多个人,我通常建议默认1:1,一个Agent绑定一个人。一个人的所有权限、偏好、数据边界都清晰挂在这个Agent下面。如果确实有“多人共用一个Agent”的诉求,比如一个会议助手,那就给Agent增加multi-principal状态,但每个具体动作仍然要落实到某个自然人身上。这个设计原则能最大程度避免“AI到底替谁做主”的责任模糊问题。
3.2 Agent之间说同一种语言:消息协议与会话上下文管理
多个Agent协作,第一个要解决的是“沟通语言”。我给所有Agent定义了一个统一的消息信封(Envelope),所有消息都套这个壳:
{ "trace_id": "a1b2c3d4e5f6", "from": "agent.rd.alice", "to": "agent.qa.bob", "type": "task.propose", "task_id": "order-stress-test-2025", "payload": { "summary": "计划下周二完成压测方案", "deps": ["接口文档"], "deadline": "2025-06-10T18:00:00Z" }, "auth": { "principal": "user.alice", "approved": false, "signature": "sha256-xxx" }, "created_at": "2025-06-03T09:30:00Z", "schema_version": "1.0" }type字段我至少预设了task.create、task.assign、task.deny、task.complete、approval.request、approval.approve、approval.reject、message.negotiate、message.notify。一定不要把消息类型设计得太少,否则所有业务逻辑都挤在“通用消息”里,后面做路由和审计的时候会痛不欲生。
上下文管理是多人多AI协同里最容易翻车的环节。我把它切成三档:公共上下文是所有参与者共用的一份任务说明,包括目标、背景、约束、产出物标准;私有上下文属于每个Agent自己的草稿、过程数据、内部想法,其他Agent不准读;角色上下文是每个Agent的职责说明书,比如“你是测试Agent,你只负责压测方案和执行结果,不负责写业务代码”。三档分开存,公共上下文在任务空间里,私有上下文在Agent自己的存储里,角色上下文在Agent配置里。
我还专门做了“上下文瘦身”机制。多个Agent在同一任务里进进出出,如果每个Agent都把公共上下文完整拷贝进自己的提示词,几个来回之后上下文体积就成倍膨胀,模型质量急剧下降,Token消耗也扛不住。我的规则是:给Agent的提示词里只放它真正需要的摘要,原始上下文存放在存储层,Agent需要细节时按ID查询,绝不整体复制粘贴。这个规则在原型阶段帮我省了大量调参时间。
3.3 任务分配、仲裁机制和人工介入通道
任务分配不适合写死“这个任务一定给Agent B”,因为Agent会挂、会忙、可能会新增。我用一个简单的路由策略:能力标签匹配 + 当前状态过滤 + 历史成功率排序。一个任务的描述先经过一个轻量提取器,提取出需要的能力标签;然后在Agent注册表里筛出有对应能力的候选者;再过滤掉状态为忙碌的和近24小时成功率过低的;最后把任务派给综合排序最靠前的那个。整个过程看起来不复杂,但能把“谁的Agent能干这活”从静态配置变成动态决策。
仲裁机制往往是人们最容易忽略的。两个Agent各执一词,一个说按方案A走,一个说按方案B走,系统怎么办?我们的第一版设计是让两个Agent继续协商,后来发现简单问题两轮之内还能达成一致,关键分歧上就会互相礼貌地僵住。第二版改成了“协商不超过两轮,第三轮自动升级给人类裁决”。这个机制配合人工审批中心,把决策权归还给真正需要对人负责的人。
人工介入通道本身也必须认真设计。我强烈建议做一个统一的审批中心,可以是Web面板,也可以是IM机器人。所有approval.request事件都汇聚到这个入口,人类在手机上一眼看到“谁在代表谁、要做什么、需要什么信息、有几个选项”。可选项的设计一定不能少,我在实践中发现,光是“同意”和“拒绝”两个按钮是不够的,“修改后批准”这个选项能避免很多次反复提交。
3.4 可观测性不是可选项,它是生命线
到了多人多Agent协同时代,“黑盒”是不可接受的。我的做法是把所有Agent动作当作事件流落库,这里用到的是Event Sourcing的思路:谁、在什么时候、以什么身份、调用了哪个工具、输入输出是什么、有没有被人类批准,全量记录。任何一次线上事故都可以通过回放事件流定位到具体环节,而不是靠某个人回忆。
这条链路里trace_id是绝对的灵魂。一次跨部门周报任务会牵动数据Agent、内容Agent、审批人、可能还有外部API,如果没有一个贯穿始终的trace_id,排查问题就像大海捞针。我的习惯是:任务创建时生成一个trace_id,之后这条任务链路里的每一条消息、每一次工具调用、每一次审批请求都带上它。排查的时候只需要在日志系统里输入一个ID,所有相关记录就全出来了。
审计日志要额外标记每个对外动作的审批状态:未审批、已批准、已拒绝、已强制回滚。这四个状态至少要在事件表里占一个独立字段,因为很多后续的追踪和报表都依赖这个分类。我在实际排查过的案例里,90%的疑难问题最终靠的都是“按trace_id过滤事件日志”这一个操作,这个习惯真的一定要养成。
4. 从零搭一个最小可运行版本:选型、步骤与验证
4.1 技术选型:一套可以用到生产环境的组合
原型阶段不要一上来就铺很重的技术栈,但也不能太玩具,否则验证完结论也不可信。我推荐一套既快又稳的组合:
| 组件 | 推荐选型 | 选型理由 |
|---|---|---|
| 消息总线 | NATS + JetStream | 部署简单、支持持久化、内置请求回复模式,后续可平滑换Kafka |
| Agent运行时 | LangGraph 或 自写状态机 | LangGraph适合快速验证,自写状态机适合深度控制 |
| 模型接入 | OpenAI兼容API + 本地模型(Qwen/Llama) | 云端API灵活,本地模型省钱且数据不出域 |
| 结构化存储 | PostgreSQL | 存Agent档案、任务状态、权限元数据 |
| 向量存储 | pgvector | 让Agent能做记忆检索,又不用单独维护一套向量库 |
| 审批面板 | FastAPI + 单个HTML页面 | 原型阶段够用,面板只是审批入口 |
| 容器部署 | Docker Compose | 一条命令拉起全部组件,x86和ARM架构都能跑 |
为什么我把NATS排在通信层首选?因为它真的轻,一个二进制文件就能跑,JetStream提供持久化相当于自带消息队列,request-reply模式又天然适配“Agent问Agent答”的场景。相比RabbitMQ和Kafka,NATS几乎没有运维心智负担,非常适合原型起步。模型接入这块我提醒一句,不要只盯着云端API,“AI代理助手加本地模型”的组合现在很成熟,本地模型的输出质量在不少垂直任务上已经够用,还能避免把内部数据送到外部。
如果想探索Agent控制真实设备的场景,还可以把ROS生态接进来,让Agent通过ROS话题订阅和发布控制指令。ROS本质上也是一套消息总线,和协同层的事件总线是同构的,做好协议适配后,Agent既能写文档也能操作设备,扩展性一下子就打开了。
4.2 最小可运行版:搭一个能跑通全链路的骨架
第一步,定义消息协议。先写一个消息定义文件,把Envelope的字段、消息类型、任务状态枚举全部定好,用Python dataclass或TypeScript接口都行。协议是整个系统的“宪法”,这一步不要省。
第二步,搭Agent运行时骨架。每个Agent类至少实现四个方法:receive(收消息)、think(调模型做决策)、act(执行工具调用或动作)、remember(写记忆),再加上一个空的事件循环,从总线订阅自己的消息。这个阶段不要让Agent做复杂工具调用,先用一个简单的工具证明链路能走通。
class BaseAgent: def __init__(self, agent_id, principal_id, capabilities): self.agent_id = agent_id self.principal_id = principal_id self.capabilities = capabilities self.status = "idle" async def receive(self, envelope): self.status = "busy" decision = await self.think(envelope) await self.act(decision) self.status = "idle" async def think(self, envelope): # 调用模型,返回动作描述 pass async def act(self, decision): # 执行工具,发送消息 pass第三步,搭协同总线。启动NATS容器,让每个Agent订阅自己相关的主题。订阅规则要严格,比如task.{agent_id}.assign和approval.{agent_id}.request,不要用通配符一把梭。第四步,加人工介入面板。FastAPI写一个/approvals端点,从队列里读待审批事件,渲染成简单列表;审批动作发布approval.approve或approval.reject事件。
第五步是很多人会忽略的:跑场景验证。建两个Agent,一个发起任务,一个接收执行,中间过一道人工审批。把全链路所有事件打印到屏幕上,肉眼确认每条消息从哪里来、流向哪里、状态怎么变化。这一步能快速暴露协议漏洞和路由错误,我每次搭新系统都会先做一遍。
4.3 一个场景跑通的全过程
我拿“三Agent协作出一份跨部门周报”来演示整个事件流。
第一步,Agent R(代表产品经理Alice)在接入层发起任务“生成跨部门周报”,公共上下文包含目标、时间范围、汇报对象。协同层创建任务,生成trace_id,状态置为pending。第二步,路由器按能力标签把“数据收集”子任务派给Agent D,“文本合并”子任务派给Agent W,双方都收到task.assign事件。第三步,Agent D从项目管理系统、周报系统里拉取数据,产出一份结构化数据,发task.complete;Agent W收到数据后生成周报草稿,发起approval.request,等待Alice确认。第四步,Alice在审批面板里看到摘要、全文、以及“同意/修改后批准/退回”三个按钮,点了“同意”。审批通过后,Agent R广播一条任务完成消息,所有事件落库,任务状态变为done。
# 伪代码:审批中心处理逻辑 async def handle_approval_request(envelope): task_id = envelope["task_id"] await save_pending_approval(task_id, envelope) # 等待用户在面板点按钮 result = await human_decision(task_id) if result == "approve": await nats.publish(f"task.{task_id}.approved", envelope) else: await nats.publish(f"task.{task_id}.rejected", envelope)整个过程中trace_id始终不变,事件表里新增了十几条记录,任何一步都能重放。这就是我判断“系统可运行”的标准:不是Agent多聪明,而是人在任何节点都能看到发生了什么、都有机会介入,并且所有动作都有据可查。
5. 常见问题排查实录与三条独家技巧
5.1 我踩过的坑,都值得你记下来
第一个坑是任务重复执行。消息总线通常提供“至少一次”语义,网络抖动或消费者崩溃后,同一事件可能被再次投递。我第一版没做幂等,一个压测任务被两个Agent同时执行了两遍,环境里生成了两份完全不同的报告。后来在消息处理入口统一查task_id是否已处理,用Redis做去重,问题才解决。
第二个坑是上下文串扰。几个Agent订阅了同一个任务级别的Topic,结果各自读到了别人的私有上下文。排查时发现是订阅粒度太粗。修法是把Topic按agent_id分区,私有上下文永远留在各自存储里。
第三个坑是模型API超时。Agent请求外部大模型接口超时后,任务状态卡停在busy,后续消息全部阻塞。后来在每次调用外接模型时设置明确超时和重试策略,并把Agent状态机改成“超时后自动置为failed并通知人类”。这一点对“AI代理助手加本地模型”混合部署尤其重要,本地模型推理慢,超时容忍度要单独调,不能跟云端API用同一套参数。
第四个坑是权限失控。原型阶段Agent调用了一个外发API,结果它把内部文档直接打包外发了。虽然只是测试环境,但给我们狠狠敲了一次警钟。从那以后,所有外向型动作一律过审批闸门,无论调用的是外部HTTP API、发送邮件还是通知IM群。这个规范对所有模型都适用,本地模型也一样。
第五个坑是仲裁死循环。两个Agent就“2025年预算口径”争执不下,互相发送了十几轮协商消息,谁也不服谁。从那以后我把协商轮数上限设为两轮,两轮后自动升级给人类裁决。协商是手段,收敛才是目的,架构上一定要有收敛机制。
5.2 问题排查速查表
| 症状 | 可能原因 | 排查方法 | 修复建议 |
|---|---|---|---|
| 任务被重复执行 | 消费端未做幂等处理 | 按task_id查Redis日志 | 消费前查重,用task_id做幂等键 |
| Agent读到了不该读的消息 | Topic路由粒度太粗 | 查看Agent订阅的主题列表 | 按agent_id分区订阅 |
| 任务长时间pending | 审批请求无人处理或超时 | 查approval队列 | 给审批请求加超时提醒和升级 |
| 模型输出质量突然下降 | 上下文里混入了无关内容 | 查最近N条消息 | 启用上下文瘦身机制 |
| Agent长时间无响应 | 模型API超时或连接断 | 查调用方日志和重试记录 | 加超时、重试、状态机失败转移 |
| 协商卡住不收敛 | 没有设置协商轮数上限 | 统计协商消息数量 | 加轮数上限,超限升级给人类 |
5.3 三条独家技巧
开发阶段把“所有事件打印到屏幕”当成默认行为,跑通之后再关掉。多人多AI系统的运行方式跟单体应用完全不同,肉眼看到事件流从创建到审批到完成的全过程,比看任何文档都管用。我看到太多项目直接在日志文件里翻来翻去,把半天时间浪费在“这步根本没到这个Agent头上”这类问题上。
先做人肉审批版,跑一个月再考虑把某些环节自动化。自动化审批的边界很难一次想全,有的审批看似常规但实际暗藏风险,有的看起来高风险但流程稳定可以放开。我经历过几次自动化审批误伤事件之后,总结出一个规律:在原型和早期阶段,宁愿多审几次也不能漏审,等人肉审批积累了足够的审批数据,再根据数据决定哪些环节可以灰度自动化。
架构上预留“代理策略”开关。同一套Agent代码,通过配置切换三种模式:自动执行、需审批、禁止执行。这个开关的威力巨大,灰度发布新Agent时全部设成“需审批”,等确认它行为稳定之后,再把低风险动作切成“自动执行”。我后来又重构过整个系统几次,发现这个开关始终是架构里最值得保留的一块。
最后再分享一点个人体会。这套多人多AI协同架构我前前后后重构了三次,才逐渐明白一个道理:协同系统的难点根本不在模型,而在系统设计。模型会越来越强,但消息协议、身份边界、权限闸门、状态机、审计日志这些“笨功夫”,才是决定一个多Agent系统能不能稳定落地的关键。如果你也在做类似方向,我强烈建议先把协议定清楚、把trace_id打上、把人肉审批铺开,再回来研究模型调优。架构稳了,Agent再笨也能协作起来;架构糊了,Agent再聪明也只是在混乱里添乱。