1. 多Agent协作到底在解决什么问题单Agent跑任务跑到一定复杂度就会撞墙。我最早做文档分析流水线的时候一个Agent既要读文件、又要抽实体、还要生成摘要、最后还得做质量校验提示词写到三千字还是压不住——它会在某个环节“忘记”前面的约束或者把抽取阶段的任务和校验阶段的任务搅在一起。这不是模型不够强而是单点上下文里塞了太多互相冲突的目标。多Agent协作的核心思路很朴素把一个大任务拆成若干职责单一的子任务每个子任务交给一个独立的AgentAgent之间通过消息传递来协同。这跟软件工程里拆微服务是一个道理——不是技术炫技而是为了可控、可调试、可替换。这套东西能做什么举几个我实际跑过的场景科研论文辅助流水线一个Agent负责研究定位与文献梳理一个负责数据分析与图表生成一个负责初稿撰写最后一个专门做质量校准查逻辑漏洞、查数据引用一致性。四个Agent串起来比单个Agent硬写质量高出一大截。代码开发协同规划Agent拆需求编码Agent写实现审查Agent做静态检查和边界测试三个角色互相制衡。复杂信息抽取抽取Agent负责召回验证Agent负责过滤幻觉汇总Agent负责结构化输出。适合谁来参考如果你已经能跑通单Agent的调用想往复杂任务编排走这篇就是给你写的。如果你还没接触过Agent建议先把单Agent的提示词工程和工具调用搞明白再回来。注意多Agent不是银弹。任务本身如果三步就能搞定硬拆成三个Agent只会增加延迟和调试成本。我踩过的最大坑就是“为了多Agent而多Agent”。2. 协作架构的四种主流形态与选型逻辑2.1 顺序流水线架构最稳但最死板顺序架构就是Agent A输出给Agent BB输出给C像工厂流水线。它的优点是链路清晰、每一步可单独测试、失败点容易定位。缺点是没有反馈回路前一步错了后面全错而且整体延迟是各步之和。我做过一个合同信息抽取的项目用的就是顺序架构OCR清洗Agent → 条款切分Agent → 要素抽取Agent → 格式校验Agent。为什么选顺序而不是别的因为合同处理天然是线性的条款切分必须在抽取之前完成没有并行的必要。实测下来四步串行总耗时约12秒其中抽取Agent占了7秒是瓶颈。选型判断标准很简单如果子任务之间存在严格的输入输出依赖且不需要回头修正就用顺序架构。2.2 层级调度架构一个主管带一队执行者层级架构里有一个调度AgentOrchestrator它不干具体活只负责拆解任务、分配任务、收集结果、决定下一步。下面挂若干执行AgentWorker每个Worker只擅长一件事。这种架构的好处是调度逻辑和执行逻辑分离。调度Agent的提示词只需要关注“怎么拆、怎么派、怎么判断完成”执行Agent的提示词只需要关注“怎么把这一件事做好”。两者互不干扰调试的时候可以分别优化。我见过一个比较典型的实现调度Agent维护一个任务队列每个任务带状态pending / running / done / failed。它根据任务类型路由到对应的WorkerWorker完成后回调更新状态调度Agent检查是否所有任务都done是则汇总输出。实操心得调度Agent的提示词里一定要明确“什么情况下任务算完成”。我早期没写清楚调度Agent会在Worker已经给出完整答案后还反复追问白白烧token。2.3 对等协商架构Agent之间互相辩论对等架构没有主管Agent之间平级通信可以互相提问、质疑、补充。最典型的玩法是辩论式校验两个Agent对同一份数据分别给出结论如果结论不一致就进入辩论轮次各自陈述理由直到达成一致或达到最大轮次。这种架构在质量校准场景特别有用。比如论文写作流水线里撰写Agent写完一段校准Agent会逐条质疑“这里的数据引用和第三段矛盾”“这个结论缺少对照组支撑”。撰写Agent收到质疑后修正再交回校准Agent复核。代价是通信开销大、轮次不可控。我建议一定要设最大辩论轮次通常2-3轮足够否则两个Agent可能无限扯皮。另外辩论的收敛条件要写死比如“连续两轮无新增质疑则终止”。2.4 混合架构实际项目里最常见纯顺序、纯层级、纯对等都不多见真实项目往往是混合的。比如外层是层级调度某个Worker内部又跑一个顺序流水线关键校验环节再插入一个对等辩论。我的经验是先用层级架构搭骨架把调度和执行分开然后在需要质量把关的节点插入对等校验最后把某些高频调用的Worker内部做成顺序流水线以降低调度开销。这样既保证了整体可控又兼顾了局部灵活性。架构类型适用场景优点缺点我的推荐指数顺序流水线线性依赖任务链路清晰、易调试无反馈、延迟叠加四星层级调度任务可并行拆分职责分离、可扩展调度Agent易成瓶颈五星对等协商质量校验、争议裁决结论更可靠开销大、收敛难三星混合架构复杂真实项目兼顾可控与灵活设计复杂度高五星3. 任务调度的核心机制拆解3.1 任务拆解从一句话需求到可执行子任务调度Agent拿到的往往是一句模糊的需求比如“帮我分析这份销售数据并给出建议”。它需要把这句话拆成可执行的子任务。拆解的质量直接决定后续所有环节的成败。我的做法是给调度Agent一套拆解模板强制它按固定维度输出{ task_id: t001, goal: 分析销售数据并给出建议, subtasks: [ {id: s1, type: data_load, desc: 读取并清洗销售数据, depends_on: []}, {id: s2, type: analysis, desc: 按区域和品类做趋势分析, depends_on: [s1]}, {id: s3, type: insight, desc: 基于分析结果生成业务建议, depends_on: [s2]}, {id: s4, type: verify, desc: 校验建议是否有数据支撑, depends_on: [s3]} ] }关键点是depends_on字段。有了依赖关系调度器就知道哪些任务可以并行、哪些必须等待。s2和s3有依赖必须串行但如果再加一个“生成可视化图表”的子任务它只依赖s1就可以和s2并行跑。注意事项拆解粒度不要太细。我试过把“读取数据”拆成“打开文件”“解析表头”“逐行读取”三个子任务结果调度开销比执行开销还大。一般一个子任务对应一个Agent的一次完整调用比较合适。3.2 依赖管理与并行执行依赖管理说白了就是拓扑排序。把所有子任务按depends_on画成有向无环图然后逐层执行第一层是没有依赖的任务可以全部并行第二层依赖第一层的任务等第一层全部完成后并行执行以此类推。我用Python实现过一个简易调度器核心逻辑大概是这样import asyncio from collections import defaultdict async def run_dag(subtasks, agent_pool): # 构建依赖图和入度表 graph defaultdict(list) indegree {t[id]: 0 for t in subtasks} task_map {t[id]: t for t in subtasks} for t in subtasks: for dep in t[depends_on]: graph[dep].append(t[id]) indegree[t[id]] 1 # 逐层执行 ready [tid for tid, deg in indegree.items() if deg 0] results {} while ready: # 当前层全部并行 layer_results await asyncio.gather(*[ agent_pool.execute(task_map[tid], results) for tid in ready ]) for tid, res in zip(ready, layer_results): results[tid] res # 更新入度找出下一层 next_ready [] for tid in ready: for nxt in graph[tid]: indegree[nxt] - 1 if indegree[nxt] 0: next_ready.append(nxt) ready next_ready return results这段代码的核心是按层并行。同一层的任务互不依赖可以同时发起调用总延迟等于最慢那个任务的时间而不是所有任务之和。实测在一个8子任务的项目里并行执行比串行快了将近3倍。3.3 状态传递Agent之间怎么“交接工作”Agent之间传递的不只是文本还包括结构化状态。我一般把状态分成三类原始输入用户需求、原始数据全程只读。中间产物各Agent的输出按task_id索引后续Agent按需读取。全局上下文所有Agent共享的背景信息比如“这是一份2024年Q3的销售数据”“输出格式要求是Markdown表格”。传递方式有两种全量传递和按需传递。全量传递是把所有中间产物都塞给下一个Agent简单但浪费上下文窗口按需传递是只给下一个Agent它depends_on的那些任务的输出精准但需要调度器做路由。我推荐按需传递。上下文窗口是稀缺资源尤其是处理长文档的时候把无关的中间产物塞进去只会稀释模型的注意力。调度器在派发任务时根据depends_on字段把上游输出拼进提示词就行。实操心得中间产物最好做一次摘要压缩再传递。比如上游Agent输出了一千字的分析传给下游时压缩成两百字的关键结论加数据引用既保留了信息又省了token。我一般让上游Agent在输出末尾附一个“给下游的摘要”字段。3.4 失败重试与降级策略Agent调用失败是常态——超时、格式错误、幻觉导致输出不可解析都会发生。调度器必须有一套失败处理机制。我的策略是三级处理自动重试格式错误这类问题把错误信息附在提示词里重试一次成功率能到70%以上。降级执行重试仍失败换一个更简单的提示词或更小的模型兜底。比如抽取任务失败降级成“只输出你能确定的部分不确定的标注为unknown”。人工介入关键任务连续失败挂起并通知人工处理不要让错误往下游传播。重试次数我一般设2次超过就降级。因为Agent调用有成本无限重试既烧钱又拖时间。4. 从零搭建一个多Agent协同任务的完整实操4.1 环境准备与基础框架选型先说环境。我用的技术栈是Python 3.10 asyncio 一个大模型API。框架层面早期我用过几个开源的多Agent框架后来发现自己写调度器反而更可控因为框架的抽象层往往和实际需求对不上改起来比自己写还费劲。如果你不想从零写几个方向可以参考AgentScope这类框架提供了Agent定义、消息传递、流程编排的基础能力LangGraph适合把Agent流程画成图来管理状态AutoGen在对话式多Agent场景比较成熟。选哪个取决于你的任务形态——流程固定的用LangGraph对话协商多的用AutoGen。我的建议是先用框架跑通一个最小demo理解多Agent的通信模式然后根据项目需求决定是继续用框架还是自己写调度层。基础依赖就几个pip install openai asyncio aiohttp pydanticpydantic用来做输出格式校验这个后面会讲为什么重要。4.2 定义Agent角色与提示词模板每个Agent本质上就是一段系统提示词 一个模型调用 一套输出格式约束。我一般用一个类来封装class Agent: def __init__(self, name, system_prompt, output_schemaNone): self.name name self.system_prompt system_prompt self.output_schema output_schema async def execute(self, task_desc, context): messages [ {role: system, content: self.system_prompt}, {role: user, content: f任务{task_desc}\n\n上下文{context}} ] response await call_llm(messages) if self.output_schema: return validate_and_parse(response, self.output_schema) return response系统提示词我遵循一个模板角色定义 职责边界 输出格式 禁止事项。举个例子一个“数据校验Agent”的提示词你是一个数据校验专家。你的唯一职责是检查上游分析结论是否有数据支撑。 你不负责生成新结论只负责指出问题。 输出格式JSON数组每个元素包含 {issue: 问题描述, severity: high/medium/low, suggestion: 修正建议} 禁止事项不要输出与校验无关的内容不要臆造数据。注意事项职责边界一定要写死。我踩过的坑是校验Agent开始“顺手”帮上游改结论结果两个Agent的输出混在一起根本分不清谁对谁错。4.3 调度器实现任务分发与结果回收调度器是整个系统的中枢。它的核心职责是接收任务、拆解、按依赖关系派发、收集结果、判断完成。我实现过一个简化版调度器核心流程如下class Orchestrator: def __init__(self, agents): self.agents agents # {agent_name: Agent} async def run(self, user_request): # 第一步拆解任务 plan await self.decompose(user_request) # 第二步按DAG执行 results await self.execute_dag(plan) # 第三步汇总输出 final await self.aggregate(results) return final async def decompose(self, request): prompt f将以下需求拆解为子任务输出JSON格式{request} return await call_llm_with_schema(prompt, PLAN_SCHEMA) async def execute_dag(self, plan): # 拓扑排序 并行执行逻辑同3.2节 ... async def aggregate(self, results): prompt f汇总以下子任务结果生成最终输出{results} return await call_llm(prompt)这里有个关键设计拆解和汇总用的是同一个调度Agent但提示词不同。拆解时它关注“怎么分”汇总时它关注“怎么合”。分开写提示词比用一个通用提示词效果好得多。4.4 输出格式校验与结构化解析多Agent系统里格式错误是最常见的失败原因。上游Agent输出了一段自然语言下游Agent解析不了整条链路就断了。我的做法是强制所有Agent输出JSON并且用pydantic做校验from pydantic import BaseModel, ValidationError class AnalysisResult(BaseModel): summary: str key_findings: list[str] data_references: list[str] confidence: float def validate_and_parse(raw_output, schema): try: # 提取JSON部分模型有时会在JSON前后加解释文字 json_str extract_json(raw_output) return schema.model_validate_json(json_str) except ValidationError as e: # 校验失败返回错误信息供重试 return {error: str(e), raw: raw_output}校验失败时调度器会把错误信息附在重试提示词里让Agent重新输出。实测这个机制能把格式错误率从30%降到5%以下。实操心得提示词里一定要给输出示例。光说“输出JSON”不够模型不知道字段名和结构。给一个完整的示例格式正确率会大幅提升。4.5 完整跑通一个论文辅助流水线我用这套框架搭过一个论文辅助流水线四个Agent定位Agent输入研究主题输出研究问题、目标期刊、创新点方向。分析Agent输入研究问题输出数据分析方案和预期图表。撰写Agent输入前两步结果输出论文初稿。校准Agent输入初稿输出问题清单和修正建议。调度流程是定位 → 分析 → 撰写 → 校准 → 如果校准有问题则回到撰写修正最多循环2轮。实测下来一篇8000字的论文初稿四个Agent跑完约需6-8分钟token消耗约15万。质量上校准Agent平均能找出8-12个问题其中约60%是真实存在的逻辑或引用问题。这个比例比单Agent自检高不少因为校准Agent没有“自己写的东西舍不得改”的偏见。5. 常见问题与排查技巧实录5.1 Agent之间“踢皮球”怎么办现象调度Agent把任务派给Worker AA说“这不是我的职责”派给Worker BB也说“不该我管”任务卡死。根因职责边界模糊。调度Agent的拆解粒度和Worker的职责定义对不上。解决在调度Agent的提示词里明确每个Worker的能力清单拆解时只生成能力清单内的任务类型。同时给Worker加一个兜底规则“如果任务不属于你的职责输出{“status”: “reject”, “reason”: “...”}不要自行处理”。5.2 上下文爆炸token消耗失控现象跑一个任务烧了几十万token成本远超预期。根因全量传递中间产物每个Agent都拿到所有上游输出上下文越滚越大。解决按需传递 摘要压缩。调度器只把depends_on的直接上游输出传给下游并且要求上游在输出末尾附一个200字以内的摘要供传递用。我实测这个改动能把token消耗降低60%以上。5.3 死循环两个Agent无限辩论现象撰写Agent和校准Agent来回修改永远达不成一致。根因没有设置终止条件。解决三重保险——最大轮次限制硬性2-3轮、收敛判断连续两轮无新增问题则终止、超时熔断单任务超过N秒强制结束并输出当前最优结果。5.4 输出格式反复出错现象某个Agent总是输出格式不对重试多次仍失败。根因提示词里的格式说明不够具体或者模型能力不足以稳定输出复杂结构。解决给完整示例 用pydantic校验 降级策略格式实在出不来就退化成纯文本输出由下游做容错解析。问题类型典型现象根因解决手段踢皮球任务无人认领职责边界模糊能力清单 reject机制上下文爆炸token消耗失控全量传递按需传递 摘要压缩死循环无限辩论无终止条件轮次限制 收敛判断 超时熔断格式错误解析失败提示词不具体示例 校验 降级5.5 独家避坑技巧汇总几个我踩过坑之后总结的硬经验调度Agent用强模型Worker可以用弱模型。调度需要理解复杂依赖关系Worker只需要执行单一任务用便宜模型完全够。每个Agent的输出都加一个confidence字段。下游可以根据置信度决定是否采信低置信度的结果触发人工复核。日志要记全。每次Agent调用的输入、输出、耗时、token消耗都记下来出问题的时候能快速定位是哪个环节。先串行跑通再改并行。并行虽然快但调试难度大。我一般先用串行验证逻辑正确再改成并行优化速度。给每个Agent设超时。单个Agent调用超过60秒直接判失败不要让整个流水线卡在一个Agent上。这套多Agent协作的架子搭起来之后后面加新Agent、改调度逻辑都很方便。我现在的做法是把Agent定义和调度逻辑分开管理Agent像插件一样注册进调度器需要什么能力就挂什么Agent。这个内容后续还可以往Agent能力自动发现、动态路由、基于历史表现的Agent选择这些方向扩展等我把动态路由跑稳定了再单独写一篇。 多Agent协作架构实战：任务调度、依赖管理与避坑指南
1. 多Agent协作到底在解决什么问题单Agent跑任务跑到一定复杂度就会撞墙。我最早做文档分析流水线的时候一个Agent既要读文件、又要抽实体、还要生成摘要、最后还得做质量校验提示词写到三千字还是压不住——它会在某个环节“忘记”前面的约束或者把抽取阶段的任务和校验阶段的任务搅在一起。这不是模型不够强而是单点上下文里塞了太多互相冲突的目标。多Agent协作的核心思路很朴素把一个大任务拆成若干职责单一的子任务每个子任务交给一个独立的AgentAgent之间通过消息传递来协同。这跟软件工程里拆微服务是一个道理——不是技术炫技而是为了可控、可调试、可替换。这套东西能做什么举几个我实际跑过的场景科研论文辅助流水线一个Agent负责研究定位与文献梳理一个负责数据分析与图表生成一个负责初稿撰写最后一个专门做质量校准查逻辑漏洞、查数据引用一致性。四个Agent串起来比单个Agent硬写质量高出一大截。代码开发协同规划Agent拆需求编码Agent写实现审查Agent做静态检查和边界测试三个角色互相制衡。复杂信息抽取抽取Agent负责召回验证Agent负责过滤幻觉汇总Agent负责结构化输出。适合谁来参考如果你已经能跑通单Agent的调用想往复杂任务编排走这篇就是给你写的。如果你还没接触过Agent建议先把单Agent的提示词工程和工具调用搞明白再回来。注意多Agent不是银弹。任务本身如果三步就能搞定硬拆成三个Agent只会增加延迟和调试成本。我踩过的最大坑就是“为了多Agent而多Agent”。2. 协作架构的四种主流形态与选型逻辑2.1 顺序流水线架构最稳但最死板顺序架构就是Agent A输出给Agent BB输出给C像工厂流水线。它的优点是链路清晰、每一步可单独测试、失败点容易定位。缺点是没有反馈回路前一步错了后面全错而且整体延迟是各步之和。我做过一个合同信息抽取的项目用的就是顺序架构OCR清洗Agent → 条款切分Agent → 要素抽取Agent → 格式校验Agent。为什么选顺序而不是别的因为合同处理天然是线性的条款切分必须在抽取之前完成没有并行的必要。实测下来四步串行总耗时约12秒其中抽取Agent占了7秒是瓶颈。选型判断标准很简单如果子任务之间存在严格的输入输出依赖且不需要回头修正就用顺序架构。2.2 层级调度架构一个主管带一队执行者层级架构里有一个调度AgentOrchestrator它不干具体活只负责拆解任务、分配任务、收集结果、决定下一步。下面挂若干执行AgentWorker每个Worker只擅长一件事。这种架构的好处是调度逻辑和执行逻辑分离。调度Agent的提示词只需要关注“怎么拆、怎么派、怎么判断完成”执行Agent的提示词只需要关注“怎么把这一件事做好”。两者互不干扰调试的时候可以分别优化。我见过一个比较典型的实现调度Agent维护一个任务队列每个任务带状态pending / running / done / failed。它根据任务类型路由到对应的WorkerWorker完成后回调更新状态调度Agent检查是否所有任务都done是则汇总输出。实操心得调度Agent的提示词里一定要明确“什么情况下任务算完成”。我早期没写清楚调度Agent会在Worker已经给出完整答案后还反复追问白白烧token。2.3 对等协商架构Agent之间互相辩论对等架构没有主管Agent之间平级通信可以互相提问、质疑、补充。最典型的玩法是辩论式校验两个Agent对同一份数据分别给出结论如果结论不一致就进入辩论轮次各自陈述理由直到达成一致或达到最大轮次。这种架构在质量校准场景特别有用。比如论文写作流水线里撰写Agent写完一段校准Agent会逐条质疑“这里的数据引用和第三段矛盾”“这个结论缺少对照组支撑”。撰写Agent收到质疑后修正再交回校准Agent复核。代价是通信开销大、轮次不可控。我建议一定要设最大辩论轮次通常2-3轮足够否则两个Agent可能无限扯皮。另外辩论的收敛条件要写死比如“连续两轮无新增质疑则终止”。2.4 混合架构实际项目里最常见纯顺序、纯层级、纯对等都不多见真实项目往往是混合的。比如外层是层级调度某个Worker内部又跑一个顺序流水线关键校验环节再插入一个对等辩论。我的经验是先用层级架构搭骨架把调度和执行分开然后在需要质量把关的节点插入对等校验最后把某些高频调用的Worker内部做成顺序流水线以降低调度开销。这样既保证了整体可控又兼顾了局部灵活性。架构类型适用场景优点缺点我的推荐指数顺序流水线线性依赖任务链路清晰、易调试无反馈、延迟叠加四星层级调度任务可并行拆分职责分离、可扩展调度Agent易成瓶颈五星对等协商质量校验、争议裁决结论更可靠开销大、收敛难三星混合架构复杂真实项目兼顾可控与灵活设计复杂度高五星3. 任务调度的核心机制拆解3.1 任务拆解从一句话需求到可执行子任务调度Agent拿到的往往是一句模糊的需求比如“帮我分析这份销售数据并给出建议”。它需要把这句话拆成可执行的子任务。拆解的质量直接决定后续所有环节的成败。我的做法是给调度Agent一套拆解模板强制它按固定维度输出{ task_id: t001, goal: 分析销售数据并给出建议, subtasks: [ {id: s1, type: data_load, desc: 读取并清洗销售数据, depends_on: []}, {id: s2, type: analysis, desc: 按区域和品类做趋势分析, depends_on: [s1]}, {id: s3, type: insight, desc: 基于分析结果生成业务建议, depends_on: [s2]}, {id: s4, type: verify, desc: 校验建议是否有数据支撑, depends_on: [s3]} ] }关键点是depends_on字段。有了依赖关系调度器就知道哪些任务可以并行、哪些必须等待。s2和s3有依赖必须串行但如果再加一个“生成可视化图表”的子任务它只依赖s1就可以和s2并行跑。注意事项拆解粒度不要太细。我试过把“读取数据”拆成“打开文件”“解析表头”“逐行读取”三个子任务结果调度开销比执行开销还大。一般一个子任务对应一个Agent的一次完整调用比较合适。3.2 依赖管理与并行执行依赖管理说白了就是拓扑排序。把所有子任务按depends_on画成有向无环图然后逐层执行第一层是没有依赖的任务可以全部并行第二层依赖第一层的任务等第一层全部完成后并行执行以此类推。我用Python实现过一个简易调度器核心逻辑大概是这样import asyncio from collections import defaultdict async def run_dag(subtasks, agent_pool): # 构建依赖图和入度表 graph defaultdict(list) indegree {t[id]: 0 for t in subtasks} task_map {t[id]: t for t in subtasks} for t in subtasks: for dep in t[depends_on]: graph[dep].append(t[id]) indegree[t[id]] 1 # 逐层执行 ready [tid for tid, deg in indegree.items() if deg 0] results {} while ready: # 当前层全部并行 layer_results await asyncio.gather(*[ agent_pool.execute(task_map[tid], results) for tid in ready ]) for tid, res in zip(ready, layer_results): results[tid] res # 更新入度找出下一层 next_ready [] for tid in ready: for nxt in graph[tid]: indegree[nxt] - 1 if indegree[nxt] 0: next_ready.append(nxt) ready next_ready return results这段代码的核心是按层并行。同一层的任务互不依赖可以同时发起调用总延迟等于最慢那个任务的时间而不是所有任务之和。实测在一个8子任务的项目里并行执行比串行快了将近3倍。3.3 状态传递Agent之间怎么“交接工作”Agent之间传递的不只是文本还包括结构化状态。我一般把状态分成三类原始输入用户需求、原始数据全程只读。中间产物各Agent的输出按task_id索引后续Agent按需读取。全局上下文所有Agent共享的背景信息比如“这是一份2024年Q3的销售数据”“输出格式要求是Markdown表格”。传递方式有两种全量传递和按需传递。全量传递是把所有中间产物都塞给下一个Agent简单但浪费上下文窗口按需传递是只给下一个Agent它depends_on的那些任务的输出精准但需要调度器做路由。我推荐按需传递。上下文窗口是稀缺资源尤其是处理长文档的时候把无关的中间产物塞进去只会稀释模型的注意力。调度器在派发任务时根据depends_on字段把上游输出拼进提示词就行。实操心得中间产物最好做一次摘要压缩再传递。比如上游Agent输出了一千字的分析传给下游时压缩成两百字的关键结论加数据引用既保留了信息又省了token。我一般让上游Agent在输出末尾附一个“给下游的摘要”字段。3.4 失败重试与降级策略Agent调用失败是常态——超时、格式错误、幻觉导致输出不可解析都会发生。调度器必须有一套失败处理机制。我的策略是三级处理自动重试格式错误这类问题把错误信息附在提示词里重试一次成功率能到70%以上。降级执行重试仍失败换一个更简单的提示词或更小的模型兜底。比如抽取任务失败降级成“只输出你能确定的部分不确定的标注为unknown”。人工介入关键任务连续失败挂起并通知人工处理不要让错误往下游传播。重试次数我一般设2次超过就降级。因为Agent调用有成本无限重试既烧钱又拖时间。4. 从零搭建一个多Agent协同任务的完整实操4.1 环境准备与基础框架选型先说环境。我用的技术栈是Python 3.10 asyncio 一个大模型API。框架层面早期我用过几个开源的多Agent框架后来发现自己写调度器反而更可控因为框架的抽象层往往和实际需求对不上改起来比自己写还费劲。如果你不想从零写几个方向可以参考AgentScope这类框架提供了Agent定义、消息传递、流程编排的基础能力LangGraph适合把Agent流程画成图来管理状态AutoGen在对话式多Agent场景比较成熟。选哪个取决于你的任务形态——流程固定的用LangGraph对话协商多的用AutoGen。我的建议是先用框架跑通一个最小demo理解多Agent的通信模式然后根据项目需求决定是继续用框架还是自己写调度层。基础依赖就几个pip install openai asyncio aiohttp pydanticpydantic用来做输出格式校验这个后面会讲为什么重要。4.2 定义Agent角色与提示词模板每个Agent本质上就是一段系统提示词 一个模型调用 一套输出格式约束。我一般用一个类来封装class Agent: def __init__(self, name, system_prompt, output_schemaNone): self.name name self.system_prompt system_prompt self.output_schema output_schema async def execute(self, task_desc, context): messages [ {role: system, content: self.system_prompt}, {role: user, content: f任务{task_desc}\n\n上下文{context}} ] response await call_llm(messages) if self.output_schema: return validate_and_parse(response, self.output_schema) return response系统提示词我遵循一个模板角色定义 职责边界 输出格式 禁止事项。举个例子一个“数据校验Agent”的提示词你是一个数据校验专家。你的唯一职责是检查上游分析结论是否有数据支撑。 你不负责生成新结论只负责指出问题。 输出格式JSON数组每个元素包含 {issue: 问题描述, severity: high/medium/low, suggestion: 修正建议} 禁止事项不要输出与校验无关的内容不要臆造数据。注意事项职责边界一定要写死。我踩过的坑是校验Agent开始“顺手”帮上游改结论结果两个Agent的输出混在一起根本分不清谁对谁错。4.3 调度器实现任务分发与结果回收调度器是整个系统的中枢。它的核心职责是接收任务、拆解、按依赖关系派发、收集结果、判断完成。我实现过一个简化版调度器核心流程如下class Orchestrator: def __init__(self, agents): self.agents agents # {agent_name: Agent} async def run(self, user_request): # 第一步拆解任务 plan await self.decompose(user_request) # 第二步按DAG执行 results await self.execute_dag(plan) # 第三步汇总输出 final await self.aggregate(results) return final async def decompose(self, request): prompt f将以下需求拆解为子任务输出JSON格式{request} return await call_llm_with_schema(prompt, PLAN_SCHEMA) async def execute_dag(self, plan): # 拓扑排序 并行执行逻辑同3.2节 ... async def aggregate(self, results): prompt f汇总以下子任务结果生成最终输出{results} return await call_llm(prompt)这里有个关键设计拆解和汇总用的是同一个调度Agent但提示词不同。拆解时它关注“怎么分”汇总时它关注“怎么合”。分开写提示词比用一个通用提示词效果好得多。4.4 输出格式校验与结构化解析多Agent系统里格式错误是最常见的失败原因。上游Agent输出了一段自然语言下游Agent解析不了整条链路就断了。我的做法是强制所有Agent输出JSON并且用pydantic做校验from pydantic import BaseModel, ValidationError class AnalysisResult(BaseModel): summary: str key_findings: list[str] data_references: list[str] confidence: float def validate_and_parse(raw_output, schema): try: # 提取JSON部分模型有时会在JSON前后加解释文字 json_str extract_json(raw_output) return schema.model_validate_json(json_str) except ValidationError as e: # 校验失败返回错误信息供重试 return {error: str(e), raw: raw_output}校验失败时调度器会把错误信息附在重试提示词里让Agent重新输出。实测这个机制能把格式错误率从30%降到5%以下。实操心得提示词里一定要给输出示例。光说“输出JSON”不够模型不知道字段名和结构。给一个完整的示例格式正确率会大幅提升。4.5 完整跑通一个论文辅助流水线我用这套框架搭过一个论文辅助流水线四个Agent定位Agent输入研究主题输出研究问题、目标期刊、创新点方向。分析Agent输入研究问题输出数据分析方案和预期图表。撰写Agent输入前两步结果输出论文初稿。校准Agent输入初稿输出问题清单和修正建议。调度流程是定位 → 分析 → 撰写 → 校准 → 如果校准有问题则回到撰写修正最多循环2轮。实测下来一篇8000字的论文初稿四个Agent跑完约需6-8分钟token消耗约15万。质量上校准Agent平均能找出8-12个问题其中约60%是真实存在的逻辑或引用问题。这个比例比单Agent自检高不少因为校准Agent没有“自己写的东西舍不得改”的偏见。5. 常见问题与排查技巧实录5.1 Agent之间“踢皮球”怎么办现象调度Agent把任务派给Worker AA说“这不是我的职责”派给Worker BB也说“不该我管”任务卡死。根因职责边界模糊。调度Agent的拆解粒度和Worker的职责定义对不上。解决在调度Agent的提示词里明确每个Worker的能力清单拆解时只生成能力清单内的任务类型。同时给Worker加一个兜底规则“如果任务不属于你的职责输出{“status”: “reject”, “reason”: “...”}不要自行处理”。5.2 上下文爆炸token消耗失控现象跑一个任务烧了几十万token成本远超预期。根因全量传递中间产物每个Agent都拿到所有上游输出上下文越滚越大。解决按需传递 摘要压缩。调度器只把depends_on的直接上游输出传给下游并且要求上游在输出末尾附一个200字以内的摘要供传递用。我实测这个改动能把token消耗降低60%以上。5.3 死循环两个Agent无限辩论现象撰写Agent和校准Agent来回修改永远达不成一致。根因没有设置终止条件。解决三重保险——最大轮次限制硬性2-3轮、收敛判断连续两轮无新增问题则终止、超时熔断单任务超过N秒强制结束并输出当前最优结果。5.4 输出格式反复出错现象某个Agent总是输出格式不对重试多次仍失败。根因提示词里的格式说明不够具体或者模型能力不足以稳定输出复杂结构。解决给完整示例 用pydantic校验 降级策略格式实在出不来就退化成纯文本输出由下游做容错解析。问题类型典型现象根因解决手段踢皮球任务无人认领职责边界模糊能力清单 reject机制上下文爆炸token消耗失控全量传递按需传递 摘要压缩死循环无限辩论无终止条件轮次限制 收敛判断 超时熔断格式错误解析失败提示词不具体示例 校验 降级5.5 独家避坑技巧汇总几个我踩过坑之后总结的硬经验调度Agent用强模型Worker可以用弱模型。调度需要理解复杂依赖关系Worker只需要执行单一任务用便宜模型完全够。每个Agent的输出都加一个confidence字段。下游可以根据置信度决定是否采信低置信度的结果触发人工复核。日志要记全。每次Agent调用的输入、输出、耗时、token消耗都记下来出问题的时候能快速定位是哪个环节。先串行跑通再改并行。并行虽然快但调试难度大。我一般先用串行验证逻辑正确再改成并行优化速度。给每个Agent设超时。单个Agent调用超过60秒直接判失败不要让整个流水线卡在一个Agent上。这套多Agent协作的架子搭起来之后后面加新Agent、改调度逻辑都很方便。我现在的做法是把Agent定义和调度逻辑分开管理Agent像插件一样注册进调度器需要什么能力就挂什么Agent。这个内容后续还可以往Agent能力自动发现、动态路由、基于历史表现的Agent选择这些方向扩展等我把动态路由跑稳定了再单独写一篇。