上个季度我把一个跑了两年的单Agent自动化报告工具推倒重来升级成了Multi-Agent架构。整个过程让我最有感触的一件事是Multi-Agent不是把API多调几次也不是把一堆Prompt塞进一个线程里——而是把一个脑子里挤着的事情拆成几个脑子各管一段。要做到这一点背后其实就三件事拆任务、隔离上下文和协作机制。这三个词看着简单真正做起来每一个都能让你摔几跤。让我一个一个拆开讲然后直接把一套可以落地的代码结构给你你照着改就能用。1. 为什么我不再用单Agent硬扛复杂需求1.1 单Agent的瓶颈长提示词和上下文崩塌我最早那版工具其实是一个“全知全能”Agent一个System Prompt里塞满规则既负责查资料又负责写结论还要兼顾格式排版。前期还好一旦任务变复杂比如需要处理30份PDF再生成一份分析报告我发现它开始频繁“精神分裂”。具体表现是上下文越来越长Token费用肉眼可见地往上翻但效果越来越差指令开始互相打架比如要求“严谨引用”和“自由发挥”共存它经常选择性遗忘后半段开始出现上下文污染前面的原始数据噪音污染了后面的写作逻辑输出里混进了大量无关内容甚至直接把PDF原文片段抄进结论。单Agent最核心的问题是它只有一个“工作记忆”。所有任务、所有中间结果、所有规则都要挤在一个上下文窗口里。你无法把大任务拆到不同大脑里并行处理更没办法让某个环节用独立的状态去思考所有东西全都是纠缠在一起的。1.2 Multi-Agent核心三个支柱如何解决瓶颈后来我换了思路不再追求一个Agent搞定一切而是把一个复杂目标拆解成若干子任务交给不同的Agent去执行。这就是Multi-Agent的基本逻辑。拆任务对应的是“复杂度的解耦”。一个任务拆成N个子任务后每个子任务对模型的推理压力更小输出质量上限更高。隔离上下文对应的是“专注力的保障”。每个Agent只接收自己干活所需的那部分数据不相关的东西一律不塞给它这样它的上下文窗口更干净不容易产生幻觉也不会被无关字段带跑。协作机制对应的是“流程的一致性”。拆开的任务、隔离的上下文最后要能拼回一个完整结果就需要一套明确的信息传递和调度逻辑否则就是各自为战。我打个比方。单Agent相当于一个身兼数职的全能员工你让他既当前台又当财务再当售后他一定会出错。Multi-Agent像一个小团队有人查资料有人做分析有人写报告最后还要有个负责人统筹和校稿。每件事都有人管每个人只需要专注自己的领域。1.3 方案选型自研框架还是直接拼市面上已经有LangGraph、AutoGen、CrewAI这些现成框架我当时也研究过一圈。框架的好处是封装好了调度和记忆管理上手快坏处是编排逻辑比较重新手容易被抽象层绕晕出了问题不太容易定位到底是哪一步透传了脏上下文。我的选择是自研一套轻量级调度。原因有三点我们的任务链路并不复杂主要就是“拆解—执行—汇总—校验”用框架反而增加学习成本我们需要精细控制每一步传给Agent的上下文框架自带的历史记忆管理经常把不想要的信息也带进去自研代码更容易加日志和debug排查上下文污染时非常有用。如果你也只是跑个Demo用现成框架没问题如果你要做到生产级可控强烈建议自己写一层非常薄的调度壳把这套Multi-Agent理念亲手实现一遍。2. 拆任务如何把一个大目标切成可执行的子任务2.1 任务拆分的三个原则拆任务不是简单地把一段话切成两段需要遵循三条原则否则拆出来的都是伪任务。第一按职责边界拆。每个Agent只负责一类逻辑任务比如“资料检索Agent”只负责检索和抽取“分析Agent”只负责数据处理“撰写Agent”只负责输出文本。不要让一个Agent既写代码又写文稿职责一混Prompt就会互相污染。第二按数据域拆。不同Agent接触的数据集要尽量不重叠。比如检索Agent读原始PDF和网页分析Agent只能拿到结构化统计结果撰写Agent只能拿到分析结论。谁需要什么就给什么不需要的一律不给。第三按依赖关系拆。子任务之间必须有明确的先后或者并行关系。没有依赖的任务可以并发执行有依赖的任务则要串行排队否则会出现Agent拿到了A的输出却发现B还没产出的尴尬局面。2.2 拆解流程从目标到依赖图我实际操作时有一套固定流程每次拆任务都按这个走不容易漏。第一步写清楚目标描述。不要一句“帮我写一份报告”而要写“生成一份关于新能源汽车2024年市场规模的行业分析报告包含核心数据、趋势判断和一页PPT摘要”。目标越具体拆出来的任务越清晰。第二步使用大模型辅助生成任务列表。我会让一个“策划Agent”把目标拆成10个左右的原子子任务再用“去重合并”来精简。比如“收集2023年销量数据”和“收集2024年销量数据”可以合并成一个“收集近两年销量数据”并标注需要过滤的时间范围。第三步给每个子任务标注输入、输出和验收标准。这一步很容易被忽略但它决定了后续上下文隔离的边界。第四步建立依赖关系。比如“数据清洗”必须在“数据收集”之后“趋势分析”必须在“数据清洗”之后。依赖关系搞清楚后面做并发调度时就方便了。我习惯用一张表来记录拆解结果这里给个示例子任务负责人输入输出依赖收集销量数据检索Agent厂商官网/行业报告列表结构化CSV无数据清洗与去重清洗Agent检索Agent的CSV清洗后的数据表收集销量数据趋势分析分析Agent清洗后数据表趋势结论JSON数据清洗与去重报告撰写撰写Agent趋势结论JSON完整Markdown报告趋势分析质量校对校对Agent完整Markdown报告修订清单报告撰写2.3 案例拆解自动生成行业调研报告我用一个真实做过的案例更直观地说明。当时要给某客户做一份新能源汽车市场调研目标词条是“Multi-Agent拆任务、隔离上下文、协作机制”——拆出来的任务大概是这样首先是“资料检索任务”我让检索Agent只负责下载和抽取基本面信息比如政策文件、销量新闻、企业财报摘要。它输出的不是一堆网页全文而是经过压缩的要点式JSON。其次是“数据处理任务”清洗Agent拿到JSON后把数据转成统一格式比如把“万辆”“万台”统一成纯数字把日期格式统一剔除明显重复的新闻。然后是“研判分析任务”分析Agent只看清洗后的结构化数据它负责写结论性内容。比如“2024年第一季度销量同比增长30%主要驱动因素是补贴政策”。最后是“报告撰写任务”撰写Agent只看分析结论和少量关键图表数据它用人类编辑的口吻写出完整报告。它拿不到原始网页也拿不到脏数据所以它不会被无关信息带偏。关键是每个任务的入口都很窄。从下往上数据的抽象层级越来越高上下文越来越干净。这其实就是把拆任务和隔离上下文结合在一起用了。3. 隔离上下文每个Agent只管自己该知道的3.1 上下文污染是怎么毁掉结果的上下文污染是Multi-Agent系统里最隐蔽、最毁结果的坑。第一次遇到时我花了两天才找到原因症状是撰写Agent的产出忽然开始引用一些完全无关的内容比如在新能源汽车报告里突然冒出一句“该路段维护成本较高”。排查后发现早期我在设计时为了省事把检索Agent返回的原始资料全文直接塞给了撰写Agent。这些资料里有大量关于维修网点分布、路段基础设施的零散信息根本不是报告需要的但撰写Agent却从中“学到了”一些看起来像结论的东西。上下文污染有三种常见表现Token浪费型Agent接收了大量无关内容注意力被稀释关键信息反而不突出隐性偏差型Agent从它本不该接触的数据里发现了表面规律然后编进输出造成幻觉指令覆盖型后注入的内容覆盖了System Prompt里的规则Agent开始按“用户消息”里的要求胡来。很多人在单Agent时代觉得上下文越大越全越好但在Multi-Agent里这条经验完全失效。你必须在架构层面主动做减法。3.2 实现隔离的三个层次隔离上下文不是只靠“少传点数据”就能做到的我把它拆成三个层次每层都要落实。第一层是Prompt级隔离。每个Agent要有自己独立的System Prompt设定独立的角色、职责边界和输出格式不要共用一套。比如检索Agent的Prompt强调“只抽取事实不做判断”分析Agent的Prompt强调“只基于输入数据推导不引用外部记忆”。这层隔离是基础也是最容易被忽略的。第二层是数据级隔离。每个Agent只接收自己任务范围的数据。我实现时会在调度代码里做一层“数据门卫”Gate而不是让Agent自己去选。凡是允许传给该Agent的字段在白名单里不在白名单里的字段一律过滤掉。第三层是状态级隔离。Agent的记忆要独立。每个Agent有自己的history互不共享。不能因为Agent A之前查过某份资料Agent B就自动认为自己也见过。如果某些信息确实需要共享那就通过显式的协作消息传递而不是让上下文串味。3.3 什么时候可以共享上下文什么时候必须隔离有些信息是全局的没必要隔离。比如用户的目标描述、项目的最终交付格式、统一的术语表。这些可以在所有Agent的Prompt里都放一份能让输出风格保持一致。真正需要隔离的是中间产物和原始数据。比如检索得到的网页全文、清洗前的原始表格、会议记录、内部分析草稿这些默认都不要跨Agent共享按需传递。我总结了一个简单判定原则如果某条信息会让接收Agent产生“我可能不需要它”的想法那这条信息就该隔离。反过来如果接收Agent缺了它无法完成任务那就必须传递。这里最难的是“元信息”的隔离。元信息指的是“我知道有个Agent正在做什么”“我知道有个任务失败了”这类状态信息。我一开始会把所有状态日志都发给每个Agent结果它们开始在输出里讨论别的Agent的情况就像团队里有人在上班时间八卦同事的日程。后来我明确在System Prompt里写了“你不是调度员你不需要了解其他Agent的任务状态”这个问题立刻消失了。4. 协作机制让Agent们像团队一样不吵架4.1 消息协议用结构化JSON而不是自然语言Agent之间要协作第一步是定义通信协议。我见过有人直接让一个Agent输出自然语言然后传给下一个Agent当输入比如“请你根据我刚才说的情况生成报告”。这条路几乎必死自然语言里充满了模糊指代下一步Agent很难稳定解析。我在系统里用结构化JSON作为统一消息格式。每条协作消息包含这些字段{ request_id: 7f2a1c90, sender: master_agent, receiver: research_agent, task_id: task_001, type: execute_task, payload: { goal: 收集2024年新能源车销量数据, constraints: {time_range: 2024-01-01_2024-12-31}, output_format: json }, timestamp: 2025-04-01T10:00:00Z }用结构化消息的好处有三个路由明确sender和receiver字段让调度器知道该把消息送到哪里参数分离payload里只放本次任务需要的数据不掺其他杂质可追踪request_id让整个任务链路可以串起来查日志。自然语言协作不是不能用但它适合“讨论型”场景不适合“生产型”流水线。我在生产环境里只允许结构化消息只有在需要Agent自由发挥时才会用自然语言讨论。4.2 任务交接与仲裁谁负责发下一步指令协作机制里最核心的问题是“谁说了算”。我推荐使用“主管-专员”模式Master-Worker由一个调度Agent或一段调度代码统一裁决而不是让Agent们互相指派任务。主管-专员模式的好处是闭环可控。主管负责拆分任务、下发指令、接收结果、判断是否达标专员Agent只负责执行自己的子任务无权派活给其他Agent。这样避免了两件事一是Agent之间互相推诿迟迟不交活二是出现递归指派比如Agent A让Agent B干活B又让A干活形成死循环。你可能会问那“讨论型协作”怎么办比如需要多个Agent互相评分。我的做法是让主管发起一个“评审任务”把多个Agent的产出作为输入交给一个独立的评审Agent去打分而不是让Agent们直接互相传消息。从形式上看它们还是有互动实际控制权始终在主管手里。4.3 结果校验与人类介入协作机制里另一个重点是对的“产出”建立信任。每个Agent都会犯错尤其是当任务非常复杂或者输入数据有噪音时。如果主管直接把上一个Agent的输出当作既成事实传给下一个Agent错误就会沿着流水线不断放大。我在每个Agent输出之后加了一个校验层。校验层有两种实现方式规则校验检查输出格式是否符合JSON结构、是否包含必填字段、数字是否在合理范围内模型校验用另一个Agent作为独立“QA员”对前一个Agent的输出进行交叉验证。当校验失败时有两种处置策略。第一是让原Agent带着失败原因重新跑一遍第二是直接转人工介入。我设定了一个置信度阈值当主管Agent判断某条结果质量评分低于0.7时它会中止整条流水线进入人工review队列。这里有个细节不要让校验Agent直接修改原Agent的输出而是让它输出一份“修订建议清单”。修改动作仍然回到原Agent那里去执行否则等于让一个没看过原始信息的Agent擅自改了别人的成果数据来源就乱了。4.4 并发与顺序什么时候并行什么时候串行协作机制还面临一个效率问题任务之间如果不存在依赖关系完全应该并行执行节省整个系统的响应时间。比如在前面的调研报告里“收集销量数据”和“收集政策资料”这两个子任务互相独立我就让两个Agent同时跑。而“趋势分析”必须等“数据清洗”完成后才能开始这时候就串行。实现并行时最需要小心的还是隔离。如果多个Agent同时运行并且它们共享同一个API Key池那没问题但如果它们共享同一个Token计数器和历史记录对象就会互相踩踏。我最后的做法是每个Agent实例独立专门的调度器用一个线程池管理并发数量并发数不超过3避免模型接口限流。5. 实操过程从零搭一个最小可用的Multi-Agent系统5.1 抽象一个BaseAgent类先用一个最基础的抽象类把Agent的通用属性封装起来。这个类非常简单核心就是“一段System Prompt 一段可调用的大模型接口”。# base_agent.py from dataclasses import dataclass, field dataclass class BaseAgent: name: str system_prompt: str api_key: str model: str gpt-4o-mini history: list field(default_factorylist) max_context_tokens: int 6000 def reset_context(self): self.history [] def add_context(self, message: dict): self.history.append(message) # 简单截断策略只保留最近几轮防止Token超限 if len(self.history) 8: self.history self.history[-8:] def call_model(self, user_input: str) - str: messages [{role: system, content: self.system_prompt}] messages.extend(self.history) messages.append({role: user, content: user_input}) # 这里真正调用大模型接口 response self._llm_completion(messages) self.add_context({role: assistant, content: response}) return response def _llm_completion(self, messages): # 以OpenAI SDK为例生产环境请替换为自己的接口封装 from openai import OpenAI client OpenAI(api_keyself.api_key) resp client.chat.completions.create( modelself.model, messagesmessages, temperature0.3, ) return resp.choices[0].message.content这个类最重要的设计是max_context_tokens和history截断。只要History超过8轮就砍掉最旧的这样每个Agent的记忆窗口都被限制在自己的任务范围内不会无限膨胀。5.2 实现隔离上下文按任务创建独立Agent实例隔离的关键在于每个子任务都单独创建一个Agent实例而不是复用同一个实例。每个实例有自己的System Prompt和History。# creator.py from base_agent import BaseAgent def create_research_agent(api_key): return BaseAgent( nameresearch_agent, system_prompt( 你是一名严格的行业资料研究员。\n 你只负责抽取数据、事实和引用来源。\n 你不做判断不写结论不带个人观点。\n 只输出JSON格式的要点列表。 ), api_keyapi_key, ) def create_analyst_agent(api_key): return BaseAgent( nameanalyst_agent, system_prompt( 你是一名数据洞察分析师。\n 你只能基于input_data字段中的数据进行分析。\n 禁止参考外部记忆禁止猜测数据背后的原因。\n 输出简洁的趋势判断JSON。 ), api_keyapi_key, )可以看到每个Agent的System Prompt都清楚地划定了职责边界。这种Prompt不是一种“装饰”而是数据隔离在文本层面的最后一道防线。5.3 实现协作用任务队列和结果路由接下来是调度器它负责创建任务、下发数据、接收结果。我用一个简单的队列来管理。# orchestrator.py import json from creator import create_research_agent, create_analyst_agent class Orchestrator: def __init__(self, api_key): self.api_key api_key self.agents { research: create_research_agent(api_key), analyst: create_analyst_agent(api_key), } self.result_store {} def run_report_pipeline(self, raw_goal): # Step 1: 调研任务 research_result self.agents[research].call_model( json.dumps({task: 收集资料, goal: raw_goal, output: structured_lists}, ensure_asciiFalse) ) self.result_store[research] json.loads(research_result) # Step 2: 只把调研结果的结构化摘要传给分析Agent analyst_input { input_data: self.result_store[research], task: 生成趋势判断, } analysis_result self.agents[analyst].call_model( json.dumps(analyst_input, ensure_asciiFalse) ) self.result_store[analysis] json.loads(analysis_result) return self.result_store这一步就是协作机制的落地。主管Orchestrator控制流程每个Agent只接收前一步的结构化输出没有直接共享底层原始数据。后面如果需要加“撰写Agent”或者“校对Agent”就按同样的模式往下串。5.4 运行效果与调优过程第一版跑起来之后我对效果做了对比。同样完成一份新能源汽车调研报告单Agent版上下文很快就超过了8000 Token输出的报告里经常混入原始网页中的噪声。Multi-Agent版里检索Agent的输出被压缩成JSON要点后分析Agent和撰写Agent的上下文输入往往只有1000到2000 Token左右效果反而更稳定。过程中真的调整了不少地方。最大的一个调优是给检索Agent的输出格式加了一个max_items限制防止它一次抽取50条事实结果分析Agent挑花了眼。后来我又在分析Agent的输出里加了一个confidence字段低于0.75的数据不写进最终报告这就把质量控制提前到了协作环节。各地实测稳定运行后整条流水线的Token消耗比原来的单Agent做法降低了约40%——你可能会说这是数据集变小了的自然结果但实际上是因为我们不再把中间冗余数据全都传给每一步的Agent了。6. 常见问题与排查技巧实录6.1 上下文漂移Agent开始把别人的任务当成自己的任务症状撰写Agent在报告里写“根据我的搜索日志显示”但它的角色根本不是检索员或者分析Agent忽然引用了检索阶段的原始URL地址。排查路径先用日志把每次调用时的System Prompt和实际messages打印出来。如果messages里出现了不该有的内容比如检索Agent的历史记录被传给了分析Agent那就是History被全局复用了。我的解决方法是在BaseAgent里彻底禁用共享History改为每个Agent独立维护并且封装一个clear_history()方法在每轮任务开始时清空。6.2 死循环Agent之间互相重试和指派第二种高频坑是死循环。A让B干活B觉得缺信息就把问题原样返回给AA又把它当成新一轮任务下发于是两个Agent互相交换同一条消息直到超出时间限制。解决思路有两个一是消息类型里明确区分execute_task和request_info调度器看到request_info时直接转人工而不是转给另一个Agent二是设置最大递归深度比如同一request_id被重试3次后强制中止。我建议每个Agent输出底部都带一个needs_human字段只要某个Agent认为它收到了病态输入就把控制权交还给主管由主管决定是否人工介入。6.3 Token成本失控常见现象系统稳定运行一段时间后Token消耗莫名其妙飙升。我排查后发现很多Agent的History里存了大量已完成的旧任务数据每次调用都把这些旧数据重新发给模型。处理方式就是我前面说的每个Agent在任务开始时跑一次reset_context()只保留必要的基础信息。另外我把大份的中间数据比如清洗后的CSV放到独立的存储里Agent用数据ID去读取而不是直接把全部数据拼进Prompt。隔离的本质不只是“不共享”还包括“不常驻”。6.4 速查表问题、原因、解决方案我整理了一张速查表专门给团队成员排查用症状可能原因解决方案输出内容错乱、幻觉增多上下文被无关数据污染增加Gate过滤只传白名单字段多个Agent互相等待依赖关系定义不清绘制子任务依赖图统一由主管调度相同错误反复出现Agent的History里残留了错误示例每轮任务前清空History单个Agent耗时过长任务拆分粒度太粗进一步拆分或增加并行Agent输出风格不一致所有Agent共用了同一套System Prompt为不同Agent定制独立PromptToken成本激增中间数据被重复传递中间结果存入数据存储按需读取最后再分享一个小技巧。我在写这个系统的时候没有一开始就奔着“让Agent变得聪明”去优化而是把精力花在“让Agent别被干扰”上。因为我发现多数失败场景并不是模型能力不够而是上下文被污染、任务边界不清晰、协作链路不顺畅。你把拆任务、隔离上下文、协作机制这三件事做扎实了Multi-Agent系统的稳定性会远超你砸再多Prompt去调单个Agent的折腾。这套代码结构你拿去改一改完全可以直接用在其他领域比如自动写代码、客服工单分流、数据分析批量处理原理是一样的。