多智能体协作守护：从监控探测到自动恢复的工程化实践

系统出故障时，很多人的第一反应是登录服务器，找到进程，重启一下。这个动作在单机时代还算有效，但在微服务、容器化、多节点部署成为常态的今天，已经远远不够：你根本不知道故障是从哪个服务开始的，也不知道重启之后它会不会再次崩溃。真正的难点不是“重启”，而是三个问题：谁先发现问题，谁来判断该不该恢复，谁负责执行恢复并确认结果。

“The Caretakers”并不是某一个具体的开源框架，而是一套值得借鉴的多智能体协作守护思路。它的核心隐喻是：系统需要一群各司其职的“看护者”，有人负责巡检，有人负责决策，有人负责执行恢复，有人负责通知和留痕，而不是让一个脚本把所有事情都干完。本文会用一套最小可运行的 Python 参考实现，展示如何把“监控、判断、恢复、通知”拆成多个 Agent 协作完成。

这篇文章适合三类读者：正在做自动化运维平台、想用多 Agent 思路改造现有监控脚本的开发者；对 AI Agent 工程化感兴趣、但不想只看概念的人；以及被“自愈脚本”坑过、想理解恢复动作边界的人。读完你不仅能搭出一个可运行的多 Agent 守护系统，还能避开自动恢复里最危险的几个坑。

1. 为什么“自动守护”需要重新设计

传统的监控自愈脚本通常长这样：一个脚本里同时包含探测逻辑、故障判断、执行恢复命令、发送报警消息。看起来很方便，但随着服务数量增长，问题会逐渐暴露出来。

第一个问题是职责耦合。探测、判断、恢复、通知写在一个脚本里，意味着改报警模板要动恢复逻辑，调探测频率也可能影响恢复动作。代码一旦变多，几乎没人敢改。

第二个问题是缺少判断边界。脚本发现端口不通就重启服务，但端口不通可能是因为上游数据库挂了，也可能是因为负载过高。盲目重启不仅解决不了问题，还可能把现场破坏掉，等真正排查时连日志都找不到。

第三个问题是恢复动作没有约束。一个脚本拥有过高权限、执行没有幂等性、重复触发、缺少审计，这在生产环境非常危险。自动恢复一旦误判，就是一个新的故障源。

多 Agent 化解决的正是这几个问题。把“谁去看”“谁去想”“谁去做”“谁去说”拆开之后，每个角色都可以独立升级、独立测试、独立审计。这也符合工程上的单一职责原则。更重要的是，当你在一个 Agent 体系里加入 AI 判断时，只有职责边界清晰，模型决策才有可能被约束和验证。否则你只是给一个混乱的脚本加了一层不确定性。

下面这张对比表可以更直观地看出差异。

如果你的系统只有一个服务、一个节点，传统的脚本完全够用。当服务数量增长、故障模式变复杂，或者你想要引入 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 系统负责探测它是否存活，如果连续多次探测失败，就自动拉起服务，并把整个过程发给通知渠道。

整个流程可以拆成六个阶段：

1. 探测：Observer 每隔一段时间发起一次健康检查。健康检查必须做“多次确认”，避免一次抖动就触发恢复。
2. 判定：状态管理模块根据连续探测结果，把服务从 HEALTHY 切换到 DEGRADED，再进入 RECOVERING。这个状态转换是做自动恢复的核心，因为不是所有异常都需要立刻处理。
3. 锁定：恢复动作执行前，先申请一个全局恢复锁。这样可以防止多个 Agent 同时执行恢复，避免“恢复风暴”。
4. 动作：Recoverer 执行具体的恢复命令，比如重启进程、清理临时文件。动作必须幂等，也就是重复执行不会产生副作用。
5. 验证：恢复完成之后，重新探测服务是否恢复。如果恢复失败，进入 FAILED 状态并升级通知。
6. 通知与审计：把整条事件链发送给 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定义巡检 Agent，recoverer.py定义恢复 Agent，state.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._running

run_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, timeout=3) 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_output=True, text=True, timeout=30, ) 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_threshold=2) observer = HttpHealthObserver( name="http-health", url="http://127.0.0.1:8080/health", interval=5, ) recoverer = CommandRecoverer( name="http-recoverer", command=["python", "-m", "http.server", "8080"], interval=1, ) scheduler = CaretakerScheduler(state_manager=state_manager) scheduler.add_observer(observer) scheduler.register_recoverer(observer_name="http-health", recoverer=recoverer) print("The Caretakers 守护系统启动") scheduler.loop_forever(interval=1) 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。

验证成功与否，可以关注三个标志：

1. 连续失败达到阈值后是否进入了DEGRADED状态。
2. 调度器是否只触发了一次恢复动作，而不是反复触发。
3. 恢复完成后状态是否回到HEALTHY。

如果运行失败，优先检查 Python 版本、端口占用、以及恢复命令是否正确。不要直接怀疑代码逻辑，多数情况下是环境问题。

7. 常见问题与排查方法

这些坑里，最值得警惕的是“恢复命令执行成功但服务仍不可用”。脚本只做到了“执行了恢复”，没有做到“服务真的恢复”。这也是为什么状态机里必须有验证环节——恢复动作之后，靠下一次探测来确认结果，而不是假设成功。

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 实例之间不互相干扰；把状态机改成更丰富的状态模型，适配更多故障场景。

最后提醒一句：自动恢复系统是“最后一道防线”，不是“第一道防线”。先把监控、告警、人工响应流程梳理清楚，再逐步开放自动恢复，才是比较稳妥的工程节奏。