多AI协同系统实践:代理调度、消息路由与状态机设计
发布时间:2026/10/1 18:49:40来源:尧图网络
做AI应用做得久了你会发现真正难的不是单点能力而是协同问题。一个模型再强也只能在一个上下文里干活一个用户再熟练也没法同时盯住五六个AI助手、把每轮问答手工搬来搬去。我在实际项目中反复碰到同一个痛点后开始研究“让AI代理代为交互”这套思路也就是把用户、AI、工具、权限全部抽象成可编排的实体让代理代替人去找其他AI协作、汇总结果、再交给人确认。这套方案本质上是在设计一个多人多AI协同的系统架构它要解决的核心问题不是“哪个模型更强”而是“当一个任务涉及多个AI、多个用户、多个数据源时谁在恰当的时间调用谁上下文怎么传递权限怎么兜底”。这篇文章会把我的设计思路、分层方案、核心机制和一个可落地的最小系统实现从头到尾拆开适合正在做AI应用架构、多智能体系统、或者准备系统架构设计师相关内容的读者参考。1. 项目的核心问题与设计动机1.1 一个真实的工作场景还原先说我自己踩过的一个坑。上次做企业内部知识库项目团队里有三个人协作产品经理负责整理需求、研发负责查代码、运营负责整理话术。我们同时开了三个网页版AI助手每个助手各自擅长不同的事但问题马上来了产品经理用AI A梳理完需求要把结论复制给运营运营再把摘要粘贴到AI B里生成话术研发还得把AI C的排期建议手动转回群里。整个过程不客气地说就是个“人工路由”。更难受的是上下文根本没连续性每个AI都只看到一截对话碎片。那时候我就意识到单点模型能力不是瓶颈瓶颈在于没有人替这些AI和用户做统一调度。如果有一个“代理层”它能读取公共知识库、记住每个参与者的角色、把任务自动分发到最合适的模型、再把多个AI的输出合并成一段可读结论那整个工作流就完全不一样了。这也是我正式启动这个项目的直接原因。1.2 三个设计目标这个系统立项后我给自己定了三条硬性目标用户不再需要手工复制粘贴。用户只要发起一个会话、指派一个代理剩下的追问、数据抽取、对比分析全部交给代理代为交互。人只做决策不做搬运工。多个AI模型可以被当成一个“模型池”。不同模型有不同擅长的领域比如大参数模型擅长推理小模型擅长轻量分类本地模型擅长处理敏感数据。系统根据任务特征自动决定调用谁必要时两个模型协同完成同一件事。会议式协同而不是聊天式串联。支持“一个会议里同时有多个AI角色、多个人类角色”的场景比如主持人A发起任务研发代理B给出技术方案评审代理C提出风险点人类D做最终审批。这像一个标准的会议室只不过有些参会人是AI。这三条目标看起来简单落地时每一条都牵扯到不少架构决策。1.3 哪些人需要这套架构如果你也遇到下面至少一种情况那这套架构大概率对你有参考价值你在做一个AI功能但不想被某一个模型的API锁死随时打算换底座。你想让AI主动替人跨部门沟通比如自动把需求分派给不同的模型处理。你在准备“人工智能”相关的大作业、毕业设计或者想写一个有深度、有架构感的AI系统。你对“系统架构设计师”考试里的架构风格、消息路由、状态机设计感兴趣想找一个贴近前沿的业务案例。2. 整体系统架构分层设计2.1 四层架构全景整个系统我最终收敛为四个层次外加一条贯穿全流程的权限审计线。每层只干自己该干的事层与层之间通过约定好的数据对象通信。层次职责核心组件接入层接收来自Web、企微、命令行、OpenAPI的请求网关、会话适配器协调层核心大脑负责代理调度、路由、状态管理代理协调器、任务分配器模型层连接各类异构模型模型网关、适配器、本地推理服务数据层持久化上下文、记忆、日志PostgreSQL、Redis、向量数据库协调层是整个系统的灵魂。它不直接调用模型而是维护一组代理对象的状态由这些代理对象去模型层“点名”要能力。一个典型的流程是接入层收到用户消息协调层判断这是哪个代理的会话把消息放进该代理的专属上下文窗口代理通过模型网关调用模型等模型返回结果后协调层再把结果写给所有关注这个会话的其他代理。这套分层最直接的好处是换模型、换前端、加数据源都不影响其他层。我后来甚至把“图片生成”也做成一个普通模型接入上一层的代理完全无感。2.2 把消息路由做成“分布式交换机”路由层在实现时我参考了很多“分布式交换机系统架构”的思路。核心概念是不做中心化的全量共享每个代理实例挂在一个逻辑交换机上交换机之间再互联。这个概念借用到AI系统里特别顺。具体来说每个代理实例有一个固定ID所有发给它或由它发出的消息都带着路由头。协调器内部维护一张“交换表”记录“消息类别 → 应当送达的代理ID集合”。比如“技术方案评审”类消息会同时投递给研发代理和评审代理“对外发布”类消息只投递给运营代理。这比广播效率高得多也不会造成上下文污染。这种设计的另一个好处是故障隔离。某个模型网关超时了影响的只是挂在那一格交换机下的代理其他代理照常运转。我遇到过几次模型服务崩溃整个系统没有停摆就是因为路由层把故障控制住了。2.3 和单体“智能体框架”的差异早期我也试过直接把所有智能体逻辑塞进一个进程用一个全局管理器直接调用各家模型。看起来实现简单但用起来暴露两个问题一是状态非常容易错乱多个任务同时运行会互相污染上下文二是扩展性差想加一个新模型要改全局代码。后来我彻底切成“代理即实体”的思路每个代理有独立的运行上下文和配置全局只保留注册表和路由表。与之配套的是一套完善的权限机制哪个人能命令哪个代理、哪个代理能读哪个数据源全部写进配置。这也是为什么这套系统即使是在“多人多AI”并发场景下还能保持每个协作流互不干扰。3. 代理实体与全生命周期管理3.1 代理到底是什么在这套架构里代理不是一段简单的提示词模板而是一个有状态、有记忆、有能力的逻辑实体。它由四部分组成身份ID、能力卡片、记忆单元、运行状态。身份ID是全局唯一的用于路由和审计。能力卡片描述这个代理能调用哪些模型、哪些工具、哪些数据源。记忆单元保存短期会话上下文和长期向量记忆。运行状态则记录了它现在是在等待任务、执行中、还是在等待人工确认。我经常跟朋友解释可以把代理想象成一个带工牌的员工。工牌上写着岗位和权限口袋里揣着笔记本干完活会把关键信息记录下来。它替人在系统里跑来跑去但重大决策必须回头找真人签字。3.2 代理状态机设计代理生命周期我设计成五个状态流转关系不复杂但很关键状态含义可流转到IDLE空闲等待被激活RUNNINGRUNNING正在调用模型或工具执行任务WAITING、COMPLETEDWAITING等待人工审批或外部数据返回RUNNINGCOMPLETED本次任务结束结果已归档IDLEERROR执行异常需要干预或重试IDLE、RUNNING核心代码里就是一个状态机加事件驱动的回调class AgentStateMachine: def __init__(self): self.state State.IDLE self._handlers {} def register(self, event, handler): self._handlers[event] handler def dispatch(self, event, payload): if event not in self._handlers: raise UnhandledEvent(event) new_state, output self._handlers[event](self.state, payload) self.transition(new_state, output)这个结构的好处是任何外部事件——新消息、模型返回、定时器唤醒、人工审批通过——都能统一走dispatch入口。我在项目里很少直接改状态字段所有状态变更都通过事件触发这样容易被审计也容易回溯问题。3.3 能力卡片与记忆单元每个代理都带一份能力配置文件我用的格式是JSON因为哪个语言都能解析。下面是我常用的最小示例{ agent_id: research-assistant, owner: product-manager, model_routes: [ {task_type: code, model: deepseek-coder}, {task_type: long_text, model: claude-reasoner}, {task_type: local_private, model: local-llama3-8b} ], tools: [knowledge_base.search, git.repo_overview, doc.summary], memory: { short_term: {policy: last 20 rounds}, long_term: {vector_store: pgvector, collection: research_mem} } }记忆单元的设计上短期记忆直接放在Redis里用代理ID加会话ID作为键结构是消息数组超过轮数自动裁剪。长期记忆则落到向量库里每次对话结束会做一次摘要嵌入之后新任务进来可以先做语义检索把相关历史片段注入提示词。这一步很影响效果尤其当代理需要回顾前几天讨论过的“技术选型结论”时没有长期记忆就只能靠人重新喂一遍。3.4 人的权限边界与控制代理代替人去交互不代表把人踢出决策链。在这套系统里代理默认没有最终决策权。所有对外发布、资金操作、强删除类动作都要走人工审批。审批的方式也很简单代理在跑到需要决策的步骤时会生成一个确认请求推送到接入层的消息队列。用户回复“同意”或附上修改意见代理拿到结果再继续执行。这个机制我至今没有省掉因为模型幻觉是客观存在的AI代理做得再顺溜最终责任还是在人身上。权限粒度上我至少设计了三级系统管理员、会话管理员、成员。会话管理员可以增删代理参与者成员只能发送消息和接收结果。如果想要更细可以按“代理能力”再细分比如A用户只能把任务派给研发代理不能命令运营代理发内容。这个一般写在注册表里避免越权乱调。4. 多AI协同的关键机制4.1 会话空间建模多人多AI协同第一个要解决的是“上下文归属”问题。不能简单搞一个全局黑盒因为不同人对同一个AI说的话不能混在一起。我引入了“会话空间”这个概念。一个协作项目对应一个会话空间空间里可以有多个频道每个频道绑定一组代理和用户。比如“技术频道”绑定了研发代理和测试代理“运营频道”绑定了运营代理。消息在频道内流转跨频道的消息必须显式转发。这样做的效果是研发代理在技术频道里反复讨论代码不会影响运营代理在运营频道里整理话题。如果确实需要跨频道引用系统会把引用内容的摘要注入目标频道而不是把整段原始对话复制过去避免上下文被无意义撑爆。4.2 消息路由与模型选型策略模型层接入了多个大模型API也接入了本地部署的模型。接入不难难的是怎么分配任务。我维护了一张“任务类型 → 模型”的路由表并在模型网关层做统一分发任务特征路由规则长文章总结、复杂推理调用强推理大模型温度调低代码生成/审查调用代码专长模型附带仓库相关文件涉及客户隐私的文本处理只允许走本地模型禁止外发简单分类/信息抽取调用低成本模型批量并行这个策略的核心动机是成本和质量平衡。我实测过很多次所有任务都丢给最强模型不仅贵有些场景效果反而不如专用模型。本地模型在私有化部署场景下更重要一些敏感数据本身就不该出内网。路由表不是写死的而是从代理的能力卡片里读取。每个代理知道自己能用的模型列表和适用场景机器人在启动时会根据任务类型做一次匹配如果没匹配到就报“无可用模型”并挂起而不是乱调用。4.3 任务卡片带来的前后端一致性多个AI协同过程中最大的问题不是“谁去做”而是“做到哪一步了”。我在系统里设计了一个“任务卡片”对象所有协同环节都围绕这个卡片更新状态。一张任务卡片包含任务编号、发起人代理、当前处理代理、依赖数据、输出结果、审批状态。卡片创建时会进入协调层的待办池处理代理开始工作后卡片状态变成RUNNING模型返回结果后变成PENDING_REVIEW人工批准后变成CLOSED。所有代理在收到下一步指令前都会先查一下任务卡片的最新状态。这个机制解决了一个我很头疼的问题多AI协同很容易出现两个代理同时改同一个结果导致产生冲突。有了任务卡片就相当于给每项工作加了“锁”同一时刻只有一个代理能持有卡片的写权限。说白了就是把几个人工智能同桌开会的事变成了规范化的工作流。4.4 可控范围内的自动交互虽然整套系统的卖点叫“代理代为交互”但我不想让它彻底变成无人值守的自动流水线。代理之间的交互默认是可见的A代理发了什么、B代理回了什么、中间调用了哪些工具用户都能在界面上看到。用户可以在任何节点插入指令也可以随时中止某个代理的循环。这一点在多人协同场景里特别重要因为它让人类保持着“随时接管”的掌控感。纯粹的自动驾驶会让人不安有方向盘、能随时踩刹车的辅助驾驶大家才敢放心用。5. 最小可用系统实操记录5.1 部署环境与我选用的组件这个系统的第一次稳定部署我放在了国产ARM服务器上。当时还是有点波折的因为不少中间件在aarch64架构下的安装包并不齐全需要自己编译或者找兼容版本比如Node.js 18以上版本在麒麟系统上就需要用特定RPM包或源码安装有些Python库的预编译轮子也找不到。最终部署环境大致是这样操作系统国产银河麒麟aarch64架构服务框架FastAPI Uvicorn消息队列Redis Stream RabbitMQ数据库PostgreSQL pgvector向量模型本地部署的向量化模型服务如果用x86服务器整个过程会省心很多。但既然要聊架构落地避不开适配性这个话题我也顺便说一句在国产化环境下搞定部署尽量选择对ARM友好的组件遇到问题多看看源码编译思路别死等现成安装包。5.2 核心服务骨架我会把最核心的协调服务简化成三部分代码代理注册、消息入队、调模型。代理注册表agents {} def register_agent(card): 注册一个新的代理进入路由交换表。 if card.agent_id in agents: raise AgentAlreadyExists(card.agent_id) agents[card.agent_id] { card: card, state_machine: AgentStateMachine(), context: [] } route_table.add_subscriber(card.channel, card.agent_id) return card.agent_id消息分发到代理async def dispatch_to_agent(agent_id, message): state_machine agents[agent_id][state_machine] state_machine.dispatch(MESSAGE_RECEIVED, { sender: message.sender, payload: message.payload }) route_result await orchestrator.route_forward(message) if route_result NEED_MODEL: reply await model_gateway.call( agent_idagent_id, route_keysmessage.route_keys, contextagents[agent_id][context] ) await commit_result(agent_id, reply, message.channel)这段代码很朴素但它体现了整套系统的灵魂代理先收到消息再经状态机判断是否要调用模型最后把输出提交。模型网关的调用细节被隔离在model_gateway.call里任何模型升级都不影响上层。5.3 跑通一次多人多AI协作假设我们有三个参与方人类用户小王产品经理、代号“研发”的代理、“评审”的代理。小王在会话空间里发起请求“帮我写这段需求的技术实现方案并评估风险。”流程如下接入层收到消息识别发起人小王消息归属到“技术频道”会话空间。协调层根据能力路由表把消息同时投递给“研发”代理和“评审”代理。“研发”代理更新到RUNNING状态从知识库检索架构历史记录。“研发”代理调用代码模型生成技术方案草稿。协调层把草稿注入“评审”代理要求做风险清单。“评审”代理调用低成本的分类模型产出风险列表。两个输出合并成一份结构化报告回到小王客户端。小王在报告上点了“通过”任务卡片CLOSED。整个过程中小王没有手动复制任何内容代理之间自动完成了信息流转。就算后来换了模型或者新增了一类代理只要能力卡片和路由配置对应上流程照样能跑。6. 常见问题与排查技巧实录6.1 两个AI代理反复“甩锅”我第一次联调时就碰到一个很经典的问题研发代理说“方案可以了”评审代理说“建议优化”研发代理又回复“已经优化了”评审又回复“还需要优化”两边在循环里死磕谁也说服不了谁。这背后的原因是代理的提示词里缺乏“终止条件”。我后来在评审代理的能力卡里加了明确规则“如果连续两轮修改意见不包含新的具体问题将状态移交人工审批”。同时任务卡片也引入了“最大交互轮数”超过三轮自动升级到人类由人做终结判断。这个经验非常实用多AI协同不是对话越多越好得有机制兜底。6.2 上下文过长导致Token爆炸本地模型尤其容易踩这个坑。多轮协作后代理的上下文数组越攒越多发给模型时直接超长轻则报错重则把整块内存吃光。我的排查步骤很简单先看代理的短期记忆策略是不是“永远全量保留”如果是立刻改成按轮数裁剪加“重要结论摘要”。然后对长文本任务强制走“先检索、再注入”的流程而不是把整个历史都塞给模型。最后在模型网关层加了输入长度预检超过阈值直接拒绝调用并报错避免模型接口进入不可控状态。6.3 模型接口不稳定时好时坏联调第三方API时接口超时和限流几乎是每天都会发生的。每条请求都可能失败但如果整个任务因此挂起用户体验就很差。我在模型网关层做了三件事超时机制、指数退避重试、降级路由。如果A模型三次重试都失败网关自动把请求切到B模型。因为这个降级是发生在网关层上层代理完全无感。同时所有失败记录都写进监控日志方便事后复盘是哪家服务不稳定。6.4 权限与审计的坑多人多AI协同权限一旦做粗很容易出乱子。我之前为了图省事让所有代理共用一套数据源读取权限结果运营代理把内部研发数据也翻阅了一遍虽然没有造成实际破坏但已经很危险了。后来我坚持“最小权限”原则每个代理只能读取自己角色需要的数据源跨域读取必须申请临时授权。所有代理的调用链都记录到审计表谁在什么时间、什么会话里访问了哪个数据源一查便知。做企业级应用时这一条不能省。7. 个人实操体会与扩展方向这套架构是我边踩坑边迭代出来的不是坐在工位上凭空画的。我的体会是多人多AI协同真正难的不是模型能力而是把“人的决策”、“AI的生成能力”和“系统的状态管理”拧成一股绳。代理代为交互本质上是在给AI画边界、给协作定规则。如果你想让这套系统更完整我建议按这三个方向扩展一是把任务卡片升级成可视化看板让人和AI的活动一目了然二是加入反馈回路让代理在执行完任务后主动总结“哪里不顺、下次怎么改进”三是把模型网关接入更多模型家族并沉淀一套离线评估集每次换模型前先跑一遍测试防止无脑升级导致效果倒挂。我自己正在把这套东西做成一个开源脚手架接下来会继续在真实的跨部门协作场景里磨等跑得更稳了再把细节放出来跟大家同步。
网站建设高端定制企业官网