1. 从一台吃灰的机器说起为什么要折腾多 Agent 集群手里有台 24 小时开机的机器可能是家里的旧笔记本也可能是云上的一台常驻实例这件事本身就意味着一种资源浪费——它每天耗着电、占着带宽却只在你偶尔登录的时候干点活。我最初的想法很朴素既然机器不停机那能不能让它替我持续地做一些重复性的开发辅助工作比如代码审查、日志巡检、文档生成、依赖升级检查这类不需要我实时盯着的事情。单 Agent 能做的事情其实很有限。一个 Agent 跑起来它本质上是一个请求-响应的循环你给它一个任务它调用模型、执行工具、返回结果然后结束。如果任务链条长一点比如先拉取仓库、再分析改动、再跑测试、再生成报告单 Agent 很容易在中间某一步卡住或者因为上下文太长而丢失早期信息。更麻烦的是单 Agent 是串行的一个任务没跑完下一个任务只能排队。多 Agent 集群解决的正是这个问题。它的核心思路是把一个大任务拆成若干可以并行、可以互相校验的子任务每个子任务交给一个独立的 Agent 实例去跑Agent 之间通过消息队列或者共享状态来协作。这样一来机器 24 小时不停机这件事就从浪费变成了产能——你可以让三个 Agent 同时跑三条不同的流水线也可以让它们互相 review 彼此的产出。这篇文章适合两类人看。一类是手里有常驻机器、想把它利用起来的开发者另一类是对多 Agent 协作感兴趣、但还没找到落地场景的人。我会把整套集群的搭建过程拆开讲包括机器怎么准备、Agent 怎么隔离、任务怎么分发、失败怎么兜底以及我在实际跑的过程中踩过的那些坑。所有涉及远程连接的部分我都会用 SSH 这条最稳妥的路径来讲因为它是让一台机器随时可被调度的基础。2. 集群的整体设计先想清楚边界再动手写代码2.1 为什么是集群而不是一个更聪明的 Agent很多人第一反应是我能不能写一个超级 Agent让它自己规划、自己拆解、自己执行理论上可以但实际跑下来问题很多。单个 Agent 的上下文窗口是有限的任务链条一长早期的关键信息比如这个仓库的测试命令是 pnpm test 而不是 npm test就会被挤掉。而且单 Agent 一旦在某一步陷入死循环整个流程就卡死了没有冗余。多 Agent 集群的思路完全不同。它不追求单个 Agent 有多聪明而是追求分工明确 互相兜底。我把它类比成一个小的开发团队有人负责写有人负责审有人负责跑测试。每个人能力不需要顶尖但流程清晰、职责分明整体产出反而更稳定。具体到我的设计集群里有三类角色调度 Agent负责接收任务、拆解任务、把子任务分发给执行 Agent并汇总结果。它不直接干活只做编排。执行 Agent真正干活的比如跑代码分析、生成文档、执行测试。每个执行 Agent 跑在独立的进程或容器里互不干扰。审查 Agent对执行 Agent 的产出做二次校验比如检查生成的代码有没有语法错误、文档有没有遗漏关键字段。这三类角色通过一个轻量的消息队列通信。我选的是基于文件系统的队列而不是引入 Redis 或 RabbitMQ原因很简单常驻机器资源有限多一个中间件就多一份运维负担而文件队列在单机场景下足够可靠重启也不丢消息。2.2 任务分发的核心逻辑状态机 心跳集群能不能稳定跑关键看任务分发。我一开始用的是最简单的轮询文件夹方案调度 Agent 往一个目录里写任务文件执行 Agent 定时扫描目录、取走任务、执行完写回结果。这个方案跑了两天就出问题了——某个执行 Agent 卡在一个死循环里任务被它取走后永远不返回调度 Agent 却以为任务还在处理中整个队列就堵住了。后来我改成了状态机 心跳的方案。每个任务在文件里带一个状态字段取值是pending、running、done、failed。执行 Agent 取走任务时把状态改成running同时写一个心跳时间戳。调度 Agent 每隔一段时间扫描所有running状态的任务如果发现某个任务的心跳超过阈值我设的是 5 分钟没更新就判定这个执行 Agent 挂了把任务状态改回pending重新分发。这个改动看起来简单但它是整个集群从玩具变成能长期跑的分水岭。没有心跳机制任何一个 Agent 的偶发卡死都会拖垮整条流水线。2.3 机器与网络SSH 是让集群活起来的血管集群里的 Agent 不一定都跑在同一台机器上。我实际部署时调度 Agent 跑在一台常驻的 Linux 实例上执行 Agent 有一部分跑在同一台机器有一部分跑在我本地的工作站上——因为有些任务需要访问本地才有的代码仓库和凭证。这就引出了 SSH 的作用。SSH 在这里不是远程登录工具这么简单它是让不同机器上的 Agent 能够互相调度、互相传文件的通道。调度 Agent 需要能通过 SSH 连到执行 Agent 所在的机器把任务文件推过去再把结果拉回来。整个过程是自动化的不需要人工登录。这里有个关键点SSH 必须配置成免密登录否则每次调度都要输密码自动化就无从谈起。免密登录靠的是密钥对公钥放在目标机器的~/.ssh/authorized_keys里私钥留在调度机上。我后面会详细讲怎么配以及怎么避免常见的认证失败问题。3. 环境准备把每台机器都变成可被调度的节点3.1 基础环境清单与选型理由在动手之前先把每台机器的基础环境统一。我踩过的最大坑就是不同机器上的环境不一致导致同一个 Agent 脚本在这台能跑、在那台报错。下面是我最终定下来的清单组件选型为什么选它操作系统Ubuntu 22.04 LTS长期支持软件源稳定SSH 默认配置合理运行时Python 3.11Agent 脚本用 Python 写3.11 的异步支持成熟远程通道OpenSSH Server系统自带无需额外安装配置透明任务队列文件系统 JSON零依赖重启不丢调试直观进程管理systemd系统自带开机自启崩溃自动重启日志标准输出 journalctl不引入额外日志系统systemd 直接接管选 Ubuntu 而不是 CentOS 系是因为我在 CentOS 8 上遇到过 SSH 服务默认配置偏严、需要手动调整的情况而 Ubuntu 的默认配置对自动化更友好。当然如果你手上只有 CentOS 8也不是不能用后面我会提一下需要改哪里。Python 3.11 是硬性要求因为我的 Agent 脚本用到了asyncio.TaskGroup这个特性在 3.11 才稳定。如果你机器上是 3.8 或 3.9要么升级要么把脚本里的并发部分改成asyncio.gather。3.2 SSH 免密登录一步步配到能通SSH 免密是整个集群的地基这一步不通后面全白搭。我把它拆成四个动作每台机器都要做一遍。第一步在调度机上生成密钥对。如果已经有~/.ssh/id_ed25519就跳过没有的话执行ssh-keygen -t ed25519 -C agent-cluster -f ~/.ssh/id_ed25519这里选 ed25519 而不是 RSA是因为它密钥更短、签名更快而且现代 OpenSSH 都支持。-C后面跟的是注释写清楚用途方便以后管理多把密钥。第二步把公钥推到目标机器。用ssh-copy-id是最省事的ssh-copy-id -i ~/.ssh/id_ed25519.pub user10.11.225.193这条命令会把公钥追加到目标机器的~/.ssh/authorized_keys。如果目标机器没装ssh-copy-id也可以手动来先cat ~/.ssh/id_ed25519.pub复制内容再登录目标机器把内容粘贴进authorized_keys。第三步验证免密是否生效。ssh -i ~/.ssh/id_ed25519 user10.11.225.193 echo ok如果直接输出ok而不提示输密码说明通了。如果还提示输密码说明公钥没生效往下看排查部分。第四步写进 SSH 配置文件简化后续调用。每次都敲一长串 IP 和密钥路径太累在调度机的~/.ssh/config里加一段Host agent-node-1 HostName 10.11.225.193 User user IdentityFile ~/.ssh/id_ed25519 ServerAliveInterval 30 ServerAliveCountMax 3配好之后ssh agent-node-1就能直接连上。ServerAliveInterval和ServerAliveCountMax这两行很关键它们让 SSH 连接在空闲时定期发心跳避免被中间网络设备掐断——集群里的连接经常是长时间空闲的没有这个配置很容易莫名其妙断线。3.3 目标机器上的 SSH 服务配置光配客户端还不够目标机器的 SSH 服务也得确认几个关键项。打开/etc/ssh/sshd_config检查这几行PubkeyAuthentication yes AuthorizedKeysFile .ssh/authorized_keys PasswordAuthentication noPubkeyAuthentication必须是yes否则公钥认证直接被拒。PasswordAuthentication建议设成no因为集群里所有连接都走密钥关掉密码认证能减少被暴力尝试的风险。改完配置要重启服务sudo systemctl restart ssh注意Ubuntu 上服务名是ssh有些系统上是sshd。如果restart ssh报 the ssh or sshd service is unavailable先systemctl list-units | grep ssh看看实际的服务名是什么。提示改sshd_config之前务必保持一个已经登录的 SSH 会话不要关。万一配置写错导致服务起不来你还能用这个会话改回去。这是我在一台远程机器上把自己锁在外面之后学到的教训。4. Agent 的隔离与编排让每个 Agent 各司其职4.1 用独立进程 独立工作目录做隔离集群里最怕的就是 Agent 之间互相污染。比如执行 Agent A 正在跑一个任务中途改了某个共享文件执行 Agent B 读到的就是脏数据。我的做法是给每个 Agent 分配一个独立的工作目录目录结构长这样/opt/agent-cluster/ ├── scheduler/ │ ├── tasks/ # 待处理任务 │ ├── running/ # 处理中任务 │ └── done/ # 已完成任务 ├── worker-1/ │ ├── workspace/ # 工作目录 │ └── logs/ ├── worker-2/ │ ├── workspace/ │ └── logs/ └── shared/ └── config.json # 共享配置每个 worker 只在自己的workspace里干活任务文件通过调度器在tasks和running之间移动。这样即使某个 worker 把工作目录搞乱了也不会影响别人。进程隔离靠 systemd 的 service 单元。每个 worker 一个 service 文件比如/etc/systemd/system/agent-worker-1.service[Unit] DescriptionAgent Worker 1 Afternetwork.target [Service] Typesimple Useruser WorkingDirectory/opt/agent-cluster/worker-1 ExecStart/usr/bin/python3 /opt/agent-cluster/worker-1/worker.py Restartalways RestartSec5 StandardOutputjournal StandardErrorjournal [Install] WantedBymulti-user.targetRestartalways是关键worker 崩了 systemd 会自动拉起来RestartSec5避免频繁重启刷日志。StandardOutputjournal让日志进 journalctl查起来方便。4.2 调度器的核心循环调度器是整个集群的大脑它的逻辑其实不复杂就是一个循环扫描任务、分发任务、检查心跳、回收超时任务。核心代码大概长这样import json import time import shutil from pathlib import Path TASKS_DIR Path(/opt/agent-cluster/scheduler/tasks) RUNNING_DIR Path(/opt/agent-cluster/scheduler/running) DONE_DIR Path(/opt/agent-cluster/scheduler/done) HEARTBEAT_TIMEOUT 300 # 5 分钟 def dispatch(): for task_file in TASKS_DIR.glob(*.json): task json.loads(task_file.read_text()) # 找一个空闲的 worker worker pick_idle_worker() if worker is None: continue task[status] running task[worker] worker task[heartbeat] time.time() target RUNNING_DIR / task_file.name target.write_text(json.dumps(task)) task_file.unlink() # 通过 SSH 把任务推给 worker push_to_worker(worker, target) def check_heartbeat(): now time.time() for task_file in RUNNING_DIR.glob(*.json): task json.loads(task_file.read_text()) if now - task.get(heartbeat, 0) HEARTBEAT_TIMEOUT: # 超时回收任务 task[status] pending task.pop(worker, None) (TASKS_DIR / task_file.name).write_text(json.dumps(task)) task_file.unlink() print(f任务 {task_file.name} 超时已回收) if __name__ __main__: while True: dispatch() check_heartbeat() time.sleep(10)这段代码里有两个设计点值得说。第一任务文件在tasks、running、done三个目录之间移动而不是原地改状态。这样做的好处是任何时候你ls一下目录就能直观看到有多少任务在排队、多少在处理、多少已完成调试的时候特别省心。第二心跳检查是独立的一轮不和分发混在一起这样即使分发逻辑出问题超时回收依然能正常工作。4.3 通过 SSH 推送任务与拉取结果调度器和 worker 不在同一台机器时任务推送和结果拉取都走 SSH。推送用scp或者rsync都行我选rsync因为它支持增量传输任务文件小的时候差别不大但结果文件可能很大比如生成的报告增量传输能省不少时间。rsync -az -e ssh -i ~/.ssh/id_ed25519 \ /opt/agent-cluster/scheduler/running/task-001.json \ user10.11.225.193:/opt/agent-cluster/worker-1/inbox/拉取结果反过来rsync -az -e ssh -i ~/.ssh/id_ed25519 \ user10.11.225.193:/opt/agent-cluster/worker-1/outbox/task-001-result.json \ /opt/agent-cluster/scheduler/done/这里有个细节rsync的-e参数指定 SSH 命令如果你在~/.ssh/config里配好了agent-node-1可以直接写-e ssh agent-node-1更简洁。注意rsync传输前会先建立 SSH 连接如果目标机器的 SSH 服务没起来或者密钥不对会报认证失败。批量操作的时候建议先写一个健康检查脚本把所有节点的 SSH 连通性跑一遍确认都通了再开始分发任务。5. 让集群真正24 小时不停稳定性与自愈5.1 心跳机制之外还需要哪些兜底心跳解决的是worker 卡死的问题但集群要长期跑还得考虑更多失败场景。我整理了一张表把实际遇到过的故障和对应的兜底策略列出来故障场景表现兜底策略worker 进程崩溃任务永远 runningsystemd 自动重启 心跳超时回收SSH 连接断开任务推送失败重试 3 次间隔递增磁盘写满任务文件写不进去定时清理 done 目录保留最近 7 天模型接口超时任务卡在调用处单次调用设超时超时后任务标记 failed任务死循环CPU 跑满worker 内部设执行时间上限超时自杀机器重启所有进程丢失systemd 开机自启任务状态从文件恢复这张表里的每一条都是我在实际运行中真实遇到过的。尤其是磁盘写满这一条一开始我没在意结果跑了大概两周done目录里堆了几万个结果文件把磁盘占满了新任务全部失败。后来加了一个定时清理脚本只保留最近 7 天的结果问题才解决。5.2 任务重试与幂等性任务失败后重试是必须的但重试有个前提任务必须是幂等的。也就是说同一个任务跑一次和跑两次结果应该一样不能因为重试就产生重复的副作用。我的做法是在任务文件里带一个task_idworker 执行前先检查这个task_id有没有已经处理过。如果处理过直接返回上次的结果不重复执行。这个检查用一个简单的 SQLite 表就能实现import sqlite3 def is_processed(task_id): conn sqlite3.connect(/opt/agent-cluster/shared/processed.db) cur conn.execute(SELECT result FROM processed WHERE task_id ?, (task_id,)) row cur.fetchone() conn.close() return row[0] if row else None def mark_processed(task_id, result): conn sqlite3.connect(/opt/agent-cluster/shared/processed.db) conn.execute(INSERT OR REPLACE INTO processed VALUES (?, ?), (task_id, result)) conn.commit() conn.close()有了幂等性保证重试就可以放心大胆地做。我的重试策略是失败后等 30 秒重试最多重试 3 次3 次都失败就标记为failed进人工处理队列。5.3 日志与可观测性出问题时能快速定位集群跑起来之后最怕的是出问题了但不知道哪里出的。我的经验是日志要满足三个条件能按任务查、能按时间查、能按 worker 查。按任务查靠的是每个任务的所有日志都带task_id前缀。按时间查靠的是 journalctl 自带的时间过滤。按 worker 查靠的是每个 worker 一个独立的 systemd servicejournalctl -u agent-worker-1就能看这个 worker 的全部日志。我常用的几条命令# 看某个 worker 最近的日志 journalctl -u agent-worker-1 -n 100 --no-pager # 看某个时间段所有 worker 的日志 journalctl -u agent-worker-* --since 1 hour ago # 实时跟踪调度器日志 journalctl -u agent-scheduler -f--no-pager这个参数在脚本里调用时很有用否则 journalctl 会尝试用分页器导致脚本卡住。6. 常见问题与排查实录6.1 SSH 认证失败从报错到定位的完整路径SSH 认证失败是集群搭建过程中最高频的问题没有之一。我把遇到过的几种情况和排查方法整理成表报错信息可能原因排查方法Permission denied (publickey)公钥没放到目标机检查~/.ssh/authorized_keys内容Permission denied (publickey)权限不对chmod 700 ~/.ssh chmod 600 ~/.ssh/authorized_keysHost key verification failed目标机指纹变了ssh-keygen -R 10.11.225.193清除旧指纹Connection refusedSSH 服务没起sudo systemctl status sshConnection timed out网络不通或防火墙ping目标机检查防火墙规则No such identity密钥路径写错检查-i参数或 config 里的IdentityFile权限问题特别隐蔽。~/.ssh目录必须是 700authorized_keys必须是 600~/.ssh的属主必须是当前用户。如果权限太开放SSH 会直接拒绝使用这个密钥而且报错信息不会告诉你具体是权限问题只会说Permission denied。我在这上面浪费过整整一个下午。6.2 任务卡住不返回怎么快速定位是哪个环节任务卡住的时候不要急着重启整个集群按下面的顺序排查通常几分钟就能定位看任务状态ls /opt/agent-cluster/scheduler/running/确认任务确实在 running 状态。看心跳时间cat任务文件看heartbeat字段距今多久。超过 5 分钟基本可以判定 worker 卡了。看 worker 日志journalctl -u agent-worker-X -n 50看最后几行停在哪里。看 worker 进程ps aux | grep worker.py确认进程还在不在。看系统资源top或htop确认是不是 CPU 或内存跑满了。大部分情况下卡住的原因是 worker 在等一个永远不会返回的网络请求。我的解决办法是给所有网络调用设超时Python 里用asyncio.wait_for包一层import asyncio async def call_with_timeout(coro, timeout60): try: return await asyncio.wait_for(coro, timeouttimeout) except asyncio.TimeoutError: raise RuntimeError(调用超时)6.3 批量操作时的几个实用技巧集群节点多了之后手动一台台操作效率太低。我总结了几个批量操作的技巧批量执行命令用for循环加 SSHfor host in agent-node-1 agent-node-2 agent-node-3; do echo $host ssh $host systemctl is-active agent-worker done批量传文件用rsync配合循环for host in agent-node-1 agent-node-2; do rsync -az worker.py $host:/opt/agent-cluster/worker/ done批量重启服务for host in agent-node-1 agent-node-2; do ssh $host sudo systemctl restart agent-worker done这些命令看起来简单但要注意一点如果某个节点连不上循环会卡在那里。加个超时参数更稳妥ssh -o ConnectTimeout5 $host commandConnectTimeout5表示 5 秒连不上就放弃避免整个批量操作被一个坏节点拖死。6.4 关于模型接口调用的几个坑集群里的 Agent 最终都要调用模型接口来干活这块也有不少坑。我遇到过的包括接口偶发超时、返回格式不稳定、并发调用被限流。超时的问题前面说了用asyncio.wait_for兜底。返回格式不稳定的问题靠的是严格的解析和校验——不要假设返回一定是合法 JSON先 try 解析解析失败就重试。并发限流的问题靠的是在调度器层面控制并发数不要一次性把所有任务都推出去。我在调度器里加了一个简单的令牌桶限制同时 running 的任务数不超过 worker 数量MAX_CONCURRENT 3 def dispatch(): running_count len(list(RUNNING_DIR.glob(*.json))) if running_count MAX_CONCURRENT: return # ... 继续分发逻辑这个限制看起来保守但它保证了集群不会因为突发任务量而雪崩。任务排队就排队总比全部失败强。7. 跑了一个月之后我的一些真实体会集群稳定跑了一个多月处理了大概几千个任务。回过头看最有价值的不是代码本身而是几个设计上的取舍。第一简单优先。我一开始想过用 Redis 做队列、用 Docker 做隔离、用 Prometheus 做监控后来全砍掉了只留文件队列 systemd journalctl。原因很简单每引入一个组件就多一个可能出故障的地方而单机场景下这些组件的收益远小于它们的运维成本。第二幂等是重试的前提。没有幂等保证的重试本质上是在制造重复副作用。我见过太多人加了重试逻辑结果任务跑了两遍数据写重了反而更难排查。第三SSH 配置要一次做对。免密登录、心跳保活、超时控制这三件事在集群搭建初期花半小时配好后面能省下无数个小时的排查时间。尤其是ServerAliveInterval不配的话长时间空闲的连接会被中间设备悄悄掐断而报错信息往往指向别的地方非常误导。最后分享一个小技巧给每个 worker 的日志加一个启动横幅打印 worker 名称、启动时间、当前加载的配置。这样journalctl里一眼就能看出某个 worker 是什么时候重启的、重启后加载的配置对不对。这个横幅我是在 worker 脚本开头加了一行print(f[worker-1] started at {datetime.now()})成本几乎为零但排查问题时特别有用。
