1. 先从“多开Agent”这个误区说起1.1 为什么“多开”看起来最像多智能体看到这个标题可能有人会觉得我在抬杠多智能体协作不就是多开几个Agent让它们各自干活吗真不是。我在实际项目里踩过这个坑一开始也觉得只要多开几个Agent实例把不同任务丢给它们就能实现“协作”结果跑起来发现它们不是在协作是在互相添乱。先说清楚我理解的“多开Agent”是指同一个Agent系统或者同一套LLM能力被复制出多个实例每个实例有独立上下文执行独立任务看起来像是一支团队在并行工作。比如你开一个Agent写方案开一个Agent做翻译再开一个Agent当审校让它们各自输出结果最后拼在一起。这种模式在很多Demo里很常见也确实有产出所以“多开 多智能体”这个印象特别容易形成。但操作过的人应该都有感觉这种模式一旦任务互相依赖结果就会变得不可控。写方案的Agent不知道翻译Agent的术语口径审校Agent看的可能还是旧版本。表面上人人都忙产出却对不上。这就像把几个人关进同一间办公室各干各的没有项目经理没有晨会没有共享看板最后你拿到的是一堆相互矛盾的碎片。从架构角度看多开Agent只是“多实例”不是“多智能体协作”。协作的前提是共享目标、信息同步、冲突消解和结果收敛。这四个能力恰恰不是靠多开就能自动获得的它们需要一个明确的协调者。这个协调者就是中介者模式里的“中介者”。1.2 “多开”的真实代价上下文割裂与控制权失控很多人觉得多开Agent成本低、上手快确实它不需要额外设计协调逻辑。但你只要把任务复杂度往上提一档代价就出现了。第一个代价是上下文割裂。每个Agent实例有独立的上下文窗口它不知道其他Agent已经做了什么、正在做什么。如果任务要求多个Agent围绕同一份文档协作你只能把公共背景塞进每个Agent的Prompt里结果是同样的大段上下文被重复加载Token消耗成倍上涨而且不同Agent对同一段文字的理解还可能不一致。更麻烦的是当Agent A修改了一段内容Agent B手里的还是旧版本最后合并时冲突一堆。第二个代价是控制权失控。多开之后谁来定义“任务完成”A认为完成了B还在等数据或者两个Agent同时执行了同一项操作产生重复副作用。没有统一的中介者任务的发起、追踪、终止、补偿都没有出口。你可能要把这些逻辑硬编码在业务代码里而业务代码越写越像意大利面。第三个代价是观测性差。多开模式下每个Agent的日志是独立的消息流转路径不清晰。一旦出错你很难知道是哪一个Agent产出错了、哪一条消息没有送达、哪一步决策导致了级联失败。我见过很多团队在这个阶段挣扎最后都要回头补一套中心化的流程编排。这里可以给一张对比表帮助理解维度多开Agent中介者协作共享目标无各干各的由中介者统一定义和分解上下文相互隔离重复加载中介者统一维护共享上下文消息流转直接调用或完全无序由中介者统一路由调度控制依赖外部硬编码集中调度、可追踪冲突消解基本不存在中介者裁决可观测性差良好所以我的结论很直接多智能体协作的关键不是“多开”了多少个Agent而是这些Agent之间的交互有没有被正确建模。中介者模式就是最适合干这件事的架构骨架。2. 中介者模式多智能体协作的正解2.1 中介者模式到底在解决什么问题中介者模式Mediator Pattern是经典设计模式里的一个。它的核心思想很朴素当多个对象需要互相通信时不要让它们直接引用彼此而是引入一个中介者对象所有通信都通过中介者转发。这样对象之间的网状依赖就变成了星型依赖每个对象只认识中介者不认识其他对象。生活里最典型的中介者就是客服中转。用户不需要知道售后、技术、物流分别是哪个部门、由谁负责只跟客服说问题客服负责把消息转给对应的人再把结果带回给用户。整个过程里用户和各个部门没有直接耦合部门之间也不需要互相知道对方的内部逻辑。客服就是这个中介者。放到多智能体系统里Agent之间需要进行的交互比人之间还要复杂。它们要传递任务指令、交换中间结果、共享工具调用状态、协商资源分配还会出现结果冲突。如果每个Agent都直接调用其他Agent的接口系统会变成一张复杂的网状图谁依赖谁根本理不清。引入中介者之后Agent只管向中介者发消息、从中介者收结果一切交互路径都被收拢到一点这才是可维护、可观测、可控制的结构。有人可能会问中介者模式是不是把所有逻辑都集中到一个大对象里面那岂不是把复杂度转移了这个问题很关键。中介者模式不是把逻辑堆砌进一个上帝类而是把“交互规则”从各个Agent中抽离出来单独管理。Agent仍然是执行任务的主体保留自己的推理能力、工具能力和局部状态中介者只负责协调比如决定消息怎么路由、任务怎么分配、结果怎么汇总、冲突怎么裁决。这是“协调层”和“执行层”的分离。2.2 中介者不等于“中心化一切”我在设计多智能体系统时最常被问的一句话是“中介者会让系统变成单点是不是就不适合大规模并发”这里有一个误区中介者模式里的“中介者”是逻辑上的协调入口不等于物理上的单进程单机器。它可以基于消息队列、事件总线、分布式调度组件来实现本身可以水平扩展。更重要的是中介者并不替代Agent内部的能力。比如你要做一个内容创作流水线规划Agent负责拆解选题写作Agent负责起草审校Agent负责检查质量。每个Agent内部的Prompt设计、工具调用、模型推理逻辑都应该保留在Agent自身中介者只负责决定“当前该派哪个Agent上场”“上一轮结果是否合格”“如果审校不过要不要退回给写作Agent重写”。这种分层其实很像项目制团队项目负责人中介者负责排期、协调、验收不负责写具体代码每个工程师Agent负责自己模块的深度执行。如果项目负责人开始替工程师写代码那架构就失衡了。中介者要把“协作的复杂性”包下来而不是把“任务的专业性”抢走。所以我会把系统分成两条线Agent线负责单点能力中介者线负责全局协作。Agent线可以不断扩展比如加新的专业Agent中介者线则保持稳定只有协作规则发生变化时它才需要改。这也是中介者模式给多智能体系统带来的长期价值——扩展成本被限制在了两条线各自的边界内不会互相污染。3. 用中介者模式搭建一套多智能体协作框架3.1 先定义Agent和中介者的边界真正落地的时候第一步不是写代码而是把Agent和中介者的边界画清楚。我在项目里习惯先定四个抽象Agent执行实际任务的最小单元有唯一的名称和能力描述能接收消息、处理任务、返回结果。Mediator中介者管理Agent注册、消息路由、任务调度、状态维护。Message统一消息结构至少包含消息ID、发送方、接收方或主题、消息类型、载荷、时间戳。Context共享上下文由中介者维护用来存放跨Agent需要共享的信息。这套抽象的核心约束是Agent之间禁止直接互相调用。如果一个Agent需要另一个Agent的结果它只能把需求作为消息发送给中介者由中介者决定怎么路由。这样做的直接好处是Agent可以被单独替换、单独测试不会因为某个Agent的接口变了导致连锁故障。我在实际项目中还加了一条约定Agent不感知其他Agent的存在。它可以感知“有一个中介者”但它完全不知道“另一端有谁”。这会让Agent的Prompt设计变得很干净因为它只需要描述自己的专业能力不需要理解整个团队的拓扑结构。这个约定在项目规模变大后特别值钱因为你新加一个Agent的时候不需要去改任何老Agent的代码。3.2 一个最小可运行的Python实现下面我给出一个极简但完整的中介者模式实现核心目的不是生产可用而是帮助理解架构关系。我会用代码把“Agent只认识Mediator不认识彼此”这件事说清楚。from dataclasses import dataclass, field from typing import Dict, List, Optional import uuid dataclass class Message: sender: str content: str msg_type: str task receiver: Optional[str] None msg_id: str field(default_factorylambda: uuid.uuid4().hex[:8]) result: Optional[str] None class Agent: def __init__(self, name: str, mediator: Mediator): self.name name self.mediator mediator self.mediator.register(self) def receive(self, msg: Message) - None: # 子类实现具体任务处理逻辑 raise NotImplementedError def send(self, msg: Message) - None: msg.sender self.name self.mediator.route(msg) def reply(self, original: Message, content: str) - None: self.send(Message( senderself.name, receiveroriginal.sender, contentcontent, msg_typeresult, )) class Mediator: def __init__(self): self.agents: Dict[str, Agent] {} self.waiting_results: Dict[str, Message] {} self.pending_queue: List[Message] [] def register(self, agent: Agent) - None: self.agents[agent.name] agent def route(self, msg: Message) - None: if msg.receiver: # 点对点路由 self._deliver(msg.receiver, msg) else: # 没有指定接收者时入队由中介者调度 self.pending_queue.append(msg) def _deliver(self, receiver: str, msg: Message) - None: agent self.agents.get(receiver) if agent: agent.receive(msg) def process_queue(self) - None: while self.pending_queue: msg self.pending_queue.pop(0) receiver self._select_target(msg) if receiver: self._deliver(receiver, msg) def _select_target(self, msg: Message) - Optional[str]: # 最简单的调度策略按消息中的receiver或者按能力名匹配 if msg.receiver: return msg.receiver for name, agent in self.agents.items(): if hasattr(agent, can_handle) and agent.can_handle(msg.content): return name return None这个例子虽然简单但已经把关键结构体现出来了Agent初始化时注册到Mediator发送消息时只调用self.mediator.route()接收消息时只面对Message对象。Agent之间完全没有直接引用。Mediator可以扩展出更复杂的调度策略比如基于任务类型、基于优先级、基于当前负载。再补一个具体的业务场景示例规划Agent拆解任务写作Agent执行写作审校Agent负责质检。规划Agent发送一个“写一篇技术文章”的任务给写作Agent写作Agent产出后发送结果给中介者中介者再把结果转给审校Agent。所有流程都由中介者串联你加一个新的Agent只需要注册到Mediator改一条路由规则。3.3 从中介者再往前一步加入角色路由与任务队列上面的最小实现只解决了结构问题真正到生产环境还需要补三个能力路由策略、任务队列和状态管理。路由策略决定了消息到达Mediator之后应该交给哪个Agent。最简单的是按receiver字段定向转发但多智能体系统里更多场景是“我想找一个能做XX的Agent”这时候就需要能力路由。我常用的做法是给每个Agent声明能力标签capabilitiesMediator维护一个能力到Agent的映射。任务到来时Mediator根据任务类型匹配对应能力的Agent如果一个Agent正在忙就进入等待队列等它空闲后再派发。这样比硬编码receiver灵活得多。任务队列很重要。多Agent系统里请求往往是异步的Agent A在等Agent B的结果Agent B可能还在处理上一个任务。如果Mediator是同步调用整个系统很容易被卡死。我在实际项目里会为Mediator增加一个pending_queue和超时机制任务入队后如果超过指定时间没有被消费就触发超时回调把任务重新分配或上报异常。这类机制可以降低Agent之间互相等待造成的死锁风险。状态管理则要回答“当前整体任务进展到什么程度了”。我的做法是维护一个全局会话Session里面包含任务版本号、每个Agent最近一次输出、共享上下文指针。Agent回复时带上任务版本号Mediator就能判断结果是否过期。如果版本号不匹配说明有其他Agent修改了共享上下文旧结果直接丢弃避免脏数据污染整个流程。4. 主流多智能体框架里的中介者痕迹4.1 AutoGen / CrewAI / LangGraph 各自怎么实现如果你的项目不打算完全从零自研也可以用现成的多智能体框架但我建议先看懂它们内部的中介者设计否则很容易被“Agent”这个词带偏。AutoGen是微软开源的多Agent对话框架。它里面有一个GroupChatManager这个角色做的事就是典型的中介者管理多个ConversableAgent决定下一轮谁发言把消息路由给合适的Agent并且维护聊天的历史上下文。AutoGen里的Agent之间也不直接对话而是通过GroupChatManager统一调度。如果你用过AutoGen你会发现它的消息流完全依赖这个Manager它就是一套中介者实现。CrewAI的思路不太一样它强调“角色扮演”。Crew团队包含多个Agent每个Agent有角色、目标和背景故事然后通过Process比如sequential或hierarchical协同工作。在顺序流程里前一个Agent的输出自动成为后一个Agent的输入这条链路其实是Crew在内部维护的在层级流程里还有一个Manager Agent负责任务分配和结果审查这个Manager Agent就是中介者。LangGraph则把多Agent协作建模成图。节点是Agent或函数边是状态转移图的执行逻辑由一个全局State对象驱动。这里的“中介者”不是某个具体类而是图的执行引擎和State。每个节点从State里读取输入把输出写回State再由引擎决定下一步走哪条边。这比显式的中介者更抽象但逻辑上它承担了同样的职责收敛交互、统一状态、控制流转。我见过不少团队选型时只看Agent的名号不看编排层。其实你要重点考察的不是Agent怎么定义而是框架怎么管理Agent之间的消息、上下文和调度。框架的协作能力恰恰由它的“中介者组件”决定。我把这几个框架放一起对比框架中介者载体协作模型适用场景AutoGenGroupChatManager群聊式多轮对话对话、角色扮演、多步推理CrewAICrew / Process流程式任务链路内容创作、流程化业务LangGraphStateGraph State图状态机复杂条件分支、可控流程自研Mediator自建路由与队列灵活定制强定制、深度集成业务4.2 自研框架 vs 框架组件选型时你要先问自己你的核心业务复杂度在哪里如果只是几个固定的Agent轮流干活用CrewAI这种顺序编排就够了如果Agent之间需要多轮讨论、动态决策AutoGen的对话模式更合适如果流程有很多条件分支、循环、回退LangGraph的图模型更可控。但如果你对协作过程有很强的定制要求比如要对接自己团队的权限体系、任务系统、运维基础设施或者要精细控制每条消息的生命周期那我建议自研一个轻量中介者。不要迷信框架框架解决了通用问题但定制成本可能远超你的想象。核心是可以从框架的架构思路里借鉴但双亲的职责一定要想清楚。我已经开发过的几个项目经验是团队越靠近业务侧越应该自研中介者团队越靠近研究侧、Demo侧越适合直接用框架。承接业务系统时你需要的不只是消息流转还有权限、超时、审计、重试、成本控制这些很难靠通用框架直接满足。4.3 记忆和工具该挂在哪里很多多Agent项目最终出现问题不是Agent能力不够而是“记忆”和“工具”的位置放错了。常见的错误是把记忆放在Agent内部。每个Agent独占一份历史记忆结果Agent A不知道Agent B已经做过的决策重复提问、重复输出甚至得出矛盾结论。正确的做法是把共享记忆放在中介者这一层由中介者决定哪些内容可以给哪些Agent看到。Agent只需要在消息里带上自己对上下文的需求中介者从共享记忆里取出相应片段返回给它。这既可以节省上下文窗口也能保证信息一致性。工具也一样。不要在每个Agent里各自绑定一套工具应该让Agent通过中介者去调用工具。中介者维护工具注册表Agent发出“我想搜索”“我想读取文件”的请求中介者执行对应工具并把结果返回。这个设计的好处是第一工具可以被统一审计谁在什么时候调用了什么工具都有记录第二Agent不需要关心工具的API细节只关心自己需要什么降低了Agent开发的复杂度第三如果某个工具需要切换版本或者降级不需要改所有Agent。我见过一个失败的案例团队让每个Agent都直接绑定了外部API Key结果某个Agent误调用了一个高消耗接口账单涨了好几倍。后来把所有工具收口到中介者统一管理加上配额和审批问题才解决。这看起来是工程问题本质上就是把应该放在中介者这一层的权限控制权给了Agent。5. 常见问题与排查技巧实录5.1 消息风暴与循环调用多智能体系统跑起来之后最恶心的故障就是消息风暴。想象一下写作Agent把初稿发给审校Agent审校Agent觉得不行退回重写写作Agent改完再发过去审校又提出新问题两边就来回高频交互消息越积越多最后系统吞吐被拖垮甚至整个流程停止响应。这个现象在AutoGen里尤其明显群聊式协作特别容易陷入对话循环。我排查这类问题时会先看中介者的消息日志检查是否存在同一对Agent之间反复流转同一任务的情况。解决手段有几个第一给消息加跳数限制比如一条任务最多流转5次超过后自动终止并上报给人工第二加去重机制如果发送的Content和前一轮相同直接拦截判定为无进展循环第三设置冷却时间同一个Agent处理同一个任务的频率不能超过一定阈值。这些都可以放在Mediator层实现不需要改动Agent内部的逻辑。在实际项目里我还会在消息里带一个trace_id用来串联整条链路的每一次流转。这样即使出现循环也能通过trace_id快速定位到是哪个环节开始打转。不然你面对的是几十条相似日志根本无从下手。5.2 死锁、超时和重试多Agent系统还有一个隐蔽问题死锁。Agent A请求Agent BAgent B又请求Agent A两边都在等对方先返回结果结果谁也没继续执行。如果系统是同步阻塞调用这种循环等待会直接把进程卡死。我在设计Mediator时会刻意把Agent之间的消息传递设计成异步模式。Agent发消息后立即返回不阻塞自己的执行Mediator把消息入队后续通过回调或者事件机制通知Agent。同时每个任务都要有明确的超时时间比如一个任务最多等30秒超时后Mediator自动把任务标记为失败并触发补偿逻辑比如重新分配或通知上层。另一个容易忽略的是重试。Agent 调用大模型接口时经常因为网络波动、限流失败直接失败显然不行盲目重试也不行可能放大成本。我会把重试策略收敛到Mediator层统一控制比如同一任务最多重试2次重试间隔采用指数退避。同时把Agent的幂等性做起来Agent处理消息时带上消息ID如果收到重复消息可以直接返回上一次的结果避免重复执行副作用。当然最稳妥的方式是让所有关键消息持久化比如写入本地数据库或消息队列。这样即使系统崩溃也能从持久化记录里找到流程中断的位置手动恢复不至于整个任务重跑。5.3 中介者单点问题与可观测性中介者本身也可能成为瓶颈。如果你只有一个Mediator实例所有Agent的消息都要经过它它一旦挂了整个协作系统就瘫痪了。要解决这个问题第一选择是给Mediator加持久化把消息队列放到Redis或RabbitMQ等成熟组件里Mediator进程崩溃后可以重新消费未完成任务。第二选择是让Mediator无状态化把Agent注册信息和共享上下文放到外部存储Mediator只负责计算路由和调度逻辑这样多个Mediator实例可以水平扩展。可观测性是中介者模式一个非常重要的加分项。既然所有消息都经过Mediator你就可以在这里埋点记录消息总数、Agent响应时长、任务失败率、重复流转次数。这些指标是优化整个系统的数据基础。我会在Mediator里把关键事件全部打日志消息入队、路由目标、Agent开始处理、Agent返回结果、超时触发。运行一段时间后用这些日志分析瓶颈比如哪个Agent响应最慢、哪条链路频繁失败然后针对性优化。我自己在调试时还有个习惯给每个会话配一个独立的中介者实例。也就是说不是全局一个Mediator管所有任务而是每个业务会话创建一个Mediator内部维护这个会话的Agent列表和上下文。这样既避免不同会话之间的消息串扰也让会话级别的隔离和恢复变得简单。这个做法在并发需求高的场景下尤其好用。最后说点个人体会做了几年多智能体项目我的感觉是真正决定系统能不能跑得顺的往往不是单个Agent的模型多强、Prompt写得多好而是它们之间有没有一套清晰的协作协议。中介者模式的价值不在于它多高深而在于它强制你回答几个关键问题消息从哪来、到哪去、由谁决策、冲突怎么解、状态怎么存。有一次我临时接了一个项目对方说已经“多开Agent”跑起来了结果我一看代码每个Agent直接在业务逻辑里互相new对象、调公有方法改一个Agent的构造函数另外三个地方报错。我花了三天时间把所有交互收口到中介者新加功能时就再也不用担心牵一发动全身。那种舒坦感只有踩过坑的人才懂。所以我给正在做多智能体的朋友一个建议别急着堆Agent数量先想清楚你的中介者长什么样。哪怕一开始只有一个简单的Mediator类先把结构立起来后面再慢慢加路由、加队列、加持久化路会顺很多。这个内容后续还可以扩展的方向也很多比如把中介者做成可视化编排面板、引入自动路由策略学习、把消息轨迹做成复盘工具。但所有扩展都要有一个前提——你的协作根基建在中介者模式上而不是建在多开的幻觉上。