Multi-Agent系统实战：任务拆解与上下文隔离的工程实践

1. 为什么单Agent迟早会撞上天花板

1.1 从一个真实翻车现场说起

去年我接手了一个内部工具的需求：自动读取一份几十页的产品需求文档，拆出功能点，生成对应的接口定义、测试用例和前端组件骨架。一开始我的思路很直接——写一个超级Prompt，把文档全文塞进去，让模型一次性输出所有产物。

结果呢？第一次跑，模型把需求文档里第三章的内容和第七章的内容混在一起，生成的接口定义引用了根本不存在的字段。第二次跑，我加了更详细的指令，它倒是把功能点拆对了，但生成测试用例时完全忘了前面定义的接口签名，自己编了一套。第三次我学乖了，分步骤调用，但每一步都把完整上下文传进去，Token消耗直接爆炸，而且模型在长上下文里开始“注意力涣散”，前面明确说过的约束到后面就丢了。

这个经历让我彻底明白一件事：单Agent架构在处理复杂任务时，瓶颈不在模型能力，而在上下文管理。一个Agent既要理解全局、又要拆解任务、还要保证每一步的输出一致性，这就像让一个人同时当项目经理、架构师和程序员，还得记住所有会议内容——不现实。

1.2 单Agent的三个死穴

我把踩过的坑归纳成三个核心问题：

上下文污染。当所有信息都塞在一个上下文窗口里，不同任务的信息会互相干扰。比如你在处理用户认证模块时，上下文里还残留着前面数据库设计的讨论，模型很容易把两件事搅在一起。这不是模型笨，是注意力机制本身的特性决定的——上下文越长，每个Token分到的注意力越少。

错误累积。单Agent模式下，如果第一步任务拆解就偏了，后面所有步骤都会沿着错误方向走。而且因为没有隔离机制，你很难定位到底是哪一步出了问题。我遇到过最离谱的情况是：模型在生成API文档时，把一个字段类型从string改成了integer，原因是它在更早的上下文里看到了一个不相关的数字示例。

无法并行。复杂任务里有很多子任务其实是独立的，比如前端组件生成和后端接口定义可以同时进行。但单Agent只能串行处理，效率上不去。更关键的是，串行处理意味着每一步都要等上一步完全结束，整体延迟是累加的。

1.3 Multi-Agent的核心思路：分而治之

Multi-Agent架构的本质，就是把一个复杂任务拆成多个相对独立的子任务，每个子任务交给专门的Agent处理，每个Agent有自己的上下文窗口，互不干扰。这就像从“一个人全包”变成“一个团队协作”——有人负责拆需求，有人负责写接口，有人负责写测试，各干各的，最后汇总。

但这里有个关键问题：拆任务不是随便拆的。拆得不好，Agent之间信息传递成本比单Agent还高。我见过有人把任务拆成十几个Agent，结果光是协调它们之间的通信就耗掉了80%的Token预算。所以拆任务的核心原则是：高内聚、低耦合。每个Agent的职责要清晰，输出要标准化，依赖关系要明确。

下面这张表是我总结的单Agent和Multi-Agent的对比，你可以直观看到差异：

2. 拆任务：怎么拆才不翻车

2.1 拆任务的三个层次

拆任务不是简单地把大任务切成小块，而是要按依赖关系和信息流来拆。我通常分三个层次来考虑：

第一层：按阶段拆。比如一个完整的“需求文档转代码”任务，可以拆成：需求解析 → 功能点提取 → 接口设计 → 代码生成 → 测试用例生成。每个阶段是一个独立的Agent，输出是下一个阶段的输入。这种拆法适合流程清晰、步骤固定的任务。

第二层：按角色拆。同一个阶段里，不同角色关注的点不一样。比如“接口设计”阶段，可以拆成：数据模型Agent（负责定义数据结构）、接口签名Agent（负责定义函数签名）、错误处理Agent（负责定义异常和错误码）。这三个Agent可以并行工作，最后合并。

第三层：按数据拆。当处理的数据量很大时，可以按数据分片拆。比如一份几百页的文档，可以按章节拆成多个Agent并行解析，最后汇总。这种拆法适合数据密集型任务。

我实际项目里最常用的是混合拆法：先按阶段拆，阶段内再按角色拆，数据量大时再按数据拆。但要注意，拆得越细，协调成本越高。我的经验是：单个Agent的输出Token控制在2000以内，Agent总数控制在7个以内。超过这个数，协调开销会吃掉大部分收益。

2.2 任务拆解的实操模板

下面是我常用的任务拆解模板，你可以直接套用：

task: 需求文档转代码 stages: - name: 需求解析 agent: requirement_parser input: 原始需求文档 output: 结构化需求列表 dependencies: [] - name: 功能点提取 agent: feature_extractor input: 结构化需求列表 output: 功能点清单（含优先级） dependencies: [requirement_parser] - name: 接口设计 agents: - name: data_model_designer input: 功能点清单 output: 数据模型定义 - name: api_signer input: 功能点清单 output: 接口签名列表 - name: error_handler input: 功能点清单 output: 错误码定义 dependencies: [feature_extractor] merge: 合并三个Agent的输出为完整接口定义 - name: 代码生成 agent: code_generator input: 完整接口定义 output: 代码文件 dependencies: [api_signer, data_model_designer, error_handler] - name: 测试生成 agent: test_generator input: 完整接口定义 + 代码文件 output: 测试用例 dependencies: [code_generator]

这个模板的关键在于：每个Agent的输入输出都明确定义，依赖关系清晰，可以并行的部分明确标注。实际运行时，你可以根据这个模板动态调度Agent。

2.3 拆任务的常见坑

坑一：拆得太细。我见过有人把“生成一个函数”都拆成一个Agent，结果Agent数量爆炸，协调成本远超收益。记住：拆分的粒度应该以“一个Agent能独立完成且不需要频繁与其他Agent通信”为标准。

坑二：依赖关系没理清。如果Agent A依赖Agent B的输出，但B的输出格式没定义好，A就得做大量解析工作。我的做法是：所有Agent的输出都用JSON Schema定义，这样下游Agent可以直接解析，不需要额外理解。

坑三：忽略合并成本。并行Agent的输出合并往往比想象中复杂。比如三个Agent分别生成了数据模型、接口签名和错误码，合并时可能出现命名冲突、类型不匹配等问题。我的经验是：在拆任务时就定义好合并规则，比如统一命名规范、统一类型系统。

提示：拆任务时先画一张依赖图，把每个Agent的输入输出标清楚。如果发现某个Agent的输入来自超过3个上游Agent，说明拆得太细了，考虑合并。

3. 隔离上下文：让每个Agent只关注自己的事

3.1 上下文隔离的三种策略

上下文隔离是Multi-Agent的核心优势，但怎么隔离有讲究。我实践下来有三种策略：

完全隔离。每个Agent有独立的上下文窗口，只接收自己需要的输入，不共享任何历史信息。这种策略适合子任务完全独立的场景，比如并行处理多个文档章节。优点是Token效率最高，缺点是Agent之间无法共享隐性知识。

部分隔离。每个Agent有独立的工作上下文，但可以访问一个共享的“全局知识库”（比如项目规范、命名约定）。这种策略适合需要保持一致性的场景，比如多个Agent生成代码时都要遵循同一套编码规范。实现方式通常是把共享知识放在系统Prompt里，或者通过一个专门的“知识检索Agent”按需提供。

层级隔离。有一个“协调者Agent”负责管理全局上下文，每个“工作者Agent”只接收协调者分配的任务和必要上下文。协调者负责汇总结果、处理冲突。这种策略适合任务复杂、需要动态调度的场景。缺点是协调者可能成为瓶颈。

我实际项目里最常用的是部分隔离：每个Agent独立上下文，但共享一个精简的“项目规范”文档。这个文档通常不超过500字，包含命名规范、类型定义、错误码规范等。

3.2 上下文传递的实操方法

上下文隔离不等于不传递信息。关键是传递什么、怎么传递。我的做法是：

只传必要信息。下游Agent不需要知道上游Agent的完整思考过程，只需要知道最终输出。比如代码生成Agent不需要知道需求解析Agent是怎么理解需求的，只需要拿到结构化的功能点清单。

用结构化格式传递。所有Agent之间的通信都用JSON格式，字段名和类型提前定义好。这样下游Agent可以直接解析，不需要额外理解自然语言。

控制传递链长度。如果A→B→C→D，D拿到的信息可能已经失真了。我的做法是：关键信息直接传递，不经过中间Agent。比如原始需求文档的关键约束，直接传给所有相关Agent，而不是让每个Agent从上游输出里推断。

下面是一个上下文传递的示例：

{ "task_id": "req2code_001", "stage": "api_design", "agent": "data_model_designer", "input": { "features": [ {"id": "F001", "name": "用户登录", "priority": "P0"}, {"id": "F002", "name": "用户注册", "priority": "P0"} ], "constraints": { "naming": "snake_case", "id_type": "uuid", "timestamp_format": "iso8601" } }, "output_schema": { "models": [ { "name": "string", "fields": [ {"name": "string", "type": "string", "required": "boolean"} ] } ] } }

这个结构里，constraints就是共享知识，直接传给每个Agent，不经过中间环节。output_schema定义了输出格式，下游Agent可以直接按这个格式解析。

3.3 上下文隔离的注意事项

注意一：共享知识要精简。我见过有人把整个项目文档作为共享知识传给每个Agent，结果每个Agent的上下文都被占满了。共享知识应该只包含跨Agent的约束和规范，不包含具体任务信息。

注意二：避免信息孤岛。完全隔离可能导致Agent之间信息不一致。比如两个Agent分别定义了同一个数据结构，但字段名不一样。解决办法是：关键数据结构由专门的Agent定义，其他Agent引用。

注意三：上下文窗口要留余量。每个Agent的上下文窗口不要塞满，留20%左右的余量给模型推理。我实测下来，上下文利用率超过80%时，模型输出质量会明显下降。

提示：可以用一个简单的规则判断上下文是否过载——如果Agent的输出开始出现“忘记前面约束”的情况，说明上下文太满了，需要精简输入或拆分任务。

4. 协作机制：Agent之间怎么配合

4.1 三种协作模式

Multi-Agent的协作模式决定了整体效率。我实践下来有三种模式：

流水线模式。Agent按顺序执行，前一个的输出是后一个的输入。这种模式最简单，适合线性任务。缺点是串行执行，延迟高。优化方法是：识别可并行的阶段，把流水线变成“流水线+并行分支”。

黑板模式。所有Agent共享一个“黑板”（共享存储），每个Agent从黑板上读取自己需要的信息，处理完后写回黑板。这种模式适合任务边界模糊、需要频繁交互的场景。缺点是黑板可能成为竞争资源，需要加锁机制。

市场模式。有一个“调度Agent”根据任务需求动态分配Agent。比如有10个任务和5个Agent，调度Agent根据每个Agent的负载和能力分配任务。这种模式适合任务动态到达的场景，但实现复杂度最高。

我实际项目里最常用的是流水线+并行分支。比如前面说的需求文档转代码，整体是流水线，但接口设计阶段有三个Agent并行工作。实现方式是用一个简单的DAG调度器，按依赖关系触发Agent。

4.2 协作中的通信协议

Agent之间怎么通信，直接决定了协作效率。我的做法是定义一套简单的通信协议：

消息格式。所有Agent之间的消息都用JSON，包含sender、receiver、type、payload四个字段。type可以是request、response、event。

同步与异步。需要立即返回结果的用同步调用，不需要的用异步消息。比如代码生成Agent需要接口定义，这是同步调用；而日志记录Agent接收事件通知，这是异步消息。

错误处理。每个Agent的输出都包含status字段，success表示成功，error表示失败并附带错误信息。下游Agent根据status决定是否继续。

下面是一个通信示例：

{ "sender": "feature_extractor", "receiver": "api_designer", "type": "response", "payload": { "status": "success", "features": [...], "metadata": { "confidence": 0.95, "warnings": [] } } }

4.3 协作中的冲突处理

多个Agent并行工作时，冲突不可避免。常见的冲突有：

命名冲突。两个Agent定义了同名的数据结构或函数。解决办法是：在共享知识里定义命名规范，比如所有数据模型以Model结尾，所有接口以API结尾。

类型冲突。一个Agent把字段定义为string，另一个定义为integer。解决办法是：关键类型由专门的Agent定义，其他Agent引用。

逻辑冲突。两个Agent对同一业务逻辑的理解不一致。解决办法是：在任务拆解阶段就明确业务规则，把规则作为共享知识传给所有Agent。

我处理冲突的流程是：先检测（对比各Agent输出），再分类（命名/类型/逻辑），最后按预设规则解决。如果规则解决不了，就升级到人工介入。

4.4 协作效率的优化技巧

技巧一：批量通信。不要每个小结果都立即传递，攒一批再传。比如代码生成Agent生成了10个文件，一次性传给测试生成Agent，而不是一个一个传。

技巧二：缓存中间结果。如果某个Agent的输出会被多个下游Agent使用，缓存起来避免重复计算。比如功能点清单被接口设计、代码生成、测试生成三个Agent使用，缓存后只需要计算一次。

技巧三：超时与重试。每个Agent调用设置超时时间，超时后重试或降级。我通常设置30秒超时，重试2次。如果还失败，就记录错误并跳过，避免阻塞整个流程。

技巧四：监控与日志。每个Agent的输入输出都记录日志，方便排查问题。我通常记录task_id、agent_name、input_tokens、output_tokens、duration、status六个字段。

提示：协作机制的设计原则是“简单优先”。不要一开始就上复杂的市场模式，先用流水线跑通，再根据瓶颈逐步优化。

5. 实战：从零搭建一个Multi-Agent系统

5.1 技术选型与架构设计

搭建Multi-Agent系统，技术选型很关键。我的选型原则是：用最少的组件跑通核心流程。

Agent框架。我试过几个主流框架，最后选择自己写一个轻量级的调度器。原因是大框架往往封装太厚，调试困难，而且很多功能用不上。自己写的话，核心代码不超过500行，完全可控。

通信机制。用内存队列（比如Python的queue.Queue）做Agent之间的消息传递。简单、快、不需要额外依赖。如果Agent分布在多台机器上，再换成消息队列（如RabbitMQ）。

存储。中间结果用JSON文件存储，简单直接。如果需要频繁读写，用SQLite。不需要上数据库，除非数据量真的很大。

调度器。用一个简单的DAG调度器，按依赖关系触发Agent。核心逻辑是：维护一个待执行Agent列表，每次检查依赖是否满足，满足就执行。

整体架构如下：

[输入] → [调度器] → [Agent池] ↓ [共享存储] ↓ [输出汇总]

5.2 核心代码实现

下面是我实际项目里的核心代码，你可以直接参考：

import json import queue from dataclasses import dataclass, field from typing import Any, Callable @dataclass class AgentTask: task_id: str agent_name: str input_data: dict dependencies: list = field(default_factory=list) output: Any = None status: str = "pending" class MultiAgentOrchestrator: def __init__(self): self.tasks = {} self.agents = {} self.message_queue = queue.Queue() self.shared_store = {} def register_agent(self, name: str, handler: Callable): self.agents[name] = handler def add_task(self, task: AgentTask): self.tasks[task.task_id] = task def _dependencies_met(self, task: AgentTask) -> bool: for dep_id in task.dependencies: dep = self.tasks.get(dep_id) if not dep or dep.status != "success": return False return True def _build_input(self, task: AgentTask) -> dict: input_data = dict(task.input_data) for dep_id in task.dependencies: dep = self.tasks[dep_id] input_data[f"dep_{dep_id}"] = dep.output return input_data def run(self): while True: pending = [t for t in self.tasks.values() if t.status == "pending"] if not pending: break ready = [t for t in pending if self._dependencies_met(t)] if not ready: raise RuntimeError("Deadlock detected: no task can proceed") for task in ready: task.status = "running" try: handler = self.agents[task.agent_name] input_data = self._build_input(task) task.output = handler(input_data) task.status = "success" except Exception as e: task.status = "failed" task.output = {"error": str(e)} return {tid: t.output for tid, t in self.tasks.items()}

这段代码的核心是run方法：每次找出所有依赖已满足的任务，并行执行。实际使用时，你可以把handler换成调用LLM的函数。

5.3 一个完整的运行示例

下面是一个完整的示例，模拟“需求文档转代码”的流程：

def requirement_parser(input_data): # 实际项目中这里调用LLM return { "features": [ {"id": "F001", "name": "用户登录", "priority": "P0"}, {"id": "F002", "name": "用户注册", "priority": "P0"} ] } def feature_extractor(input_data): features = input_data["dep_task_1"]["features"] return {"features": features, "count": len(features)} def data_model_designer(input_data): features = input_data["dep_task_2"]["features"] models = [] for f in features: models.append({ "name": f"{f['name']}Model", "fields": [ {"name": "id", "type": "uuid", "required": True}, {"name": "created_at", "type": "iso8601", "required": True} ] }) return {"models": models} def api_signer(input_data): features = input_data["dep_task_2"]["features"] apis = [] for f in features: apis.append({ "name": f"{f['name']}API", "method": "POST", "path": f"/api/{f['id'].lower()}" }) return {"apis": apis} def code_generator(input_data): models = input_data["dep_task_3"]["models"] apis = input_data["dep_task_4"]["apis"] return { "code": f"// Generated {len(models)} models and {len(apis)} APIs", "models": models, "apis": apis } # 组装 orch = MultiAgentOrchestrator() orch.register_agent("requirement_parser", requirement_parser) orch.register_agent("feature_extractor", feature_extractor) orch.register_agent("data_model_designer", data_model_designer) orch.register_agent("api_signer", api_signer) orch.register_agent("code_generator", code_generator) orch.add_task(AgentTask("task_1", "requirement_parser", {"doc": "..."})) orch.add_task(AgentTask("task_2", "feature_extractor", {}, ["task_1"])) orch.add_task(AgentTask("task_3", "data_model_designer", {}, ["task_2"])) orch.add_task(AgentTask("task_4", "api_signer", {}, ["task_2"])) orch.add_task(AgentTask("task_5", "code_generator", {}, ["task_3", "task_4"])) result = orch.run() print(json.dumps(result, indent=2, ensure_ascii=False))

这个示例里，task_3和task_4是并行的，因为它们都只依赖task_2。task_5依赖task_3和task_4，所以会等它们都完成后再执行。

5.4 性能调优与监控

跑通之后，下一步是调优。我通常关注三个指标：

Token消耗。每个Agent的输入输出Token数。优化方向是精简输入、控制输出长度。我实测下来，把共享知识从2000字压缩到500字，整体Token消耗降低30%。

执行时间。每个Agent的耗时。优化方向是并行化、缓存中间结果。我遇到过某个Agent耗时特别长，排查发现是输入数据太大，精简后耗时降低60%。

成功率。每个Agent的成功率。优化方向是改进Prompt、增加重试。我通常设置成功率低于90%就告警，低于80%就暂停流程。

监控方面，我写了一个简单的日志装饰器：

import time import functools def monitor(agent_name): def decorator(func): @functools.wraps(func) def wrapper(input_data): start = time.time() try: result = func(input_data) status = "success" except Exception as e: result = {"error": str(e)} status = "failed" duration = time.time() - start print(f"[{agent_name}] status={status} duration={duration:.2f}s") return result return wrapper return decorator

把这个装饰器加到每个Agent上，就能看到每个Agent的执行情况。

6. 常见问题与排查技巧实录

6.1 问题速查表

6.2 独家避坑技巧

技巧一：先跑通再优化。不要一开始就追求完美的架构。先用最简单的流水线跑通，再根据瓶颈逐步优化。我见过太多人一开始就设计复杂的市场模式，结果连基本流程都跑不通。

技巧二：给每个Agent写单元测试。每个Agent的输入输出都是确定的，可以单独测试。我通常用几个典型输入测试每个Agent，确保输出符合预期。这样出问题时能快速定位是哪个Agent的问题。

技巧三：保留中间结果。每个Agent的输出都存下来，方便排查问题。我通常存成JSON文件，文件名包含task_id和agent_name。出问题时直接看文件，比看日志快。

技巧四：设置合理的超时。每个Agent调用设置超时，避免一个Agent卡住整个流程。我通常设置30秒超时，重试2次。如果还失败，记录错误并跳过。

技巧五：监控Token消耗。Token是成本大头，要实时监控。我通常设置一个预算上限，超过就告警。优化Token的方法包括：精简Prompt、控制输出长度、缓存中间结果。

技巧六：用真实数据测试。不要用构造的简单数据测试，要用真实数据。我遇到过用简单数据测试没问题，用真实数据就翻车的情况。真实数据往往有各种边界情况，能暴露更多问题。

6.3 一个真实的排查案例

有一次，我的Multi-Agent系统在代码生成阶段频繁失败。排查过程如下：

第一步，看日志。发现code_generator的输入里，models字段是空的。说明上游的data_model_designer没有输出。

第二步，看data_model_designer的日志。发现它的输入里，features字段是空的。说明上游的feature_extractor没有输出。

第三步，看feature_extractor的日志。发现它的输入里，dep_task_1是空的。说明requirement_parser没有输出。

第四步，看requirement_parser的日志。发现它报错了，错误信息是“输入文档为空”。

第五步，检查输入文档。发现文档路径写错了，实际文件不存在。

整个排查过程花了10分钟，因为每个Agent的输入输出都有日志。如果没有日志，可能要花几个小时。

这个案例的教训是：日志要记录每个Agent的完整输入输出，不要只记录状态。我现在的做法是，每个Agent的输入输出都存成JSON文件，文件名格式是{task_id}_{agent_name}_{timestamp}.json。

6.4 性能优化的几个实用技巧

技巧一：批量处理。如果多个任务可以合并，就合并处理。比如10个文档章节的解析，可以合并成一次LLM调用，而不是10次。

技巧二：缓存。如果某个Agent的输出会被多次使用，缓存起来。我用一个简单的字典做缓存，key是输入数据的哈希，value是输出。

技巧三：异步执行。如果某个Agent的调用不需要立即返回结果，用异步执行。比如日志记录、监控上报，都可以异步。

技巧四：降级策略。如果某个Agent失败，不要直接报错，而是降级处理。比如代码生成失败，可以返回一个模板代码，而不是让整个流程失败。

技巧五：预热。如果某个Agent的初始化耗时较长，提前预热。比如LLM连接池，提前建立连接，避免第一次调用时等待。

提示：性能优化要基于数据，不要凭感觉。先用监控工具找到瓶颈，再针对性优化。我见过有人优化了半天，结果优化的是不是瓶颈的部分，白费功夫。

7. 一些个人体会

Multi-Agent这套东西，我踩过的坑比走过的路还多。最开始我以为难点在技术实现，后来发现技术实现反而是最简单的——真正难的是任务拆解的粒度和上下文隔离的边界。

拆得太粗，Agent之间信息传递成本高，跟单Agent没区别。拆得太细，协调成本爆炸，整体效率反而下降。我现在的经验是：先按阶段拆，阶段内如果某个Agent的输出超过2000 Token，就考虑再拆。这个阈值不是绝对的，但对我大部分项目都适用。

上下文隔离也是，完全隔离会导致信息不一致，完全不隔离又失去Multi-Agent的意义。我现在的做法是：每个Agent独立上下文，但共享一个精简的“项目规范”文档。这个文档不超过500字，包含命名规范、类型定义、错误码规范等。实测下来，这个方案在一致性和Token效率之间取得了不错的平衡。

还有一个体会是：不要追求一步到位。我见过太多人一开始就设计复杂的Multi-Agent架构，结果连基本流程都跑不通。我的建议是：先用单Agent跑通，遇到瓶颈再拆成Multi-Agent。拆的时候先用最简单的流水线，跑通后再优化并行和协作。

最后分享一个小技巧：给每个Agent起一个有意义的名字。不要用agent1、agent2这种，用requirement_parser、code_generator这种。这样看日志、排查问题时，一眼就能知道是哪个环节出了问题。这个习惯帮我省了很多时间。