我最近在折腾多Agent系统时有朋友问“你不是天天说单Agent不行那多个Agent一起上不就行了吗”这让我意识到很多人对多Agent的理解还停留在“多调用几次API”的层面。实际上多Agent的核心从来不是数量而是协作架构和任务调度。没有这两样东西你只会得到一群各说各话的智能模块而不是一套能自动完成复杂AI协同任务的生产力团队。这篇内容我想讲讲如何从零设计一套多Agent系统包括最常用的协作架构、任务调度机制以及一个完整可复现的实操案例。项目代号我习惯叫22.7取自2022年7月那版最粗糙的原型。适合手里有基础大模型开发经验、想往Agent方向深挖的读者也适合正在带团队做AI应用的架构师。1. 多Agent协作架构先搞懂为什么需要它1.1 单Agent的天然瓶颈以及多Agent的价值前面说了多调用几次API不是多Agent。单Agent最大的瓶颈是上下文窗口。你让一个模型既要读数据、又要写报告它会把数据和分析搅在一起结果就是报告开头引用的数据到后面自己打脸。另一个问题是角色冲突同一个模型用同一套提示词扮演多个角色输出会趋向折中反而失去专业性。比如你让模型既当“严谨的数据分析师”又当“激进的营销文案”它很可能会给出一篇既不严谨也不激进的四不像内容。多Agent的价值在于把复杂任务拆开让每个Agent只做一件相对简单的事并配上独立的提示词与工具。比如信息采集Agent只负责搜索和清洗分析Agent只负责提炼指标撰写Agent只负责组织语言。每个Agent的上下文窗口里装的都是它真正需要的内容模型活在自己的专业领域内效果会稳定得多。更重要的是多Agent还能做“交叉验证”。A Agent产出的结论可以让B Agent复核避免单一模型幻觉被直接当成事实。1.2 主流协作模式对比与选型思路架构层面我见过三类常见模式。第一类是集中式Orchestrator-Worker一个调度Agent负责拆任务、分发结果、汇总输出Worker各自干活。好处是可控性强坏处是调度Agent本身可能成为瓶颈和单点。第二类是分散式Agent-to-Agent没有唯一中心Agent之间相互发消息像办公室同事七嘴八舌讨论。好处是灵活坏处是容易陷入“聊跑了”的状态需要非常强的通信协议。第三类是层级式调度Agent下面再派生出子调度Agent适合处理庞大的任务树。新手阶段我强烈建议从集中式入手。原因很直接任务调度的问题还没解决前不要引入自由通信的复杂度。集中式可以先把“拆解、执行、汇总”跑通再考虑更复杂的模式。我遇到过不少团队一上来就搞“全自主多Agent协商”最后项目直接烂尾。不是协商不好而是基础不稳。下面是三类模式的对比模式复杂度可控性典型场景集中式低高流程清晰的中等复杂任务分散式高中开放探索类任务如头脑风暴层级式中较高超大型项目需要逐层拆解选型时我还有一个额外标准调度Agent会不会看到全量上下文。如果所有结果都回流到调度Agent它的上下文很快会被撑爆那就要考虑层级式或引入中间存储。合理的设计是让调度Agent只看到每个Worker返回的“小结”而不是原始长篇结果。1.3 通信协议与数据格式架构的血肉协作架构定完之后最容易被忽视的是Agent之间的通信协议。这一点做实了多Agent系统才算真正连通。我习惯把所有Agent消息统一为JSON结构包含四个字段task_id、agent_id、payload、meta。payload里放真正要传递的数据meta里放时间戳、意图、状态码等控制信息。这里的关键“为什么”是固定数据结构后下游Agent才能稳定解析输入而不是靠“读自然语言猜意思”。如果你让Agent A输出一段散文让Agent B从散文里提取信息总有一天会漏掉重要字段。我给自己的规矩是Agent之间的自由发挥只允许发生在每个节点的内部处理过程一旦要跨节点传递就必须走强类型结构。一个典型的消息结构大概是这样的{ task_id: collect_001, agent_id: collector, payload: {vendor: Tesla, metrics: [sales, price]}, meta: {ts: 2022-07-11T08:00:00Z, status: success} }2. 任务调度多Agent系统的“操作系统”2.1 先对任务做拆解从列表到依赖图任务调度不是把所有子任务放进一个队列就完事。很多子任务之间存在依赖关系。比如信息采集必须先于数据分析数据分析先于文案撰写。如果只按先后顺序排队遇到并行任务时要么过度保守、要么顺序错误。我的做法是把任务描述成一张有向无环图DAG。每个节点是一个子任务每条边是依赖关系。调度器按拓扑排序执行先跑没有依赖的任务再解锁后续节点。这里有个很实用的细节DAG里要标注关键路径因为整个任务的完成时间基本被这条路径决定。如果给关键路径上的Agent配更强模型或更快工具整体效率提升会非常明显。举一个具体例子生成市场分析报告的任务图有5个节点A数据采集无依赖、B行业背景调研无依赖、C数据分析依赖A、D竞品对比依赖B、E形成报告依赖C和D。调度器可以同时执行A和B等C和D完成后再执行E。如果只用自然语言描述因果逻辑模型往往会串行执行白白浪费时间显式DAG能让并行真正发生。2.2 从串行到并行调度策略怎么选调度策略要回答两个问题什么时候启动哪个任务以及任务资源冲突怎么处理。我见过的常用策略有四种。串行调度实现最简单适合任务量小、依赖强的场景缺点是慢。轮询调度让多个Worker排队轮流取任务适合执行Agent能力接近的情况。优先级调度给任务标记优先级优先执行影响关键路径的节点适合有明确主次的任务。动态调度则根据Agent返回结果由调度Agent现场决定下一步适合结果不确定的任务。我自己通常组合使用宏观上按DAG拓扑顺序调度微观上在可并行队列里按优先级排序。比如给关键路径上的节点标成P0普通背景调查标成P1可做可不做的信息增强标成P2。调度器永远先执行P0避免整个项目被拖到最后一刻。有一个容易踩的坑有些Agent比如调用外部API的采集Agent执行时间不稳定可能几秒到几十秒波动。如果不加超时控制调度器会一直等在某个慢Agent上整个系统都被拖住。2.3 状态管理与上下文同步调度过程中有一类问题很多人忽略数据流。每个Agent的输入输出以及Agent之间怎么同步直接决定系统稳定性。我的做法是把系统分成两个层面任务状态层和消息传递层。任务状态层用Task对象保存任务ID、状态、输入参数、输出结果、依赖关系。消息传递层让Agent完成任务后把结果写入共享Storage或直接作为下一个任务的输入参数。这里的关键“为什么”是不要依赖Agent自己“记住”别人的产物。LLM的记忆非常不可靠真正可靠的做法是把结果显式写进下一轮的提示词中。我常用一个接地气的说法Agent之间的通信只走快递不走脑电波。让Agent甲“记住刚才的结果然后传给乙”很容易传歪但把甲的输出序列化到JSON再原封不动注入乙的上下文就非常稳。状态管理还要注意并发安全。多个Worker同时写入共享存储时最好每个任务完成后立刻写入独立key不要让多个Agent同时修改同一个对象否则会出现结果互相覆盖的灵异问题。2.4 中断、恢复与人工介入调度系统不可能永远顺风顺水。真实场景里用户可能中途修改需求、某个外部API连续失败、或者某个Agent输出质量实在太差。这时候需要一套中断与恢复机制。我的做法是在调度器里维护一个检查点每完成一个任务就把它的输出结果持久化到本地或数据库。这样整个系统失败后可以从最近的检查点恢复而不是从头再来。人工介入也很重要。我会给调度器加一个“暂停并等待审核”的状态。当评审Agent认为某个结果不合格且自动重试超过阈值时就把任务挂起通知人来决定是继续修改还是跳过。这比让系统死循环重试高效得多。因为有些问题本质上来自任务定义不清晰光靠重试解决不了。3. 完整构建复杂AI协同任务22.7的实操过程3.1 场景定义与Agent角色设计接下来拿一个具体案例做实操演示。假设任务是“构建一份新能源车企竞品分析报告”。这个任务在真实工作里需要多个岗位协作我把它映射成四个Agent规划Agent负责把用户需求拆解成子任务并调度采集Agent负责检索厂商信息、销量数据、公开新闻分析Agent负责对采集到的数据做对比分析提炼指标撰写Agent负责把分析结果写成结构化报告。每个Agent的提示词就是它的岗位说明。比如采集Agent的提示词我会写成你是新能源车企调研员。你的职责是围绕指定厂商收集其主打车型、销量、价格区间、近期核心动作。只输出结构化JSON不要输出无关段落。数据来源需要给出置信度标记。如果数据缺失请在对应字段填写null不要猜测。规划Agent的提示词里则要强调拆解粒度不要一次性拆出几十个小任务建议控制在5到8个子任务每个子任务语义完整否则调度复杂度会指数上升。3.2 编写调度核心任务队列与执行引擎实操环节我用Python写一个极简调度引擎。不用引入复杂框架核心思想很简单把Agent封装成可调用对象把任务封装成数据对象调度器循环扫描就绪任务并执行。from dataclasses import dataclass, field dataclass class Task: id: str agent: str inputs: dict dependencies: list field(default_factorylist) status: str pending class SimpleScheduler: def __init__(self, agents: dict): self.agents agents self.tasks {} self.results {} def add_task(self, task: Task): self.tasks[task.id] task def _ready_tasks(self): ready [] for task in self.tasks.values(): if task.status pending and all(self.tasks[d].status done for d in task.dependencies): ready.append(task) return ready def run(self): while True: ready self._ready_tasks() if not ready: if all(t.status done for t in self.tasks.values()): break raise RuntimeError(no ready task but unresolved dependencies) for task in ready: task.status running agent self.agents[task.agent] result agent.run(**task.inputs) self.results[task.id] result task.status done这段代码已经很能说明调度的骨架。实际生产我会加超时控制、失败重试和并发线程池。比如用ThreadPoolExecutor并行执行无依赖关系的任务并给每个Task单独设置超时时间避免一个慢Agent拖死全场。3.3 关键参数与提示词工程细节提示词写作上我坚持“三段式”结构角色与目标、输入数据区、输出格式区。角色与目标告诉Agent它是谁、要做什么输入数据区显式放入上游Agent输出输出格式区规定JSON或Markdown。我强烈建议在最前面放“你是……”因为角色设定会影响后面所有推理。还有一个容易踩的坑如果不允许Agent说不知道它就会编造数据。所以提示词最后固定加一句如果数据缺失请在对应字段填写null不要猜测。这句话能极大提升系统可信度。参数方面我一般把温度设低0.1到0.3确保调度和格式输出稳定把max_tokens预留充足避免长报告被截断。给不同Agent分配不同模型也很常见采集Agent用速度快的小模型分析Agent用推理强的旗舰模型。3.4 给系统加上失败重试与日志上面那个简单调度器没有异常处理实际只能算是玩具。真实系统一定要加三样东西超时控制、重试机制、完整日志。超时控制保证一个Agent卡住不会拖死全局重试机制让瞬时错误有机会自愈日志记录每轮任务的状态变化便于事后排查。我把日志分成三层系统日志记录调度器自身行为任务日志记录每个Task的状态流转模型日志记录发往Agent的完整messages。这三层日志联合起来基本能定位所有问题。有人觉得日志麻烦但多Agent系统一旦跑起来你会感谢这些记录因为你根本不可能靠“重新跑一次”来复现问题。3.5 运行实测与效果分析在22.7项目里我跑过一版上述系统。任务下发后规划Agent先拆出了7个子任务采集Agent用一轮就把公开信息整合成结构化JSON分析Agent识别出三家厂商的定位差异撰写Agent最后产出一份约2000字的报告。整个流程耗时约90秒大头在采集Agent的外部请求。对比单Agent直接生成多Agent版本的数据一致性更好报告里不再出现前后打架的销量数据。代价是工程复杂度明显上升需要花更多时间处理异常。这个结论并不意外多Agent不是为了“快”而是为了“稳”和“可控”。如果你需要超低延迟多Agent反而可能不适合。4. 常见问题与排查技巧实录4.1 Agent“答非所问”或上下文污染最常见的现象是Agent B收到了Agent A的输出但生成内容却混进了上一轮无关信息。排查时先打印本轮实际发给Agent的messages确认有没有混入旧消息。我的做法是每一轮任务都重新构造上下文只包含该任务需要的输入。不要因为图方便而把历史消息全部保留那会让Agent误以为自己在连续对话最终输出的主题越来越偏。还有一个隐蔽问题如果多个任务共用一个Agent实例而这个Agent内部维护了记忆缓存前一个任务的脏数据会残留。必要时给每个任务初始化一份干净的上下文。4.2 任务死锁与超时控制多Agent任务依赖复杂时容易出现“等待循环”。两个Agent的任务相互依赖就会在DAG里形成环。运行时表现为调度器一直等不到就绪任务。排查方法是记录任务状态流转日志一旦发现所有任务都处于pending但没有任何一个变成ready就检查依赖关系是否成环。Agent执行超时也很常见。外部API偶尔会卡几十秒所以我会给每个Task设置单独超时时间超时后按失败处理并重试。重试次数建议最多2次超过就跳过或通知规划Agent重新调整任务。超时时间需要根据任务类型区分数据采集类可以放宽简单计算类要收紧否则系统响应会变得很慢。4.3 结果校验与格式纠错让Agent输出JSON时模型偶尔会加一段说明或把字段名改掉导致下游解析失败。通用解法是强格式约束提示词加解析兜底。解析失败后不要直接放弃把报错信息拼回提示词告诉模型“上轮输出格式不对请重新生成”。这一招在多数模型上都能救回来。另一个思路是加评审Agent专门检查上一轮输出是否符合要求不合格就打回重做。评审Agent不需要很强推理但提示词要非常严格避免它“顺手帮改内容”只给通过或不通过加原因。4.4 成本控制与资源优化多Agent系统的token消耗通常比单Agent高很多因为任务拆解、结果汇总、评审重试都在消耗token。我控制成本的两个办法一是尽量复用子任务结果比如数据采集类任务可以按天缓存二是给不同Agent分配不同规模的模型简单任务用小参数模型复杂分析用大模型。还有一个容易忽略的优化方向是结果压缩。Agent A返回的原始结果可能有几千字如果直接传给Agent B上下文会被塞满。可以先让A自己生成一份摘要再把摘要传给B。这本质上是用一次小模型调用换取全局token的大幅下降。常见问题速查表问题症状排查思路上下文污染Agent输出混入旧任务内容检查messages是否包含历史消息依赖成环所有任务pending且一直不执行检查DAG是否有环Agent超时整个任务卡住加超时控制设置重试JSON解析失败下游报错强格式约束解析纠错循环结果互相覆盖共享存储数据异常每个任务独立key避免并发写token超标成本骤增结果摘要化、任务缓存、小模型分流5. 再往前走一步评估与自适应5.1 评估指标怎么定多Agent系统不能只看最终输出还要看单次任务之间的稳定性。我给系统设计的基础指标包括任务成功率多少任务一次执行成功、结果一致性同一任务跑两遍的差异度、调度延迟每个节点执行时间、token消耗。有了这些指标你才能知道改动一个提示词或换一个模型到底有没有变好。结果一致性尤其容易被忽略。LLM的抽样特性决定同一提示词每次输出都不同放在多Agent里会被放大。如果系统频繁给出方向不同的结果用户很难信任。我的做法是对关键子任务固定温度接近0甚至固定随机种子对非关键任务保留一定随机性。评估指标可以这样列指标计算方式合理区间任务成功率一次成功任务数 / 总任务数大于0.9结果一致性两次运行结果的语义相似度大于0.8调度延迟子任务平均执行时间视任务而定token消耗每次完整任务的总体token数持续优化5.2 从固定调度到自适应协商如果任务流程经常变化固定DAG就不够用。可以进一步做自适应调度让规划Agent每一轮都根据当前任务状态决定下一步让哪个Agent做什么。这本质上是一种动态规划但代价是每轮都会消耗规划Agent的token而且需要非常强的任务状态总结能力。我的建议是先把固定调度跑稳定再逐步增加动态分支。我最近在尝试的方向是让Agent之间通过结构化消息自主协商“谁来做下一步”。这已经接近分散式协作但我仍然坚持把协商结果落到一个可追踪的任务日志里否则出了错根本没法复盘。协商机制可以用来处理边界情况而不是替代基础调度。最后分享一点个人体会多Agent系统最大的敌人是“我明明每一步都做了但为什么全乱了”的失控感。解决失控的唯一方法就是把一切状态显式化、可观测化。记录所有任务流转缓存所有中间结果让超时和重试自动化。等这些基础都扎实了再谈更复杂的架构。22.7这个项目能跑起来靠的不是精巧算法而是踏踏实实的工程化管理。希望这篇内容能帮你少踩几个坑。