1. 这套系统到底在解决什么问题多人多AI协同说白了就是让多个用户和多个AI代理在同一个任务空间里干活而且AI代理不只是被动应答而是能主动代表用户去跟其他AI或其他人交互。我最早接触这个方向是因为一个很具体的痛点团队里三个人同时用各自的AI助手处理同一份需求文档结果三个AI给出的方案互相矛盾版本对不上最后还得人工去对齐。那时候我就在想能不能让AI代理之间先协商好再把人拉进来做决策这个系统的核心价值在于把“人跟AI一对一聊天”升级成“人跟AI代理团队协作”。适合谁来参考如果你正在做多智能体编排、企业级AI工作流、或者需要多个AI角色分工完成复杂任务的场景这套架构思路可以直接拿去改。如果你只是想让一个AI帮你写写邮件那确实用不上但理解一下代理间协商的机制也没坏处。我实测下来这套架构最直接解决的问题有三个第一多个AI代理之间的状态同步和冲突消解第二人类用户如何在不被信息淹没的前提下介入关键决策第三整个交互过程的内容合规性怎么在架构层面保证而不是靠事后过滤。2. 整体架构设计与核心思路拆解2.1 为什么是“代理代为交互”而不是“人直接指挥每个AI”一开始我也想过让用户直接给每个AI下指令不就行了但实际操作中你会发现当AI代理数量超过三个用户根本管不过来。就像你不可能同时跟五个下属分别交代任务还保证他们互相不打架。所以核心思路是用户只需要跟一个“主代理”交互主代理负责把任务拆解、分发给其他专业代理并在代理之间出现分歧时进行仲裁。这个设计的关键在于代理层级。我采用的是三层结构用户层、协调层、执行层。用户层只有一个入口协调层负责意图理解和任务分解执行层是多个专业代理各司其职。协调层的代理我称之为“调度代理”它不直接干活而是决定谁干活、怎么干、干到什么程度算完。注意调度代理的权限必须明确边界否则它会变成一个瓶颈。我的做法是给调度代理设置最大递归深度和最大代理调用数防止它无限拆解任务导致系统卡死。2.2 多AI协同的通信机制选型代理之间怎么说话这是架构里最容易踩坑的地方。我试过三种方案共享内存、消息队列、以及基于事件总线的发布订阅。最后选了事件总线原因很直接共享内存虽然快但多个代理同时读写同一块状态时锁竞争严重而且调试极其困难消息队列适合点对点但多代理广播场景下要维护大量队列运维成本高。事件总线的优势在于解耦。每个代理只管往总线发事件谁关心谁订阅。比如“任务分解完成”这个事件执行代理订阅了就开始干活日志代理订阅了就开始记录合规检查代理订阅了就开始扫描内容。新增一个代理不需要改其他代理的代码插上就能用。具体实现上我用的是基于Redis Stream的事件总线每个代理有自己的消费者组。为什么不用Kafka因为这套系统的消息量级没那么大Redis Stream足够而且部署简单本地开发环境跑一个Redis实例就行。消息格式统一用JSON包含事件类型、发起代理ID、目标代理ID可选为空表示广播、时间戳、负载数据。2.3 人类介入点的设计逻辑多人多AI协同里人不是旁观者而是关键决策的触发者。但什么时候该让人介入我的经验是设置置信度阈值和冲突检测两个触发条件。当某个代理对某个决策的置信度低于阈值我设的是0.7或者两个代理对同一问题给出了矛盾结论系统就暂停自动流程向人类用户推送一个“决策请求”。这个决策请求不是简单地把问题抛给人而是附带上下文谁提出的、依据是什么、其他代理怎么看、如果选A会怎样、选B会怎样。用户只需要点一下或者简短回复系统就继续跑。实测下来一个复杂任务里人类介入次数控制在3到5次是比较舒服的太多会烦太少又容易跑偏。3. 核心模块的实操细节与避坑要点3.1 代理身份与权限管理每个AI代理必须有唯一身份标识这个标识不只是名字还包括它的能力范围、可访问的数据权限、以及它能调用的工具列表。我见过有人图省事所有代理共用一个API Key结果一个代理误操作把另一个代理的数据删了排查了半天。我的做法是给每个代理分配独立的凭证并在调度层维护一张权限表。这张表长这样代理名称能力标签可访问数据域可调用工具最大并发任务数调度代理任务分解、仲裁全局任务元数据无1检索代理信息检索、摘要知识库只读搜索API5生成代理文本生成、改写知识库读写无3合规代理内容审核全局输出流规则引擎2这张表在系统启动时加载运行时不动态修改。如果需要调整权限走配置更新流程重启对应代理。这样做虽然不够灵活但避免了运行时代理越权的问题。实操心得代理的“最大并发任务数”这个参数很关键。我一开始没设限制结果检索代理被同时调用十几次API直接限流了。后来根据每个代理背后模型或服务的承载能力来设稳定多了。3.2 任务分解与依赖编排调度代理接到用户需求后第一步是任务分解。这里我用的是“目标-子目标-动作”三层分解法。举个例子用户说“帮我写一份竞品分析报告”调度代理会分解成确定竞品范围、收集竞品信息、分析对比、生成报告。每个子目标再分解成具体动作比如“收集竞品信息”分解成“检索竞品官网”“检索行业报告”“检索用户评价”。分解完成后调度代理会生成一个有向无环图DAG节点是动作边是依赖关系。这个DAG不是给用户看的而是给执行代理看的。每个执行代理从DAG里认领自己能做的节点做完之后更新节点状态触发下游节点。这里有个坑DAG的粒度不能太细否则代理之间通信开销巨大也不能太粗否则一个代理干太久其他代理闲着。我的经验是每个动作节点的预期执行时间控制在30秒到2分钟之间超过2分钟的就再拆低于30秒的就合并。3.3 代理间冲突消解机制多个代理协同冲突是必然的。比如检索代理找到的数据和生成代理之前生成的内容矛盾或者两个执行代理对同一个子任务给出了不同方案。我的冲突消解分三步走第一步自动检测。合规代理和调度代理都会监听事件总线发现矛盾结论时触发冲突事件。第二步协商。冲突双方代理各自陈述理由和依据调度代理根据预设规则如数据来源权威性、时效性、与用户历史偏好的一致性给出建议。第三步如果调度代理无法裁决或者冲突涉及关键决策就升级到人类用户。我实测下来80%的冲突可以在自动协商阶段解决剩下20%需要人介入。这个比例跟代理的能力配置有关如果检索代理的数据源质量高冲突率会明显下降。4. 完整实操流程与关键环节实现4.1 环境准备与基础组件搭建这套系统对硬件要求不算高我是在一台16核32G的机器上跑的同时运行8个代理实例加一个Redis和一个PostgreSQL。如果你只是做原型验证8核16G也够用。基础组件清单Redis 7.0用作事件总线和代理状态缓存PostgreSQL 15存储任务元数据、代理配置、审计日志Python 3.11代理运行时我用的是asyncio做异步处理一个本地模型或API接入每个代理背后可以接不同的模型我测试时用了本地部署的小模型做检索代理大模型API做生成代理安装依赖的命令很简单pip install redis asyncpg httpx pydanticRedis的配置需要注意开启Stream功能默认是开的但持久化策略要调一下。我建议用AOF持久化appendfsync设为everysec兼顾性能和安全。4.2 代理基类的实现要点每个代理都继承一个基类基类负责事件总线的连接、消息的序列化反序列化、心跳上报、以及优雅退出。这个基类我改了大概五版才稳定核心是处理好异步任务的取消和异常传播。import asyncio import json import redis.asyncio as redis from pydantic import BaseModel class AgentMessage(BaseModel): event_type: str source_agent: str target_agent: str | None timestamp: float payload: dict class BaseAgent: def __init__(self, agent_id: str, redis_url: str): self.agent_id agent_id self.redis redis.from_url(redis_url) self.running True async def publish(self, event_type: str, payload: dict, target: str None): msg AgentMessage( event_typeevent_type, source_agentself.agent_id, target_agenttarget, timestampasyncio.get_event_loop().time(), payloadpayload ) await self.redis.xadd(agent_bus, {data: msg.model_dump_json()}) async def subscribe(self, consumer_group: str): # 创建消费者组如果已存在则忽略错误 try: await self.redis.xgroup_create(agent_bus, consumer_group, id0, mkstreamTrue) except Exception: pass while self.running: messages await self.redis.xreadgroup( consumer_group, self.agent_id, {agent_bus: }, count10, block1000 ) for stream, msgs in messages: for msg_id, data in msgs: await self.handle_message(json.loads(data[data])) await self.redis.xack(agent_bus, consumer_group, msg_id) async def handle_message(self, msg: dict): raise NotImplementedError这个基类里subscribe方法用的是消费者组模式保证同一个消息不会被同一个代理重复消费。block1000表示没有消息时阻塞1秒避免空转消耗CPU。注意xgroup_create在消费者组已存在时会抛异常所以要用try-except包住。我一开始没处理这个每次重启代理都报错虽然不影响运行但日志很乱。4.3 调度代理的任务分解逻辑调度代理是整个系统的大脑它的核心方法decompose_task接收用户原始需求输出一个DAG。我用的分解策略是“先广度后深度”先识别需求涉及哪些领域每个领域生成一个子目标再对每个子目标做深度分解。async def decompose_task(self, user_request: str) - dict: # 第一步领域识别 domains await self.identify_domains(user_request) # 第二步为每个领域生成子目标 sub_goals [] for domain in domains: goals await self.generate_sub_goals(user_request, domain) sub_goals.extend(goals) # 第三步构建DAG dag {nodes: [], edges: []} for i, goal in enumerate(sub_goals): dag[nodes].append({ id: fnode_{i}, goal: goal, agent_type: self.match_agent_type(goal), status: pending }) # 根据依赖关系添加边 deps await self.infer_dependencies(goal, sub_goals[:i]) for dep in deps: dag[edges].append({from: dep, to: fnode_{i}}) return dagidentify_domains和generate_sub_goals都是调用模型完成的但我在prompt里加了约束领域数量不超过5个子目标总数不超过15个。这是为了防止DAG过大导致执行代理过载。4.4 执行代理的任务认领与执行执行代理启动后先向调度代理注册自己的能力标签然后进入事件循环。当调度代理发布“任务就绪”事件时执行代理检查DAG中是否有自己可执行的节点前置依赖都已完成且自己有能力执行有就认领并执行。认领机制我用的是乐观锁执行代理先读取节点状态如果状态是pending且自己匹配就尝试用Redis的原子操作把状态改为claimed。改成功了就执行改失败了说明被别的代理抢了跳过。async def try_claim_node(self, node_id: str) - bool: key fnode_status:{node_id} # 使用Redis的SET NX实现原子认领 result await self.redis.set(key, self.agent_id, nxTrue, ex300) return result is not Noneex300是认领超时时间5分钟。如果代理认领后崩溃了5分钟后其他代理可以重新认领避免任务永久卡住。执行完成后代理更新节点状态为completed并发布“节点完成”事件。调度代理监听这个事件检查DAG中是否有新的节点变为可执行状态有就发布新的“任务就绪”事件。4.5 人类介入的推送与响应处理当系统需要人类决策时调度代理会生成一个决策请求推送到用户界面。推送渠道我试过邮件、企业微信机器人、以及自建的WebSocket推送。邮件太慢企业微信机器人适合团队场景WebSocket适合有自己前端的情况。我最后用的是WebSocket加企业微信机器人双通道确保用户能及时看到。决策请求的格式{ request_id: dec_001, type: conflict_resolution, summary: 检索代理和生成代理对竞品A的市场份额数据存在分歧, options: [ {id: opt_1, desc: 采用检索代理的数据来源2024年Q2行业报告, confidence: 0.85}, {id: opt_2, desc: 采用生成代理的数据来源模型推理, confidence: 0.62} ], context: 该数据将用于报告第三章的市场分析部分, deadline: 2025-01-15T18:00:00Z }用户回复后调度代理根据选择更新相关节点的数据继续执行。如果超时未回复系统会采用置信度更高的选项并记录一条“超时自动决策”日志。5. 内容合规的架构级保障5.1 为什么合规不能只靠事后过滤很多人做AI系统合规就是最后加一个敏感词过滤。但这套系统里多个代理互相交互内容在代理之间流转如果只在最终输出时过滤中间过程可能已经产生了不合规的中间结果甚至被其他代理引用放大。所以我的做法是合规检查嵌入到事件总线的每一个环节。具体来说合规代理订阅所有事件对每个事件的payload做实时扫描。扫描分三级第一级是关键词匹配速度快但误报率高第二级是语义分析用一个小模型判断内容是否涉及风险第三级是人工复核队列只有前两级都拿不准的才进队列。5.2 合规检查的性能优化全量扫描所有事件对性能有影响。我实测过如果每个事件都走语义分析系统吞吐量下降约40%。优化方案是分级处理关键词匹配对所有事件都做语义分析只对包含文本内容且长度超过50字的事件做人工复核队列每天清理一次。另外合规代理的扫描是异步的不阻塞主流程。但如果扫描发现高风险内容会立即发布“合规告警”事件调度代理收到后暂停相关任务等待人工确认。实操心得合规规则的更新频率很高我建议把规则配置放在数据库里合规代理定期拉取最新规则而不是硬编码在代码里。这样调整规则不需要重启代理。6. 常见问题与排查技巧实录6.1 代理之间消息丢失或重复这是最常见的问题。消息丢失通常是因为消费者组没有正确ack或者Redis内存满了被淘汰。我的排查步骤先看Redis的info memory确认used_memory是否接近maxmemory再看消费者组的pending列表xpending agent_bus consumer_group如果有大量pending说明代理处理不过来。消息重复则是因为ack时机不对。我一开始是处理完才ack但如果处理过程中代理崩溃消息会被重新投递。后来改成处理前先ack但这样如果处理失败消息就丢了。最终方案是处理前ack但处理失败时把消息重新发布到总线并记录失败日志。6.2 调度代理任务分解不合理有时候调度代理会把一个简单任务拆成十几个子任务导致执行代理频繁切换上下文效率反而低。排查方法是看DAG的节点数量和平均执行时间。如果节点数超过15个或者平均执行时间低于20秒就说明分解过细。调整方法是在调度代理的prompt里加约束或者在调度层加一个后处理步骤合并那些可以顺序执行且属于同一代理的小节点。我后来加了一个“节点合并”逻辑把同一代理的连续小节点合并成一个效果很明显。6.3 人类介入请求被忽略用户可能因为忙没看到决策请求导致任务卡住。我的解决方案是设置多级提醒第一次推送后5分钟没响应发第二次提醒15分钟没响应发第三次并抄送备用联系人30分钟没响应自动采用默认选项并记录。默认选项的选择逻辑是如果有置信度超过0.9的选项选它否则选保守方案即不引入新数据、不改变现有结论的方案。6.4 合规误报导致任务频繁暂停合规代理的语义分析模型如果太敏感会把正常内容判为风险导致任务频繁暂停。我踩过这个坑后来调整了策略只有置信度超过0.95的风险才触发暂停0.8到0.95之间的只记录日志不暂停低于0.8的忽略。这样既保证了安全又减少了误报干扰。问题现象可能原因排查方法解决措施代理收不到消息消费者组未创建或Redis连接断开检查xgroup是否存在ping Redis重建消费者组加连接重试任务卡在pending认领超时或代理崩溃查看node_status的TTL调大ex值加代理心跳检测合规频繁暂停语义分析阈值过低查看合规日志的置信度分布调整阈值加白名单机制DAG执行缓慢节点粒度过细或依赖过多统计节点数和平均执行时间合并小节点优化依赖推断7. 这套架构的扩展方向与个人体会这套系统跑通之后我陆续做了一些扩展。一个是加了“代理记忆”每个代理维护自己的短期记忆和长期记忆短期记忆存在Redis里长期记忆存PostgreSQL这样代理在处理相似任务时能复用之前的经验。另一个是加了“代理能力动态注册”新代理上线时自动向调度代理注册自己的能力标签不需要改配置文件。我个人在实际操作中的体会是多AI协同系统的难点不在AI本身而在协调机制。代理之间的通信协议、冲突消解规则、人类介入时机这些才是决定系统好不好用的关键。模型能力可以慢慢升级但协调机制设计不好再强的模型也白搭。最后分享一个小技巧如果你刚开始做这套系统不要一上来就搞五六个代理。先从两个代理加一个调度代理开始把通信和冲突消解跑通再逐步增加代理数量和类型。我一开始贪多同时上了八个代理结果调试了两周才让它们不互相打架。后来砍到三个代理重新跑一天就通了然后再慢慢加回去反而更快。
在3D动画中的应用)
