上周有朋友问我一个挺实际的问题现在团队里每个人手里至少有一两个AI助手如果我想让五个AI同时参与一个项目、相互之间还要传递资料和结论应该怎么组织我当时第一反应是这不只是选哪个模型的问题而是要用一套系统架构来承接“多人多AI协同”。后来我把这套思路落地成了一个原型踩了不少坑也沉淀出一些可复用的架构设计。今天这篇文章就把我自己的做法拆开讲聊清楚AI代理是怎么作为人的替身去和其他代理交互的以及协同系统里最核心的注册、路由、编排、隔离这几个层面该怎么设计。这篇内容更适合正在做企业内部AI平台、想引入多智能体协作的团队或者对AI代理架构感兴趣的个人开发者。我会从概念解释到最小系统搭建再讲真实运行时会遇到的问题尽量让还没有做过类似系统的同学也能照着搭出一版可以跑的骨架。1. 从“一人一模型”到“多人多代理”架构问题的起点1.1 人的角色变了从操作者变成委托者过去我们使用AI通常是打开一个对话框输入问题、获得答案人始终处于“操作者”位置。但在协同场景里人会变得非常忙如果要让AI A去查资料、AI B去做分析、AI C去审校人不可能同时开三个对话框来回搬运结果。于是AI代理的作用就变了它不再只是“回答问题”而是要替用户去和其他代理沟通、接收任务、反馈进度。这里说的AI代理本质是一个能感知任务上下文、调用工具、并与其他代理交换消息的程序。它最关键的属性是“代为交互”代理以用户授权代表的身份代用户向其他代理或模型发起请求用户只在关键节点做审核或决定。举个例子一个产品团队有三个人各有一个代理。产品经理代理发起“整理本周竞品动态”的任务开发代理负责补技术参数设计代理负责输出交互建议人最后只看汇总结果。这个过程中人并没有直接一句一句地跟每个模型对话交互动作全部由代理完成。我在设计这类系统时最先想明白的一件事是多AI协同的架构重点不在模型而在代理之间如何通信、如何组织。模型今天可以用A厂商明天可以换成本地模型但代理之间的对话规则、消息格式、审核流程应该保持稳定。这也是为什么我需要把架构重心放在“代理层”而不是把精力都花在调prompt上。1.2 为什么必须在“人—模型”中间插一个代理层如果直接把模型接口暴露给多个用户会立刻遇到四个问题权限没法精细控制。每个用户能访问哪些工具、能读到哪些数据直接在模型接口层很难管理。上下文会越串越乱。用户A任务里的内部文档如果不做隔离可能被用户B的代理在生成过程中无意识引用。任务状态不可追踪。模型接口是无状态的谁来发起、当前走到哪一步、谁处理过没有任何记录。策略没法统一收敛。有的任务需要人工确认有的任务可以自动执行没有一个统一的地方做判断。代理层就是夹在人和模型之间的管理层。它像公司前台所有对外沟通都经过前台前台知道什么人应该找哪个部门什么话能说、什么话不能说。在这个架构里代理层承担四类封装身份封装每个代理有唯一的AgentID归属到某个真实用户或某个系统角色。策略封装代理根据任务类型决定是用快速模型还是强推理模型是否允许调用工具。记忆封装代理可以读取短期任务记忆和长期团队偏好但不会把不该带的上下文带出去。审计封装所有代理间的往来消息全部留痕出了问题可以回溯。所以与其说这是“多模型调度系统”不如说是一个“代理组织系统”。模型只是执行单元真正参与协同的最小实体是代理。2. 多人多代理协同架构的主干模块2.1 代理注册与身份边界先给每个人一个“助手小号”多人多代理协同的起点不是写消息队列而是先定义清楚“谁是谁”。我建议先设计一个代理注册表每条记录包含AgentID、真实用户、角色、权限范围、可用工具、绑定的模型策略等。这个注册表可以落在数据库里也可以启动时加载到内存但一定要有持久化因为后续审计需要。下面是一个简化的注册记录结构{ agent_id: agent_zhang_pm, owner_user: user_zhang, display_name: 张产品-助手, role: product_manager, permissions: { read_scope: [team_alpha/documents/pm, public], tool_scope: [search, calendar], model_policy: balanced }, max_hop: 3, created_at: 2025-01-01T00:00:00Z }我为什么要额外设计“身份边界”因为在实际协同中代理之间交换信息很容易越界。用户A的代理想获取用户B的私有日程如果不做校验这种请求会被直接转给模型模型很可能毫无保留地输出。有了注册表之后每次代理发起跨用户请求时路由层先去查“源代理”是否有权限访问“目标范围”没有就直接拒绝。这个检查和普通IT系统的权限校验一样只是执行方从人变成了代理。我踩过的一个经验是权限范围不要太粗。刚开始我只写了“可以读团队文档”结果一个代理把整库文档全捞了一遍。后来改为“读 /team_alpha/documents/pm 目录下的文档且单次不超过20篇”效果立刻收敛。给代理的权限应该像给人开的工单权限越具体越好。2.2 消息总线与分布式交换把代理间的通信看作“交换机”代理之间通信不能全部点对点否则每加一个代理就要改一堆接口。我的做法是引入一条消息总线所有代理只跟总线交互。这里可以借鉴传统网络里的交换机思想交换机不关心每台电脑具体在做什么它只按报文里的目标地址查表转发。代理系统里的“目标地址”就是主题topic。我会把主题设计成带层级的结构例如/agent/{agent_id}/task给某个指定代理下发任务。/team/{team_id}/broadcast向整个团队广播消息。/task/{task_id}/result某个任务的结论回传通道。/system/orchestrator/command编排器下发的控制命令比如暂停、重试。这其实就是发布订阅模型。为什么我不用点对点的HTTP轮询因为协同过程是异步的、多对多的。代理A执行任务可能需要30秒在这期间代理B和代理C也可能在并行工作HTTP请求会让所有节点彼此等待还会产生大量无用连接。用Redis Streams或者MQTT这类的消息中间件生产者把消息丢到主题里消费者按自己的节奏处理天然适合这种松耦合结构。我会在消息里固定一批字段后面排查问题全靠它字段作用示例msg_id唯一消息IDmsg_001234trace_id链路追踪ID一次任务一个trace_alpha_001from发送方AgentIDagent_zhang_pmto接收方AgentID广播时为空agent_li_devreply_to回复主题/task/task_001/resulttype消息类型request / response / command / eventrequestpayload具体内容体结构化JSONschema_version消息体版本1.0timestamp时间戳...hop_limit剩余转发次数3消息体字段是协同系统最容易忽略的部分。我建议第一版就把trace_id、hop_limit、reply_to设计好不然后面做追踪和防循环时要回改所有代理代码非常痛苦。2.3 模型网关屏蔽不同厂家的差异代理层不应该直接对接每家模型服务商中间需要一个模型网关。网关的作用和银行统一支付接口类似不管底层是本地部署的开源模型还是第三方API统一暴露一个/v1/chat/completions接口给上层代理使用。代理只需要知道模型网关地址不需要关心实际调用哪家服务。网关内部需要做几件事协议转换、超时控制、重试、令牌消耗统计、按策略路由模型。比如一个“快速分类”任务路由到响应速度最快的模型一个“复杂推理”任务路由到推理能力更强的模型。我整理了下面几种任务类型对应的模型偏好任务类型模型偏好原因简单摘要、分类轻量级模型响应快、成本低多步推理、代码生成强推理模型需要逻辑链稳定中文文书润色中文能力强的模型表达更自然涉及隐私数据的分析本地模型数据不出内网模型网关还会统一做“失败收敛”。模型接口偶发抖动是家常便饭没有网关的话每个代理都要自己写重试逻辑很容易出现雪崩。我的经验是网关层重试2次间隔1秒和3秒如果还失败就返回一个结构化错误给编排器由编排器决定是否换模型或降级处理。3. 编排策略任务路由、上下文隔离与冲突仲裁3.1 谁来拆任务、派任务、收结果有了代理和消息总线系统只是有了“能说话的人”但还需要一个“项目经理”。我把它叫做编排器负责把一个大任务拆成子任务、按主题派发给对应代理、收集结果、判断下一步。没有编排器的话代理之间会各说各话最终没有谁把碎片信息汇总成结论。编排器的工作逻辑可以抽象成一个状态机。一个子任务通常经历以下状态pending - assigned - running - review - merged | | - failed - blocked (需要人工介入)具体实现时我会为每个任务维护一份轻量记录包含任务ID、目标描述、当前状态、候选执行AgentID、父任务ID、结果摘要。如果子任务失败编排器可以选择重新分配或标记为人工处理。这里要特别注意分配策略。最简单的分配是“按角色匹配”任务标签是“代码评审”就发给开发角色的代理“文案审核”就发给编辑角色代理。更复杂一点可以根据代理的历史成功率、当前负载做加权随机。我第一版只做角色匹配够用了没必要一开始就上智能调度。3.2 上下文隔离防止“越聊越串”多人多AI协同中最头疼的不是模型不够聪明而是上下文串味。模型本身没有“用户边界”概念只要消息里出现了别人的数据它就会当作有效信息使用。所以必须在架构层面做上下文隔离。我的做法是在消息里强制携带namespace起到数据分类的作用具体我拆成三个维度隔离维度隔离键典型场景用户级user_zhang个人私密资料团队级team_alpha团队共享资料团队成员可读任务级task_001仅某个任务上下文可读其他任务不可见每个代理生成回复之前编排器会把“当前代理可用的上下文”组装好再发给模型网关。组装规则很简单用户级上下文只能来自自己团队级上下文只能来自同一个团队任务级上下文只能来自当前trace_id所对应的任务。任何代理传过来的消息都要先确认源消息的namespace和目标代理的权限范围匹配才能进入上下文组装。这里分享一个实战教训我早期为了让团队目标一致把团队相关资料全部拼到系统prompt里结果代理A在写代码时引用了代理B的客户合同内容。后来改成“按角色组装上下文”——产品角色只注入用户反馈和需求文档开发角色只注入技术方案和代码库索引。上下文瘦身后串味问题大幅减少token开销也降了一截。3.3 冲突仲裁当两个AI给出不同答案协同系统里经常出现这种情况代理A基于资料说“方案X可行”代理B同时指出“方案X有风险”。两个结论都是模型生成的都挺有说服力该信谁我目前的做法是把冲突分为两类。第一类属于信息不全需要触发二次补充。编排器会要求双方补充依据来源再各自重新生成一次结论。第二类属于角色立场差异比如成本和体验之间的权衡这时我会做一个加权投票产品和设计角色在体验指标上权重高点开发和运维角色在成本稳定性上权重高点按加权分决定默认采纳哪个结论。下面是一段简化的仲裁逻辑我在原型里跑过效果还算稳定def arbitrate(results, weights): scores {} for task_id, agent_id, conclusion, confidence, role in results: scores[conclusion] weights.get(role, 1) * confidence top_conclusion, top_score max(scores.items(), keylambda x: x[1]) second_score sorted(scores.values())[-2] if len(scores) 1 else 0 if top_score - second_score 0.2: return need_review, None return accept, top_conclusion需要说明的是这类仲裁并没有真正解决“谁对谁错”它解决的是“决定怎么推进”。真正高风险场景本质还是要让人拍板。给仲裁器设一个最高权限一旦拿不准就转人工确认而不是强行合并。这样可以避免代理间无休止争论也给整个系统留出安全阀。4. 最小可用系统搭建选型、目录结构与核心代码4.1 技术选型为什么我选消息中间件而不是直连搭建一个最小可用的多人多代理系统不建议第一版就上Kafka或者大型分布式框架会陷入运维泥潭。我推荐三级依赖FastAPI做代理和编排器接口Redis Streams做消息总线模型网关用轻量Python服务封装。这套组合的好处是数据量不大时能跑得很顺组件复用率高出了问题也好排查。为什么选Redis Streams而不是只写在内存里因为系统崩溃后消息还能恢复代理重启后能继续消费未完成任务。为什么不用HTTP定时轮询因为轮询会造成无谓的请求开销和延迟而且代理无法实时感知任务到达。Redis Streams本身就支持消费组天然支持多个代理实例同时消费同一个主题还能保证一条消息不会被重复消费未确认前。目录结构我用了比较稳妥的分层方式multi-agent-system/ ├── agent/ # 代理逻辑 │ ├── worker.py # 消费任务并调用模型 │ └── profile.py # 代理注册与权限加载 ├── orchestrator/ # 编排器 │ ├── dispatcher.py # 任务拆解与分配 │ └── arbitrate.py # 冲突仲裁 ├── gateway/ # 模型网关 │ ├── providers/ # 不同模型协议适配 │ └── router.py # 按任务策略选模型 ├── bus/ # 消息总线封装 │ ├── schema.py # 消息字段定义 │ └── redis_stream.py # Redis Streams读写 └── config/ # 配置文件4.2 消息体设计与代理侧接收循环我会在bus/schema.py里固定消息结构然后让所有模块共用。代理侧接收循环的核心逻辑是从队列里读一条消息校验hop_limit组装上下文调模型发结果到reply_to。伪代码大致长这样def process_next_task(stream_reader, gateway_client): message stream_reader.read_blocking() if message.hop_limit 0: return reject(message, reasonhop_limit exceeded) context build_context(message.namespace, message.agent_id) prompt render_prompt(message.payload.task_description, context) result gateway_client.chat( model_policymessage.model_policy, messagesprompt, toolsload_available_tools(message.agent_id) ) stream_writer.publish( topicmessage.reply_to, payload{task_id: message.payload.task_id, result: result}, trace_idmessage.trace_id, hop_limitmessage.hop_limit - 1 )这里最关键的点有两个。一个是幂等消费同一个任务可能因为网络原因被重复推送所以message_payload里要带task_id处理前先查结果表如果已经存在就直接返回不做重复计算。另一个是hop_limit每次转发减一减到0就丢弃这是防止代理间消息形成循环的最廉价手段。4.3 把本地模型和第三方模型都接进网关模型网关是所有代理访问模型的唯一入口。配不同模型服务商只需要在providers目录里加一个适配器然后写配置。例如providers: - name: local_main type: openai_compatible base_url: http://localhost:8080/v1 api_key_env: LOCAL_API_KEY timeout: 30 - name: cloud_fast type: vendor_sdk model: fast-v2 api_key_env: CLOUD_API_KEY timeout: 10我把本地模型优先用在涉及隐私数据的任务上。所谓本地模型接入点本质上就是一个在局域网内运行的推理服务模型网关通过内网地址调用它。这样一来敏感数据不会流出公司网络同时上层代理的代码不受影响。这里有个小经验模型网关里一定要记录每次请求的token消耗和耗时并按trace_id汇总。看起来只是日志其实它能帮你发现“某个任务token消耗异常高”往往意味着prompt里拼进去了不必要的历史消息。5. 实测中的失败模式与排查调优5.1 代理间循环对话与任务风暴我第一次把两个代理放开互相通信后大概半小时就出故障了消息队列里的消息越来越多代理的CPU居高不下但任务并没有完成。通过查看trace_id我发现同一个任务多次出现在处理日志里而且代理A发给代理B的消息最后又回到了代理A自己。根因很简单当时消息协议里没有hop_limit代理B觉得“信息不足”又向代理A发了一个补充请求代理A也这样认为于是双方不停互问。这个现象就像一个电话座机串线两边都以为对方有资料实际上两边都没有。修复方式有两步。第一步所有消息强制带hop_limit转发一次减一到0禁止继续转发。第二步编排器为每个任务设置deadline超过时间未完成就终止并转人工。之后我还在监控面板里加了“同一任务的消息数”指标只要超过10条就高亮告警。这套组合下来消息风暴问题基本绝迹。5.2 上下文串味与隐私泄漏一条完整的排查链路这个坑我印象最深因为它是隐蔽的。某次测试中用户A的代理想用户A整理一份会议纪要结果纪要里出现了一句“根据用户B手中的客户报价单本轮方案建议降价5%。”用户A当时就懵了他从未授权读取用户B的资料。排查路径是这样的第一步从用户A代理的回复里往回找确认用的是哪个trace_id。第二步看这次trace_id关联的prompt组装记录发现模型网关确实收到了“客户报价单”相关文本。第三步查这段文本来自哪里。结果显示编排器在组装用户A上下文时把团队context给过滤错了——它按team_alpha授权把团队所有文档都允许读而用户B的客户报价单也存储在同一个团队目录下。第四步确认根因权限范围没有细分到“子目录”或“文档标签”只分到了“团队目录”这一层导致同一团队下所有可读数据都被代理当作上下文。修复方案是调整权限模型团队级上下文再细分文档标签比如“报价单”“人事材料”“技术方案”作为独立权限项代理只有在其任务标签匹配时才能引用。同时我在网关里加了一道审计提示当某个代理的回复里出现其他代理独有的命名实体时自动进行告警。这个机制虽然无法完全阻止泄漏但能显著提高暴露速度。5.3 模型幻觉被协同放大必须要有质量闸门单独用AI时幻觉可能只是一句错误描述协同系统里一个代理的幻觉会被下一个代理当成既定事实继续推理出更离谱的结论。我遇到过一次技术代理的模型在结果里声称“已完成数据库连接测试”实际上它根本没有工具权限只是顺着代码评审任务“编”了一个结论产品代理看到“测试已完成”马上输出“可以进入上线准备”。整条链路非常流畅但所有结论都是空中楼阁。这个问题的本质是代理的中间结论缺少“证据校验”。我后来在编排器里加了一个简单的质量闸门步骤一要求每个代理返回结果时附带“依据来源”可以引用具体文档ID、搜索结果链接或工具调用记录。步骤二如果代理声称调用了某个工具编排器会检查该代理的任务日志中是否有对应的工具调用记录没有则打回。步骤三高风险操作指令比如“删除文件”“发送对外邮件”“修改生产配置”不直接执行而是生成“待批准指令”由人工确认后通过另一个通道执行。这个改动确实降低了自动化程度但换来了可信度。做多AI协同我认为“慢一点但可控”永远好过“快要错得上天花板”。5.4 没有trace就谈不上协同排查前面积累的排错经验有一点共通之处能快速定位问题全靠消息里的trace_id串起整条链路。每一轮协同从用户发起任务到最终结果返回中间可能有十几次代理间消息流转如果日志没有统一的trace_id出了问题根本无从查起。所以我会在系统第一版就做结构化日志每个日志行至少包含timestamp, trace_id, from_agent, to_topic, msg_type, status, latency_ms, token_cost不需要上很重的链路追踪系统先把日志落盘后续量大了再接入更专业的平台。实际操作的时候我会用这个日志做两件事查看某个trace_id下的全链路消息和时间分布以及统计每个代理的平均处理延迟。只要这两个数据是准的绝大多数协同异常都能定位到具体节点。6. 落地扩展记忆层、评估体系与人机回环6.1 短期任务记忆与长期团队记忆代理系统不能每轮都是“失忆状态”。我把记忆层分成两段。短期记忆对应当前任务的生命周期例如“任务中已确认的约束条件”“临时决定采用的技术方案”这些数据可以放在Redis里key为task_id过期时间跟任务deadline一致。长期记忆对应团队的复利知识例如“团队更偏好中文输出”“过往项目中对某个方案的否决原因”这些需要持久化到向量库或普通数据库。注意长期记忆必须按namespace隔离否则又会串味。写入长期记忆时我建议要有人在环确认代理解析出一个“可复用结论”时先呈报给创建者确认再写入未经确认的结论只能作为临时参考。6.2 评估体系每个协同结果都该有一份质检报告多代理协同会大幅提高“产出速度”但也容易淹没“产出质量”。我在实际落地时给编排器加了一个评估环节每次大任务汇总后由一个独立的评估代理检查最终结果按四个维度打分评估维度检查内容完成度是否覆盖最初任务里所有子项一致性各代理结论之间是否冲突、是否有明显前后矛盾可追溯性每个关键结论是否有依据来源安全性是否引用越权数据是否包含高风险指令评估不会阻止结果发送给用户但会把低分结果标记出来要求任务发起人确认。这相当于给协同系统加了一道“出厂质检”虽然多一步操作但对系统建立信任非常关键。另外评估代理自身也可能出错所以关键决策仍然需要人来审不能形成“AI评估AI”的闭环。6.3 人机回环把“全自动”换成“默认可控”我做了几轮原型之后最大的变化是对“自动”这件事变得保守。多人多AI协同真正有价值的地方不是让AI们在一分钟内生成一万字报告而是让人类团队能够用更少的沟通成本把大量候选方案和信息快速收敛成可执行结论。这里面每一步都可能出错所以人的参与不能省。我会在任务流里保留两个人工介入点一个是高风险工具调用前的确认点另一个是冲突仲裁无法收敛时的最终决策点。在这两个点上系统暂停下来等人工指令其他环节尽量自动。这套“默认可控”的设计让团队在试运行期间没有出现过一起因误操作导致的严重事故。我对接团队时的建议也一直是不要追求完全无人值守先做到“人不在也能安全跑人回来还能接得上”就好。如果后续要继续扩展我建议从记忆层和评估体系入手先把团队偏好沉淀成结构化数据再把质检规则从人工抽检逐步推向自动化检查。架构上第一步的代理注册、消息总线和编排策略已经打好底子往上叠加就会顺很多。我自己在这个项目里最深的体会是AI代理协同系统本质上是在模拟一个靠谱的项目组而系统架构就是那套“靠谱”的保障机制。把通信边界、任务状态、审计链路先立住后面换什么模型都不会乱。