这段时间我在折腾一个内部AI协作平台核心命题就一句话当团队里同时有十几个会说话但彼此不通气的AI怎么让它们替我们干活而不是替我们添乱。更麻烦的是人这边也不是一个人产品、研发、运营都要进来提需求。于是我把问题拆成“多人多AI协同”核心思路是引入一个AI代理层让所有人和所有AI之间的交互都由代理代为完成。这套系统我内部代号叫Agent Mediation Hub项目标题里的“AI代理代为交互”说的就是这一层。这套架构折腾了大概三个月中间踩了不少坑也推翻过两版设计。这篇文章我把整体架构、核心模块、一个能跑起来的最小版本以及实际运行中遇到的典型问题都写出来。适合正在自建AI平台的架构师和后端工程师也适合想理解多AI协作逻辑的产品同学。看完你应该能明白为什么多AI系统需要的不是更聪明的模型而是更可靠的交互层。1. 为什么要在人和AI之间加一个代理层1.1 多AI并存的真实困境现在企业内部同时用好几个AI模型是很正常的事有的模型擅长代码有的擅长结构化输出有的本地部署用来处理敏感数据还有一个云端API用来跑复杂推理。问题在于这些模型互相之间不认识会话上下文也不共享。一个人要完成一个需求可能先让模型A列提纲再把A的输出喂给模型B写代码中间还要手动复制粘贴、整理格式。如果团队里有五个人同时这么干效率不是提升是灾难。我一开始也试过直接做一个模型聚合网关把多个API统一封装一下让用户自己选模型。但实测下来没什么用。因为用户根本不知道自己该选哪个模型即便选了也要自己管理上下文。真正的问题不是接口不统一而是“一个需求被拆成多个子任务时需要有一个东西记住来龙去脉并负责调度”。这就引出了代理层的价值它不只是转发请求而是代替用户理解意图、选择模型、维护上下文、汇总结果。1.2 代理层到底代做了什么用一个生活化的类比。多AI协同系统就像一个繁忙的研发中心每个模型是一名工程师。你不能让客户直接冲到工位上去找工程师因为客户说不清楚自己想找谁工程师也没空听客户从头讲一遍背景。这时候就需要一个项目经理坐在前台客户只需要跟项目经理讲需求项目经理负责排期、分活、收集结果、汇报。这个项目经理就是AI代理。具体到系统实现代理层代做了三件事。第一件是意图理解用户一句“帮我写个Python脚本处理这批Excel”代理要解析出这是代码生成任务而不是闲聊。第二件是任务分发代理要根据各AI的能力、当前负载、成本约束决定交给谁。第三件是上下文管理代理要把同一会话里的历史消息、文件摘要、业务规则统一维护起来而不是让每个模型各自维护一套断裂的上下文。这三件事听起来简单真正落地时每一个都有不少细节。1.3 哪些场景适合这套架构也不是所有场景都需要加代理层。如果只是一个人偶尔用一两个AI直接对话就够了加一层反而增加延迟和故障点。这套架构适合的场景有几个共同特征参与人数超过三个、AI数量超过两个、任务需要跨模型接力、并且有权限审计或成本管控需求。最典型的就是企业内部的需求协作平台、自动化研发助手、以及需要多个专用模型配合的智能客服系统。我见过很多团队在这类场景里直接用多模型聊天群比如把四个AI拉到一个群里开会。结果就是上下文混乱、模型互相说车轱辘话、用户也不知道该听谁的。问题的根源不是模型不够聪明而是缺少一个替所有人做信息路由和协调的中间层。加一个AI代理代为交互不是增加复杂度而是把复杂度集中到可控的位置。2. 架构整体拆解四层模型与一条事件总线2.1 四层架构模型我最终采用的是四层架构接入层、代理内核、模型服务层、数据与记忆层。接入层负责接收来自Web端、IM机器人、API调用等渠道的请求代理内核是整个系统的大脑负责意图解析、路由、编排和结果汇总模型服务层是各种AI能力的提供方可以是本地推理服务也可以是云端API数据与记忆层存放会话、事件、向量索引和权限配置。这四层之间的通信不是传统的REST同步调用而是基于消息总线的事件驱动。我为每一层都定义了标准化接口层与层之间通过事件解耦。这样做的直接好处是新增一个模型时不需要改代理内核的代码只要注册能力并订阅相应主题即可新增一个人机交互渠道时也不需要动模型层只要把渠道回执接入总线。层级核心职责关键技术选型参考接入层多端接入、用户认证、请求标准化WebSocket、IM回调、REST API代理内核意图理解、路由决策、任务编排、结果合并FastAPI、LangGraph风格的状态机、规则引擎模型服务层大模型推理、专用模型调用、工具执行Ollama、vLLM、云端模型API、内部模型网关数据与记忆层会话存储、长期记忆、向量检索、审计日志Redis、PostgreSQL、向量数据库2.2 为什么选事件总线而不是REST直连最早一版我图省事让代理内核直接调用各模型API同步等结果。代码确实好写但很快出了问题某个模型API超时整个请求被卡住新增一个模型时要改代理内核里的硬编码调用想实现“两个模型并行跑再比较结果”这种模式同步调用会非常别扭。这些痛点让我把通信方式改成了事件驱动。事件总线的逻辑可以理解成“发邮件而不是打电话”。打电话需要双方同时在线一方慢一点整通电话就悬在那里发邮件则把消息丢进队列收件方什么时候处理都行发件方也不用干等。在多人多AI协同场景里消息天然是异步的用户发一条需求进来代理把需求拆成子任务发给多个模型各模型返回结果后由代理汇总再把最终回复推送回用户整个过程各环节都不需要强同步。2.3 消息结构设计我踩过的最大的一个坑是消息结构不统一。一开始不同的模型返回格式千奇百怪代理内核要写一堆兼容代码。后来我把所有事件统一成一套schema核心字段包括trace_id链路追踪用、session_id会话隔离用、actor发起人、topic事件类型和payload业务载荷。无论接入层、代理内核还是模型层都只认这一套消息结构。{ trace_id: trace_20250601_001, session_id: session_10086, actor: { type: user, id: alice, tenant: tech }, topic: agent.request, payload: { message: 帮我写一个Python脚本处理这批Excel, attachments: [], preferred_model: null } }有了统一结构之后排查问题变得容易得多。出问题时我先按trace_id拉出整条链路看事件在哪一步丢失或超时而不是到处翻日志拼凑用户到底发了什么、模型到底回了什么。这个习惯一定要在一开始就养成否则后面多AI协同起来你连问题出在哪个环节都会搞不清楚。3. 核心模块代理内核里的五脏六腑3.1 能力注册表代理内核第一件要做的事是维护一份“模型能力清单”。每个模型在接入时都要声明自己能干什么、延迟多少、成本多少、上下文窗口多大。我把这些信息统一放在能力注册表里路由引擎做决策时直接查表而不是靠一堆if-else硬编码。这里有个容易忽略的细节模型的“能力”不仅是“擅长代码”还是“擅长写作”还要包含更结构化的信息比如是否支持工具调用、是否支持流式输出、是否适合处理图片等。我用JSON描述每个AI能力{ model_id: local-coder-7b, name: 本地代码模型, capabilities: [code_generation, code_explain], endpoint: http://127.0.0.1:11434/v1/chat/completions, context_window: 32768, max_output_tokens: 8192, cost_per_1k_tokens: 0.001, latency_p50_ms: 800, status: healthy }注册表不能写死要有管理接口让运维同学在新增模型时直接注册不用改代码。我在这块用的是配置中心加一个小型管理页面注册信息变更后广播给所有代理实例。实际上如果你只是做MVP用PostgreSQL表或Redis Hash也能撑住但一定要有“模型退役”和“一键下架”的能力否则线上模型出故障时路由还在往它那边发请求事故就会扩大。3.2 意图识别与路由策略意图识别是整个代理内核里最容易做得花里胡哨、也最容易翻车的地方。我试过用大模型做全动态路由也就是把用户消息、所有模型的能力描述一股脑丢给一个“路由模型”让它决定交给谁。效果不稳定因为路由模型会撒谎它经常给自己“加戏”比如明明没有某个能力却因为上下文里提到了类似关键词就乱指派。更稳的做法是分层路由。第一层用规则加少量关键词做粗筛比如消息里出现“代码”“脚本”“bug”就划分到代码类需求出现“规划”“方案”“步骤”就划分到规划类需求。第二层再让一个能力较强的模型做细粒度判断尤其是那些规则命不中的模糊请求。第三层是兜底策略如果模型判断置信度低就把请求发给一个默认的通用模型而不是直接报错。路由不是只选一个模型。我后来把路由结果设计成“候选模型列表加权重”例如代码类请求里本地代码模型权重0.7云端代码模型权重0.3。当权重最高的模型超时或挂了代理能自动切到备选模型这对生产稳定性非常重要。路由决策的每一步都要记录日志因为模型随时可能抽风没有日志等于让一个间歇性作妖的调度员替你拍板还不用负责。3.3 任务编排与多AI协同模式多AI协同听着高级实际做起来其实就是几种固定模式。我把它们总结成四类顺序接力、并行分发、竞争择优、评委汇总。顺序接力适用于一个任务需要多步完成比如先让规划模型拆任务再让代码模型写实现最后让测试模型审查代码。并行分发给独立子任务比如让多个模型分别负责不同文档章节。竞争择优是同一个子任务发给多个模型选结果最好的那个。评委汇总则是让一个“评委模型”综合多个模型的输出生成最终答案。编排模式适用场景注意事项顺序接力复杂流程分步处理上一步输出要做schema校验再传入下一步并行分发互不依赖的多模块任务注意并发配额和成本峰值竞争择优关键决策需要质量保障需要统一评审标准否则选不出来评委汇总多个模型意见需要融合评委模型要能识别冲突并给理由编排模式不要嵌死在代码里我把每个任务的工作流声明为配置。比如“代码审查”流程就是代码生成模型输出代码然后测试模型审查最后汇总模型给出修改意见。配置化之后产品调整流程不用发版业务对接也灵活得多。3.4 上下文与记忆管理多AI协同系统最容易被低估的模块是上下文管理。用户和代理之间的会话是一套上下文代理和每个模型之间的调用是另一套上下文多个模型接力时还要传递中间产物。我一开始天真地以为把历史消息全量塞给每个模型就行结果很快撞上上下文窗口爆炸的问题。一个业务场景聊了二十轮后光历史消息就有上万token再塞给模型既慢又贵。现在的做法是“短期会话加长期记忆”两层结构。短期会话保存最近若干轮原始的完整消息确保模型能理解当前语境。长期记忆则通过总结和向量化把关键信息和结论存到向量数据库里。每次请求时代理先从长期记忆中检索与当前问题相关的片段再和最近几轮原始消息拼在一起形成精简上下文。这套做法能明显降本增效代价是额外多一次向量检索和总结调用。上下文管理还有一个安全问题不同租户的数据绝对不能串。Redis里的key一律带上tenant_id数据库查询必须带租户过滤条件模型调用时通过system prompt声明数据边界。这些看起来是基本功但在多AI协同场景里特别容易因为“代理转发”而忘掉一旦串了数据就是严重事故。4. 实操过程搭一个最少可用的Agent Mediation Hub4.1 技术选型别一上来就上重武器如果只是验证“AI代理代为交互”的思路不建议一开始就上Kubernetes、消息队列、分布式链路追踪这些东西太重了。我的MVP只用了三个组件FastAPI做代理内核入口Redis做会话存储和轻量级队列Ollama在本地跑一个代码模型。云端模型API就先不接等流程跑通了再扩展。用FastAPI是因为它异步支持好、带自动文档调试起来省心。Redis在这里同时承担了两个角色一是存会话上下文二是通过Redis Streams做简单的事件总线。如果你已经有Kafka也可以直接替换但MVP阶段别把精力浪费在搭建消息集群上。4.2 最小代理内核代码下面的代码是一个可以直接跑起来的版本。它包含三个核心文件逻辑注册表、路由、调用。注释我尽量写清楚你把这套代码跑起来后就拥有了一个最小可用的多AI协同代理。# app.py import asyncio import json import httpx from fastapi import FastAPI, HTTPException from pydantic import BaseModel import redis.asyncio as aioredis app FastAPI() redis_client aioredis.from_url(redis://127.0.0.1:6379) # 能力注册表简化版 MODEL_REGISTRY { local-coder: { url: http://127.0.0.1:11434/v1/chat/completions, model: qwen2.5-coder:7b, capabilities: [code_generation, code_explain], fallback: None, }, general: { url: http://127.0.0.1:11434/v1/chat/completions, model: qwen2.5:7b, capabilities: [general], fallback: None, }, } class ChatRequest(BaseModel): session_id: str user_id: str message: str class ChatResponse(BaseModel): session_id: str reply: str model_used: str def route_message(message: str) - str: # 第一层规则路由 if any(kw in message for kw in [代码, 脚本, 函数, 修复bug, python]): return local-coder # 第二层兜底 return general async def call_model(model_key: str, history: list) - str: model_conf MODEL_REGISTRY[model_key] async with httpx.AsyncClient(timeout30) as client: payload { model: model_conf[model], messages: [ {role: system, content: 你是企业内部AI协作代理中的专用模型只负责自己擅长的事。}, *history, ], } resp await client.post(model_conf[url], jsonpayload) if resp.status_code ! 200: raise HTTPException(status_code502, detailmodel call failed) return resp.json()[choices][0][message][content] app.post(/v1/chat, response_modelChatResponse) async def chat(req: ChatRequest): # 会话隔离按session_id取历史 redis_key fsession:{req.session_id}:messages raw_history await redis_client.lrange(redis_key, -5, -1) history [json.loads(item) for item in raw_history] model_key route_message(req.message) history_for_model history [{role: user, content: req.message}] reply await call_model(model_key, history_for_model) # 写回会话 await redis_client.rpush(redis_key, json.dumps({role: user, content: req.message}, ensure_asciiFalse)) await redis_client.rpush(redis_key, json.dumps({role: assistant, content: reply}, ensure_asciiFalse)) return ChatResponse(session_idreq.session_id, replyreply, model_usedmodel_key)这段代码的骨架就反映了代理层的核心思想用户只管发消息路由和调用都由代理代为完成。实际生产版本会在route_message前加一层大模型路由在call_model后加结果校验和成本统计但框架是一样的。4.3 会话隔离与权限设计多人多AI场景下最怕两个用户之间的上下文互相串。我在代码里所有Redis key都以session:开头这只是第一步。真正的问题是用户能不能访问这个sessionId如果接口只校验登录不校验session归属那么用户A只要拿到用户B的session_id就能读到B的历史消息甚至继续以B的上下文提问。这是多人系统的常见漏洞。我的做法是要求每个请求都带user_id并且session在创建时就绑定用户或团队。访问Redis前先查一张session_acl表确认该用户对该session有读写权限。如果session属于团队则把团队所有用户加入白名单。这个逻辑虽然简单但必须在架构第一版就做后续再补漏洞往往要改一大片调用链。4.4 本地模型联动验证代码写完后用curl发一条消息测一下curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {session_id:s1,user_id:alice,message:帮我写一个Python函数读取某目录下所有csv文件}如果本地Ollama服务正常代理会把这句请求路由到local-coder模型返回代码和解释。你再发一句“今天天气怎么样”它会落入general模型。这个简单验证能确认规则路由、会话存储、模型调用都通了。接下来再慢慢加第二个模型、第三个模型然后把路由从规则升级到模型判断。5. 真实踩坑运行三个月后遇到的典型问题5.1 路由结果不稳定我第一个压测出来的问题就是让大模型做动态路由时它经常把“帮我写个汇报PPT”这种任务路由到代码模型因为“写”和“Python脚本”产生了相关性。排查下来发现单纯用文本向量相似度或者关键词匹配都不够稳必须让路由模型输出一个结构化的“路由决策JSON”包含candidate_models和confidence并且加一个置信度阈值低于阈值就转人工默认模型。{ candidate_models: [local-coder, cloud-general], confidence: 0.78, reason: 用户要求写代码并处理文件 }低置信度请求不能硬路由这是我在这个模块上最大的教训。宁可让你的通用模型多回答几次也不能让一个不对口的专用模型强行输出一个看起来很专业但实际上完全偏题的答案。5.2 多个AI输出打架并行分发时两个模型对同一个问题的结论经常完全不同。比如一个模型说方案A可行另一个说方案A有严重风险。如果没有合并策略用户会收到两个矛盾答案然后陷入更大的困惑。我在实践中发现不能简单把“多数投票”当万能药因为多数模型可能共享同一套训练偏见。我的解决方式是引入“评委模型”或“综合模型”。当编排模式是竞争择优时必须给评委模型一个统一的评审标准比如“更符合用户目标”“更节省资源”“风险更低”。评委要输出结论并说明理由用户如果觉得结论不对可以追加追问触发重新评审。这套机制虽然多一次模型调用但在重要决策场景下非常值得。5.3 上下文窗口被撑爆最典型的现象是用户连续对话十几轮后请求变慢甚至模型报“context length exceeded”。原因是历史消息无脑堆积。我刚开始把Redis里所有历史消息都拼进去一个长会话可能好几千条消息直接把模型上下文窗口塞爆。后来补了两道防线。第一道是对历史消息做摘要压缩超过20轮就触发一次总结用摘要替换更早的原始消息。第二道是向量检索把与当前问题相关的历史片段找出来而不是全量加载。这两道防线合在一起基本上能支撑日常长会话。要注意的是摘要本身也可能丢失细节所以我保留了“展开摘要”的能力用户可以要求看完整历史。5.4 代理之间形成循环调用多AI协同里还有一类隐蔽问题代理A调用代理B代理B又因为某种原因调用代理A形成死循环。我遇到过因为两个代理都在等待对方完成上下文整理结果互相发事件队列被塞满整个系统变卡。解决办法是在事件头部加一个max_depth字段每次代理调用代理时递减减到0直接终止并返回“调用链过深”提示。另外要给事件总线加消息去重同一个trace_id的事件如果已经处理过直接丢弃。这个防循环逻辑在架构设计阶段就要考虑到别等到线上出事了再补。5.5 权限校验被藏在代理后面我把权限校验放在代理内核里后来审计时发现一个问题某些模型服务端日志里能看到用户原始数据而模型层本身不做鉴权等于数据跑到一个不设防的下游。多人多AI协同系统里这个问题会被放大因为代理会同时给多个下游模型提供上下文任何一个下游系统的安全短板都会变成你的短板。补救方式是两层校验代理入口做一次身份和权限校验模型服务层的网关上再做一次租户校验。本地部署的模型还好外部API只能通过请求头传递租户信息并在系统prompt中明确“未经授权的数据不得输出”。这块没有捷径只能做流程管控。6. 从Demo到生产还差的几块硬骨头6.1 可观测性链路追踪比日志更重要多AI协同系统的问题绝大多数是“跨环节”的。表面上用户说我没收到回复内部实际是路由选择了模型A、模型A超时、代理自动切到模型B、模型B返回格式异常、代理重试、最终成功但延迟超时。这种情况光看应用日志很难定位必须靠链路追踪。我在生产环境里给每个事件都写入trace_id用类似Jaeger的组件收集agent.request到model.response的全链路数据。每个环节都要记录三类信息时间、输入输出摘要、异常原因。尤其要记录路由决策的理由因为模型路由本身有随机性如果不记录你根本不知道一条请求为什么会被派给某个模型。我现在排查问题的标准动作是先拉链路的span再决定是看路由日志还是模型调用日志还是看Redis会话状态。6.2 审计谁在什么时候对哪个AI说了什么多人多AI协同系统本质上是一个数据交换平台所有人和所有AI之间都隔着一个代理这反而是做审计的好机会。每次交互都完整记录用户ID、会话ID、选中的模型、完整请求内容、返回内容摘要、耗时和成本。我会定期跑一份审计报表看哪些场景消耗了最多token哪些模型频繁触发失败回退有没有异常的数据访问模式。审计日志的一个隐藏价值是“模型行为阵地”如果你想切换某个模型可以基于历史请求重放对比新旧模型的效果。这几乎不需要额外开发只要你从一开始就保存了完整的请求和响应记录。这项工作对模型选型和成本控制帮助极大。6.3 把本地模型和Agent生态接进来现在很多同学在尝试“本地模型加Agent助手”的组合也就是让一个轻量级本地模型承担大部分日常交互云端模型只处理复杂任务。这种混合架构和本文说的事件总线是一致的本地模型就是模型服务层里一个节点代理根据成本和隐私要求优先调度它。更进一步的是把非聊天类Agent接进来比如机器人场景下的ROS。ROS里各个部件本身就是通过话题通信的把ROS消息桥接到代理事件总线上AI代理就可以一边感知传感器状态一边调度模型做决策再把控制指令写回ROS话题。这个方向本质和多人多AI协同是同一个架构难题不同系统之间如何通过一个中立代理进行标准化交互。6.4 成本与延迟治理最后补一句成本。多AI协同最耗钱的地方不是单次调用而是“无意义的重复调用”。比如同一个上下文打包发给三个模型做竞争择优结果三个模型答案差不多钱就白花了。我给每个编排模式都设置了“成本预算”低优先级任务只用本地模型高优先级任务才走竞争择优并行分发的模型数量一般不超过两个评委模型使用成本最低的轻量模型而不是把最强的模型拿来汇总。延迟方面HTTP长连接和流式输出是必须的但流式输出在多AI协同里的处理要小心不能让用户看到半成品。最简单的做法是代理把所有子任务结果汇集完毕后再一次性流式输出最终回复。虽然用户感知会慢一点但体验一致性比“快但乱七八糟”重要得多。我个人在实际操作中的体会是这套架构最值钱的地方不是“用了多聪明的模型”而是“把交互边界划清楚了”。AI代理代为交互本质上是给混乱的多人多AI生态立规矩。你先让一个模型、一个场景、一个团队跑通再慢慢扩不要一上来就搞全家桶式编排。等扩到三四个模型时你会发现能力注册表、事件总线、链路追踪这些看似不起眼的模块才是系统能不能稳定跑下去的关键。现在再回头看当初踩的那些路由漂移、上下文爆炸、循环调用的坑基本都是因为没在架构层把边界和规范先立住。