1. 从一台机器到一支队伍多 Agent 集群到底在解决什么问题最早折腾自动化脚本的时候我的思路很朴素写一个 Python 脚本定时跑一遍把该抓的数据抓下来该生成的文件生成出来该提交的代码提交上去。这套东西在任务单一、流程固定的时候确实够用但只要任务一多、分支一复杂脚本就会变成一团乱麻。比如我同时要处理代码审查、文档生成、依赖升级、Issue 分类这四件事如果全塞进一个脚本里任何一个环节卡住整条流水线就停了。更麻烦的是这四件事需要的“思维方式”完全不同——代码审查要严谨文档生成要通俗依赖升级要保守Issue 分类要快。用一个脚本硬扛最后就是什么都做不好。多 Agent 集群的思路本质上就是把“一个人干所有活”变成“一支团队分工干”。每个 Agent 有自己的角色、自己的工具集、自己的上下文窗口彼此之间通过消息或任务队列协作。这样做的好处非常直接单个 Agent 的提示词可以写得很聚焦工具权限可以收得很紧出错时也容易定位是哪个环节的问题。我打造的这个集群核心目标就是让它在没有人盯着的情况下24 小时持续运转把重复性的开发运维工作吃掉我只在关键节点做决策。这套东西适合谁参考如果你已经在用单个自动化脚本处理日常任务但感觉维护成本越来越高或者你手头有多个需要不同“角色”才能处理好的任务那多 Agent 集群就是下一步。它不需要你一开始就上很重的框架用最朴素的消息传递加任务队列也能跑起来。下面我会把整个设计思路、核心细节、实操过程和踩过的坑一层层拆开讲。2. 集群整体设计与角色拆解2.1 为什么是“集群”而不是“一个大 Agent”很多人第一反应是我写一个超级长的提示词让一个 Agent 同时具备代码审查、文档生成、依赖升级的能力不就行了我试过结论是行不通。原因有三个。第一上下文窗口是有限的你把所有角色的规则都塞进去真正干活时留给任务本身的上下文就少了模型容易“忘事”。第二工具权限没法收窄一个 Agent 如果既能改代码又能删文件还能发请求一旦它判断失误破坏面很大。第三调试困难出了问题你根本不知道是哪个角色的逻辑错了。集群模式把这些问题拆开了。每个 Agent 只关心自己那一亩三分地提示词短、工具少、职责清晰。Agent 之间通过一个调度层通信调度层负责把任务分发给合适的 Agent并收集结果。这个调度层可以简单到一个 Redis 队列也可以复杂到带优先级和重试机制的任务系统。我选择的是“轻调度 重角色”的方案调度层只做分发和状态记录所有业务逻辑都在各个 Agent 内部。2.2 我划分了哪几个核心角色经过几轮调整我最终稳定下来的角色有五个。第一个是调度 Agent它不干具体活只负责接收外部事件比如定时触发、Webhook、手动投递然后根据任务类型分发给对应的执行 Agent。第二个是代码 Agent负责代码审查、格式化、简单的重构建议。第三个是文档 Agent负责根据代码变更生成或更新文档。第四个是依赖 Agent负责检查依赖版本、生成升级建议、跑兼容性测试。第五个是巡检 Agent负责定时检查各个 Agent 的健康状态、任务积压情况并在异常时发出告警。这五个角色不是拍脑袋定的而是根据我日常任务的出现频率和相互独立性来划分的。代码和文档经常联动但它们的输出标准完全不同所以必须分开。依赖升级和代码审查有重叠但依赖升级需要跑测试工具链更重也单独拆开。巡检 Agent 是后来加的因为有一次调度队列积压了几百个任务没人发现导致后续任务全部延迟。2.3 通信机制的选择与取舍Agent 之间怎么通信是个关键决策。我试过三种方式直接函数调用、HTTP 接口、消息队列。直接函数调用最简单但耦合太紧一个 Agent 崩了会连带影响调用方。HTTP 接口解耦好一些但需要维护服务发现和重试逻辑对于个人项目来说偏重。最后我选了消息队列具体用的是 Redis 的 List 结构做简易队列配合一个状态 Hash 记录任务进度。为什么是 Redis 而不是更专业的消息中间件因为我的任务量级不大每天几千条消息Redis 完全扛得住而且部署简单一条命令就能跑起来。消息格式我用的是 JSON包含任务 ID、任务类型、载荷、创建时间、重试次数这几个字段。调度 Agent 往队列里推执行 Agent 从队列里拉处理完把结果写到另一个结果队列调度 Agent 再根据结果决定下一步。这个模式的好处是任何一个执行 Agent 挂掉任务不会丢重启后继续从队列里拉就行。注意用 Redis List 做队列时一定要用 BRPOP 而不是 RPOP前者是阻塞式弹出没有任务时会挂起等待不会空转消耗 CPU。这个细节在任务量小的时候看不出来任务量一大空转的代价就很明显。3. 核心细节解析与实操要点3.1 每个 Agent 的提示词怎么写才聚焦提示词是 Agent 的灵魂。我踩过的最大坑就是一开始把提示词写得太“全能”。比如代码 Agent 的提示词里我既让它审查代码风格又让它检查安全漏洞还让它给重构建议。结果它每次输出都很长但真正有用的信息被淹没在废话里。后来我把代码 Agent 拆成两个子角色一个只做风格和规范检查输出格式固定为“文件:行号:问题描述”另一个只做安全扫描输出格式固定为“风险等级:文件:行号:建议”。这样每个 Agent 的输出都是结构化的后续处理起来非常方便。文档 Agent 的提示词重点是“通俗”和“准确”的平衡。我给它定的规则是先读代码变更的 diff再读现有的文档然后只更新受影响的部分不要重写整篇。输出格式要求是 Markdown并且必须保留原有的标题层级。这个约束很重要否则文档 Agent 很容易把一篇结构良好的文档改成流水账。依赖 Agent 的提示词里我明确要求它“保守”。具体来说它只能建议升级到当前大版本内的最新小版本跨大版本升级必须标记为“需要人工确认”。这个规则帮我避免了好几次因为大版本不兼容导致的构建失败。3.2 工具权限的最小化原则每个 Agent 能调用的工具我都做了严格限制。代码 Agent 只能读文件、写文件、执行格式化命令不能执行任意 shell 命令。文档 Agent 只能读代码、读写 Markdown 文件不能碰代码文件。依赖 Agent 可以执行包管理器的查询和安装命令但不能直接改代码。巡检 Agent 只能读状态、发通知不能修改任何业务数据。这个最小化原则看起来麻烦但实际收益很大。有一次代码 Agent 因为提示词里的一个歧义试图执行一个删除临时目录的命令因为权限限制被拦下来了。如果当时给它开了任意 shell 权限后果可能就是误删重要文件。工具权限的配置我建议写在一个独立的配置文件里每个 Agent 启动时加载自己的权限列表这样调整起来也方便。3.3 任务状态机与重试策略任务从创建到完成会经历几个状态待处理、处理中、成功、失败、重试中。我用 Redis Hash 来记录每个任务的状态Key 是任务 IDValue 是状态和更新时间。调度 Agent 在分发任务前会先检查任务状态避免重复分发。执行 Agent 处理完后会更新状态并写入结果。重试策略我设的是最多三次每次间隔翻倍第一次等 10 秒第二次等 30 秒第三次等 90 秒。为什么是翻倍而不是固定间隔因为很多失败是瞬时的比如网络抖动或临时资源不足翻倍等待能给系统恢复的时间。如果三次都失败任务会被标记为“最终失败”并触发告警由我人工介入。实操心得重试次数不要设太多否则一个坏任务会反复占用资源。我一开始设了十次结果一个格式错误的任务在队列里反复横跳把其他任务都堵住了。三次是个比较平衡的值。4. 实操过程与核心环节实现4.1 环境准备与依赖安装整个集群跑在一台常驻的 Linux 小主机上配置不高四核八G但胜在稳定。操作系统用的是 Ubuntu 22.04Python 版本是 3.11。核心依赖只有几个redis用于队列和状态存储requests用于调用模型接口pyyaml用于读配置文件schedule用于定时任务。模型接口我用的是兼容 OpenAI 格式的本地服务这样不依赖外部网络响应也快。安装过程很直接先装 Redis再建一个 Python 虚拟环境把依赖装进去。Redis 的配置我改了两个地方一是把maxmemory设成 512MB防止内存无限增长二是把appendonly打开这样重启后队列里的任务不会丢。Python 这边我用systemd把每个 Agent 做成一个服务这样开机自启崩了也能自动拉起。# 安装 Redis sudo apt update sudo apt install redis-server -y # 修改 Redis 配置 sudo sed -i s/^# maxmemory .*/maxmemory 512mb/ /etc/redis/redis.conf sudo sed -i s/^appendonly no/appendonly yes/ /etc/redis/redis.conf sudo systemctl restart redis-server # 创建虚拟环境并安装依赖 python3 -m venv venv source venv/bin/activate pip install redis requests pyyaml schedule4.2 调度 Agent 的核心逻辑调度 Agent 是整个集群的入口。它启动后做两件事一是监听一个 HTTP 端口接收外部投递的任务二是定时扫描任务队列把待处理的任务分发给对应的执行 Agent。分发逻辑很简单读任务类型查路由表找到对应的队列名把任务推过去。路由表我写在 YAML 文件里长这样routes: code_review: queue: queue:code agent: code_agent doc_update: queue: queue:doc agent: doc_agent dep_check: queue: queue:dep agent: dep_agent health_check: queue: queue:patrol agent: patrol_agent调度 Agent 每 5 秒扫一次待处理任务用 Redis 的SCAN命令遍历状态 Hash找到状态为“待处理”的任务然后根据路由表推送到对应队列。这里有个细节推送到队列和更新状态必须在一个事务里完成否则可能出现任务推了但状态没更新导致重复分发。我用 Redis 的MULTI和EXEC包起来保证原子性。4.3 执行 Agent 的通用骨架每个执行 Agent 的代码结构都差不多区别只在处理函数。骨架是这样的启动后连 Redis然后进入一个循环用BRPOP从自己的队列里拉任务。拉到任务后先更新状态为“处理中”然后调用处理函数处理函数返回结果后把结果写到结果队列更新状态为“成功”。如果处理函数抛异常就更新状态为“重试中”并把重试次数加一如果超过三次就标记为“最终失败”。处理函数的签名统一为def handle(payload: dict) - dict输入是任务载荷输出是结果字典。这样每个 Agent 只需要实现自己的handle函数其他逻辑都是复用的。我写了一个基类BaseAgent把循环、状态更新、异常处理都封装好了子类只需要继承并实现handle就行。import json import redis import time class BaseAgent: def __init__(self, queue_name, result_queue, redis_client): self.queue_name queue_name self.result_queue result_queue self.redis redis_client def run(self): while True: _, raw self.redis.brpop(self.queue_name, timeout5) if raw is None: continue task json.loads(raw) task_id task[id] self.redis.hset(task_status, task_id, processing) try: result self.handle(task[payload]) self.redis.hset(task_status, task_id, success) self.redis.lpush(self.result_queue, json.dumps({ id: task_id, result: result })) except Exception as e: retry task.get(retry, 0) 1 if retry 3: self.redis.hset(task_status, task_id, failed) else: task[retry] retry time.sleep(10 * (2 ** (retry - 1))) self.redis.lpush(self.queue_name, json.dumps(task)) self.redis.hset(task_status, task_id, retrying) def handle(self, payload): raise NotImplementedError4.4 巡检 Agent 与告警机制巡检 Agent 每 60 秒跑一次检查三件事一是各个队列的长度如果某个队列积压超过 100 条就发告警二是各个 Agent 的心跳每个 Agent 每分钟会往一个特定的 Key 写时间戳如果某个 Agent 超过 3 分钟没写就认为它挂了发告警三是任务失败率如果最近 100 个任务的失败率超过 20%也发告警。告警方式我用的是邮件加本地日志。邮件通过 SMTP 发送配置写在 YAML 里。本地日志用 Python 的logging模块按天切分保留 30 天。为什么不用更花哨的告警方式因为个人项目最重要的是可靠和简单邮件虽然老土但不会漏看而且不依赖第三方服务。注意事项心跳 Key 一定要设过期时间比如 5 分钟。否则 Agent 挂了之后心跳 Key 还在巡检 Agent 会误判为正常。我一开始没设过期结果一个 Agent 挂了半天都没发现。5. 常见问题与排查技巧实录5.1 任务重复执行怎么办这是最常见的问题原因通常是状态更新和队列推送没有原子性。比如调度 Agent 推送任务到队列后还没来得及更新状态就崩了重启后扫描到状态还是“待处理”就会再推一次。解决办法有两个一是用 Redis 事务保证原子性二是给任务加唯一 ID执行 Agent 在处理前先检查这个 ID 是否已经处理过。我两个都用了。事务保证大部分情况下不会重复唯一 ID 检查作为兜底。具体做法是执行 Agent 在处理前先往一个 Set 里SADD任务 ID如果返回 0 说明已经处理过直接跳过。这个 Set 也要设过期时间比如 24 小时避免无限增长。5.2 Agent 卡死无响应怎么排查Agent 卡死通常有三个原因一是模型接口超时二是死循环三是 Redis 连接断开。排查顺序是先看日志如果日志停在某一行不动了说明卡在那里再用py-spydump 一下堆栈看看卡在哪个函数最后检查 Redis 连接是否正常。模型接口超时是最常见的我的做法是给所有模型调用加超时默认 30 秒超时就抛异常让任务进入重试。死循环一般出现在处理函数里比如while条件写错了这个只能靠代码审查和单元测试来防。Redis 连接断开的话redis-py默认会自动重连但如果网络彻底断了就需要重启 Agent这个用systemd的自动重启功能兜底。5.3 队列积压怎么快速清理队列积压的原因通常是某个 Agent 处理速度跟不上生产速度。临时解决办法是多开几个同类型的 Agent 实例一起从同一个队列里拉任务。Redis 的BRPOP是支持多消费者的多个实例同时拉任务会被均匀分配。长期解决办法是优化处理函数减少单任务耗时或者把大任务拆成小任务。我遇到过一次依赖 Agent 积压原因是它在跑兼容性测试时每个任务都要装一遍依赖非常慢。后来我改成用一个共享的虚拟环境测试前只更新变化的依赖速度提升了十几倍。这个优化思路就是找出处理函数里最耗时的步骤想办法复用或缓存。5.4 常见问题速查表问题现象可能原因排查方法解决办法任务重复执行状态更新非原子检查 Redis 事务加唯一 ID 检查Agent 卡死模型接口超时看日志、py-spy加超时、自动重启队列积压处理速度慢看队列长度多开实例、优化处理心跳丢失Agent 挂了检查心跳 Key设过期时间、自动拉起任务失败率高提示词歧义看失败任务日志收紧提示词、加约束5.5 我踩过的三个大坑第一个坑是提示词太长导致模型“失忆”。有一次我把代码审查的所有规则都写进一个提示词结果模型在处理长文件时只看了前半部分规则后半部分完全忽略。后来我把规则拆成多个短提示词分多次调用每次只关注一个方面效果就好了很多。第二个坑是 Redis 内存爆了。因为任务结果都往 Redis 里写时间一长内存就满了。后来我改成结果只保留最近 1000 条旧的自动淘汰用LTRIM命令实现。这个细节在任务量小的时候完全想不到但一旦跑起来内存增长是很快的。第三个坑是 Agent 之间的循环依赖。文档 Agent 更新文档后会触发代码 Agent 检查代码注释是否同步代码 Agent 检查完又触发文档 Agent 更新形成死循环。解决办法是给任务加一个“来源”字段如果任务是由另一个 Agent 触发的就不再触发反向任务。这个规则很简单但能避免很多麻烦。6. 集群的扩展与日常维护6.1 怎么加一个新 Agent加新 Agent 的流程很固定第一步在路由表里加一条新路由指定队列名和 Agent 名第二步写一个新的处理类继承BaseAgent实现handle函数第三步写一个systemd服务文件把新 Agent 注册成服务第四步启动服务观察日志确认它正常从队列里拉任务。整个过程最花时间的是写handle函数其他都是模板化的。我建议把新 Agent 的提示词先在小范围测试确认输出格式稳定后再接入集群。否则一个格式不稳定的 Agent 会污染结果队列影响后续处理。6.2 日常维护清单每天花五分钟检查这几项队列长度是否正常、失败任务是否有异常增长、磁盘空间是否充足、日志里是否有反复出现的错误。每周做一次深度检查更新依赖版本、清理旧日志、检查 Redis 内存使用、回顾失败任务的原因分布。每月做一次演练手动停掉一个 Agent看巡检 Agent 是否能正确告警以及自动重启是否生效。这套维护流程看起来繁琐但实际做起来很快因为大部分检查都是看几个数字。关键是养成习惯不要等出了问题才去查。我现在的做法是把这些检查项写成一个脚本每天定时跑结果发到我邮箱我扫一眼就行。6.3 后续可以扩展的方向这个集群目前只覆盖了开发运维的几个环节后续可以往两个方向扩展。一是增加更多角色比如测试 Agent负责根据代码变更自动生成和运行测试用例或者安全 Agent负责定期扫描依赖漏洞。二是增强调度能力比如支持任务优先级让紧急任务插队或者支持任务依赖让某个任务必须等另一个任务完成后才能开始。这两个方向都不需要改动现有架构只需要加新的 Agent 或扩展调度逻辑。这也是集群模式的好处扩展是加法不是改法。你不需要动已经跑通的部分只需要在旁边加新的模块就行。最后分享一个小技巧给每个 Agent 的日志加上统一的 Trace ID这样排查问题时可以把一个任务在所有 Agent 里的日志串起来看。这个 Trace ID 在任务创建时生成随任务一路传递写日志时带上。没有这个 ID 的时候排查跨 Agent 的问题非常痛苦有了之后效率提升很明显。