多Agent集群架构实战：7×24小时任务流处理系统设计

1. 从单兵作战到集群协作我为什么要折腾多 Agent 系统最早接触 Agent 这个概念的时候我和大多数人一样停留在“给大模型套个壳让它能调用几个工具”的认知层面。一个 Agent 就是一个提示词模板加上几个函数调用能查天气、能算数学题、能搜网页感觉已经挺厉害了。但真正把它放到实际业务场景里跑起来问题就暴露了一个 Agent 既要理解用户意图又要规划任务步骤还要调用工具、处理异常、汇总结果上下文越堆越长到了后面它自己都“忘了”前面在干什么。这就像你开了一家一人公司老板是你客服是你财务是你技术也是你。小打小闹还行订单一多你连自己上一秒在做什么都记不清。单 Agent 的瓶颈不在于模型不够聪明而在于角色过载——它被迫在一个上下文窗口里同时扮演太多角色每个角色都需要不同的知识背景、不同的工具集、不同的输出格式混在一起就是一团浆糊。所以我开始琢磨能不能把一个大而全的 Agent 拆成多个小而专的 Agent让它们各司其职、互相配合这就是多 Agent 集群的起点。我打造的这个系统核心目标就一个——让一群 Agent 像一支训练有素的团队一样7×24 小时不间断地处理任务流。它适合那些需要持续处理大量重复性但又有一定认知复杂度任务的场景比如内容审核与分发、数据采集与清洗、客户咨询分级响应、监控告警的初步研判等等。不管你是刚听说 Agent 的新手还是已经写过几个单 Agent Demo 的开发者这套思路都能给你一些可以直接抄作业的参考。2. 整体架构设计多 Agent 集群的骨架怎么搭2.1 为什么是“集群”而不是“流水线”很多人一听到多 Agent第一反应是搭一条流水线Agent A 处理完传给 Agent BB 处理完传给 C像工厂传送带一样。这个思路没错但它只适用于步骤固定、顺序明确的场景。真实业务里更多的是分支、循环、并行、回退——有些任务需要同时查三个数据源再汇总有些任务发现异常要回退到上一步重新处理有些任务需要多个 Agent 投票决定下一步走向。所以我选择的是“集群”模式而非“流水线”模式。集群的核心特征是有一个调度中枢负责任务分发和状态管理多个工作 Agent各自独立运行、通过消息队列通信还有一个监督 Agent负责质量检查和异常处理。这就像一家公司有项目经理、执行团队和质量控制部门而不是一条死板的流水线。具体来说我的架构分为四层接入层负责接收外部任务请求做初步的格式校验和优先级标记然后丢进任务队列。这一层我用了一个轻量的消息队列服务保证任务不丢、可追溯。调度层核心是一个调度 Agent它不直接干活而是根据任务类型、当前各工作 Agent 的负载情况、历史处理成功率决定把任务分配给谁。调度层还维护一个全局状态表记录每个任务当前处于哪个阶段、由哪个 Agent 处理、处理结果是什么。执行层这是真正干活的部分包含多个专业化的工作 Agent。每个 Agent 有自己独立的系统提示词、工具集和上下文窗口。它们从队列里取任务处理完把结果写回状态表然后通知调度层。监督层一个监督 Agent 持续扫描已完成的任务做质量抽检和一致性校验。如果发现异常它会生成一个“返工任务”重新丢回队列并标记原任务的问题类型供后续分析。2.2 Agent 之间的通信协议怎么定多 Agent 系统最容易翻车的地方就是通信。如果 Agent 之间直接互相调用函数耦合度太高加一个新 Agent 就要改一堆代码。我的做法是所有 Agent 只和消息队列以及状态表打交道彼此之间不直接通信。每个任务在状态表里有一个结构化的记录包含任务 ID、当前状态、输入数据、输出数据、处理历史、优先级、超时时间等字段。Agent 从队列里拿到任务 ID 后去状态表读取自己需要的信息处理完再把结果写回去。这样每个 Agent 都是无状态的可以随时重启、扩容、替换不影响其他 Agent。消息格式我用的是 JSON字段设计上遵循一个原则必填字段尽量少扩展字段用命名空间隔离。比如基础字段只有 task_id、task_type、status、payload 四个不同 Agent 特有的字段放在 payload 下面各自的命名空间里。这样新增 Agent 类型时不需要改消息格式定义只需要约定好 payload 的结构就行。注意消息队列一定要设置死信队列和重试上限。我一开始没设重试上限结果一个格式错误的任务被反复重试了几百次把整个队列堵死了。后来改成最多重试 3 次超过就丢进死信队列人工介入系统稳定性立刻上了一个台阶。2.3 7×24 小时运行的关键心跳、熔断与自愈“24 小时不停工作”听起来很酷但实现起来要解决三个问题Agent 挂了怎么办、任务卡住了怎么办、系统过载了怎么办。我的方案是心跳检测 熔断机制 自动重启三件套。每个工作 Agent 每隔 30 秒往一个心跳表里写一条记录调度 Agent 每隔 1 分钟扫描一次心跳表发现某个 Agent 超过 90 秒没有心跳就把它标记为“不健康”暂停向它分发新任务同时触发一个重启流程。重启不是简单的 kill 再 start而是先把该 Agent 正在处理的任务重新放回队列根据状态表判断哪些任务处于“处理中”状态且属于该 Agent然后再重启进程。熔断机制针对的是任务级别。如果某个 Agent 连续处理失败超过阈值我设的是 5 次调度层会暂时把它从可用列表中移除所有新任务分配给其他同类 Agent。同时监督 Agent 会介入分析失败原因如果是提示词问题就自动调整提示词模板如果是工具调用问题就检查工具服务状态。自动重启这块我踩过一个坑一开始用简单的进程守护挂了就拉起来结果发现有些 Agent 是“假死”——进程还在但内部逻辑卡在一个死循环里心跳照发但任务不处理。后来改成心跳里带任务处理计数如果连续三个心跳周期计数没有增长即使心跳正常也判定为假死触发重启。3. 核心 Agent 的角色设计与提示词工程3.1 调度 Agent不做具体事只做决策调度 Agent 是整个集群的大脑但它的工作内容其实很“虚”——它不处理任何具体业务数据只负责看状态表、做决策、发指令。它的提示词设计核心是规则明确、边界清晰。我给调度 Agent 的系统提示词里明确写了它的职责范围只做任务分配、优先级调整、异常任务标记三件事不允许它直接修改任务内容或代替工作 Agent 处理数据。这样做的好处是调度逻辑和工作逻辑解耦调度 Agent 的提示词可以写得很短很精炼决策速度快不容易出错。调度 Agent 的决策依据主要有四个维度任务优先级、任务类型与 Agent 专长的匹配度、各 Agent 当前队列长度、各 Agent 历史成功率。这四个维度加权计算一个分数选分数最高的 Agent 分配任务。权重我调了好几版目前用的是优先级 0.4、匹配度 0.3、负载 0.2、成功率 0.1。优先级权重最高是因为业务上紧急任务必须优先处理成功率权重最低是因为所有 Agent 的成功率都在 95% 以上差异不大。3.2 工作 Agent专业化分工与工具集隔离工作 Agent 是真正干活的我目前部署了五类采集 Agent、清洗 Agent、分析 Agent、生成 Agent、审核 Agent。每一类的提示词、工具集、输出格式要求都不一样。以采集 Agent 为例它的核心能力是调用外部数据源接口、解析返回结果、提取结构化字段。它的工具集里只有 HTTP 请求工具、JSON 解析工具、正则提取工具没有文本生成工具。这样做是故意的——限制 Agent 的能力范围反而能提高它的稳定性和输出一致性。如果给采集 Agent 也配上文本生成能力它有时候会“自作主张”把采集到的原始数据改写一遍导致下游 Agent 拿到的是二手信息。分析 Agent 则相反它的工具集里有数据统计工具、图表生成工具、文本摘要工具但没有外部请求工具。它只处理上游传过来的结构化数据做趋势分析、异常检测、关键信息提取输出一份分析报告给生成 Agent。审核 Agent 比较特殊它不处理业务数据只做质量检查。它的提示词里写满了各种检查规则数据完整性检查、格式合规性检查、逻辑一致性检查、敏感内容检查。它输出的是一个审核结果对象包含通过/不通过标记、问题类型列表、修改建议。3.3 监督 Agent质量兜底与反馈闭环监督 Agent 是我后来才加上的起因是有一次发现生成 Agent 输出的内容里有一处数据引用错误但审核 Agent 没查出来因为审核 Agent 只检查了格式和逻辑没有做事实核查。监督 Agent 的职责就是跨 Agent 的端到端质量抽检。它不检查每一个任务而是按比例随机抽检抽检比例我设的是 10%。抽检时它会重新走一遍关键校验逻辑比如把生成内容里的数据点和原始数据源做比对、把分析结论和原始数据做一致性验证。如果发现问题它会生成一个“返工任务”附带具体的问题描述和修改建议重新丢回队列。监督 Agent 还有一个重要功能是反馈优化。它会定期统计各类问题的出现频率如果发现某一类问题频繁出现就自动调整对应工作 Agent 的提示词模板。比如发现采集 Agent 经常漏掉某个字段就在它的提示词里加一条“务必提取 XX 字段”的强调。这个自动优化机制让系统有了自我进化的能力运行时间越长整体质量越高。实操心得监督 Agent 的抽检比例不要设太高否则会大幅增加系统负载也不要设太低否则起不到兜底作用。我试过 5%、10%、20% 三档10% 是性价比最高的既能发现大部分系统性问题又不会让监督 Agent 成为瓶颈。4. 任务流转与状态管理的实操细节4.1 任务生命周期从入队到归档的完整链路一个任务从进入系统到最终归档会经历多个状态。我把状态定义得很细因为状态越细排查问题时越容易定位。主要状态包括待分配、已分配、处理中、待审核、审核中、待返工、已完成、已归档、已失败。任务入队时状态是“待分配”调度 Agent 分配后变成“已分配”工作 Agent 开始处理时变成“处理中”处理完变成“待审核”审核 Agent 接手后变成“审核中”审核通过变成“已完成”超过保留期后变成“已归档”。如果审核不通过变成“待返工”重新进入分配流程。如果重试超过上限变成“已失败”进入死信队列。每个状态变更都会在状态表里记录时间戳和操作 Agent 的 ID这样任何一个任务出了问题我都能回溯出完整的处理链路。有一次发现某个任务的处理时间异常长查状态表发现它在“处理中”状态停留了 40 分钟进一步查日志发现是采集 Agent 调用外部接口时超时了但超时重试逻辑有 bug导致它一直在等。这个问题如果状态定义不细根本查不出来。4.2 并发控制与资源隔离多 Agent 集群跑起来之后最大的挑战是并发控制。如果同时有几百个任务在跑每个 Agent 都在调外部接口、读写数据库、生成内容资源竞争会非常严重。我的做法是按 Agent 类型做资源隔离。每一类 Agent 有独立的并发上限采集 Agent 最多同时处理 20 个任务因为外部接口有速率限制分析 Agent 最多 50 个计算密集型但不需要外部资源生成 Agent 最多 30 个模型调用有配额限制审核 Agent 最多 100 个轻量级检查。这些上限不是拍脑袋定的是根据实际压测结果调的。压测方法很简单逐步增加并发任务数观察每个 Agent 的处理延迟和错误率。当错误率开始上升或延迟明显增加时就说明接近上限了。我一般会把上限设在压测峰值的 80% 左右留 20% 的余量应对突发流量。资源隔离还体现在数据库连接池上。每个 Agent 类型有独立的连接池互不抢占。这样即使采集 Agent 因为外部接口慢导致连接池满了也不会影响分析 Agent 的正常工作。4.3 超时处理与优雅降级7×24 小时运行的系统超时是常态而不是异常。外部接口会超时、模型调用会超时、数据库查询会超时关键是怎么处理超时。我的策略是分级超时 优雅降级。每个任务有一个总超时时间默认 5 分钟每个处理步骤有各自的步骤超时时间比如采集步骤 60 秒、分析步骤 30 秒、生成步骤 90 秒。如果步骤超时先尝试重试一次重试还超时就跳过这个步骤用默认值或空值填充继续往下走同时在任务记录里标记“降级处理”。这样做的好处是不会因为一个环节卡住导致整个任务失败。比如采集 Agent 调某个数据源超时了重试一次还是不行那就用缓存里的旧数据代替虽然数据新鲜度差一点但任务能继续跑完总比整个任务失败强。当然降级处理的任务会被监督 Agent 重点标记后续人工复核。注意降级处理一定要有明确的标记和告警。我一开始没做告警结果有些任务一直在降级处理数据质量慢慢下滑过了两周才发现。后来加了降级率监控每小时统计一次降级任务占比超过 5% 就发告警。5. 常见问题与排查技巧实录5.1 Agent 之间“踢皮球”怎么办这是多 Agent 系统里最隐蔽也最烦人的问题任务在几个 Agent 之间来回传递每个 Agent 都觉得不是自己的责任状态表里显示任务一直在“处理中”但实际上没有任何 Agent 在真正处理它。我遇到过一次典型场景采集 Agent 把任务标记为“处理中”后发现数据格式不符合预期于是把任务状态改回“待分配”但没有清空自己的处理标记。调度 Agent 看到任务状态是“待分配”又分配给采集 Agent采集 Agent 看到处理标记还在以为是自己正在处理的任务就跳过了。结果任务就在“待分配”和“处理中”之间反复横跳。解决办法是引入任务锁和超时释放机制。任何 Agent 在开始处理任务前必须先获取任务锁在状态表里把 locked_by 字段设为自己的 ID处理完释放锁。如果任务锁超过步骤超时时间还没有释放调度 Agent 会强制释放并重新分配。这样就杜绝了踢皮球的可能性。5.2 上下文膨胀导致 Agent “失忆”工作 Agent 处理长任务时上下文会越来越长到了后面模型开始“忘记”前面的指令输出质量急剧下降。这个问题在生成 Agent 身上特别明显因为它需要参考前面采集和分析的结果来生成最终内容。我的解决方案是上下文分段 关键信息摘要。不让生成 Agent 直接读取完整的采集数据和分析报告而是让分析 Agent 在输出时额外生成一份“关键信息摘要”只包含生成 Agent 真正需要的字段和结论。这样生成 Agent 的上下文长度能控制在 2000 token 以内输出质量稳定很多。另外每个 Agent 的系统提示词里都加了一条“如果发现上下文过长优先关注最近 5 条消息和系统提示词中的核心指令”。这是给模型的一个显式提醒实测能减少“失忆”现象。5.3 模型调用配额耗尽后的应急方案生成 Agent 和分析 Agent 都依赖模型调用而模型调用是有配额限制的。有一次半夜配额用完了所有生成任务全部卡住系统虽然没崩但任务积压了几百个。后来我加了一套配额监控和应急降级机制。系统实时监控模型调用的剩余配额当配额低于 20% 时自动降低生成 Agent 的并发上限优先保证高优先级任务当配额低于 5% 时生成 Agent 切换到“模板模式”用预置的模板加变量填充来生成内容虽然质量不如模型生成但至少能保证任务不积压。等配额恢复后再把积压的降级任务重新跑一遍。下面这张表是我整理的核心问题速查表涵盖了实际运行中遇到的大部分典型问题问题现象可能原因排查方法解决方案任务长时间停留在“处理中”Agent 假死或任务锁未释放检查心跳表和处理计数强制释放锁并重启 Agent任务在多个 Agent 间反复传递状态标记不完整或锁机制缺失查看状态变更历史引入任务锁和超时释放Agent 输出质量突然下降上下文膨胀或提示词被污染检查上下文长度和提示词版本分段处理并加摘要机制模型调用频繁超时配额不足或网络波动查看配额监控和调用日志降级到模板模式或重试系统整体吞吐量下降资源竞争或死信队列积压检查各 Agent 并发数和队列长度调整并发上限并清理死信5.4 日志与可观测性建设多 Agent 系统如果没有好的日志和监控出了问题就是两眼一抹黑。我在这上面投入了不少精力最终搭建了一套三层可观测性体系。第一层是任务级追踪。每个任务有一个全局唯一的 trace_id所有 Agent 处理这个任务时产生的日志都带上这个 trace_id。这样我可以在日志系统里输入一个 trace_id看到这个任务从入队到归档的完整处理链路包括每个 Agent 的输入输出、处理耗时、状态变更。第二层是Agent 级指标。每个 Agent 暴露一组指标当前队列长度、处理中任务数、平均处理耗时、成功率、错误类型分布。这些指标每 10 秒采集一次存到监控系统里可以画趋势图、设告警阈值。第三层是系统级健康度。包括消息队列积压量、死信队列长度、数据库连接池使用率、模型调用配额剩余量等。这些指标反映的是整个系统的运行状态任何一个异常都可能导致系统整体性能下降。实操心得日志里一定要记录 Agent 的“思考过程”也就是它为什么做出某个决策。比如调度 Agent 分配任务时日志里要写清楚它给每个候选 Agent 打了多少分、为什么选了这个。这样出问题时才能知道是决策逻辑有问题还是执行环节有问题。我一开始只记录结果不记录过程排查问题时全靠猜效率极低。6. 性能调优与成本控制的一些实战经验6.1 批量处理与流水线并行单任务逐个处理效率很低因为每个任务都有固定的启动开销读取状态、加载上下文、初始化工具等。我后来引入了批量处理机制调度 Agent 不再一个一个分配任务而是攒一批同类型任务默认 10 个一起分配给工作 Agent。工作 Agent 批量读取状态、批量处理、批量写回固定开销被摊薄了整体吞吐量提升了将近 3 倍。流水线并行是另一个优化点。一个任务在采集阶段时下一个任务可以同时进入分析阶段再下一个任务进入生成阶段。这样不同类型的 Agent 可以同时工作而不是串行等待。实现方式是在状态表里维护一个“就绪队列”每个 Agent 处理完自己的步骤后把任务标记为“下一步就绪”调度 Agent 扫描就绪队列把任务分配给下一阶段的 Agent。6.2 缓存策略哪些数据值得缓存多 Agent 系统里有很多重复计算比如同一个数据源可能被多个任务查询、同一段文本可能被多次分析。合理的缓存策略能大幅降低成本和延迟。我的缓存分三级结果缓存、中间数据缓存、提示词缓存。结果缓存针对的是完全相同的任务请求如果发现新任务和已完成任务的输入完全一致直接返回缓存结果。中间数据缓存针对的是采集 Agent 获取的原始数据同一个数据源在 5 分钟内只请求一次后续任务复用缓存。提示词缓存针对的是系统提示词因为系统提示词很长且不变缓存后能减少模型调用的 token 消耗。缓存失效策略我用的是TTL 主动失效结合。TTL 默认 5 分钟同时当数据源更新时主动清除相关缓存。这样既保证了数据新鲜度又避免了频繁请求。6.3 成本控制哪些钱该花哪些钱能省多 Agent 系统跑起来之后成本主要来自三块模型调用、服务器资源、外部接口调用。我的原则是核心环节用最好的模型辅助环节用便宜甚至免费的方案。生成 Agent 和分析 Agent 是核心环节用的是能力最强的模型因为它们的输出质量直接决定最终结果。采集 Agent 和审核 Agent 是辅助环节采集 Agent 主要做数据抓取和格式转换不需要太强的语言理解能力用轻量模型就够了审核 Agent 做的是规则检查很多检查项可以用正则和规则引擎实现根本不需要调模型。服务器资源方面工作 Agent 用按量付费的实例因为负载波动大调度 Agent 和监督 Agent 用包月实例因为它们需要 7×24 小时稳定运行。外部接口调用尽量用批量接口代替单次接口批量接口的单位成本通常只有单次接口的几分之一。6.4 扩容与缩容什么时候加机器什么时候减系统的负载不是恒定的白天任务多、晚上任务少工作日多、周末少。如果一直保持高配成本浪费如果一直保持低配高峰期又扛不住。我实现了一套基于队列长度的自动扩缩容机制。调度 Agent 每分钟检查一次各类型 Agent 的队列长度如果某个类型的队列长度持续 3 分钟超过阈值比如 50 个待处理任务就自动增加该类型 Agent 的实例数如果队列长度持续 10 分钟低于阈值比如 5 个就减少实例数。扩容上限是初始实例数的 3 倍缩容下限是初始实例数的 1 倍。这套机制运行下来高峰期系统能自动撑住 5 倍于平时的负载低谷期成本能降到平时的 40% 左右。关键是阈值和持续时间要调好太敏感会导致频繁扩缩容浪费资源太迟钝会导致高峰期响应慢。7. 这套系统还能怎么扩展跑了一段时间之后我发现这套多 Agent 集群的架构其实不局限于我最初设计的那些任务类型。只要把工作 Agent 的提示词和工具集换一换就能适配全新的业务场景。比如把采集 Agent 换成“代码拉取 Agent”把分析 Agent 换成“代码审查 Agent”把生成 Agent 换成“修复建议 Agent”就变成了一套自动代码审查系统。再比如把采集 Agent 换成“日志读取 Agent”分析 Agent 换成“异常检测 Agent”生成 Agent 换成“告警通知 Agent”就变成了一套智能运维系统。架构本身是中性的关键在于Agent 的角色定义和它们之间的协作协议。只要把这两样设计好剩下的就是不断调优和迭代。我目前正在尝试的一个方向是让监督 Agent 具备“主动学习”能力——它不仅发现问题还能从问题中总结规律自动生成新的检查规则让整个系统的质量保障能力随着运行时间不断增强。最后分享一个我在实际运行中体会最深的小技巧不要追求一次设计完美而是让系统先跑起来在运行中发现问题、解决问题。我最初版本的调度逻辑只有简单的轮询分配后来才慢慢加上优先级、负载均衡、成功率加权。如果一开始就想着设计一个完美的调度算法可能到现在还没上线。多 Agent 系统的魅力就在于它是一个活的系统你可以在运行中不断调整它、优化它看着它一点点变得更好。