做AI应用这一年多我越来越觉得单打独斗的智能体已经到了天花板。让它查资料、写代码、做总结确实都能出活可任务稍微复杂一点比如调研行业现状-输出竞品分析-再基于分析给一份落地执行方案单个智能体就开始顾此失彼前面查到的资料写到后面就丢了角色姿态切换不干净上下文一长输出质量肉眼可见地崩塌。于是我把重心移到了AI智能体协同平台——不是让一个智能体干所有事而是让多个智能体像一支团队那样分工、协作、互检。这篇文章把我从选型、搭建到踩坑的完整过程整理出来适合正在做智能体应用、或者正琢磨着把多个智能体串起来落地的开发者参考也适合刚接触多智能体协作、想理解背后原理的朋友。1. 单个智能体常常不够用协同要解决的真问题先讲明白一个底层逻辑多智能体协同不是为了多而多而是单智能体在某些现实约束下确实做不到位。把这层想清楚后面搭平台时你就知道哪些地方必须做重、哪些地方可以做得轻。1.1 单智能体的三堵墙上下文、角色与工具链第一堵墙是上下文墙。模型有上下文窗口但真实场景中不是窗口越大越好。窗口一大检索进来的无关信息会稀释注意力出现中间遗忘。我实测过一个五万字上下文的单智能体任务它回答最后一问时引用的居然是文档前两部分的数据完全没注意中间章节已经更新了结论。你让一个智能体同时负责检索、分析、写报告、复核它就不得不在一次上下文里塞进所有中间结果信息之间互相干扰最后出来的东西形似而神不至。第二堵墙是角色墙。单智能体可以通过System Prompt切换角色但切换不彻底。让它先当产品经理再当开发它经常把两个视角混在一起要求它严格批判自己刚写的方案它大多时候只是换个说法夸自己。因为没有真正的另一个视角自我批评很容易变成自我确认。我试过用先写方案再自评的方式倒逼质量效果远不如写方案的人和评审的人分开来得明显。第三堵墙是工具链。一个智能体要挂搜索、要开数据库、要调内部API、要写代码执行全塞给它之后工具调用的策略空间非常大模型常常在工具选择上犯迷糊轻则多调几次重则选错工具产生一连串错误结果。而拆成多个智能体各管一摊之后每个智能体的工具面很窄决策空间小出错率明显降下来。1.2 协同带来的不只是效率更是正确性很多人以为协同就是为了让任务跑得快一点。其实快是次要的正确性才是关键。举一组我实际测过的数据一个竞品分析流程单智能体一次跑完的事实性错误率大概在10%到15%拆成采集智能体、分析智能体、复核智能体之后采集只负责抓原文分析只基于采集给的原文片段做推理复核拿着分析结论与原始数据逐条核对最终错误率压到了5%以下。多花一轮调用换来回溯和纠正的机会这笔账非常划算。另一个隐性收益是并行度。调研类任务常常是天然的树状结构单智能体只能串行处理多智能体可以让一个查市场、一个查技术、一个查政策。我做一个行业扫描时三个智能体并行跑了六轮总耗时比单智能体串行少了一半还多而且每个分支的上下文都保持干净互不污染。我踩过一阵子之后总结出一条经验能拆则拆但要拆在信息类型不重叠的地方。如果两个智能体干的事高度重叠拆了只会增加消息成本和协调成本真正值得拆的是那些输入源不同、判断逻辑不同、输出形态不同的环节。拆错了协同平台会变成一个翻倍的烧钱机器。2. 协同平台的骨架角色、编排、消息与共享记忆一个协同平台无论用开源框架还是自研底层都是四件事给智能体定义角色、把任务编排成结构、让智能体之间能传消息、让它们能共享必要的记忆。这四件事想明白了具体用什么技术栈反而不重要。2.1 角色模型Planner、Executor与Critic怎么分我用的角色模型基本是三类Planner负责拆解任务和排期Executor负责执行具体子任务Critic负责复核与汇总。但这不是死的。实际项目中我会再加两个辅助角色Router负责消息分发批量任务时有用Memory Keeper负责管理共享记忆。小项目三个角色就够大项目才需要Router和Memory Keeper。关键不是角色名字而是权责边界。谁是决策者Planner拍板任务拆解方案。谁对输出负责每个Executor对自己的产物负责但Critic有被打回的权利。这里最容易踩的坑是两个角色都想定义问题——比如分析智能体总觉得研究背景没查清楚不停要求采集智能体补充材料结果陷入无限往返。我的解决办法是让Planner在拆解阶段就把信息边界写清楚采集到什么颗粒度算完成、分析依赖哪些字段、哪些情况直接跳过。边界写进任务描述比事后靠大模型临场判断可靠得多。2.2 编排模式管线、星型与黑板常见的编排结构有三种。管线式最直接A写完交给BB处理完交给C。适合有固定先后顺序的任务比如生成-执行-校验。缺点是一环断了全链停链路越长单点故障概率越高。星型则有一个中心智能体通常是Planner做调度周围多个Worker并行干自己的活汇报给中心。适合调研、扫描这类可以并行展开的任务。缺点是中心容易成为瓶颈所有消息都过它一旦消息量大Planner的上下文会迅速膨胀。黑板式是另一个思路所有智能体不直接对话而是往一块共享空间黑板里写内容、读内容。有点像团队共用一个白板谁需要谁来看。这种模式解耦性最好适合参与者多、协作关系动态变化的场景。代价是要设计好黑板数据的读写规范和冲突处理——两个智能体同时写一个条目以谁为准实际项目我推荐混合使用主流程按管线走支线任务用星型并行跨智能体的中间产物统一放共享仓库相当于一个轻量黑板。不要迷信单一模式协作结构应该跟着任务形态走。2.3 消息协议与共享记忆让智能体之间说人话智能体之间通信不是简单地传字符串。我早期吃过亏让两个智能体直接对话A说根据上面的内容做总结B根本不知道上面指什么。后来统一成结构化消息每条消息必须有明确的sender、receiver、task_id、content、source_ref。source_ref是关键用来标明这条内容来自哪次检索、哪个文件、哪段原文这样下游智能体可以做溯源而不是盲信。共享记忆的设计更考验功夫。全量共享不可行上下文放不下完全不共享智能体之间又失去协同基础。我目前的做法是分层记忆记忆层内容谁能写生命周期工作记忆当前子任务的中间结果所有相关执行者任务结束后清理项目记忆本次任务的关键结论和决策记录只有Planner和Critic项目结束时归档长期知识跨任务的领域经验Memory Keeper维护长期保留定期复审写入权限必须做限制。在我的框架里只有Planner和Critic有权修改项目记忆Executor的产出只进工作记忆避免中间智能体把个人推测写进全局结论污染下游。这看起来是个很小的设计实际运行中救了无数次命。3. 从零搭一个可跑的协同平台选型与最小实现理论讲再多不如跑一个最小闭环。这一章我把选型和代码都给出来你照着改就能用。3.1 框架选型自研还是基于LangGraph/CrewAI先把结论放前面如果你的项目是原型验证直接用LangGraph或CrewAI别自己造轮子如果你的场景有复杂的消息审计、权限控制、私有化部署要求那就基于LangGraph的底层原语做二次开发或者干脆自研一个极简消息总线。我整理了一个对比表供你按场景挑方案适合场景主要优点主要代价自研消息总线 多智能体循环轻量内部工具、流程相对固定可控性高、依赖少、容易加审计编排能力弱复杂图谱要自己写LangGraph需要图状编排、状态管理、人机确认状态模型清晰支持条件分支和循环学习曲线稍陡概念多CrewAI偏角色扮演式协作快速原型上手极快角色和任务抽象友好复杂度高时控制力不足Coze等低代码平台非代码人员快速搭建零代码、发布即用深度定制和私有化受限我最终选了LangGraph做骨干因为我的流程里需要条件分支——比如Critic打回时要回到Executor重跑LangGraph的图模型正好支持这种非线性的控制流。但如果你只想验证多智能体协作到底行不行我建议先走CrewAI半小时就能出一个能跑的Demo。3.2 最小实现一个Python多智能体协作Demo给一个可以跑的最小骨架不用任何框架用asyncio模拟一个Planner拆任务、Executor并行执行、Critic复核的流程。目的是让你理解协同的内核理解之后替换成LangGraph只是API层面的差别。import asyncio from dataclasses import dataclass, field dataclass class Message: sender: str receiver: str task: str payload: dict field(default_factorydict) class Bus: def __init__(self): self.agents {} self.history [] def register(self, agent): self.agents[agent.name] agent async def dispatch(self, msg: Message): self.history.append(msg) print(f[{msg.sender} - {msg.receiver}] {msg.task}) if msg.receiver in self.agents: await self.agents[msg.receiver].on_message(msg) class Agent: def __init__(self, name, bus): self.name name self.bus bus self.bus.register(self) async def send(self, receiver: str, task: str, payload: dict): await self.bus.dispatch(Message(self.name, receiver, task, payload)) async def on_message(self, msg: Message): raise NotImplementedError然后定义一个Planner它把任务拆成两路并行class Planner(Agent): async def on_message(self, msg: Message): if msg.task start: # 拆成两个并行子任务 await self.send(executor_market, market_research, {product: msg.payload[product]}) await self.send(executor_tech, tech_research, {product: msg.payload[product]})两个Executor分别处理处理完把结果发给Criticclass Executor(Agent): def __init__(self, name, bus, kind): super().__init__(name, bus) self.kind kind async def on_message(self, msg: Message): # 这里替换成真实的大模型调用 result f{self.kind}_result_for_{msg.payload[product]} await self.send(critic, submit, {self.kind: result})Critic汇总两份结果输出终稿并且有权要求重跑class Critic(Agent): async def on_message(self, msg: Message): self.results getattr(self, results, {}) self.results.update(msg.payload) if len(self.results) 2: # 校验两份结果的一致性通过则输出终稿 if self.results[market_research] and self.results[tech_research]: print(FINAL REPORT:, self.results)跑起来只需要注册Agent再发一条start消息async def main(): bus Bus() planner Planner(planner, bus) ex_m Executor(executor_market, bus, market) ex_t Executor(executor_tech, bus, tech) critic Critic(critic, bus) await planner.send(planner, start, {product: 智能客服}) asyncio.run(main())这个骨架虽然简化到极致但已经具备协同平台最核心的三件事消息路由、角色分工、状态汇总。你要做的就是把每个Executor的on_message换成真实的大模型调用把Critic的校验逻辑换成依赖上的核对清单。3.3 与模型层的磨合函数调用、结构化输出与成本控制多智能体协同下你调的不是一个模型而是一批模型所以必须从一开始就约束输出格式。我强烈建议所有智能体的产出都走JSON Schema用模型的函数调用或结构化输出模式来强制而不是靠Prompt提示请输出JSON。靠Prompt的幻想很快会被打破模型偶尔会给你夹带注释或者Markdown围栏你的下游解析器就崩了。另一个重要教训是不是每一步都需要最强的模型。我的流水线里Planner和Critic用能力强的模型Executor里纯抽取类任务用便宜的小模型分析类任务再用中档模型。分级调用之后整条流程的成本降了大概40%质量几乎没有变化。这个优化越早做越好等流量上来再改改造成本高很多。4. 协同平台五个高频翻车点与完整排查思路协同平台跑起来之后问题比单智能体多一个维度。单智能体出问题基本是模型问题协同出问题往往是系统问题。下面这几个翻车点我全遇到过按出现频率排列。4.1 上下文污染A的输出把B带偏了现象Executor B的分析结果突然风格大变或者反复引用一些明明不存在的中间结论。排查链路我建议按这个顺序走第一步看消息内容本身。把所有发给B的Message打出来检查是不是有额外的prompt内容串进去了。很多人会把Prompt模板和消息内容拼在一个字符串里一不留神就把上一个任务的指令带了过来。第二步查共享记忆。是不是某个智能体把个人看法写进了项目记忆被下游当成了事实。我遇到过采集智能体在项目记忆里写了该产品定位高端但实际上它只是猜测下游分析智能体拿这个猜测当基础做了整页推理全盘跑偏。第三步查消息组装代码。很多人图省事把整个历史对话一股脑塞给下游智能体而不是只传与当前task相关的部分。结果就是前面几次任务的内容全变成了干扰噪声。根治手段是消息最小化原则每条消息只携带当前任务需要的最小上下文并强制附source_ref。另外建议给每条消息加一个可信度字段标明内容是原文引用还是智能体推理下游对低可信度的内容保持警惕。4.2 死循环与幻觉放大终止条件与校验器怎么设计死循环最典型的就是Critic反复打回Executor每次都改但改完又有新问题。这不是模型能力问题是评审标准不明确。我后来的做法是给Critic一份checklist让它只能按清单打回每一条打回必须注明命中清单哪一项同时设置最大迭代轮次我一般设2轮最多3轮超过就当次放弃交由人工处理。另外给Critic的指令里明确写只评审新产出不扩散新需求防止评审变成无限加需求。幻觉放大是协同特有的风险A无中生有了一条信息B基于它继续推理C又基于B的结果再次放大。这比单智能体幻觉更危险因为错误会层层叠加。解决办法就是前面说的trace与source_ref再加一道事实锚点校验——所有数字、日期、引用必须能从source_ref指向的原文中找到找不到就标记为低置信度。低置信度的内容可以保留但必须在终稿里明确标注不允许混在事实里直接输出。4.3 Token成本失控协同模式的钱是怎么烧掉的协同模式的钱是这样烧掉的每个智能体都有独立的上下文但中间结果反复传递信息重叠度很高。比如三个智能体各读了一遍五千字的原文再加上互相传消息单次任务消耗就从5k变成了5k乘以3再加消息量。看起来每次调用不贵累积起来很吓人我第一个月跑协同平台的时候账单直接翻了三倍。控制手段我总结成四条能不传原文就不传原文传摘要或只传结论。分级模型把重模型的调用集中在Planner和Critic上。在共享记忆层做去重和裁剪同一份数据不被重复写入。给每个子任务设置token预算超预算就强制结束并上报。这些都是老生常谈但真能省下真金白银。尤其是第一条很多人舍不得信息损耗总觉得传全原文最稳妥结果就是上下文越来越长、推理越来越慢、账单越来越贵。4.4 可复现性差非确定性输出下的回归测试方案大模型输出天然有随机性协同又会放大这种随机性一个智能体的随机波动可能改变下游的整个走向。同一份输入上午跑和下午跑结论可能不一样。这在生产环境里非常头疼。我的做法有三层。第一层关键节点用温度0能降一点是一点。第二层给每个智能体的输入做快照记录完整的Prompt版本和模型版本出问题能精确回溯是哪个版本引入的。第三层建一套回归用例集每次修改流程后跑一遍对比关键输出字段的漂移程度。别追求完全一致重点是漂移可控——比如结论方向一致、数字误差在可接受范围内就算合格。5. 从能协同到协同得好度量、监控和演进方向平台能跑不算本事跑得稳定、跑得可解释才算。这一章是我的运营视角讲怎么把协同平台当成一个长期工程来维护。5.1 协同质量怎么量化我常用的指标有四个任务完成率最终产出通过Critic验收的比例。低于80%说明拆解或指令有问题。平均迭代轮次一个任务从开始到验收要多少轮。轮次过多说明Planner的拆解颗粒度不对或者Critic的checklist不够清晰。信息损耗率上游关键结论在最终产出里保留的比例。这个靠人工抽检估算比如抽五条核心结论看终稿里保留了几条。单位任务成本Token开销除以任务数用来盯成本趋势。不用追求复杂先盯这四个能暴露大部分问题。我每周会看一眼这个表哪项异常就往哪项定位。5.2 可观测性与行为审计协同平台跑起来之后最怕黑盒。我一开始也没做日志后来排查一个数据错误翻了半天不知道是哪个智能体引入的脏数据。现在我的平台对所有消息做全量审计日志类似一个行为审计层记录每个智能体收到了什么、产出了什么、最终哪些内容进了终稿。这不只是为了排查问题也为了复盘和合规。社区里关于智能体行为审计的讨论越来越多安全领域也出现了针对智能体应用的专项清单其中智能体之间的数据暴露不当信任边界都是重点条目。我的原则是早做留痕不吃亏日志字段宁多勿少等出问题时再补就晚了。5.3 演进方向从人编排到自主协同边界在哪很多人设想的终极形态是智能体全自主协作人只负责给目标。我现在的观点是自主度要渐进给。第一步先把流程固定、消息规范让人可以在关键节点介入第二步让人只在关键节点确认比如Critic打回超轮次、外部行动要执行时第三步再尝试把一些常规决策交还智能体。别一上来就追求全自主否则你收获的不是省心是失控。我目前维护的协同平台最大收益不是省了多少人力而是把思考过程变成了可审计、可回放、可优化的工程资产。什么时候你会觉得这套东西真的值了当业务方问你这个结论是怎么来的时你能点开一条消息链路从原始采集数据一路追到终稿每一环都说得清楚。到这一步AI智能体协同平台才算真正在你的团队里立住了。最后再分享一个小技巧给每个智能体起一个带具体职责的名字而不是Agent_1这种代号。日志可读性会大幅提升排查问题时你一眼就能看出是哪条链路出的问题。就这一个改动帮我省下了无数翻日志的时间。