1. 为什么单Agent在复杂任务面前总是掉链子如果你已经用大模型做过一些实际项目大概率经历过这样的场景让一个Agent去完成调研某个技术方向、整理数据、写一份分析报告、再自查一遍逻辑漏洞这种复合任务结果它要么在中途忘了最初的目标要么把前面几步的结论丢得一干二净要么在自查环节敷衍了事。这不是模型不够聪明而是单Agent架构本身存在结构性瓶颈。单Agent的问题可以归结为三个层面。第一是上下文窗口的物理限制。一个Agent要同时承载任务规划、中间结果、工具调用记录、历史对话token消耗极快一旦超出窗口早期信息就被截断导致失忆。第二是角色冲突。同一个Agent既要当执行者又要当审核者它在审核自己产出时天然带有确认偏误很难真正挑出自己的毛病。第三是串行效率低下。一个Agent按顺序做五件事总耗时是五件事之和而如果这五件事里有三件互不依赖串行就是纯粹的浪费。多Agent协作架构要解决的核心问题就是把一个大而全的Agent拆成多个小而专的Agent通过明确的协作协议和任务调度机制让它们各司其职、并行推进、互相校验。这背后的思路其实和人类团队分工一模一样你不会让一个人同时当产品经理、程序员、测试和项目经理而是拆成不同角色用流程和沟通机制把它们串起来。这篇内容面向的是已经了解大模型基本调用、想进一步搭建多Agent系统的开发者。我会从协作架构的底层设计讲到任务调度的具体实现再到实际搭建中那些文档里不会写的坑。整套思路不绑定某个特定框架你可以用AgentScope、LangGraph、AutoGen或者自己手写调度器来实现重点是理解架构决策背后的逻辑。2. 多Agent协作的四种主流架构与选型逻辑2.1 从谁说了算这个维度切分架构多Agent系统的架构设计本质上要回答一个问题决策权如何分配。围绕这个问题目前实践中跑得通的主要有四类架构它们不是互斥的很多生产系统是混合使用。第一类是主从架构Orchestrator-Worker。有一个中心调度Agent负责拆解任务、分配子任务、汇总结果其他Worker Agent只负责执行自己被分配的那一块。这是最容易理解和实现的结构适合任务可以被清晰拆解的场景比如写一篇论文可以拆成文献调研数据分析撰写校对四个子任务。它的缺点是中心Agent容易成为瓶颈且一旦拆解逻辑出错整个流程就跑偏。第二类是流水线架构Pipeline。Agent按固定顺序排列前一个的输出是后一个的输入像工厂流水线。适合步骤明确、顺序固定的任务比如数据清洗流程抽取Agent→清洗Agent→校验Agent→入库Agent。优点是可控性强、每步可单独调试缺点是缺乏灵活性中间某步出问题很难动态调整。第三类是辩论/评审架构Debate/Review。多个Agent对同一问题给出方案然后互相评审、迭代改进。典型的是生成Agent评审Agent的对抗组合或者多个生成Agent各自出方案再投票。这种架构特别适合需要高质量输出的场景比如代码生成、论文写作因为评审环节能有效捕捉生成环节的疏漏。第四类是黑板架构Blackboard。所有Agent共享一块黑板通常是共享内存或数据库谁有需要就往上面写谁需要信息就从上面读由一个控制Agent根据黑板状态决定激活哪个Agent。这种架构灵活性最高适合问题解决路径不固定的探索型任务但实现复杂度也最高状态管理很容易失控。2.2 选型时真正该看的三个指标很多人在选架构时只看哪个听起来高级这是本末倒置。我实际做项目时会先看三个指标指标含义影响架构选择任务可分解度任务能否被清晰拆成独立子任务高则选主从/流水线低则选黑板输出质量要求对结果准确性的容忍度高则必须加评审/辩论环节实时性要求对响应延迟的敏感度高则优先并行架构避免串行评审举个具体例子。如果你要做的是批量处理用户反馈并分类任务可分解度高、质量要求中等、实时性要求高那就用主从架构加并行Worker不需要评审环节。但如果你要做的是生成一份对外发布的技术白皮书质量要求极高、实时性无所谓那就值得上辩论架构让多个Agent反复打磨。2.3 一个容易被忽略的点Agent数量的边际递减新手常犯的错误是Agent越多越好觉得拆得越细越专业。实际上Agent数量存在明显的边际递减效应。每增加一个Agent就增加一份通信开销、一份状态同步成本、一份出错概率。我实测下来大多数任务3到5个Agent是甜点区超过7个之后协调成本会迅速吃掉分工带来的收益。判断是否需要增加Agent的标准很简单新增的这个Agent是否承担了一个现有Agent无法兼任的、且对结果有实质影响的独立职责。如果只是让它也参与一下那不如不加。比如评审Agent值得单独存在因为审核和执行是天然冲突的职责但格式化Agent就没必要格式化完全可以作为执行Agent的一个工具调用。3. 任务调度多Agent系统真正的技术难点3.1 调度器到底在调度什么很多人以为任务调度就是把任务分给Agent这只说对了一半。一个完整的调度器要处理四件事任务分解、依赖管理、资源分配、失败恢复。任务分解是把用户的高层目标拆成可执行的原子任务。这一步通常由规划Agent完成输出一个任务列表每个任务标注输入、输出、依赖关系。依赖管理是维护任务之间的先后顺序比如数据分析必须在数据抽取完成后才能开始。资源分配是决定哪个任务交给哪个Agent、什么时候执行。失败恢复是当某个任务失败或输出不合格时决定是重试、换Agent还是回退。这四件事里依赖管理和失败恢复是最容易出问题的。依赖管理做不好会出现任务B在任务A还没完成时就启动了的竞态失败恢复做不好一个环节卡住整个流程就死在那里。3.2 用DAG表达任务依赖实践中表达任务依赖最成熟的方案是有向无环图DAG。每个节点是一个任务每条边是一个依赖关系。DAG的好处是天然支持并行——没有依赖关系的节点可以同时执行同时又能保证有依赖的节点按顺序执行。一个典型的DAG任务定义长这样tasks { extract: { agent: data_agent, deps: [], input: raw_source }, analyze: { agent: analysis_agent, deps: [extract], input: extract.output }, draft: { agent: writing_agent, deps: [analyze], input: analyze.output }, review: { agent: review_agent, deps: [draft], input: draft.output }, revise: { agent: writing_agent, deps: [review], input: review.feedback } }调度器的工作就是遍历这个DAG找出所有依赖已满足的任务把它们放进执行队列。这里有个关键细节依赖满足的判断不能只看前置任务是否执行过还要看前置任务的输出是否有效。如果extract任务执行了但输出为空analyze任务不应该启动而应该触发extract的重试。3.3 并行执行与状态同步DAG里同一层的任务可以并行执行这是多Agent系统相比单Agent最大的效率优势。但并行带来一个新问题状态同步。多个Agent同时读写共享状态时很容易出现数据不一致。我的做法是给每个任务分配独立的输出槽位Agent只写自己的槽位读别人的槽位时通过调度器统一代理。这样避免了直接竞争调度器在任务完成时统一更新全局状态。具体实现上可以用一个中心化的状态存储比如Redis或者简单的内存字典每个任务完成后写入{task_id: output}下游任务启动时从状态存储里按需读取。注意并行任务的数量要受限于实际的API并发限制和成本预算。我见过有人一口气并行20个Agent调用结果触发限流一半任务失败反而比串行还慢。建议并行度控制在3到5之间根据你的API配额调整。3.4 失败恢复的三种策略失败恢复是调度器里最考验设计的地方。我总结下来有三种策略按复杂度递增重试策略是最简单的任务失败就重新执行最多重试N次。适合偶发性失败比如网络抖动导致的API超时。但要注意重试要加退避不能立即重试否则容易连续撞上同一个问题。降级策略是重试失败后换一个方案。比如主Agent调用失败切换到备用Agent或者复杂模型调用失败降级到轻量模型。这要求系统里对每个任务都准备好备选方案。回退策略是最复杂的任务失败后不是简单重试而是回退到上游重新规划。比如review任务发现draft质量太差不是让review重试而是触发draft重新生成。这需要调度器能识别失败根因在上游这种情况并支持DAG的动态调整。实际项目里我通常三层都用先重试重试不行降级降级还不行才回退。这样既保证了鲁棒性又不会因为一点小问题就大动干戈重新规划。4. 从零搭建一个多Agent协同系统的实操路径4.1 环境准备与框架选型动手之前先明确技术栈。多Agent系统的核心组件包括大模型调用层、Agent定义层、调度器、状态存储、工具层。你可以用现成框架也可以自己组装。现成框架里AgentScope 2.0对多Agent协作的支持比较完整内置了消息传递和分布式部署能力LangGraph擅长用图结构表达Agent流程适合DAG式调度AutoGen的对话式协作模式适合辩论/评审架构。如果你想要最大控制权直接用Python手写调度器加OpenAI/通义/本地模型的API调用也完全可行我很多项目就是这么干的因为框架的抽象层有时候反而碍事。环境上Python 3.10以上装好你选用的模型SDK准备一个状态存储开发阶段用内存字典就够生产环境上Redis。如果涉及本地模型部署Ollama或vLLM都是成熟选择前者适合快速验证后者适合高并发生产。4.2 Agent的角色定义与提示词设计多Agent系统里每个Agent的提示词设计比单Agent更讲究因为角色边界必须清晰。一个Agent的提示词要包含四部分角色身份、职责范围、输入输出规范、协作约束。角色身份要具体不要写你是一个助手而要写你是一个负责数据清洗的Agent专门处理结构化数据的缺失值填充和异常值检测。职责范围要明确边界写清楚你只负责X不负责Y避免Agent越界。输入输出规范要规定格式比如输入是JSON格式的原始数据输出是清洗后的JSON字段保持不变。协作约束要说明当你发现数据问题超出你的处理范围时输出ESCALATE: 原因不要自行处理。这里有个实操心得给每个Agent的输出加上结构化标记比如用JSON或者带标签的文本。这样调度器解析起来稳定不会因为模型输出格式飘忽而崩溃。我一般要求Agent输出严格的JSON如果解析失败就触发重试重试时在提示词里强调必须输出合法JSON。4.3 调度器的核心循环实现调度器的主循环逻辑其实不复杂核心就是找可执行任务→执行→更新状态→重复。下面是一个简化但可运行的实现import json from concurrent.futures import ThreadPoolExecutor class Scheduler: def __init__(self, tasks, agents, stateNone): self.tasks tasks self.agents agents self.state state or {} self.completed set() self.failed set() def get_ready_tasks(self): ready [] for tid, task in self.tasks.items(): if tid in self.completed or tid in self.failed: continue if all(d in self.completed for d in task[deps]): ready.append(tid) return ready def run_task(self, tid): task self.tasks[tid] agent self.agents[task[agent]] inputs {d: self.state.get(d) for d in task[deps]} try: result agent.execute(task[input], inputs) self.state[tid] result self.completed.add(tid) except Exception as e: self.failed.add(tid) self.state[tid] {error: str(e)} def run(self, max_parallel3): while len(self.completed) len(self.failed) len(self.tasks): ready self.get_ready_tasks() if not ready: break with ThreadPoolExecutor(max_workersmax_parallel) as pool: pool.map(self.run_task, ready) return self.state这段代码的关键点在get_ready_tasks它只返回那些所有依赖都已完成的任务。run方法用线程池实现并行max_parallel控制并发度。实际生产里你还需要加上重试逻辑、超时控制、日志记录但骨架就是这个。4.4 工具层与外部能力接入Agent要真正干活必须能调用外部工具。常见的工具包括搜索引擎、代码执行器、文件读写、数据库查询、API调用。工具层的设计原则是统一接口、明确schema。每个工具定义成一个函数附带清晰的参数说明和返回格式。Agent通过function calling或者ReAct模式来调用工具。这里有个坑不要让所有Agent都能访问所有工具。数据Agent不需要代码执行器写作Agent不需要数据库写权限。按角色分配工具权限既安全又能减少Agent的误用。工具调用的结果要统一处理。我的做法是工具层返回标准结构{success: bool, data: any, error: str}Agent根据success决定下一步。这样调度器和Agent都不用关心具体工具的内部实现。5. 那些文档里不会写的踩坑记录5.1 Agent之间踢皮球的循环依赖我遇到过一个典型问题写作Agent把稿子交给评审Agent评审Agent提了修改意见写作Agent改完又交回去评审Agent又提意见来回十几轮停不下来。这就是循环依赖导致的死循环。根因是评审Agent的提示词里没有通过标准它总能找到可以改进的地方。解决方案是给评审环节设置明确的收敛条件要么设定最大迭代次数比如3轮要么让评审Agent输出一个明确的通过/不通过判断通过就结束。我现在的做法是两者结合评审Agent必须输出{pass: true/false, score: 0-10, feedback: ...}score达到阈值或者迭代到上限就强制结束。5.2 上下文在传递中的信息衰减多Agent系统里信息在Agent之间传递时会衰减。A Agent的输出经过调度器传给B AgentB Agent可能只提取了它认为重要的部分再传给C Agent时信息又少一层。几轮下来最初的约束条件可能就丢了。对策是关键约束要显式传递不能依赖Agent自己记忆。比如最终报告不超过2000字这个约束不能只在第一个Agent的提示词里写而要在每个相关Agent的输入里都带上。我通常维护一个global_constraints字段每次调用Agent时都注入进去确保约束不丢失。5.3 成本失控并行不等于免费多Agent系统最容易被低估的是成本。单Agent一次任务可能几千token多Agent系统因为要传递上下文、多轮评审token消耗轻松翻5到10倍。如果并行度高短时间内大量调用账单会很难看。控制成本的手段有几个用轻量模型做简单任务分类、抽取用7B模型就够别上大模型压缩上下文传递时只传必要信息不要整个历史都塞进去设置预算上限调度器里加token计数超过阈值就降级或终止。我一般会给每个任务预估token消耗调度时累计接近预算就切换到更省的策略。5.4 输出格式不稳定导致的解析崩溃模型输出格式飘忽是多Agent系统的常见故障源。你要求输出JSON它有时候给你包一层markdown代码块有时候字段名拼错有时候多写一段解释。调度器解析失败整个流程就断了。我的应对是三层防护第一层提示词里用few-shot示例强化格式第二层解析时做容错比如先尝试提取代码块内容再解析第三层解析失败触发重试重试时把错误信息反馈给Agent让它修正。实测下来加了第三层之后格式问题的解决率能到95%以上。6. 让多Agent系统真正跑稳的几个进阶思路6.1 给Agent加记忆而不是每次从零开始基础的多Agent系统里每个Agent每次被调用都是无状态的这导致它无法从历史交互中学习。进阶做法是给Agent加上分层记忆短期记忆存当前任务的上下文长期记忆存跨任务的经验。短期记忆好实现就是维护一个对话历史。长期记忆需要向量数据库把过去的成功案例、失败教训存进去Agent执行前先检索相关经验注入提示词。比如评审Agent可以检索过去这类稿子常犯的错误写作Agent可以检索过去被评审打回的原因。这能显著提升多轮任务的质量。6.2 动态调整Agent的参与策略固定的Agent配置适合稳定任务但面对变化的任务动态调整更有效。核心思路是根据任务复杂度决定启用哪些Agent。简单任务只走执行轻量校验复杂任务才启用完整的多轮评审。实现上可以在调度器前面加一个任务评估环节用一个轻量模型判断任务复杂度输出一个配置建议调度器据此决定DAG的结构。这样既保证了简单任务的效率又保证了复杂任务的质量。6.3 可观测性没有日志的多Agent系统是黑盒多Agent系统一旦出问题排查难度远高于单Agent因为涉及多个Agent的交互。必须建立完善的可观测性。我要求系统记录每个Agent的输入、输出、耗时、token消耗、工具调用记录以及调度器的每次决策。这些日志不仅用于排错还能用于优化。比如分析哪些Agent经常失败哪些环节耗时最长哪些提示词效果差。有了数据支撑优化才有方向。我一般用结构化日志JSON格式输出到文件再用简单的脚本做统计分析。6.4 人机协同关键节点留人工确认完全自动化的多Agent系统在高质量要求的场景下风险很高。更稳妥的做法是在关键节点设置人工确认。比如评审Agent判定不通过时不是自动触发重写而是把评审意见推给人工由人决定是重写还是接受。这种人机协同模式在论文写作、代码生成这类场景特别实用。Agent负责繁重的初稿和迭代工作人负责关键决策既保证了效率又保证了质量。实现上就是在调度器里加人工节点执行到该节点时暂停等待人工输入后再继续。7. 一个完整协同任务的落地示例7.1 任务场景技术调研报告生成假设我们要搭建一个系统输入一个技术主题输出一份结构化的调研报告。这个任务可以拆成五个子任务主题拆解、资料检索、要点提炼、报告撰写、质量评审。对应的Agent配置是规划Agent负责主题拆解检索Agent负责资料收集分析Agent负责要点提炼写作Agent负责报告撰写评审Agent负责质量把关。DAG结构是规划→检索→分析→撰写→评审其中评审不通过则回到撰写最多迭代3轮。7.2 关键配置与参数规划Agent用中等模型即可它的输出是任务列表不需要太强的生成能力。检索Agent需要接入搜索工具提示词里要强调返回来源和摘要不要编造。分析Agent用强模型因为要点提炼质量直接决定报告质量。写作Agent用强模型评审Agent也用强模型因为评审需要判断力。参数上检索Agent的返回条数控制在10到15条太多会稀释重点。分析Agent的输出限制在1000字以内避免信息过载。评审Agent的通过阈值设为8分满分10分低于8分触发重写。7.3 运行效果与调优记录第一版跑下来主要问题是检索Agent返回的资料质量参差导致分析Agent提炼的要点偏离主题。调优措施是在检索Agent后面加了一个相关性过滤步骤用一个轻量模型对检索结果打分只保留高分结果传给分析Agent。加了这一步之后报告质量明显提升。第二个问题是评审环节耗时太长每轮评审要几分钟。优化措施是把评审拆成快速检查和深度评审两级快速检查用轻量模型做格式和基本逻辑检查通过后再进深度评审。这样大部分明显有问题的稿子在快速检查阶段就被拦下节省了深度评审的调用。7.4 可复用的经验总结这个案例里最值得复用的经验有三条。第一在信息流转的关键节点加过滤不要让低质量信息污染下游。第二评审分级用轻量检查拦截明显问题用重量评审保证最终质量。第三迭代要有上限任何循环都必须有终止条件否则系统会失控。这套架构不只适用于调研报告换成代码生成、数据分析、内容创作思路是一样的拆解任务、分配角色、定义依赖、设置评审、控制迭代。变的只是Agent的具体提示词和工具配置不变的是协作和调度的底层逻辑。我在实际项目里踩过的最大一个坑是早期太迷信全自动什么环节都想让Agent自己搞定结果系统脆弱得不行一个小异常就全盘崩溃。后来想明白了多Agent系统的价值不是取代人而是把人从重复劳动里解放出来让人专注于真正需要判断力的决策。把人工确认节点设计好系统的稳定性和输出质量都会有质的提升。 多Agent协作架构与任务调度实战：从设计到避坑
1. 为什么单Agent在复杂任务面前总是掉链子如果你已经用大模型做过一些实际项目大概率经历过这样的场景让一个Agent去完成调研某个技术方向、整理数据、写一份分析报告、再自查一遍逻辑漏洞这种复合任务结果它要么在中途忘了最初的目标要么把前面几步的结论丢得一干二净要么在自查环节敷衍了事。这不是模型不够聪明而是单Agent架构本身存在结构性瓶颈。单Agent的问题可以归结为三个层面。第一是上下文窗口的物理限制。一个Agent要同时承载任务规划、中间结果、工具调用记录、历史对话token消耗极快一旦超出窗口早期信息就被截断导致失忆。第二是角色冲突。同一个Agent既要当执行者又要当审核者它在审核自己产出时天然带有确认偏误很难真正挑出自己的毛病。第三是串行效率低下。一个Agent按顺序做五件事总耗时是五件事之和而如果这五件事里有三件互不依赖串行就是纯粹的浪费。多Agent协作架构要解决的核心问题就是把一个大而全的Agent拆成多个小而专的Agent通过明确的协作协议和任务调度机制让它们各司其职、并行推进、互相校验。这背后的思路其实和人类团队分工一模一样你不会让一个人同时当产品经理、程序员、测试和项目经理而是拆成不同角色用流程和沟通机制把它们串起来。这篇内容面向的是已经了解大模型基本调用、想进一步搭建多Agent系统的开发者。我会从协作架构的底层设计讲到任务调度的具体实现再到实际搭建中那些文档里不会写的坑。整套思路不绑定某个特定框架你可以用AgentScope、LangGraph、AutoGen或者自己手写调度器来实现重点是理解架构决策背后的逻辑。2. 多Agent协作的四种主流架构与选型逻辑2.1 从谁说了算这个维度切分架构多Agent系统的架构设计本质上要回答一个问题决策权如何分配。围绕这个问题目前实践中跑得通的主要有四类架构它们不是互斥的很多生产系统是混合使用。第一类是主从架构Orchestrator-Worker。有一个中心调度Agent负责拆解任务、分配子任务、汇总结果其他Worker Agent只负责执行自己被分配的那一块。这是最容易理解和实现的结构适合任务可以被清晰拆解的场景比如写一篇论文可以拆成文献调研数据分析撰写校对四个子任务。它的缺点是中心Agent容易成为瓶颈且一旦拆解逻辑出错整个流程就跑偏。第二类是流水线架构Pipeline。Agent按固定顺序排列前一个的输出是后一个的输入像工厂流水线。适合步骤明确、顺序固定的任务比如数据清洗流程抽取Agent→清洗Agent→校验Agent→入库Agent。优点是可控性强、每步可单独调试缺点是缺乏灵活性中间某步出问题很难动态调整。第三类是辩论/评审架构Debate/Review。多个Agent对同一问题给出方案然后互相评审、迭代改进。典型的是生成Agent评审Agent的对抗组合或者多个生成Agent各自出方案再投票。这种架构特别适合需要高质量输出的场景比如代码生成、论文写作因为评审环节能有效捕捉生成环节的疏漏。第四类是黑板架构Blackboard。所有Agent共享一块黑板通常是共享内存或数据库谁有需要就往上面写谁需要信息就从上面读由一个控制Agent根据黑板状态决定激活哪个Agent。这种架构灵活性最高适合问题解决路径不固定的探索型任务但实现复杂度也最高状态管理很容易失控。2.2 选型时真正该看的三个指标很多人在选架构时只看哪个听起来高级这是本末倒置。我实际做项目时会先看三个指标指标含义影响架构选择任务可分解度任务能否被清晰拆成独立子任务高则选主从/流水线低则选黑板输出质量要求对结果准确性的容忍度高则必须加评审/辩论环节实时性要求对响应延迟的敏感度高则优先并行架构避免串行评审举个具体例子。如果你要做的是批量处理用户反馈并分类任务可分解度高、质量要求中等、实时性要求高那就用主从架构加并行Worker不需要评审环节。但如果你要做的是生成一份对外发布的技术白皮书质量要求极高、实时性无所谓那就值得上辩论架构让多个Agent反复打磨。2.3 一个容易被忽略的点Agent数量的边际递减新手常犯的错误是Agent越多越好觉得拆得越细越专业。实际上Agent数量存在明显的边际递减效应。每增加一个Agent就增加一份通信开销、一份状态同步成本、一份出错概率。我实测下来大多数任务3到5个Agent是甜点区超过7个之后协调成本会迅速吃掉分工带来的收益。判断是否需要增加Agent的标准很简单新增的这个Agent是否承担了一个现有Agent无法兼任的、且对结果有实质影响的独立职责。如果只是让它也参与一下那不如不加。比如评审Agent值得单独存在因为审核和执行是天然冲突的职责但格式化Agent就没必要格式化完全可以作为执行Agent的一个工具调用。3. 任务调度多Agent系统真正的技术难点3.1 调度器到底在调度什么很多人以为任务调度就是把任务分给Agent这只说对了一半。一个完整的调度器要处理四件事任务分解、依赖管理、资源分配、失败恢复。任务分解是把用户的高层目标拆成可执行的原子任务。这一步通常由规划Agent完成输出一个任务列表每个任务标注输入、输出、依赖关系。依赖管理是维护任务之间的先后顺序比如数据分析必须在数据抽取完成后才能开始。资源分配是决定哪个任务交给哪个Agent、什么时候执行。失败恢复是当某个任务失败或输出不合格时决定是重试、换Agent还是回退。这四件事里依赖管理和失败恢复是最容易出问题的。依赖管理做不好会出现任务B在任务A还没完成时就启动了的竞态失败恢复做不好一个环节卡住整个流程就死在那里。3.2 用DAG表达任务依赖实践中表达任务依赖最成熟的方案是有向无环图DAG。每个节点是一个任务每条边是一个依赖关系。DAG的好处是天然支持并行——没有依赖关系的节点可以同时执行同时又能保证有依赖的节点按顺序执行。一个典型的DAG任务定义长这样tasks { extract: { agent: data_agent, deps: [], input: raw_source }, analyze: { agent: analysis_agent, deps: [extract], input: extract.output }, draft: { agent: writing_agent, deps: [analyze], input: analyze.output }, review: { agent: review_agent, deps: [draft], input: draft.output }, revise: { agent: writing_agent, deps: [review], input: review.feedback } }调度器的工作就是遍历这个DAG找出所有依赖已满足的任务把它们放进执行队列。这里有个关键细节依赖满足的判断不能只看前置任务是否执行过还要看前置任务的输出是否有效。如果extract任务执行了但输出为空analyze任务不应该启动而应该触发extract的重试。3.3 并行执行与状态同步DAG里同一层的任务可以并行执行这是多Agent系统相比单Agent最大的效率优势。但并行带来一个新问题状态同步。多个Agent同时读写共享状态时很容易出现数据不一致。我的做法是给每个任务分配独立的输出槽位Agent只写自己的槽位读别人的槽位时通过调度器统一代理。这样避免了直接竞争调度器在任务完成时统一更新全局状态。具体实现上可以用一个中心化的状态存储比如Redis或者简单的内存字典每个任务完成后写入{task_id: output}下游任务启动时从状态存储里按需读取。注意并行任务的数量要受限于实际的API并发限制和成本预算。我见过有人一口气并行20个Agent调用结果触发限流一半任务失败反而比串行还慢。建议并行度控制在3到5之间根据你的API配额调整。3.4 失败恢复的三种策略失败恢复是调度器里最考验设计的地方。我总结下来有三种策略按复杂度递增重试策略是最简单的任务失败就重新执行最多重试N次。适合偶发性失败比如网络抖动导致的API超时。但要注意重试要加退避不能立即重试否则容易连续撞上同一个问题。降级策略是重试失败后换一个方案。比如主Agent调用失败切换到备用Agent或者复杂模型调用失败降级到轻量模型。这要求系统里对每个任务都准备好备选方案。回退策略是最复杂的任务失败后不是简单重试而是回退到上游重新规划。比如review任务发现draft质量太差不是让review重试而是触发draft重新生成。这需要调度器能识别失败根因在上游这种情况并支持DAG的动态调整。实际项目里我通常三层都用先重试重试不行降级降级还不行才回退。这样既保证了鲁棒性又不会因为一点小问题就大动干戈重新规划。4. 从零搭建一个多Agent协同系统的实操路径4.1 环境准备与框架选型动手之前先明确技术栈。多Agent系统的核心组件包括大模型调用层、Agent定义层、调度器、状态存储、工具层。你可以用现成框架也可以自己组装。现成框架里AgentScope 2.0对多Agent协作的支持比较完整内置了消息传递和分布式部署能力LangGraph擅长用图结构表达Agent流程适合DAG式调度AutoGen的对话式协作模式适合辩论/评审架构。如果你想要最大控制权直接用Python手写调度器加OpenAI/通义/本地模型的API调用也完全可行我很多项目就是这么干的因为框架的抽象层有时候反而碍事。环境上Python 3.10以上装好你选用的模型SDK准备一个状态存储开发阶段用内存字典就够生产环境上Redis。如果涉及本地模型部署Ollama或vLLM都是成熟选择前者适合快速验证后者适合高并发生产。4.2 Agent的角色定义与提示词设计多Agent系统里每个Agent的提示词设计比单Agent更讲究因为角色边界必须清晰。一个Agent的提示词要包含四部分角色身份、职责范围、输入输出规范、协作约束。角色身份要具体不要写你是一个助手而要写你是一个负责数据清洗的Agent专门处理结构化数据的缺失值填充和异常值检测。职责范围要明确边界写清楚你只负责X不负责Y避免Agent越界。输入输出规范要规定格式比如输入是JSON格式的原始数据输出是清洗后的JSON字段保持不变。协作约束要说明当你发现数据问题超出你的处理范围时输出ESCALATE: 原因不要自行处理。这里有个实操心得给每个Agent的输出加上结构化标记比如用JSON或者带标签的文本。这样调度器解析起来稳定不会因为模型输出格式飘忽而崩溃。我一般要求Agent输出严格的JSON如果解析失败就触发重试重试时在提示词里强调必须输出合法JSON。4.3 调度器的核心循环实现调度器的主循环逻辑其实不复杂核心就是找可执行任务→执行→更新状态→重复。下面是一个简化但可运行的实现import json from concurrent.futures import ThreadPoolExecutor class Scheduler: def __init__(self, tasks, agents, stateNone): self.tasks tasks self.agents agents self.state state or {} self.completed set() self.failed set() def get_ready_tasks(self): ready [] for tid, task in self.tasks.items(): if tid in self.completed or tid in self.failed: continue if all(d in self.completed for d in task[deps]): ready.append(tid) return ready def run_task(self, tid): task self.tasks[tid] agent self.agents[task[agent]] inputs {d: self.state.get(d) for d in task[deps]} try: result agent.execute(task[input], inputs) self.state[tid] result self.completed.add(tid) except Exception as e: self.failed.add(tid) self.state[tid] {error: str(e)} def run(self, max_parallel3): while len(self.completed) len(self.failed) len(self.tasks): ready self.get_ready_tasks() if not ready: break with ThreadPoolExecutor(max_workersmax_parallel) as pool: pool.map(self.run_task, ready) return self.state这段代码的关键点在get_ready_tasks它只返回那些所有依赖都已完成的任务。run方法用线程池实现并行max_parallel控制并发度。实际生产里你还需要加上重试逻辑、超时控制、日志记录但骨架就是这个。4.4 工具层与外部能力接入Agent要真正干活必须能调用外部工具。常见的工具包括搜索引擎、代码执行器、文件读写、数据库查询、API调用。工具层的设计原则是统一接口、明确schema。每个工具定义成一个函数附带清晰的参数说明和返回格式。Agent通过function calling或者ReAct模式来调用工具。这里有个坑不要让所有Agent都能访问所有工具。数据Agent不需要代码执行器写作Agent不需要数据库写权限。按角色分配工具权限既安全又能减少Agent的误用。工具调用的结果要统一处理。我的做法是工具层返回标准结构{success: bool, data: any, error: str}Agent根据success决定下一步。这样调度器和Agent都不用关心具体工具的内部实现。5. 那些文档里不会写的踩坑记录5.1 Agent之间踢皮球的循环依赖我遇到过一个典型问题写作Agent把稿子交给评审Agent评审Agent提了修改意见写作Agent改完又交回去评审Agent又提意见来回十几轮停不下来。这就是循环依赖导致的死循环。根因是评审Agent的提示词里没有通过标准它总能找到可以改进的地方。解决方案是给评审环节设置明确的收敛条件要么设定最大迭代次数比如3轮要么让评审Agent输出一个明确的通过/不通过判断通过就结束。我现在的做法是两者结合评审Agent必须输出{pass: true/false, score: 0-10, feedback: ...}score达到阈值或者迭代到上限就强制结束。5.2 上下文在传递中的信息衰减多Agent系统里信息在Agent之间传递时会衰减。A Agent的输出经过调度器传给B AgentB Agent可能只提取了它认为重要的部分再传给C Agent时信息又少一层。几轮下来最初的约束条件可能就丢了。对策是关键约束要显式传递不能依赖Agent自己记忆。比如最终报告不超过2000字这个约束不能只在第一个Agent的提示词里写而要在每个相关Agent的输入里都带上。我通常维护一个global_constraints字段每次调用Agent时都注入进去确保约束不丢失。5.3 成本失控并行不等于免费多Agent系统最容易被低估的是成本。单Agent一次任务可能几千token多Agent系统因为要传递上下文、多轮评审token消耗轻松翻5到10倍。如果并行度高短时间内大量调用账单会很难看。控制成本的手段有几个用轻量模型做简单任务分类、抽取用7B模型就够别上大模型压缩上下文传递时只传必要信息不要整个历史都塞进去设置预算上限调度器里加token计数超过阈值就降级或终止。我一般会给每个任务预估token消耗调度时累计接近预算就切换到更省的策略。5.4 输出格式不稳定导致的解析崩溃模型输出格式飘忽是多Agent系统的常见故障源。你要求输出JSON它有时候给你包一层markdown代码块有时候字段名拼错有时候多写一段解释。调度器解析失败整个流程就断了。我的应对是三层防护第一层提示词里用few-shot示例强化格式第二层解析时做容错比如先尝试提取代码块内容再解析第三层解析失败触发重试重试时把错误信息反馈给Agent让它修正。实测下来加了第三层之后格式问题的解决率能到95%以上。6. 让多Agent系统真正跑稳的几个进阶思路6.1 给Agent加记忆而不是每次从零开始基础的多Agent系统里每个Agent每次被调用都是无状态的这导致它无法从历史交互中学习。进阶做法是给Agent加上分层记忆短期记忆存当前任务的上下文长期记忆存跨任务的经验。短期记忆好实现就是维护一个对话历史。长期记忆需要向量数据库把过去的成功案例、失败教训存进去Agent执行前先检索相关经验注入提示词。比如评审Agent可以检索过去这类稿子常犯的错误写作Agent可以检索过去被评审打回的原因。这能显著提升多轮任务的质量。6.2 动态调整Agent的参与策略固定的Agent配置适合稳定任务但面对变化的任务动态调整更有效。核心思路是根据任务复杂度决定启用哪些Agent。简单任务只走执行轻量校验复杂任务才启用完整的多轮评审。实现上可以在调度器前面加一个任务评估环节用一个轻量模型判断任务复杂度输出一个配置建议调度器据此决定DAG的结构。这样既保证了简单任务的效率又保证了复杂任务的质量。6.3 可观测性没有日志的多Agent系统是黑盒多Agent系统一旦出问题排查难度远高于单Agent因为涉及多个Agent的交互。必须建立完善的可观测性。我要求系统记录每个Agent的输入、输出、耗时、token消耗、工具调用记录以及调度器的每次决策。这些日志不仅用于排错还能用于优化。比如分析哪些Agent经常失败哪些环节耗时最长哪些提示词效果差。有了数据支撑优化才有方向。我一般用结构化日志JSON格式输出到文件再用简单的脚本做统计分析。6.4 人机协同关键节点留人工确认完全自动化的多Agent系统在高质量要求的场景下风险很高。更稳妥的做法是在关键节点设置人工确认。比如评审Agent判定不通过时不是自动触发重写而是把评审意见推给人工由人决定是重写还是接受。这种人机协同模式在论文写作、代码生成这类场景特别实用。Agent负责繁重的初稿和迭代工作人负责关键决策既保证了效率又保证了质量。实现上就是在调度器里加人工节点执行到该节点时暂停等待人工输入后再继续。7. 一个完整协同任务的落地示例7.1 任务场景技术调研报告生成假设我们要搭建一个系统输入一个技术主题输出一份结构化的调研报告。这个任务可以拆成五个子任务主题拆解、资料检索、要点提炼、报告撰写、质量评审。对应的Agent配置是规划Agent负责主题拆解检索Agent负责资料收集分析Agent负责要点提炼写作Agent负责报告撰写评审Agent负责质量把关。DAG结构是规划→检索→分析→撰写→评审其中评审不通过则回到撰写最多迭代3轮。7.2 关键配置与参数规划Agent用中等模型即可它的输出是任务列表不需要太强的生成能力。检索Agent需要接入搜索工具提示词里要强调返回来源和摘要不要编造。分析Agent用强模型因为要点提炼质量直接决定报告质量。写作Agent用强模型评审Agent也用强模型因为评审需要判断力。参数上检索Agent的返回条数控制在10到15条太多会稀释重点。分析Agent的输出限制在1000字以内避免信息过载。评审Agent的通过阈值设为8分满分10分低于8分触发重写。7.3 运行效果与调优记录第一版跑下来主要问题是检索Agent返回的资料质量参差导致分析Agent提炼的要点偏离主题。调优措施是在检索Agent后面加了一个相关性过滤步骤用一个轻量模型对检索结果打分只保留高分结果传给分析Agent。加了这一步之后报告质量明显提升。第二个问题是评审环节耗时太长每轮评审要几分钟。优化措施是把评审拆成快速检查和深度评审两级快速检查用轻量模型做格式和基本逻辑检查通过后再进深度评审。这样大部分明显有问题的稿子在快速检查阶段就被拦下节省了深度评审的调用。7.4 可复用的经验总结这个案例里最值得复用的经验有三条。第一在信息流转的关键节点加过滤不要让低质量信息污染下游。第二评审分级用轻量检查拦截明显问题用重量评审保证最终质量。第三迭代要有上限任何循环都必须有终止条件否则系统会失控。这套架构不只适用于调研报告换成代码生成、数据分析、内容创作思路是一样的拆解任务、分配角色、定义依赖、设置评审、控制迭代。变的只是Agent的具体提示词和工具配置不变的是协作和调度的底层逻辑。我在实际项目里踩过的最大一个坑是早期太迷信全自动什么环节都想让Agent自己搞定结果系统脆弱得不行一个小异常就全盘崩溃。后来想明白了多Agent系统的价值不是取代人而是把人从重复劳动里解放出来让人专注于真正需要判断力的决策。把人工确认节点设计好系统的稳定性和输出质量都会有质的提升。