前阵子我把一个多 Agent 集群从实验环境搬到了正式任务里到现在连续跑了一个多月没重启过。这个集群解决的痛点是以前团队成员轮班盯任务、手工调度、反复重跑现在交给一批 Agent 自动协作处理7x24 小时不间断。说白了就是用一套消息驱动 工作流编排的机制把大任务拆成子任务让不同专长的 Agent 各自干活、互相交接、失败自动重试全程不需要人守。这篇文章就聊聊整个系统的设计思路、核心实现、部署参数以及我踩过的坑。适合正在做 AI Agent 落地、或者想把自己的自动化任务升级成“无人值守”状态的开发者参考纯经验复盘没有空话。1. 项目起因与整体设计思路1.1 为什么我会需要一套多 Agent 集群其实一开始我没想搞这么复杂。最初的需求是团队里有一个数据整理与报告生成流水线每天要处理大量来源不同的非结构化信息分类、清洗、提炼关键点、生成固定格式的摘要和日报。人工做的话每人每天大概要花两小时做重复劳动而且人的注意力撑不住持续盯到凌晨。后来我试着用单 Agent 串行跑发现一个问题单个 Agent 一旦遇到长任务或者外部接口慢整个流水线就卡住卡住之后如果没有超时保护和重试机制早上起来一看昨天的任务压根没跑完。我把单 Agent 改成多 Agent 之后本质上不是在堆功能而是在做“分工”和“容错”。每个 Agent 只管一件事出了问题影响面小Agent 之间通过消息队列交换结果天然解耦速度慢的任务也不会堵住后面的环节。所以这套系统的核心思路可以总结成一句话将“一条又长又脆的单线程流水线”改造成“一组各司其职、通过消息通信协作的微服务化 Agent 群”。如果你只是偶尔跑个小脚本没必要上集群但如果你也有“长期、重复、任务量大、不允许半夜崩”的需求多 Agent 集群是值得参考的方向。1.2 集群整体架构与角色划分我落地时把系统分成了四层接入层、调度层、执行层、存储层。接入层负责接收外部任务调度层负责分派和编排执行层由若干角色不同的 Agent 组成存储层负责状态、结果和日志。整个集群里目前跑着六类 Agent主控 Agent负责任务解析、拆解、分配是整个集群的大脑。采集 Agent负责从各个数据源抓取原始内容处理外部接口请求。清洗 Agent负责去重、标准化、剔除无效字段。分析 Agent负责关键信息抽取、文本分类、摘要生成。校验 Agent负责对输出的格式和内容做检查。通知 Agent负责把最终结果推送到群、邮件或归档系统。这六类 Agent 跑的实例数量不一样。采集和分析负载高我各部署了三个实例清洗和校验各两个主控和通知因为逻辑轻量各一个实例就够了。这里想强调一下并不是 Agent 种类越多越好每一种 Agent 都要能说清它存在的理由。校验 Agent 看起来“多此一举”但它帮我拦截了至少 15% 的异常输出后面你会看到它有多重要。2. 多 Agent 协作机制与核心实现2.1 Agent 之间通信到底怎么设计Agent 之间协作最关键的是通信机制。我在早期评估过几个方案直接 HTTP 调用、数据库轮询、消息队列Redis Stream / RabbitMQ以及外部编排工具比如 Prefect、Airflow。直接 HTTP 调用看起来简单但问题很致命如果下游 Agent 临时挂了上游请求直接失败如果下游处理慢上游还要傻等。数据库轮询则效率太低而且容易产生重复读取。我最终选了 Redis Stream 作为消息总线原因有三个延迟低、实现轻量、自带消费组机制可以天然实现负载均衡和断点续读。每个 Agent 启动时会创建一个独立的消费组监听特定主题。比如采集 Agent 处理完原始数据后把结果写入stream:to_cleaner清洗 Agent 从该流消费。消息结构我用 JSON 格式包含任务 ID、来源、时间戳、载荷、优先级等字段。一条消息的典型结构是这样的{ task_id: t_20241105103022_001, source: collector_01, target_queue: to_cleaner, priority: 5, payload: { raw_text: ..., url: ... }, created_at: 2024-11-05T10:30:22Z }每一步处理完成后Agent 会更新 Redis 中的任务状态哈希表这样主控 Agent 可以实时知道任务走到哪一步。这套设计带来的好处很明显凌晨两点、哪个 Agent 崩了都不会阻塞全局队列会把消息攒着Agent 恢复后继续消费。2.2 任务拆分与状态机的实现细节主控 Agent 的核心工作是把一个复杂请求拆成多个子任务。我一开始用硬编码的规则去拆但很快就发现不同任务的逻辑差异太大。后来我换了一种策略就是“模板 字段映射”。每种任务类型定义一组工作流模板模板里包含步骤编号、所需 Agent 类型、是否允许并行、超时时间、失败策略。主控 Agent 只需要识别任务类型然后从配置中心拉取对应模板逐条生成任务消息并推送到不同队列。任务状态我定义了一套简单的状态机pending-dispatching-processing-completed/failed/timeout/retrying。每个状态变更都会写进 Redis 哈希表里并保留变更时间戳方便审计和排障。实现这套状态跟踪时我踩过一个典型坑直接用 Agent 进程里的变量记录状态一旦进程重启状态就丢了。后来把所有状态迁移操作统一封装成一个独立模块调用前先加分布式锁再写 Redis最后更新本地缓存。虽然多了一次网络 IO但换来的是“任何 Agent 挂掉都不会丢失任务上下文”的可靠性。3. 保证 24 小时稳定运行的几道护栏3.1 任务调度与重试机制要让集群 7x24 小时稳定运行最核心的不是功能写得多花哨而是任务调度和失败恢复做得好。我的策略是每个任务消息都会在 Redis 里设置一个 delayed retry延迟重试机制失败的任务不会立刻重试而是进入延迟队列等待一段时间后再重新投递。具体参数我调了几轮最终采用了这样的规则表失败次数延迟时间说明130 秒短暂抖动快速恢复25 分钟可能外部接口暂不可用330 分钟大概率需要人工介入或数据源异常4停止重试标记需人工审核防止无限重试产生垃圾消息这里有个容易被忽略的点一定不要做无限重试。无限重试不仅会拖垮下游依赖还会把 Redis 里的无效消息越积越多最终影响系统性能。我当时对重试队列做了一层“最大生命周期”限制超过 2 小时仍未成功的任务直接转入人工处理列表并触发提醒。对于外部依赖比如第三方 API我单独实现了一个带熔断器的请求封装。连续失败超过 5 次就停止请求 60 秒之后进入半开状态试探请求是否恢复。这一招对防止雪崩非常有效。3.2 心跳检测、故障转移与自愈能力光有重试还不够Agent 进程本身也可能卡死或异常退出。我实现的方案是每个 Agent 进程每隔 15 秒向 Redis 写入一条心跳记录内容包括进程 ID、当前任务 ID、内存占用和最近活动时间。主控 Agent 每隔一分钟扫描一次心跳表如果发现某个 Agent 超过 60 秒没有活动就判定其失联并把它的队列消费组中的待处理消息重新分配到其他健康实例。这里我用的是 Redis Stream 消费组的特性只要消费者组里的消费者没有确认ACK某条消息消息就会一直保留在 Pending Entries List 中。失联的消费者被移除后其他消费者可以通过XAUTOCLAIM接管这些未确认消息。利用这个机制我几乎不用额外开发“任务转移”逻辑天然具备故障恢复能力。我还在部署层加了一道保险每个 Agent 都部署了至少两个副本分别跑在不同的容器里。如果某个副本所在的主机资源耗尽健康检查失败容器编排系统会自动拉起新实例。这是典型的“进程级 消息级”双保险。下面是我用 Python 实现的一个简化版心跳处理逻辑import redis import time import os r redis.Redis(host127.0.0.1, port6379, db0) async def heartbeat(agent_id): while True: r.hset( fheartbeat:{agent_id}, mapping{ ts: time.time(), mem: os.getpgid(0), status: running } ) # 设置过期时间为 45 秒避免长期占用内存 r.expire(fheartbeat:{agent_id}, 45) await asyncio.sleep(15)3.3 监控、日志与告警你必须看得见系统在干嘛一个无人值守系统如果“看不见”出了问题就只能靠猜。我的监控组件分为三层指标层、日志层、告警层。指标层用 Prometheus 采集每个 Agent 的消费速率、处理速率、失败率、队列积压量。日志层用统一的 JSON 格式输出到文件再由日志采集器收集。告警层配置了几条规则比如“失败率超过 10% 持续 5 分钟”、“积压消息数超过 1000 条”、“心跳丢失”等触发后推送到协作群里。我记得有一次凌晨 3 点收到告警提示处理速率降到了 0排查后发现是外部某个数据源接口把我们的 IP 给限流了。如果没有监控这个故障可能会让第二天的日报全部缺失。经验之谈日志一定要带任务 ID。没有任务 ID 的日志等于噪音排查多 Agent 问题时顺着任务 ID 就能把整条链路的每一步都串起来。4. 从零搭建的实操步骤与配置解析4.1 环境准备与依赖安装在正式搭建之前我把环境分成了基础组件和应用层两部分。基础组件包含Python 3.10、Redis 6.2、Docker可选、以及消息处理相关库Redis、JSON、日志库等。如果你不想用 Docker直接在一台 Linux 服务器上跑裸服务也可以但容器化对故障迁移和资源隔离确实方便得多。我实际使用的是 Docker Compose 编排一次性把基础服务拉起来version: 3.9 services: redis: image: redis:7.0-alpine command: redis-server --appendonly yes --maxmemory 512mb --maxmemory-policy allkeys-lru ports: - 6379:6379 volumes: - redisdata:/data collector_a: build: ./agents/collector environment: - AGENT_NAMEcollector_a - REDIS_URLredis://redis:6379/0 depends_on: - redis restart: unless-stopped cleaner_a: build: ./agents/cleaner environment: - AGENT_NAMEcleaner_a - REDIS_URLredis://redis:6379/0 depends_on: - redis restart: unless-stopped volumes: redisdata:这里我踩过一个细节坑Redis 的--maxmemory-policy必须设置成allkeys-lru或者allkeys-random否则长时间运行后内存会缓慢增长。由于我还把任务状态、心跳和临时数据都存在 Redis 中内存策略设置不当运行一周后系统就莫名其妙变卡。4.2 核心配置参数解析与我的调优经验配置这块很容易被低估我逐个说明几个关键参数。Redis Stream 消息长度默认每个流最多保留 1000 条历史消息。对于高频任务1000 条可能撑不了一天导致后续排查历史记录时发现消息已经被裁剪。我把MAXLEN设置成~100000并用近似裁剪方式降低开销。消费组的 消费者 数量同一个消费组里的消费者数量不要超过分区数。因为流的每条消息只能被组内的一个消费者消费消费者数量超过消息分片后多余消费者就会一直空闲。我按队列的实际吞吐量调节消费者数量。任务超时阈值外部接口超时我设置为 20 秒整体任务超时按任务类型设置 515 分钟。不要设成固定值因为不同任务处理时间差异很大固定值要么误杀慢任务要么放水让异常任务长时间占用资源。重试次数见前文表格总原则是“阶梯式延迟 最大重试次数封顶”不要做无脑无限重试。稍微详细一点讲讲监控配置。我用 Prometheus 直接通过端点抓取 Agent 指标每个 Agent 启动时都会启动一个 HTTP 服务暴露/metrics。下面的代码段展示了如何暴露一个核心指标from prometheus_client import Counter, start_http_server PROCESSED_TOTAL Counter(agent_tasks_processed_total, Total processed tasks) FAILED_TOTAL Counter(agent_tasks_failed_total, Total failed tasks) start_http_server(9100) # 在任务处理回调中埋点 PROCESSED_TOTAL.inc()这种指标做完之后一眼就能在 Grafana 上看到整个集群的健康状况。我时常盯着“队列积压”这个指标它比 CPU 占用率更能反映集群的负载状态。4.3 Agent 通用基座与插件式扩展为了方便扩展我把每个 Agent 都写成了一个通用基座程序通过环境变量加载不同的处理模块。基座程序统一处理连接 Redis、消费消息、心跳上报、重试逻辑、日志输出等公共能力只留一个process(payload)接口给业务逻辑实现。一个人工智能 Agent 的核心业务逻辑是这样的import json import asyncio class BaseAgent: async def run(self): while True: msg await self.fetch_message() if msg is None: await asyncio.sleep(2) continue # 更新心跳标记正在处理的任务 await self.update_heartbeat(msg.task_id) try: result await self.process(msg.payload) await self.ack(msg.id) await self.push_to_next(result) except Exception as e: # 记录失败并决定是否重试 await self.handle_failure(msg, e)每类 Agent 只需要继承BaseAgent并实现process()方法就可以快速接入集群。后续加新功能基本不需要动公共代码这也是集群能一直稳定维持下去的重要原因。5. 排障实战与避坑指南5.1 我踩过的几个典型问题先说消费组重复消费的问题。Redis Stream 的消费机制会在消费者异常退出时产生重复投递因此消费端必须具备幂等性。我第一版代码没有做任务 ID 去重结果清洗 Agent 宕机重启后一部分消息被重复处理原本只需要执行一次的分析任务跑了两遍不仅浪费额度还出现了重复日报。解决办法是在每个业务方法开头检查 Redis 里是否存在“任务唯一执行标记”只有第一次处理才写标记并加上过期时间async def process(self, payload): task_id payload[task_id] ok await self.redis.set(fdedup:{task_id}, 1, nxTrue, ex3600) if not ok: # 已经处理过直接跳过 return None # 继续处理 ...其次是死信队列缺失问题。早期系统里所有进入重试且超过最大次数的消息会被直接丢弃。这导致偶尔有几天数据缺失事后还不知道是哪一步丢了。后来我增加了一组死信队列stream:dead_letter所有无法继续处理的消息统一转入这里并附上最终异常日志。每天早上我会手动扫一眼死信队列发现适合恢复的任务就手动重新投递。5.2 问题速查表从症状到对策的快速定位症状可能原因处理方式队列积压持续上升下游 Agent 消费缓慢或挂掉检查心跳查看该 Agent 日志必要时扩容实例任务处理成功后没有下游动作消息未 ACK或目标队列名称配置错误查询目标队列是否存在检查消息体中的目标队列字段Redis 内存持续上涨内存策略设置不当或历史消息无人清理设置maxmemory-policy对历史流做定期裁剪某 Agent 频繁重启资源占用过高或代码异常崩溃查看容器退出码增加内存限制检查日志栈信息重复执行同一任务消费端非幂等或消费者挂掉后重新投递增加任务 ID 去重标记保证业务逻辑幂等外部接口限流导致整体变慢没有熔断机制所有请求同时并发增加熔断器连续失败快速熔断进入半开试探这里有一个很深刻的体会排障时最有效率的手段不是反复看一行行日志而是先看“任务流转状态”。因为多 Agent 系统的故障往往发生在任务从一个环节到另一个环节的交接处而不是 Agent 内部。我一直保留一个类似tracing的表格记录每个任务在每个环节的进入时间、离开时间、处理结果。故障发生时按这个表格从头扫一遍基本都能定位到卡住的环节。5.3 我特别想提醒你的一条经验不要过度设计 Agent 之间的协作协议。早期我为了让 Agent 之间“智能协商”引入了一套复杂的自定义协作协议结果开发周期拉长出问题时又难排查。后来我把这套系统回归“消息队列 状态机”的朴素模型稳定性立刻提升一个档次。Agent 之间只需要约定消息格式、目标队列和 ACK 语义就已经能完成绝大多数自动化协作需求。控制节点根据任务类型和状态去调度而不是让每个 Agent 互相“找活干”这种集中编排模式在维护性和可控性上优势明显。6. 运行效果与我个人的实操体会6.1 实际运行数据集群跑起来之后我之前最担心的“夜间无人值守”场景终于有了底气。现在的运行数据是日均处理任务 4000 到 6000 条平均单任务总耗时 1 分钟以内整个集群的 99% 任务都在目标时间内完成。过去遇到外部接口或网络波动可能一天要人工介入好几次现在基本稳定在每周一两次——那两次通常也是因为数据源那边彻底挂了需要人工换数据源。成本方面因为整体是 Redis 多进程 轻量模型推理没有上重型 GPU 服务一个月基础设施开销很可控。如果你也有类似需求不要一上来就规划大而全的平台先用这种轻量方案跑通业务闭环再根据真实瓶颈去优化性价比会高很多。6.2 我记得的一次凌晨故障复盘有一次晚上 11 点通知 Agent 推送失败率突然升高。我凌晨起来看了一眼监控发现问题出在通知 Agent 的客户端因为令牌过期连接中断。由于我的通知 Agent 当时还没有实现令牌自动刷新就一直在高频重试。那次经历让我彻底意识到任何外部依赖都必须做“有效期管理 自动续期 独立令牌存储”。现在的实现里通知 Agent 启动时会先加载令牌然后在有效期中位线时自动刷新并且把令牌单独存到 Redis避免每个副本各持一份带来不一致。现在类似的故障基本不会再出现了。做完这套多 Agent 集群我最大的体会是真正稳定的无人值守系统不是靠某一个聪明的 Agent而是靠合理的任务拆解、明确的交接语义、完善的心跳与重试机制以及一套能看得到整体的监控。技术上没有什么神秘的黑魔法就是把消息队列、状态机、幂等去重和熔断重试这些基本功扎扎实实落在每个环节上。如果你正打算做类似的东西我建议先从一个业务闭环跑起来加上完整的跟踪日志再逐步扩展 Agent 种类。集群的复杂度涨得很快但每一份复杂度都应该对应真实的业务收益不然宁愿保持单 Agent 串行。
