多Agent协作架构设计与任务调度实战指南

我们团队最近在做一个内部知识库问答系统本来用单个大模型跑得好好的但随着需求越堆越多——既要查资料、又要生成周报、还要自动归档邮件——单个Agent开始顾此失彼。于是在一次迭代中我们引入了多Agent协作架构把任务拆给不同角色并行处理再通过任务调度来统一协调。项目代号22.7是一套完整落地、能从零搭建的复杂AI协同任务方案。如果你也在纠结多个Agent到底怎么配合任务调度用什么策略协作架构怎么选这篇文章就是从实际项目中总结出来的完整方案包含架构选型、调度机制、完整实操步骤和一堆踩坑经验适合已经跑通单Agent、准备进阶到多Agent协作的同学参考。1. 多Agent协作架构的本质与设计思路1.1 为什么要从单Agent走向多Agent协作单个大模型Agent的能力边界其实很清晰上下文窗口有限、单次任务专注度有限、工具调用链路过长容易出错。拿我们之前的单Agent方案举例用户提交一个问题后这个Agent要依次完成意图识别、资料检索、文档分析、报告生成最后还要自查一遍。整个过程是一条长链路任何一个环节出错后面全部白费。多Agent协作的思路是把这条长链路拆成多个短链路每个Agent只负责一个环节。比如专门的检索Agent只用处理查什么、去哪查、怎么查报告Agent只管怎么组织语言、怎么排版。每个Agent的提示词可以做得非常精简上下文也不会被无关内容撑爆。实测下来单Agent的完整任务成功率达到90%左右就已经不错了拆成多Agent协作后整体成功率反而能稳定在95%以上因为错误被局部化了。1.2 协作架构的三种基本模型多Agent协作架构没有标准答案我们实践下来最常用的是三种模型。第一种是流水线模型最适合任务有明显先后顺序的场景。Agent A的输出作为Agent B的输入A做完B接B做完C接跟工厂流水线一样。优点是结构简单、容易调试缺点是整体耗时等于所有Agent耗时之和而且上游出错会直接阻塞下游。第二种是编排器-执行器模型也是我们22.7方案最常采用的。一个中央调度Agent负责任务分解和结果汇总多个执行Agent负责具体干活。调度Agent本身不做重活但必须具备很强的规划力和判断力。它的优势是灵活性极高某一环节失败后调度器可以重新分配缺点是调度Agent容易成为瓶颈如果它的规划能力不够整个系统的上限就会被拉低。第三种是黑板模型多个Agent共享一块黑板公共消息区各自读取感兴趣的讯息处理后再写回结果。这种模型适合探索性、开放性任务比如头脑风暴或多方案比较但现在工程落地还不多原因后面会讲。1.3 为什么我们选择了编排器-执行器模型在22.7项目中我们最终选了编排器-执行器模型主要是因为这套模型对业务场景的覆盖最全面。我们的知识库问答任务既有明确的流水线环节检索→分析→生成又有需要动态变化的场景用户追问、问题发散纯流水线模型处理发散场景非常笨拙纯黑板模型又难以保证结果质量和收敛速度。编排器-执行器还有一个隐藏优势便于做权限和能力隔离。执行Agent可以分别绑定不同的工具、不同的知识库、甚至不同的模型。比如检索Agent用一个速度快、价格低的模型就够了报告Agent则用更强的大模型保证输出质量整体成本比所有任务都用同一个大模型省了约四成。2. 任务调度的核心机制与策略选择2.1 任务分解从粗粒度到细粒度的拆解规则任务调度第一步是任务分解这也是整个多Agent系统里最体现功底的部分。把一个复杂问题拆成多个子任务并不难难的是拆出来的子任务边界清晰、依赖关系明确、粒度大小适中。我们总结出一套比较实用的拆解规则每个子任务的产出必须是可验证的不能是开放式答案子任务之间的依赖关系尽量控制在DAG有向无环图范围内避免出现循环依赖单个子任务的耗时控制在十几秒到几十秒之间太长的任务继续拆太短的任务合并掉。以生成一份项目周报为例我们的编排器会拆成以下子任务拉取本周代码提交记录、汇总各成员的进展反馈、生成周报初稿、校验格式和错别字、输出最终版本。其中前两个子任务互不依赖可以并行后面三个必须串行。这种拆解方式让整个周报任务从原本单Agent要几分钟缩短到几十秒。2.2 调度策略并行、串行与条件分支任务调度不能只会一个个来。我们22.7方案里实现了四种调度策略实际使用频率从高到低排列。串行调度用得最多适用于有明确先后依赖关系的任务。调度器维护一个执行队列一个Agent完成并返回结果后下一个Agent才启动。并行调度用于无依赖关系的任务组比如同时检索多个知识库、同时拉取多份数据调度器用并发控制机制把多个任务同时派发给不同的执行Agent。条件分支调度稍微复杂一点。调度器根据上一个Agent的输出结果判断下一步走哪条分支。比如用户提问今天天气怎么样检索Agent返回的结果如果命中本地知识库就直接走答案生成分支如果没命中就额外派一个Agent去调用天气API。这种动态路由能力是单Agent很难做好的。还有一种是聚合调度多个Agent各自输出结果后由一个专门的聚合Agent合并去重、消除矛盾产出最终答案。这在多来源信息汇总场景里非常好用但要给聚合Agent明确好处理冲突的规则比如以最新数据为准两处来源矛盾时标注不确定性。2.3 上下文管理与消息路由多Agent协作中最麻烦的其实是上下文管理。每个Agent能看到的上下文不同相互之间传递什么信息、过滤什么信息直接影响最终质量。我们最早犯过一个错误调度器把完整对话历史广播给所有Agent结果执行Agent的注意力被无关内容分散输出质量直线下滑。后来改成按需投影策略——调度器根据子任务类型决定给每个Agent只看哪一部分上下文。检索Agent只拿到用户原始问题和知识库访问凭证报告Agent只拿到检索结果和格式要求各自所需的上下文最小化。消息路由也是容易被低估的模块。我们用了一套轻量级的事件总线所有Agent通过它收发消息消息带类型标签和优先级。调度器根据不同消息类型投递给不同Agent高优先级的任务可以插队。这套设计让我们后续加新Agent时不用改动核心代码只要注册一个新角色ID和消息处理逻辑就能接入。3. 完整实操用工人组合构建一个多Agent协同任务3.1 场景定义与角色配置下面用一个真实例子完整走一遍22.7的落地过程。场景是自动生成一篇科技行业早报要求每天早上9点前产出包含AI大模型领域的前三条重要动态、每条的简要分析和一句评论。我们在这个场景里配置了四个执行Agent和一个编排器。检索Agent负责从预置的RSS源和新闻API拉取当日资讯筛选Agent负责对新闻做去重、打分、排序选出与AI大模型领域相关度最高的前五条分析Agent负责对入选新闻生成深度点评汇总Agent负责把点评整理成最终早报格式。编排器负责任务调度和全流程监控。四个Agent的提示词各不同。检索Agent的提示词里只强调追踪哪些信息源、返回原始链接和时间戳筛选Agent的提示词里重点写清楚评分维度比如涉及大模型发布/融资/开源动态加5分涉及泛AI但不聚焦大模型加1分分析Agent的提示词则要求必须包含事实描述、背景铺垫和独立观点三个部分。角色边界越清晰协作效率越高。3.2 22.7环境配置与依赖安装动手之前先交代环境。我们用的是Python 3.10 Docker Compose跑的整套服务模型层接的是本地部署的开源大模型Qwen系列7B和14B各一个Agent之间通过Redis Stream传消息编排器逻辑用Python的asyncio实现负责调度的核心代码大约三百行。依赖安装没什么特殊之处核心的几个包是langgraph编排框架、redis消息队列、openai兼容层用于统一调用本地和远程模型、httpx异步HTTP请求。pip install langgraph redis openai httpx需要说明的是我们没有全部依赖框架编排器的调度逻辑是手写的。原因很实在框架封装的调度策略是通用的但我们内部的任务类型、消息格式、上下文投影规则都是定制过的用通用框架反而要花更多时间适配手写反而更简洁。如果你的任务类型比较标准直接用langgraph这类框架也可以不用纠结。3.3 编排器的核心调度代码这一节直接放关键代码。我们的编排器里最重要的一个函数是run_schedule它接收一个任务图用DAG表示然后按依赖关系逐层执行。from dataclasses import dataclass, field from typing import Callable, Dict, List import asyncio dataclass class TaskNode: name: str handler: Callable dependencies: List[str] field(default_factorylist) retry_count: int 3 class Scheduler: def __init__(self): self.nodes: Dict[str, TaskNode] {} self.results: Dict[str, any] {} def register(self, node: TaskNode): self.nodes[node.name] node async def execute(self, node_name: str): if node_name in self.results: return self.results[node_name] node self.nodes[node_name] for dep in node.dependencies: if dep not in self.results: await self.execute(dep) for attempt in range(node.retry_count): try: dep_results {dep: self.results[dep] for dep in node.dependencies} self.results[node_name] await node.handler(dep_results) return self.results[node_name] except Exception as e: if attempt node.retry_count - 1: raise print(fAgent {node_name} 第{attempt 1}次执行失败: {e}) async def run(self, root_node: str): await self.execute(root_node) return self.results[root_node]这段代码的核心就是递归遍历任务图、按照依赖顺序执行、失败自动重试三次。我们的22.7方案里所有Agent任务都被封装成TaskNode注册进Scheduler调度器本身不关心Agent内部逻辑。实际配置早报任务时注册代码如下scheduler Scheduler() scheduler.register(TaskNode(feeds, feed_handler)) scheduler.register(TaskNode(filter, filter_handler, dependencies[feeds])) scheduler.register(TaskNode(analyze, analyze_handler, dependencies[filter])) scheduler.register(TaskNode(collect, collect_handler, dependencies[analyze])) result await scheduler.run(collect)四个节点串成一条链feeds做完才能做filterfilter做完才能做analyze最后汇总。这就是流水线模型的一个典型实现。当然实际项目中我们在外层还包了一层调度器来处理并行分支但核心逻辑原理一致。3.4 消息路由与上下文投影的实现调度器能正常串联Agent还要靠消息路由和上下文投影两个模块。我们给每个Agent定义了消息的input_schema例如检索Agent要求输入是一个包含query和max_items的字典筛选Agent要求输入是包含raw_items和score_rules的字典。上下文投影的实现思路其实很简单在Agent启动前做一个轻量级过滤函数把调度器汇总的结果按需裁剪。def project_context(task_type: str, full_context: dict) - dict: if task_type retrieval: return {query: full_context[query], content_base: full_context[content_base_id]} if task_type filter: return {raw_items: full_context[raw_items][:20], rules: full_context[filter_rules]} if task_type report: return {filtered_items: full_context[filtered_items], format: tech_report} return full_context这个函数被编排器在每个Agent执行前调用。实测下来上下文裁剪掉后每个Agent的prompt长度平均缩短50%以上模型响应速度明显提升而且输出质量更稳定。这就是上下文投影的实际好处。3.5 运行结果与效果对照整个早报任务运行起来后效果对照如下表指标单Agent方案多Agent方案22.7平均完成耗时210秒85秒首次成功率82%96%失败后恢复用时需人工介入自动重试降级约10秒每任务Token消耗约12K约7K日报质量人工评分3.8/54.5/5耗时缩短是因为检索和预筛选两个环节实际上走了并行Token消耗降低是因为每个Agent的上下文都被裁剪过。成功率提升则是因为失败被局部隔离了——单Agent方案里如果某个环节上下文溢出整个任务必须重来而多Agent方案里失败只会发生在单环节点上重试代价小很多。4. 多Agent协作中的关键问题与实操心法4.1 死锁与循环依赖的识别与规避任务调度最怕的就是任务图出现循环依赖A等B、B等A两个Agent永远没有结果返回。我们在22.7初期就踩过一次这个坑某个分析任务的分支逻辑写错了导致Agent A的结果被送给Agent B后又传回Agent A调度器不断重跑这两个任务但始终无法收敛。后来我们加了两个防御手段。第一任务图构建时做一次DAG校验发现环直接拒绝启动第二所有Agent执行都设超时时间超过30秒没返回就标记失败并触发降级逻辑。这两个机制帮我们在后续迭代中躲掉了至少三次类似的坑。检查依赖关系可以用拓扑排序代码量很小from collections import deque def has_cycle(nodes: Dict[str, TaskNode]) - bool: indeg {name: 0 for name in nodes} for node in nodes.values(): for dep in node.dependencies: indeg[node.name] 1 queue deque([name for name, deg in indeg.items() if deg 0]) visited 0 while queue: cur queue.popleft() visited 1 for node in nodes.values(): if cur in node.dependencies: indeg[node.name] - 1 if indeg[node.name] 0: queue.append(node.name) return visited ! len(nodes)注意如果任务数量达到几十个建议直接在构建任务图的时候就做循环检测不要等到运行时。4.2 上下文丢失与信息过载的平衡多Agent协作下上下文管理是最容易出问题的地方。信息过载导致Agent抓不住重点信息不足则导致输出粗糙这个平衡很微妙。我们的经验是给Agent能完成当前任务的最少信息而不是尽可能多的信息。比如筛选Agent只需要知道有哪些新闻候选和评分规则它就不需要看到用户原始对话汇总Agent只需要知道筛选后的内容和格式要求它也不需要看评分过程。上下文裁剪得越狠各Agent的专注度越高最终质量越稳定。但这里有一个例外如果下游Agent需要理解的背景较多比如做深度分析上游就应该把关键中间结论和推理摘要一并传递而不是只给最终结果。我们在分析Agent的输入里附加了筛选Agent的总结评语效果比只给原始新闻链接好很多。4.3 成本控制模型分级与并发限制多Agent方案比单Agent多了一个优势可以做模型分级。我们22.7方案的默认配置是检索类、筛选类Agent用本地7B模型分析类Agent用14B模型汇总类Agent使用更强的商用大模型API。整体下来每任务成本比统一用商用API的方案降低60%以上。并发限制同样重要。如果编排器把所有子任务一次性并发出去本地模型的显存会直接被打满延迟反而飙升。我们实测下来的建议是本地模型并发数控制在2-4之间远程API并发数控制在5-8之间这是性价比比较高的区间。并发控制可以用Python的asyncio.Semaphore轻松实现。sem asyncio.Semaphore(4) async def limited_call(agent_func): async with sem: return await agent_func()4.4 调试多Agent系统最实用的三个手段多Agent系统的调试比单Agent要难受得多因为问题可能出在任何一个环节。我们后来总结出三个实用的调试手段。第一个是全链路日志跟踪。每个Agent执行前打印输入摘要执行后打印输出摘要调度器记录每个节点的耗时和重试次数。有了这些日志绝大多数问题都能定位到具体节点。第二个是局部复现。发现某个Agent输出质量差时直接用它的输入样本单独调用它把它的输出跟完整链路里的输出对比看是不是环境参数不同导致的。第三个是**最小复现用例思想**。如果整个链路失败但不知道原因先砍掉调度器用一套写死的固定输入逐个单独跑Agent问题很快就能浮出水面。5. 常见问题速查与经验补遗为了让你在实际搭建的时候少走弯路我把这个过程中比较高频率遇到的问题整理成一个速查参考表都是我们自己或者朋友团队实际碰到过的现象可能原因解决方案Agent之间互相等待、任务卡住任务图存在循环依赖启动前做DAG校验或拓扑排序日志里排查依赖链某个Agent输出格式不稳定提示词约束太弱上下文被无关内容干扰加强JSON Schema约束裁剪输入上下文整体耗时反而比单Agent更慢串行节点太多或者并发参数太小检查DAG结构把无依赖节点改成并行调度Token消耗比预期高很多上下文投影没生效每个Agent都在接收全量上下文检查调度器是否调用了上下文裁剪函数Agent偶尔返回空结果或报错模型服务偶发超时加重试机制建议至少3次间隔指数退避结果质量不稳定时好时坏模型温度和Top-P设置太高把生成温度降到0.2左右高确定性场景甚至可以设0Agent自带偏见或幻觉提示词没有强调事实约束在提示词里加必须基于给定内容回答不允话凭空补充等约束词还有一些经验性的东西不属于问题但值得知道。比如不要把Agent名字取得太抽象我们之前管检索Agent叫AgentAlpha调试的时候看到日志完全不知道谁在跑。后来统一改成内容检索Agent相关性筛选Agent这种一目了然的命名团队沟通效率都提升了不少。再比如提示词里给Agent一个人设往往比给一堆规则更好用。分析Agent的人设是资深科技记者筛选Agent的人设是严苛的编辑生成的风格马上就不一样了。这条经验在很多场景里都很实用。还有一点跟模型选型相关多Agent场景对模型的要求和单Agent不完全一样。单Agent强调整体推理能力多Agent更看重指令遵循能力和输出格式稳定性。我们试过用7B模型做分析类任务卡在格式上的次数很多后来14B模型专门跑分析情况立刻好转。如果你的Agent经常在有格式要求的时候翻车可以考虑单独给这类Agent升级模型。6. 从多Agent到复杂AI协同任务的一些心得6.1 好的编排器不是会分配任务而是会提前预判用了一段时间之后我越来越觉得编排器这个角色才是多Agent系统的灵魂。它不直接产出内容但它的规划能力决定了整体产出效率。我们在22.7的第二版迭代里给编排器加了一个前置判断逻辑先评估用户问题的难度和涉及范围简单问题直接走轻量级链路两个Agent搞定复杂问题才走全链路四个Agent全力跑。这个小改动带来了两倍多的效率提升——大约一半的简单问题不再需要启动全链路整体资源消耗下来一大截。如果你的多Agent系统还在追求所有任务都精细拆分反而可以考虑一下反方向能少分就少分在恰当节点简化链路。6.2 拆解任务是在理解业务不完全是在设计技术坦白讲做多Agent系统的过程中花时间最多的不是写代码而是理解业务。任务拆得好不好答案很简单你对这个业务的环节是否真的了解。如果一个业务链条本身模糊不清再优秀的调度器也调度不出来合理的结果。我们做早报任务的时候一开始总是拆得很细——多到十几个Agent。跑了一阵子发现很多Agent重叠严重该合并的没合并白白浪费资源。后来重新梳理了业务流程把十几个Agent压回四个效果反而更好了。任务不是拆得越细越好拆的粒度要跟业务的自然阶段对齐。6.3 Agent之间也需要团队磨合多Agent协作还有一个比较感性的经验Agent之间需要磨合。不是调一次就能稳定我们会根据运行日志定期调整每个Agent的提示词有时候只是加一句注意引用来源就能让下游汇报Agent的输出质量大幅提升。这其实跟带团队很像。把每个Agent当成一个有性格的组员你会更容易理解为什么它输出质量不稳、为什么它偶尔跑偏。知道这几个Agent各自的脾气之后反而更好管理了。我现在的做法是每周跑一次全量任务集把每个Agent的输出质量、耗时、失败率拉一张表对比着看。哪个Agent退步了就单独调它的提示词哪个Agent超时就考虑要不要换模型。这套运营思维的引入让整个系统的稳定性一直在缓慢上升。如果你正准备做多Agent项目我的建议是先搭一个能跑通的骨架再逐步加细节。不要一上来就追求复杂的编排策略、完美的上下文管理——先让两个Agent协作起来体会一下协作带来的问题和收益然后再逐步扩展。多Agent真正难的不是技术而是你对任务本身的理解是否足够透彻。