系统出故障时很多人的第一反应是登录服务器找到进程重启一下。这个动作在单机时代还算有效但在微服务、容器化、多节点部署成为常态的今天已经远远不够你根本不知道故障是从哪个服务开始的也不知道重启之后它会不会再次崩溃。真正的难点不是“重启”而是三个问题谁先发现问题谁来判断该不该恢复谁负责执行恢复并确认结果。“The Caretakers”并不是某一个具体的开源框架而是一套值得借鉴的多智能体协作守护思路。它的核心隐喻是系统需要一群各司其职的“看护者”有人负责巡检有人负责决策有人负责执行恢复有人负责通知和留痕而不是让一个脚本把所有事情都干完。本文会用一套最小可运行的 Python 参考实现展示如何把“监控、判断、恢复、通知”拆成多个 Agent 协作完成。这篇文章适合三类读者正在做自动化运维平台、想用多 Agent 思路改造现有监控脚本的开发者对 AI Agent 工程化感兴趣、但不想只看概念的人以及被“自愈脚本”坑过、想理解恢复动作边界的人。读完你不仅能搭出一个可运行的多 Agent 守护系统还能避开自动恢复里最危险的几个坑。1. 为什么“自动守护”需要重新设计传统的监控自愈脚本通常长这样一个脚本里同时包含探测逻辑、故障判断、执行恢复命令、发送报警消息。看起来很方便但随着服务数量增长问题会逐渐暴露出来。第一个问题是职责耦合。探测、判断、恢复、通知写在一个脚本里意味着改报警模板要动恢复逻辑调探测频率也可能影响恢复动作。代码一旦变多几乎没人敢改。第二个问题是缺少判断边界。脚本发现端口不通就重启服务但端口不通可能是因为上游数据库挂了也可能是因为负载过高。盲目重启不仅解决不了问题还可能把现场破坏掉等真正排查时连日志都找不到。第三个问题是恢复动作没有约束。一个脚本拥有过高权限、执行没有幂等性、重复触发、缺少审计这在生产环境非常危险。自动恢复一旦误判就是一个新的故障源。多 Agent 化解决的正是这几个问题。把“谁去看”“谁去想”“谁去做”“谁去说”拆开之后每个角色都可以独立升级、独立测试、独立审计。这也符合工程上的单一职责原则。更重要的是当你在一个 Agent 体系里加入 AI 判断时只有职责边界清晰模型决策才有可能被约束和验证。否则你只是给一个混乱的脚本加了一层不确定性。下面这张对比表可以更直观地看出差异。维度传统自愈脚本多 Agent 守护系统探测与恢复混在一起分离为不同 Agent故障判断固定 if-else可插拔的策略或模型决策恢复动作直接执行无约束幂等、可回滚、有权限边界通知与审计通常缺失独立 Agent 负责扩展方式复制脚本注册新的 Agent 类型如果你的系统只有一个服务、一个节点传统的脚本完全够用。当服务数量增长、故障模式变复杂或者你想要引入 AI 决策时就需要考虑这套拆分了。2. The Caretakers 的核心概念与适用场景在展开代码之前先明确几个关键概念。理解了这些术语后面的代码才不容易看晕。Caretaker看护者对某个特定指标或服务负责的最小执行单元。它可以是一个巡检探针也可以是一个恢复动作的执行器。比喻来说它就像病房里负责监测体温、换药、记录数据的护士每个人只负责自己那一摊事。Observer观察者负责探测和采集状态只报告“发生了什么”不做恢复决策。例如定时请求一个 HTTP 接口、检查磁盘使用率、读取日志中的错误关键字。Recoverer恢复者负责根据故障类型调用对应的恢复动作。它不关心故障是怎么侦察出来的只关心“这个动作能不能执行、是不是幂等、有没有超出权限”。Notifier通知者负责把故障、恢复动作、最终结果发送到钉钉、企业微信、邮件或 Webhook。它会保留恢复过程中的关键信息形成审计线索。Auditor审计者记录完整的决策链和动作链。每次故障从发现到处理结束发生了什么、谁执行了什么、结果如何都要留痕。这个角色在线上环境非常重要但很多团队会忽略。从设计模式上看The Caretakers 采用了一种“管道-过滤器”的协作方式Observer 产出事件策略层消费事件并给出决策Recoverer 执行动作Notifier 和 Auditor 负责传播和留痕。适用场景比较明确你有多个服务实例需要统一的故障发现与恢复入口。你希望恢复动作是可配置、可审计、可灰度开放的。你想在自愈流程中加入 AI 判断而不是靠一堆 if-else 堆到底。你的团队需要把“监控脚本”升级成“可治理的自动化系统”。不适合的场景也不少如果你只是监控一台服务器、想快速跑一个脚本直接用 systemd 和 crontab 更合适。如果你对自动恢复没有信心、只是想加强报警那优先完善监控和通知链路不要急着上自愈。3. 系统设计与一次故障处理的全流程一个健康的 Caretaker 系统应该围绕一次完整故障生命周期来设计。下面用一个常见的场景说明本地启动了一个 Web 服务Caretaker 系统负责探测它是否存活如果连续多次探测失败就自动拉起服务并把整个过程发给通知渠道。整个流程可以拆成六个阶段探测Observer 每隔一段时间发起一次健康检查。健康检查必须做“多次确认”避免一次抖动就触发恢复。判定状态管理模块根据连续探测结果把服务从 HEALTHY 切换到 DEGRADED再进入 RECOVERING。这个状态转换是做自动恢复的核心因为不是所有异常都需要立刻处理。锁定恢复动作执行前先申请一个全局恢复锁。这样可以防止多个 Agent 同时执行恢复避免“恢复风暴”。动作Recoverer 执行具体的恢复命令比如重启进程、清理临时文件。动作必须幂等也就是重复执行不会产生副作用。验证恢复完成之后重新探测服务是否恢复。如果恢复失败进入 FAILED 状态并升级通知。通知与审计把整条事件链发送给 Notifier同时由 Auditor 写入结构化日志。状态转换是整个系统的核心必须设计得保守一点。我强烈建议不要因为一次探测失败就立刻触发恢复至少连续失败两次再切换状态。这个“冗余确认”能挡掉很多假警报。故障处理过程中最容易被忽略的是“恢复验证”和“失败升级”。很多脚本只负责执行重启不确认重启是否真的有效。一旦恢复失败脚本就“安安静静地失败了”这是最糟糕的体验。The Caretakers 的做法是把验证当作恢复动作的一部分验证不通过就升级状态而不是结束流程。4. 环境准备与项目结构本文的 Demo 代码使用 Python 实现选择它的原因是生态成熟、表达清晰适合用来讲协作模式。版本请以实际项目为准本文重点演示通用思路。为了减少依赖核心代码主要使用标准库仅建议安装 PyYAML 用来读取配置文件。python --version # 建议 Python 3.10 及以上 mkdir caretakers-demo cd caretakers-demo python -m venv venv source venv/bin/activate pip install pyyaml目录结构如下caretakers-demo/ ├── config.yaml ├── caretaker/ │ ├── __init__.py │ ├── base.py │ ├── observer.py │ ├── recoverer.py │ ├── state.py │ └── scheduler.py └── main.py这个结构是很常见的工程划分方式base.py提供 Agent 抽象基类observer.py定义巡检 Agentrecoverer.py定义恢复 Agentstate.py维护状态机scheduler.py负责循环调度。main.py是组装入口。如果你不想用虚拟环境也可以直接全局安装 PyYAML。但考虑到 Python 多版本共存时容易产生依赖冲突还是建议用 venv 隔离。5. 核心代码实现5.1 定义 Agent 抽象基类先建立一个通用的 Agent 基类。它解决一个问题让不同种类的看护者拥有统一的生命周期和运行方式。# 文件路径caretaker/base.py from abc import ABC, abstractmethod class CaretakerAgent(ABC): 所有看护者 Agent 的基类。 def __init__(self, name: str, interval: int 10): self.name name self.interval interval self._running False abstractmethod def run_once(self) - None: 执行一次任务由子类实现。 raise NotImplementedError def start(self) - None: self._running True def stop(self) - None: self._running False property def is_running(self) - bool: return self._runningrun_once是每个 Agent 的核心动作。Observer 在run_once里做健康探测Recoverer 在run_once里执行恢复命令。这样设计的好处是调度器不需要知道 Agent 的具体逻辑只要周期性调用run_once就行。5.2 实现 Observer 探针以 HTTP 健康检查为例。假设服务提供一个/health接口Observer 每隔一段时间请求一次把结果写入共享状态对象。这里用标准库urllib请求接口不引入第三方 HTTP 客户端保证示例最小可运行。# 文件路径caretaker/observer.py import urllib.request import urllib.error from caretaker.base import CaretakerAgent class HttpHealthObserver(CaretakerAgent): 检测 HTTP 服务健康状态的观察者。 def __init__(self, name: str, url: str, interval: int 5): super().__init__(name, interval) self.url url self.state None # 由外部注入共享状态对象 def check(self) - bool: try: with urllib.request.urlopen(self.url, timeout3) as resp: return resp.status 200 except (urllib.error.URLError, TimeoutError, OSError): return False def run_once(self) - None: ok self.check() print(f[{self.name}] 探测结果: {OK if ok else FAIL}) if self.state: self.state.report_health(self.name, ok)run_once里没有恢复逻辑只上报探测结果。这就是 Observer 的职责边界只观察不决策。5.3 实现状态机与故障判定状态机是这套系统里最重要的部分。它要回答一个问题连续几次失败才允许触发恢复这里我设计一个非常简单的状态机只有 HEALTHY、DEGRADED、RECOVERING、FAILED 四种状态。状态转移规则写在report_health和mark_recovering里。# 文件路径caretaker/state.py from enum import Enum class ServiceState(Enum): HEALTHY HEALTHY DEGRADED DEGRADED RECOVERING RECOVERING FAILED FAILED class ServiceStateManager: 根据探测结果维护服务状态。 def __init__(self, fail_threshold: int 2): self.fail_threshold fail_threshold self.concurrent_failures 0 self.state ServiceState.HEALTHY self.last_decision def report_health(self, observer_name: str, ok: bool) - None: if ok: self.concurrent_failures 0 if self.state in (ServiceState.DEGRADED, ServiceState.RECOVERING): self.state ServiceState.HEALTHY self.last_decision f{observer_name} 探测恢复状态回到 HEALTHY print(f[state] {self.last_decision}) return self.concurrent_failures 1 if self.state ServiceState.HEALTHY and self.concurrent_failures self.fail_threshold: self.state ServiceState.DEGRADED self.last_decision ( f{observer_name} 连续 {self.concurrent_failures} 次失败进入 DEGRADED ) print(f[state] {self.last_decision}) def mark_recovering(self) - None: self.state ServiceState.RECOVERING self.last_decision 开始执行恢复动作状态进入 RECOVERING print(f[state] {self.last_decision}) def mark_failed(self) - None: self.state ServiceState.FAILED self.last_decision 恢复验证失败状态进入 FAILED print(f[state] {self.last_decision})这段代码的关键在于fail_threshold。它把“一次探测失败”和“需要恢复”隔离开来。网络抖动、服务启动慢、负载瞬时升高这些情况不应该触发自动恢复。5.4 实现 Recoverer 恢复执行器恢复动作必须满足两个约束幂等和可控。下面的示例里恢复动作是执行一条重启命令。真实项目中这里的动作应该对应你的服务管理方式例如 systemctl restart、docker compose restart 或调用编排平台的 API。# 文件路径caretaker/recoverer.py import subprocess from caretaker.base import CaretakerAgent class CommandRecoverer(CaretakerAgent): 执行恢复命令的恢复者。 def __init__( self, name: str, command: list, expected_returncode: int 0, interval: int 1, ): super().__init__(name, interval) self.command command self.expected_returncode expected_returncode def execute(self) - bool: print(f[{self.name}] 执行恢复命令: { .join(self.command)}) try: proc subprocess.run( self.command, capture_outputTrue, textTrue, timeout30, ) if proc.returncode self.expected_returncode: print(f[{self.name}] 恢复命令执行成功) return True print(f[{self.name}] 恢复命令执行失败: {proc.stderr.strip()}) return False except (subprocess.TimeoutExpired, OSError) as exc: print(f[{self.name}] 恢复命令异常: {exc}) return False def run_once(self) - None: # Recoverer 由调度器显式调用通常不需要自循环 return注意一个细节execute()是真正供外部调用的方法run_once()只是占位。这样设计是为了让调度器可以“按需”触发恢复动作而不是每隔固定时间盲目执行。5.5 实现调度器与主程序调度器负责把 Observer、StateManager、Recoverer 串起来。为了让示例可控这里不引入复杂的异步框架而是用循环和time.sleep模拟定时调度。# 文件路径caretaker/scheduler.py import time from caretaker.state import ServiceStateManager class CaretakerScheduler: 协调 Observer 和 Recoverer 的调度器。 def __init__(self, state_manager: ServiceStateManager): self.observers [] self.recoverers {} self.state_manager state_manager self.recover_lock False def add_observer(self, observer): observer.state self.state_manager self.observers.append(observer) return self def register_recoverer(self, observer_name: str, recoverer): self.recoverers[observer_name] recoverer return self def run_once(self) - None: for observer in self.observers: observer.run_once() if self.state_manager.state.value DEGRADED: self._trigger_recovery() def _trigger_recovery(self) - None: if self.recover_lock: print([scheduler] 已有恢复任务在执行跳过本次触发) return self.recover_lock True try: self.state_manager.mark_recovering() # 默认使用第一个恢复器真实场景可按 observer 或故障类型路由 recoverer_name list(self.recoverers.keys())[0] recoverer self.recoverers[recoverer_name] ok recoverer.execute() if ok: # 等待服务起来再由下一次探测确认 print([scheduler] 恢复动作完成等待下一次探测确认) else: self.state_manager.mark_failed() finally: self.recover_lock False def loop_forever(self, interval: int 1) - None: print([scheduler] 开始调度循环) while True: self.run_once() time.sleep(interval)调度器中加了一个recover_lock布尔值这是为了防止恢复动作重复触发。如果恢复命令本身执行时间较长而调度循环很快就可能出现“上一次还没跑完下一次又触发”的问题。实际项目中建议用分布式锁例如 Redis 锁或数据库锁。接下来是主程序它把各个 Agent 组装起来。这个文件也是你根据实际场景修改的地方。# 文件路径main.py from caretaker.observer import HttpHealthObserver from caretaker.recoverer import CommandRecoverer from caretaker.state import ServiceStateManager from caretaker.scheduler import CaretakerScheduler def main(): state_manager ServiceStateManager(fail_threshold2) observer HttpHealthObserver( namehttp-health, urlhttp://127.0.0.1:8080/health, interval5, ) recoverer CommandRecoverer( namehttp-recoverer, command[python, -m, http.server, 8080], interval1, ) scheduler CaretakerScheduler(state_managerstate_manager) scheduler.add_observer(observer) scheduler.register_recoverer(observer_namehttp-health, recovererrecoverer) print(The Caretakers 守护系统启动) scheduler.loop_forever(interval1) if __name__ __main__: main()这个主程序里有一个需要注意的点恢复命令是启动一个新的 HTTP 服务进程。但在实际项目中恢复命令通常应该保证同一个服务的唯一性例如先停机再启动否则可能启动多个实例反而造成端口冲突。更好的方式是封装一个restart_http_server.sh脚本。5.6 配置外部化把关键参数放在config.yaml里可以避免每次修改行为都去改代码。下面是一个简单的配置结构。# 文件路径config.yaml observers: - name: http-health type: http url: http://127.0.0.1:8080/health interval: 5 recoverers: - name: http-recoverer type: command command: [python, -m, http.server, 8080] state: fail_threshold: 2实际落地时建议使用配置中心或环境变量注入这些参数尤其是恢复命令和阈值。这样运维同学调整参数时不需要碰代码也方便做不同环境的差异化管理。6. 运行与效果验证现在把代码跑起来。首先启动一个可以被探测的 HTTP 服务。为了方便演示使用 Python 自带的 HTTP 服务但它没有/health接口会被探测为 404。为了让示例更真实可以先用一个简单的 Flask 服务或者直接调整 Observer 的判定逻辑把 404 也视为“不健康”。一个更轻量的做法是先用下面的命令启动一个正常的 HTTP 服务python -m http.server 8080然后启动 Caretaker 系统python main.py如果服务是正常的你会看到类似输出The Caretakers 守护系统启动 [state] http-health 探测结果: OK [scheduler] 开始调度循环 [http-health] 探测结果: OK要模拟故障可以把 HTTP 服务停掉再观察系统输出[http-health] 探测结果: FAIL [http-health] 探测结果: FAIL [state] http-health 连续 2 次失败进入 DEGRADED [scheduler] 已有恢复任务在执行跳过本次触发 [scheduler] 恢复动作完成等待下一次探测确认看到DEGRADED和恢复动作日志就说明状态机生效了。如果服务被重新拉起下一次探测会输出OK状态回到HEALTHY。验证成功与否可以关注三个标志连续失败达到阈值后是否进入了DEGRADED状态。调度器是否只触发了一次恢复动作而不是反复触发。恢复完成后状态是否回到HEALTHY。如果运行失败优先检查 Python 版本、端口占用、以及恢复命令是否正确。不要直接怀疑代码逻辑多数情况下是环境问题。7. 常见问题与排查方法问题现象可能原因排查方式解决方案探测一直返回 FAIL服务未启动或 URL 不正确手动检查接口状态和网络连通性先确保服务可访问再启动守护系统连续失败后没有进入 DEGRADEDfail_threshold 设置过高或状态未正确注入打印 state 对象检查 report_health 调用次数降低阈值或检查 observer.state 是否被注入恢复命令执行成功但服务仍不可用启动命令挂了或端口被占手动执行恢复命令观察进程是否存活封装更完善的启动脚本确保先清理旧进程调度器反复触发恢复缺少恢复锁或恢复时间过长查看日志中 recover_lock 状态使用分布式锁延长恢复冷却时间服务恢复后状态没有回到 HEALTHYObserver 没有继续运行或状态被重置确认 observer 仍在调用 report_health检查调度循环是否异常退出自动恢复误操作失败判定条件过于宽松回看探测日志和策略配置增加连续失败次数、增加人工确认开关这些坑里最值得警惕的是“恢复命令执行成功但服务仍不可用”。脚本只做到了“执行了恢复”没有做到“服务真的恢复”。这也是为什么状态机里必须有验证环节——恢复动作之后靠下一次探测来确认结果而不是假设成功。8. 工程化最佳实践Demo 可以快速跑通但放到生产环境前还有不少工作需要补上。下面这些实践来自实际运维和自动化项目中比较常见的思路按重要性排列。8.1 最小权限与命令白名单Caretaker 的 Recoverer 不能随便执行外部传入的命令。生产环境中恢复命令应该是一个固定的白名单而不是把用户输入直接拼进 shell。否则一旦配置泄露攻击者相当于拿到了一台机器的执行权限。建议把恢复命令抽象成“预定义动作”例如restart_service: nginx、clean_tmp: /data/tmp而不是任意命令。8.2 幂等性与恢复冷却恢复动作必须做到幂等。无论是重复执行还是并发执行结果都应该一致。最简单的方式是加恢复冷却时间一次故障处理完成后至少等待一段时间才能触发下一次恢复。否则服务在一个不稳定的状态下反复重启反而可能造成更严重的破坏。8.3 通知与升级机制不是所有故障都适合自动恢复。建议把恢复结果分级如果自动恢复成功走普通通知如果自动恢复失败立即升级到人工处理并附上完整的故障链路信息。这个升级通道应该独立于自动恢复逻辑保证可靠送达。8.4 审计日志所有决策和动作都应该写到独立的审计日志中。内容包括谁发现故障、状态如何变化、执行了什么命令、执行结果如何、耗时多少。审计日志不是为了追责而是为了让后续优化有据可查。没有审计的自动恢复本质上是个黑盒。8.5 灰度开放先让 Caretaker 只观察、不恢复运行一段时间确认状态判定准确。然后再对非核心服务开放自动恢复最后才扩展到核心服务。这是比较稳妥的推进路径。8.6 与监控平台联动The Caretakers 不应该替代 Prometheus、Zabbix 等监控平台。更合理的架构是监控平台负责指标采集和可视化Caretaker 负责事件消费和恢复动作。两者通过 Webhook 或消息队列对接避免重复建设。9. 总结与后续学习方向这篇文章从“自动守护”的痛点出发用一套名为 The Caretakers 的最小参考实现展示了如何把故障发现、状态判定、恢复执行和审计通知拆分成多个 Agent 协作完成。核心并不是代码本身而是三个判断一次探测失败不等于需要恢复连续失败进入 DEGRADED 才值得处理恢复动作必须幂等、可控、有验证所有的自动决策都要留痕失败时要能升级到人工。如果你想继续深入可以从这几个方向入手在 Recoverer 中接入大模型让它根据故障信息选择恢复策略但这需要严格的约束和沙箱用消息队列替换循环调度把 Observer 分散到不同节点上加入分布式锁让多个 Caretaker 实例之间不互相干扰把状态机改成更丰富的状态模型适配更多故障场景。最后提醒一句自动恢复系统是“最后一道防线”不是“第一道防线”。先把监控、告警、人工响应流程梳理清楚再逐步开放自动恢复才是比较稳妥的工程节奏。