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的对比你可以直观看到差异维度单AgentMulti-Agent上下文管理所有信息共享一个窗口每个Agent独立上下文错误隔离一处出错全局受影响错误局限在单个Agent并行能力串行执行可并行执行独立子任务调试难度难以定位问题步骤可单独调试每个AgentToken效率长上下文导致浪费按需分配更经济适用场景简单、线性任务复杂、多步骤、多角色任务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→DD拿到的信息可能已经失真了。我的做法是关键信息直接传递不经过中间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_factorylist) 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[fdep_{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, indent2, ensure_asciiFalse))这个示例里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 问题速查表问题现象可能原因排查方法解决方案Agent输出格式不对Prompt不够明确检查Prompt里的输出格式定义用JSON Schema约束输出Agent之间信息不一致共享知识没传到位对比各Agent的输入把关键约束作为共享知识整体流程卡住依赖关系有环检查DAG是否有循环依赖重新设计依赖关系Token消耗过高上下文太长统计每个Agent的Token数精简输入、拆分任务某个Agent频繁失败输入数据有问题检查该Agent的输入增加输入校验、重试并行Agent结果冲突命名/类型不统一对比并行Agent的输出统一命名规范、类型系统整体延迟高串行执行太多分析DAG的关键路径识别可并行部分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这种。这样看日志、排查问题时一眼就能知道是哪个环节出了问题。这个习惯帮我省了很多时间。