最近在做的一个项目让我对 Multi-Agent 有了完全不一样的理解。项目本身不复杂就是把一个“行业竞品分析报告”自动生成工具从单Agent改造成多Agent但真正落地之后我才发现Multi-Agent 系统能不能跑得稳根本不取决于模型选得多好而取决于三件事任务拆分拆得是否合理、上下文隔离是否彻底、协作机制是否经得起异常考验。这三个词是这类系统的地基地基不稳上层做得再花哨都会翻车。这篇文章把我实际踩过的坑、验证过的方法以及最后沉淀下来的代码结构做一个总结给正在做或准备做多Agent系统的朋友一点参考。我还会贴出一段可以运行的 Python 示例代码帮你看清楚“拆任务、隔离上下文、协作”这三件事是怎么一次性落地的。不依赖 LangGraph 这种重型框架就用标准库加一个模拟模型调用也能把一套最小的多Agent系统跑起来。1. 先想清楚Multi-Agent到底在解决什么问题1.1 单Agent的天花板不是模型不够强是上下文先撑不住了最早我做“一键生成行业分析报告”的时候用的就是单Agent加工具调用流程大概是搜索资料、整理摘要、分析数据、生成PPT大纲。看起来挺顺但一跑真实数据就露馅了。问题出在上下文上。单个模型的上下文窗口再大也架不住一个任务里塞进几十篇网页摘要、十几次工具返回、若干个中间推理过程。上下文一长模型就开始“失忆”前面检索到的数据和后面生成的结论经常对不上。到后期追问几句产品细节它甚至会把A品牌的数据安到B品牌头上。这不是模型笨而是单个上下文根本承载不了那么多中间状态。单Agent的另一个问题是没法并行。一个大任务从头到尾串行执行搜索等分析、分析等生成每一步都在消费时间。任务量一大整个流程的耗时就是所有子步骤耗时的简单相加没有任何优化空间。还有一个隐藏问题工具调用一多状态管理就乱。单Agent要在一个对话里搞清楚自己当前调到了第几个工具、上一步拿到了什么、下一步该做什么事情一多它自己都分不清自己在哪里。本质上单Agent的模式是“一个人干完所有活”对任务复杂度、上下文长度、并行能力都有硬上限。1.2 多Agent的本质把“全能选手”改造成“专业团队”多Agent解决的就是上面这些组织问题。它把一个庞大任务拆成多个有明确边界的子任务每个子任务由一个专职Agent负责。每个Agent的上下文只装自己那个环节需要的信息工具栈也只保留自己该用的工具输出结果可以独立校验。这就像把一个“全能选手”改造成一支“专业团队”有人负责检索有人负责分析有人负责写报告各管一段互不干扰。但要提醒一句多Agent不是“把多个模型拼在一起聊天”。如果只是让几个模型围在一起自由对话那大概率得到的是场面热闹但没人对结果负责的垃圾。真正的多Agent系统每一个Agent都应该有清晰职责、明确的输入输出契约、独立的上下文边界以及可预期的协作规则。这也是为什么我会把“拆任务、隔离上下文、协作机制”这三个词当成设计多Agent系统的三根支柱。后面所有内容都围绕这三根支柱展开。2. 任务拆分一个好的多Agent系统从拆任务开始2.1 拆分粒度怎么定盯住“可验证的输出”任务拆分最核心的问题不是“怎么拆”而是“拆到多细”。我见过两种极端。拆得太粗等于没拆。比如“分析竞品定价策略”这种任务如果直接丢给一个Agent去做它内部还是要重新走一遍数据抓取、清洗、对比、总结的完整流程上下文照样会爆跟你不用多Agent没有任何区别。拆得太细代价也很大。每一个子任务都意味着一次模型调用、一次上下文拼装、一次消息传递。拆到“把一句话翻译成英文”这种程度协调成本已经远超收益token费用还会成倍上涨。我自己的经验尺度是拆到“单次模型调用可以稳定产出结构化结果”的粒度。什么叫稳定产出结构化结果比如“检索10条关于某产品的最新用户评论提取时间、用户ID、情感倾向、核心观点”这种任务一次调用就能完成结果可以直接用JSON校验这就是一个合适的原子任务。而“分析这些评论中的共性问题”则可能需要先做分组、再做归纳不是一个步骤能解决的需要继续往下拆。判断拆分粒度是否合适还有一个很直观的检验方法观察每个Agent的输出是否可以被独立验证。如果子任务做完之后你无法判断它结果对不对那这个子任务就还是太大。可验证的输出才是拆分的有效边界。2.2 三种拆分策略职责、流程与数据域实际项目中我常用的拆分策略有三种它们不是互斥的经常混着用。按职责拆分是最符合直觉的。先把系统里需要出现的角色列出来检索员、分析师、文案、审核员、汇报员然后给每个角色挂对应的任务。这种拆法适合任务边界天然清晰的场景比如“客服助手 数据分析 工单处理”就不是一回事按角色把上下文彻底分开每个Agent可以独立演进。按流程拆分更适合流水线型任务。比如“收集数据、清洗数据、分析数据、生成报告”每个阶段是一个Agent前一个Agent的输出是后一个Agent的输入。这种拆法的好处是流程直观适合串行为主的任务缺点是一旦某个环节拖慢整条流水线都跟着等。按数据域拆分则用在数据隔离要求高的场景。比如按产品线拆、按地域拆、按客户类型拆每个Agent只处理自己那一片的数据互相之间不共享底层数据。这种拆法在合规、安全要求严格的场景下几乎是唯一选择。三种策略的适用场景对比如下拆分策略核心思路适用场景主要风险按职责拆按角色分工每个Agent一种能力角色边界清晰的系统角色重叠边界模糊按流程拆按阶段划分输出传递给下一个Agent固化流水线任务链路长单点阻塞按数据域拆按数据边界切分数据不跨域合规要求高的场景边界设计复杂2.3 任务编排先找出能并行的那几条支线拆完任务之后还有一个编排问题。任务之间不全是先后关系很多子任务是可以并行的。比如做竞品分析收集A品牌数据和收集B品牌数据之间没有依赖完全可以同时跑。但如果只按流程拆成“收集→分析→报告”这种天然并行就被浪费了。我的做法是把所有子任务列出来然后画一张依赖图标注清楚哪些任务没有前置依赖、哪些任务必须等特定上游完成、哪些任务的结果会被多个下游复用。这张图不需要画得多严谨只要能回答三个问题就行哪些可以同时跑、哪些必须排队、哪些是合并点。合并点尤其重要。多个并行任务的输出最终要汇聚到一个汇总Agent手里这个汇总Agent必须设计好输入结构否则等它把所有结果拼到一起时上下文又会被塞满前面做的隔离就白费了。我会在每个子任务的输出Schema里提前约定好字段名和类型保证合并的时候不需要再做一层语义转换。3. 上下文隔离每个Agent的大脑必须干净3.1 不隔离的后果串台、污染、成本三连炸上下文隔离是我这次改造中体会最深的一件事。早期版本里我曾经图省事让所有Agent共享一个Memory对象结果很快就翻车了。系统里同时跑着客服Agent和数据分析Agent共享同一个记忆池。数据分析Agent在计算月度销量的时候上下文中混进了客服对话记录里关于“退货”“差评”的讨论它硬是把这些当成市场反馈生成了一份完全跑偏的销量分析。这就是上下文污染。另一种情况是上下文串台。两个不同任务的Agent用了同一个上下文容器后一个任务的中间产物覆盖了前一个任务的前面的分析结果在后面的环节里被篡改。排查这种问题特别痛苦因为模型不报错只是输出看似合理但实际是错的。还有一个容易被忽略的问题成本。如果不做隔离所有上下文都原样传给每个Agent那么调用Token总量就是累加的任务一多费用呈线性甚至指数上升。单Agent上下文虽然大但只计算一份多Agent如果共享同一份大上下文等于每调用一个Agent都重新算一遍全部Token成本直接爆炸。3.2 三个隔离层次进程、记忆与提示词上下文隔离不是一个单一手段而是分层次的。我在项目里通常做三层隔离。第一层是进程级别的隔离。每个Agent实例持有自己独立的上下文容器其他Agent无法直接读取。这一层最重要只要在代码结构上保证每个Agent是一个独立类的实例、有自己的状态存储就已经完成了一大半隔离。具体到实现上可能是Python类里的一个列表、一个字典或者是数据库里按Agent ID隔离的记录集合。第二层是记忆级别的隔离。短期记忆和长期记忆要分开。短期记忆对应当前任务进行中的数据任务结束就清空长期记忆才放到向量库里而且向量库也要按命名空间隔离不同Agent写入不同集合检索的时候绝不跨界。如果让所有Agent共享同一个向量库而不做namespace区分检索时模型会随机拉到别的Agent的历史数据。第三层是提示词级别。每个Agent的System Prompt只包含自己岗位的职责说明、输入输出规范和少量领域知识绝不把其他Agent的任务信息塞进去。很多系统看起来做了隔离但System Prompt里把所有Agent的角色都写了一遍结果等于没隔离。每一次调用都要只给当前Agent最精简、最必要的信息。3.3 共享边界只传“交接物”不传“内心戏”强调“隔离”不是说Agent之间完全老死不相往来。任务本来就是协作的关键在于共享什么、怎么共享。我的设计原则是Agent之间只传递结构化的“交接物”不传递完整上下文。比如检索Agent只需要把“结构化搜索结果列表”递给分析Agent而不需要把自己那个长长的System Prompt、全部检索原始页面、每一个中间推理步骤都附上。交接物可以是一份JSON数据可以是一条消息记录也可以是数据库里的一行记录。生活化一点理解就是团队协作时你跟上一个人交接的应该是“这份文档的处理结果”而不是让他完整回放一遍他是怎么做的。后者信息量很大但根本没有用反而会干扰下一个环节的判断。共享边界画在哪个位置需要在设计阶段就定下来。常见做法是把用户输入、全局配置项、公共数据约束这类“全局信息”做成只读共享而把任务中间状态、临时推理、检索原始内容做成Agent私有数据流动只发生在一级消息传递里。只要坚持“交接时才传不交接不碰”上下文污染的风险就能压住。4. 协作机制让多个Agent真正配合起来4.1 消息格式别传大段自然语言传结构化载荷多Agent的协作本质是消息传递。消息格式设计得好不好直接决定系统的稳定性。我最开始也犯过错误让两个Agent之间直接传自然语言输出比如分析Agent丢给报告Agent一段没有固定结构的文字结果报告Agent每次解析方式都不一样同一批数据产出两份不同口径的报告。后来我强制统一消息格式所有Agent之间只传JSON结构化消息。一个消息至少包含这几个字段{ message_id: msg_20240611_001, sender: searcher, receiver: analyzer, task_id: task_newproduct_001, content: { search_data: [ {source: weibo, text: ..., time: 2024-06-10} ] }, meta: { schema_version: 1.0, cost: 1234, timestamp: 2024-06-11T10:00:00Z } }字段的意义很明确sender和receiver决定路由方向task_id把所有消息串成一条可追踪的任务链content是结构化载荷meta用来记录版本、成本和节点状态。这些字段不仅是为了机器读取更重要的是出了问题能沿着task_id把整条链路翻出来排查。这里有个细节content里永远只放下一个环节必要的数据不要附带额外字段。多一个字段就多一分被误用的可能。我是吃过大亏才明白这个道理的宁可多写几个消息也不要让一个消息携带过多冗余信息。4.2 四种协作模式串联、并联、混合与仲裁观察下来多Agent的协作模式主要有四种。串联模式是最简单的一个Agent处理完消息传到下一个Agent形成流水线。适合处理流程固定、阶段清晰的任务比如数据收集→清洗→分析→报告。优点是逻辑简单、容易追踪缺点是整条链路的耗时等于各环节耗时之和中间任何一个环节慢整个流程就慢。并联模式是多个Agent同时处理不同子任务最后在汇聚点合并。适合按数据域拆分的任务比如同时让三个Agent分别分析三个不同品牌的用户评论再汇总。优点是省时间缺点是对汇聚点的设计能力要求高多个结果怎么对齐、怎么去重、怎么仲裁都需要提前定好规则。混合模式更接近真实业务。一般来说先并行收集数据多个数据源同时开工然后汇合进入串行分析再到最后的报告生成。我就常用混合模式。仲裁模式比较特殊。多个Agent针对同一个问题产出候选答案由一个仲裁Agent按规则选择最优结果。适合对输出质量有要求的场景比如让三个不同风格的文案Agent各写一版营销文案再由仲裁Agent挑一个或融合一版。仲裁本身也是一次模型调用会增加成本和延迟所以不要滥用。四种模式对比协作模式典型特征适用场景代价串联串行流水线流程固定型任务耗时随链路变长并联并行处理子任务数据域拆分型任务汇聚点设计复杂混合并行串行组合复杂真实业务编排逻辑复杂仲裁多候选择优质量敏感型任务额外调用成本4.3 协调者该管什么路由、超时、校验和回滚在大多数多Agent系统里除了业务Agent还应该有一个协调者Agent我习惯叫它Coordinator也有人叫Orchestrator或Supervisor。它不是用来干活的而是用来监督和调度的。协调者的核心职责有四块。第一是路由拿到一个任务后判断应该交给哪个Agent处理同时维护每个Agent当前是否繁忙的状态。第二是超时与重试单次调用设置合理的超时时间超时了先重试一次还是不行就降级到其他Agent或直接失败退出。第三是结果校验每个Agent返回结果都要走一遍结构校验字段缺失、类型错误、长度异常都要能自动识别。第四是状态跟踪与回滚记录整个任务流的状态某个环节失败时能知道应该回滚到哪个节点而不是让错误一路蔓延。我在早期犯过一个严重错误让协调者也做成一个LLM Agent靠自然语言去理解任务并路由。结果是路由错误率大概两成两个业务Agent经常接到与职责不符的任务整个系统一跑就乱。后来我把路由改成确定性规则一个任务请求进来先看任务类型字段再查路由表命中后直接派发。确实有少量复杂情况需要语义判断但我会单独用一个轻量分类模型解决而不是让协调者自己自由发挥。稳定压倒一切协调者这层尤其不能赌概率。5. 实操用不到200行代码把三个核心全部串起来5.1 场景描述一个“新品上市分析”任务理论说了不少现在用一个最简单的例子把整套机制跑通。假设任务是做一个“新品上市分析”输入只有一个产品名输出是一份报告包含市场热度摘要、竞品动态摘要、用户反馈摘要三块。按照前面的设计思路我把这个任务拆成三个AgentSearcher 负责模拟市场信息检索输出结构化搜索结果Analyzer 负责对搜索结果做分析归类输出分析摘要Reporter 负责把分析摘要组装成最终报告文本三个Agent被刻意设计成职责单一。Searcher不知道分析规则是什么Analyzer不需要考虑报告格式Reporter只基于摘要文本生成输出。每个Agent有自己独立的context不为外界共享。5.2 代码实现Agent基类、三个角色与协调者我不依赖LangGraph、AutoGen这类框架只用Python标准库实现一个最小可用版本。逻辑很简单但三个核心机制都在里面。先看消息定义。我用dataclass定义消息保证结构化import json from dataclasses import dataclass, field from typing import Dict, List, Any, Optional dataclass class Message: message_id: str sender: str receiver: str task_id: str content: Dict[str, Any] meta: Dict[str, Any] field(default_factorydict)然后是Agent基类。每个Agent实例持有自己的context列表这是进程级上下文隔离的落地点class BaseAgent: def __init__(self, name: str): self.name name self.context: List[Dict[str, Any]] [] def add_context(self, entry: Dict[str, Any]) - None: self.context.append(entry) def process(self, msg: Message) - Message: raise NotImplementedErrorSearcher实现。这里用一段模拟函数代替真实的大模型调用和搜索API实际项目中只需要把这段替换成真正的LLM调用即可class Searcher(BaseAgent): def process(self, msg: Message) - Message: product msg.content.get(product, ) # 模拟调用搜索API的过程 search_result { product: product, hot_topics: [f{product}热度上升, f{product}讨论量增加], competitor_mentions: [竞品A发布了新版, 竞品B做了一次促销], user_feedback: [用户提到性价比, 用户提到售后] } self.add_context({type: search, result: search_result}) return Message( message_idmsg.message_id -s1, senderself.name, receiveranalyzer, task_idmsg.task_id, content{search_result: search_result}, meta{stage: search_done} )Analyzer实现class Analyzer(BaseAgent): def process(self, msg: Message) - Message: search msg.content.get(search_result, {}) # 模拟分析逻辑 analysis { market_summary: 市场营销热度呈上升趋势讨论量环比增长明显。, competitor_summary: 竞品A的版本更新值得关注竞品B的促销活动可能对份额构成压力。, product_summary: 用户整体评价正面偏多重点关注性价比与售后体验。 } self.add_context({type: analysis, result: analysis}) return Message( message_idmsg.message_id -a1, senderself.name, receiverreporter, task_idmsg.task_id, content{analysis: analysis}, meta{stage: analysis_done} )Reporter实现class Reporter(BaseAgent): def process(self, msg: Message) - Message: analysis msg.content.get(analysis, {}) report f新品上市分析报告 市场热度{analysis[market_summary]} 竞品动态{analysis[competitor_summary]} 用户反馈{analysis[product_summary]} self.add_context({type: report, result: report}) return Message( message_idmsg.message_id -r1, senderself.name, receivercoordinator, task_idmsg.task_id, content{report: report}, meta{stage: report_done} )Coordinator负责调度。它维护一个任务到Agent的路由关系并逐个派发class Coordinator: def __init__(self, agents: Dict[str, BaseAgent]): self.agents agents self.chain [searcher, analyzer, reporter] def run(self, task_id: str, payload: Dict[str, Any]) - Optional[Message]: current Message( message_idfmsg_{task_id}, sendercoordinator, receiverself.chain[0], task_idtask_id, contentpayload ) while current and current.receiver in self.agents: agent self.agents[current.receiver] current agent.process(current) return current启动整个系统if __name__ __main__: searcher Searcher(searcher) analyzer Analyzer(analyzer) reporter Reporter(reporter) coordinator Coordinator({ searcher: searcher, analyzer: analyzer, reporter: reporter }) final_result coordinator.run(task_001, {product: 智能手环X}) print(final_result.content[report])5.3 运行流程与效果分析这套代码运行时的消息流非常清晰一条消息从searcher流向coordinator再流向analyzer最后流向reporterCoordinator的run方法在while循环里完成路由直到消息的receiver不再是已知Agent也就是回到coordinator为止。关键点在于三个Agent的context是彻底隔离的。Searcher添加的搜索结果只存在于Searcher自己的列表里Analyzer添加的分析结果也只存在于Analyzer自己的列表里。Reporter做最终输出时根本拿不到原始的搜索数据它收到的只是Analyzer整理好的结构化摘要。这就是“只传交接物不传内心戏”的代码落地。实际跑起来的感受是链路清晰每层输出可控出错时能快速定位。我拿这套结构去替换项目里的旧单Agent流程后最直观的变化是分析结果的口径稳定多了不再出现报告数据前后矛盾的情况。而且由于每个Agent的上下文都很精简单次调用的Token消耗反而比原来单Agent一个大上下文低了不少。6. 常见问题与排查技巧实录6.1 上下文泄漏症状、定位与修复上下文泄漏是Multi-Agent系统里最常见的隐性Bug特征是Agent输出里出现了它本不该知道的信息。比如负责生成报告摘要的Agent竟然提到了原始网页里的某个URL那说明它拿到的上下文里混进了检索阶段的原始数据。排查思路是分级定位。先把每个Agent在处理消息前后的context完整打印出来比对它接收到的消息和实际消费数据之间的差异。接着检查消息content字段看上游是否传了多余字段。我遇到过一次泄漏是因为Searcher顺手把一段原始HTML塞进了消息的meta字段结果下游Agent在解码时误读了meta把HTML当成了正文。修复手段很直接第一约束消息字段白名单不在白名单里的字段一律丢弃第二Agent内部记录“可用字段列表”读取时只按字段名取值不展开整个消息。第三每次调用模型前做一次上下文裁剪只保留当前任务真正需要的节点数据。这三条做下来泄漏问题基本能杜绝。6.2 Agent空转拆分过细的代价有一次我在系统里加入太多细化Agent结果把“收集用户昵称”和“收集用户ID”都拆成了独立Agent。听起来边界清晰但实际上这两个任务一次调用就能同时完成。拆完之后每个Agent都只产出很少的有效信息白白多出好几次模型调用耗时和成本都上去了这就是Agent空转。判断是不是空转我会看两个指标单个任务中Agent产出的信息量与一次完整调用输入Token的比例以及每个Agent结果里“必要字段”的填充率。如果很多字段是空值或者占位符就说明这个Agent承载的任务太单薄。解决办法是反向合并。把输出字段过少的、依赖关系过近的Agent合并成一个。在上一版代码里如果Searcher和Analyzer彼此依赖太紧我会直接合并成一个“检索分析Agent”。合并之后虽然名字不那么精致但Token用量下降明显整体流程稳定很多。6.3 协作卡死超时、重试与熔断协作卡死也是多Agent系统的高频问题。症状是任务提交后石沉大海没有任何Agent返回消息或者某个Agent一直在等待一个永远不会到达的响应。最常见的诱因是循环依赖Agent A等待Agent B的结果Agent B又在等待Agent A的结果两个人都堵住了。解决办法有三个层次。第一所有Agent调用必须设置超时时间而不能无限等待。我一般把单Agent的单次模型调用超时设为30~60秒超时就抛异常。第二设置重试策略。网络抖动导致的超时重试一次即可模型长时间不响应重试也没意义应该走熔断。第三在Coordinator里维护一张依赖图运行前先检查有没有循环依赖检测到就直接拒绝任务。这三个机制配合起来卡死问题基本不会影响主流程。6.4 token成本失控估算与压缩很多人做Multi-Agent之前没算过账跑了一周之后看账单才傻眼。token成本失控的根本原因有三个上下文共享导致重复计费、任务拆分过细导致调用次数暴增、消息内容包含大量冗余字段导致输入过大。先说估算方法总Token大致等于所有Agent各自的System Prompt Token量 输入消息Token量 模型输出Token量之和再乘以每类Agent的调用次数。做了一个小时性能压测后我可以很快估算出每个任务的成本上限再乘上预计任务量就能知道每月账单大概是什么量级。再说压缩手段。每一处节省都很重要System Prompt里只保留必要的规则和少而精的Few-Shot示例参考文档只截取与任务相关段落结构化消息只保留下游用户依赖的字段输出长度尽量靠max_tokens约束。还有一个比较有效的手段是给Agent加“记忆压缩节点”每隔几轮或每个任务结束后把历史上下文提炼为一条摘要替代完整历史继续往下传。这一步能省掉的比例往往比我预想得大得多。6.5 一份可以贴在墙上的避坑清单最后把调试过程中沉淀下来的检查项整理成清单方便你排查问题时快速对照检查项判定标准异常处理上下文泄漏Agent输出中包含未分配的字段收紧消息白名单裁剪上下文上下文重复膨胀context按轮次无限制累积引入摘要压缩定期清理旧轮次Agent空转输出必要字段缺失合并过细Agent提升内聚路由错误率过高任务被派发到不匹配Agent改用确定性路由规则循环依赖消息流形成闭环运行前置依赖检查拒绝闭环Token用量异常单任务成本超出预估压缩Prompt裁剪消息字段超时响应Agent超过阈值不返回设置超时、重试和降级策略做Multi-Agent这一年多我个人最大的体会是真正能跑得稳的系统靠的从来不是某个Agent的模型有多聪明而是任务拆得是否合理、上下文管得是否干净、协作规则是否足够确定。这三件事做好了哪怕每个Agent只用一个普通模型整体效果也能超过一个用顶级模型硬撑的混乱系统。一开始我的重心全放在怎么让Agent更“智能”上后来踩遍了上面这些坑才明白工程化的基本功才是决定系统上限的东西。如果你正准备从单Agent迁到多Agent我建议把前面这张避坑清单打印出来贴在工位旁边等你能把这几个问题都控制住你的系统才算真正立住了。