1. 多Agent协作到底在解决什么问题单Agent跑任务短链路还行一旦任务链条拉长到十几个步骤问题就全暴露出来了上下文窗口塞满、工具调用互相干扰、中间结果没人校验、一个环节出错后面全盘崩溃。我最早做文档分析流水线的时候就吃过这个亏一个Agent又负责检索又负责总结又负责格式输出结果它把检索到的原文和自己的推测混在一起输出了一堆看起来合理但完全经不起核对的内容。多Agent协作的核心思路其实很朴素把一个大任务拆成若干职责明确的子任务每个子任务交给独立的Agent处理Agent之间通过结构化的消息传递来协同。这跟公司里做项目的逻辑一模一样——你不会让一个人同时干产品、开发、测试、运维而是分角色、定接口、设检查点。这套架构能解决的问题包括上下文隔离每个Agent只关心自己那部分信息不会被无关内容污染判断职责单一一个Agent只做一类事提示词可以写得非常精准输出稳定性大幅提升并行加速没有依赖关系的子任务可以同时跑整体耗时从串行的N倍压缩到接近最慢那条链路可观测性每个环节的输入输出都是独立的出问题能快速定位是哪个Agent的锅适合谁来参考这篇内容如果你已经用过大模型API做过一些单Agent的小工具现在想把它升级成能处理复杂流程的系统或者你正在做需要多步骤推理、多源信息整合的应用那接下来的内容应该能帮你少走不少弯路。完全没接触过大模型开发的朋友也能看但建议先把单Agent的调用跑通再来看协作部分理解会更深。2. 协作架构的几种典型模式与选型逻辑2.1 顺序流水线模式最简单也最常用顺序流水线就是Agent A做完传给Agent BB做完传给C像工厂流水线一样。这种模式的好处是逻辑清晰、调试容易每个环节的输入输出都能单独验证。我做过一个合同审查的流程拆成了四个Agent提取Agent负责从PDF里抽出关键条款分类Agent判断条款类型付款、违约、保密等风险Agent对照规则库标记异常条款汇总Agent生成审查报告。四个Agent串行执行每个环节的输出都是下一个环节的输入。这种模式的关键在于接口定义要严格。提取Agent输出的JSON结构必须和分类Agent期望的输入完全对齐否则后面全乱。我的做法是在每个Agent的提示词里明确写出输入格式和输出格式并且用JSON Schema做校验格式不对直接打回重试。注意顺序流水线最大的坑是错误累积。如果第一个Agent提取错了后面三个Agent全在错误的基础上工作。所以我在关键节点加了校验Agent专门检查上游输出是否合理不合理就触发重试或人工介入。2.2 层级调度模式主管Agent加执行Agent这种模式里有一个调度Agent也叫Orchestrator或Supervisor它不干具体活只负责拆解任务、分配工作、收集结果、决定下一步。下面挂着一组执行Agent每个执行Agent负责一类具体能力。调度Agent的核心能力是任务分解和路由。比如用户说“帮我分析这份财报并给出投资建议”调度Agent会把它拆成数据提取→财务指标计算→行业对比→风险评估→建议生成然后依次调用对应的执行Agent。这种模式的优势在于灵活。调度Agent可以根据中间结果动态调整后续步骤比如发现某公司现金流异常就额外调用一个“现金流专项分析Agent”。但缺点是调度Agent本身成了瓶颈它的提示词设计直接决定整个系统的上限。我实测下来调度Agent的提示词里必须包含这几样东西可用Agent的清单及各自能力描述、任务分解的规则、结果合并的策略、异常情况的处理方式。少一样都会导致调度混乱。2.3 辩论与投票模式用多个视角提升质量这种模式让多个Agent对同一个问题独立给出答案然后通过投票或辩论的方式收敛到最终结果。常见于需要高可靠性的场景比如事实核查、代码审查、方案评估。具体做法是三个Agent用不同的提示词策略一个偏保守、一个偏激进、一个偏中立分别处理同一个任务然后一个仲裁Agent对比三份输出标记分歧点针对分歧点让原Agent补充论据最后综合出最终答案。这种模式成本高——同样的任务要跑三遍以上——但在关键决策场景下很值。我做过一个技术方案评审的系统三个Agent分别从性能、成本、可维护性三个角度评审同一个架构方案仲裁Agent汇总后发现性能Agent和成本Agent在缓存策略上有冲突补充论据后找到了一个折中方案比单Agent直接给的建议靠谱得多。2.4 三种模式的对比与选型建议模式适用场景优势劣势成本顺序流水线步骤固定、依赖明确的流程逻辑清晰、易调试错误累积、不够灵活低层级调度任务复杂、需要动态决策灵活、可扩展调度Agent是瓶颈中辩论投票高可靠性要求的决策质量高、减少偏见成本高、耗时长高选型的时候先问自己三个问题任务步骤是不是固定的需不需要根据中间结果动态调整对输出质量的容忍度有多低第一个问题答案是“是”、第二个是“否”、第三个是“一般”那就用顺序流水线。如果任务复杂且需要动态决策上层用层级调度下层的关键环节嵌入辩论投票。3. 任务调度的核心机制与实现细节3.1 任务分解的粒度控制任务分解太粗Agent干不了分解太细调度开销比干活还大。我的经验是每个子任务的预期输出控制在500到1500字之间或者对应一个明确的、可用一两句话描述清楚的交付物。举个例子“分析这份用户反馈数据”这个任务太粗Agent不知道从哪下手。拆成“统计反馈中的问题类型分布”“提取出现频率最高的五个具体问题”“对每个问题给出严重程度评级”“汇总成结构化报告”就合适了。每个子任务都有明确的输出物而且可以用一两句话验收。分解的时候还要注意依赖关系。有些子任务必须等前面的完成才能开始比如必须先提取数据才能分析有些则可以并行比如同时分析不同维度的数据。调度器需要维护一个依赖图只有所有前置任务都完成的任务才能进入就绪队列。3.2 状态管理与上下文传递多Agent系统里每个Agent都是无状态的——它不记得上次干了什么。所以需要一个共享状态存储来记录整个任务的进展。我一般用Redis或者简单的JSON文件来存结构大概是这样{ task_id: xxx, status: running, subtasks: { extract: {status: done, output: ..., timestamp: ...}, analyze: {status: running, input_ref: extract.output}, report: {status: pending, depends_on: [analyze]} }, global_context: {user_query: ..., constraints: ...} }上下文传递有个关键决策传全量还是传摘要。传全量信息完整但会撑爆下游Agent的上下文窗口传摘要节省空间但可能丢失关键细节。我的做法是分层传递——全局上下文用户原始需求、约束条件始终携带上游输出只传与当前子任务相关的部分由调度器负责裁剪。3.3 超时、重试与降级策略Agent调用大模型API超时和失败是常态。必须给每个子任务设置超时时间和最大重试次数。我的配置一般是单次调用超时60秒最多重试2次重试间隔指数退避1秒、2秒、4秒。重试还失败怎么办这时候需要降级策略。比如一个负责生成详细分析的Agent挂了可以降级到一个简化版的Agent输出一个基础版本的结果保证流程能继续走完而不是整个任务卡死。实操心得重试的时候最好把上次失败的原因也带上让Agent知道上次哪里出了问题。我试过在重试提示词里加一句“上次输出因为格式不符合JSON规范被拒绝请确保输出是合法的JSON”重试成功率明显提升。3.4 并行执行与结果合并没有依赖关系的子任务并行跑能大幅缩短总耗时。实现方式很简单调度器扫描所有子任务把依赖已满足的任务同时扔进线程池或异步任务队列。但并行会带来结果合并的问题。比如三个Agent分别从不同数据源提取了信息合并的时候发现同一实体在不同来源里的属性值不一致。这时候需要一个合并Agent来处理冲突规则可以是“取多数一致的值”“取置信度最高的值”或者“标记冲突让人工确认”。我一般会在合并Agent的提示词里明确写出冲突处理规则并且要求它输出合并日志记录每个冲突是怎么解决的方便后续审计。4. 从零搭建一个多Agent协作系统的完整过程4.1 环境准备与基础框架选型先明确技术栈。大模型调用层我推荐用统一的API封装把不同厂商的模型调用统一成同一个接口方便切换和对比。Python生态里LangChain和LlamaIndex都提供了多Agent的抽象但我更倾向于自己写调度逻辑因为框架的抽象有时候会限制灵活性而且出问题了不好排查。基础依赖就这几个pip install openai httpx pydantic redisopenai用于调用兼容接口的大模型httpx做异步HTTP请求pydantic做数据校验redis做状态存储。如果你不想用Redis用SQLite或者本地JSON文件也行但并发高的时候Redis更稳。4.2 定义Agent的基类与通信协议每个Agent本质上就是一个函数接收输入调用大模型返回输出。我定义一个基类from pydantic import BaseModel from typing import Any class AgentInput(BaseModel): task: str context: dict upstream_output: Any None class AgentOutput(BaseModel): status: str # success | failed | need_retry result: Any metadata: dict {} class BaseAgent: def __init__(self, name: str, system_prompt: str, model: str gpt-4o): self.name name self.system_prompt system_prompt self.model model async def run(self, input: AgentInput) - AgentOutput: # 调用大模型解析输出返回AgentOutput ...通信协议用JSON每个Agent的输出必须能被AgentOutput解析。这样调度器不用关心每个Agent内部怎么实现的只按统一格式收发消息。4.3 调度器的核心循环实现调度器的主循环逻辑async def orchestrate(task_graph, state_store): while not task_graph.all_done(): ready_tasks task_graph.get_ready_tasks() if not ready_tasks: await asyncio.sleep(1) continue # 并行执行所有就绪任务 results await asyncio.gather(*[ execute_task(t, state_store) for t in ready_tasks ]) for task, result in zip(ready_tasks, results): task_graph.mark_done(task, result) state_store.save(task.id, result) return task_graph.get_final_result()get_ready_tasks返回所有依赖已满足且尚未执行的任务。execute_task负责调用对应的Agent、处理超时重试、更新状态。这个循环看起来简单但有几个细节要注意任务图要支持动态修改调度Agent可以在运行时添加新任务状态存储要支持并发读写多个任务同时更新状态异常要捕获并记录不能让一个任务的异常炸掉整个循环。4.4 一个完整案例技术调研报告生成系统我拿一个实际做过的项目来演示。需求是给一个技术主题自动生成一份包含现状、竞品、趋势、建议的调研报告。Agent设计搜索Agent根据主题生成搜索关键词调用搜索接口获取相关资料提取Agent从搜索结果中提取关键信息点去重、分类分析Agent对提取的信息做SWOT分析识别关键趋势撰写Agent按照报告模板组织内容生成初稿审校Agent检查事实一致性、逻辑连贯性、格式规范性任务图搜索 → 提取 → 分析 → 撰写 → 审校 ↘ ↗ → 补充搜索 →分析Agent如果发现信息不足会触发补充搜索补充搜索的结果回到提取环节重新处理。关键参数搜索Agent返回前10条结果每条截取前2000字提取Agent最多提取20个信息点每个信息点不超过200字分析Agent输出结构化JSON包含优势、劣势、机会、威胁四个数组撰写Agent目标字数3000到5000字必须引用提取的信息点审校Agent逐段检查标记有问题的段落并给出修改建议实测效果整个流程跑下来大约3到5分钟生成一份4000字左右的报告。审校Agent平均会标记5到8处需要修改的地方主要是数据引用不准确和逻辑跳跃。经过一轮修改后报告质量能达到人工撰写的70%到80%水平但速度快了十倍以上。4.5 提示词工程在多Agent场景下的特殊考量单Agent的提示词可以写得比较自由但多Agent场景下提示词必须高度结构化。因为每个Agent的输出要被下一个Agent解析格式不对整个流程就断了。我的做法是每个Agent的提示词都包含这几块角色定义你是谁你负责什么输入说明你会收到什么格式的数据任务描述具体要做什么分几步输出格式必须严格按照什么格式输出给一个示例约束条件不能做什么遇到什么情况要报错实操心得输出格式的示例一定要给全不能只给字段名。我试过只写“输出JSON格式”结果Agent输出的JSON字段名五花八门下游解析全挂。后来我把完整的示例JSON贴在提示词里格式错误率从30%降到了5%以下。5. 常见问题与排查技巧实录5.1 Agent输出格式不稳定怎么办这是最高频的问题。表现是Agent有时候输出纯JSON有时候在JSON外面包一层解释文字有时候字段名大小写不一致。排查思路检查提示词里的输出格式说明是否足够明确有没有给完整示例检查温度参数是否太高建议设成0.1到0.3检查输入里有没有干扰信息比如上游输出里带了无关的文本解决方法在提示词里加一句“只输出JSON不要输出任何其他文字”用response_format{type: json_object}强制JSON输出如果模型支持在代码里加一层容错解析先尝试直接解析失败则用正则提取JSON部分再解析5.2 任务卡死或无限循环怎么处理调度器有时候会陷入死循环比如一个任务一直重试一直失败或者任务图里出现了循环依赖。排查思路检查任务图的依赖关系是否有环检查重试次数是否设了上限检查是否有任务永远处于“running”状态解决方法给每个任务设最大重试次数我一般设3次给整个流程设最大执行时间比如10分钟超时强制终止在任务图构建阶段做环检测有环直接报错5.3 多个Agent输出矛盾怎么合并辩论模式或者多源提取场景下不同Agent对同一事实给出不同说法。排查思路检查不同Agent的输入是否一致检查是否有Agent的提示词存在偏见检查矛盾是事实层面的还是表述层面的解决方法事实层面的矛盾引入仲裁Agent让它对比多个来源选择置信度最高的表述层面的矛盾统一用结构化格式输出合并时按字段取值无法判断的矛盾标记出来输出到最终结果里让人工确认5.4 常见问题速查表问题现象可能原因排查动作解决方案Agent输出格式错误提示词不明确、温度过高检查提示词和温度参数加格式示例、降低温度、加容错解析任务卡死循环依赖、重试无上限检查任务图和重试配置环检测、设重试上限、设全局超时输出矛盾输入不一致、提示词偏见对比不同Agent的输入仲裁Agent、结构化合并、人工确认整体耗时过长串行执行、模型响应慢检查任务依赖关系并行化无依赖任务、换更快的模型成本过高重复调用、辩论轮次过多统计各Agent调用次数缓存中间结果、限制辩论轮次5.5 几个我踩过的坑坑一上下文传递太多导致下游Agent“分心”。我一开始把上游所有输出都传给下游结果下游Agent经常被无关信息干扰输出质量下降。后来改成只传相关字段质量明显提升。坑二没有做输入校验。有一次上游Agent输出了一个空数组下游Agent没做校验直接处理生成了一个完全空的报告。后来我在每个Agent的入口都加了输入校验空值、异常值直接拦截。坑三重试没有退避。一开始重试是立即执行的结果遇到API限流的时候重试请求全被拒绝。后来改成指数退避成功率大幅提升。坑四没有记录中间状态。有一次流程跑到一半挂了完全不知道跑到哪了只能从头再来。后来加了状态存储每次任务完成都落盘挂了可以从断点恢复。6. 性能优化与成本控制的实战经验6.1 模型选型的差异化策略不是所有Agent都需要用最贵的模型。我的策略是按任务复杂度分级简单任务格式转换、信息提取用便宜的小模型比如7B级别的本地模型或者低价API中等任务分析、总结用中等模型比如GPT-4o-mini级别复杂任务调度决策、仲裁、最终撰写用最强模型这样整体成本能降一半以上而质量损失很小。我实测过一个流程全用最强模型成本是$2.5分级后降到$0.8输出质量人工评估只差了5%左右。6.2 缓存与去重很多Agent的调用是可以缓存的。比如搜索Agent对同一个关键词的搜索结果在短时间内不会变可以缓存。提取Agent对同一段文本的提取结果也是确定的可以缓存。我用Redis做缓存key是“Agent名称输入内容的哈希”value是输出结果过期时间设1小时。这样重复的任务直接命中缓存省时省钱。6.3 并行度的控制并行不是越多越好。并行度太高会导致API限流、内存暴涨、调度开销增大。我的经验是并行度控制在4到8之间具体看API的限流策略和机器的内存。如果任务数量超过并行度就排队等待。调度器维护一个就绪队列和一个执行中的任务集合执行中的任务数达到上限就不再启动新任务。6.4 监控与日志多Agent系统没有监控就是盲人摸象。我至少会记录这些信息每个Agent的调用次数、成功率、平均耗时每个任务的开始时间、结束时间、状态每次调用的输入输出摘要脱敏后整体流程的耗时和成本这些数据用简单的日志文件或者SQLite存就行不需要上重型监控系统。关键是出问题的时候能快速定位。7. 扩展方向与个人体会这套架构跑通之后扩展方向其实很多。比如引入人工审核节点在关键决策点暂停流程等人确认后再继续。或者做多轮迭代让审校Agent的反馈自动触发撰写Agent的修改循环几轮直到质量达标。还可以接入外部工具让Agent能调用数据库、执行代码、操作文件系统能力边界会大很多。我个人在实际操作中的体会是多Agent系统的难点不在单个Agent有多强而在协作机制有多稳。一个中等能力的Agent加上可靠的调度、校验、重试机制比一个超强Agent单打独斗要靠谱得多。另外就是不要过度设计一开始用最简单的顺序流水线把流程跑通遇到瓶颈再引入更复杂的模式比一上来就搞层级调度加辩论投票要务实得多。最后分享一个小技巧给每个Agent起一个有意义的名字并且在日志里始终用这个名字。调试的时候看到“风险分析Agent输出异常”比看到“Agent_3输出异常”要直观得多能省不少排查时间。