容器崩溃很容易被忽略，直到错误持续数小时后才被发现。

在本教程中，Balajee 将向你展示如何构建一个 AI 代理，用于监控 Docker 容器、读取日志、利用 Claude 诊断错误等。

你还将学习如何添加 Slack 告警、速率限制、健康检查端点以及防护措施，以便在生产环境中更安全地运行。

https://freecodecamp.org/news/docker-container-doctor-how-i-built-an-ai-agent-that-monitors-and-fixes-my-containers/

Docker 容器医生：我是如何构建一个 AI 代理来监控和修复我的容器的

来源：https://www.freecodecamp.org/news/docker-container-doctor-how-i-built-an-ai-agent-that-monitors-and-fixes-my-containers/

Docker 容器医生：我是如何构建一个 AI 代理来监控和修复我的容器。也许你遇到过这样的情况：你的生产容器在凌晨 3 点崩溃了。等你醒来时，它已经连续抛出同样的错误两个多小时了。你 SSH 登录，拉取日志，解读晦涩的堆栈跟踪，去 Google 搜索错误，最后重新启动它。整个早上的二十分钟就这么没了。而最糟糕的是？下周又会出现同样的问题。

我厌倦了这个循环。我在一台 Linode 服务器上运行着 5 个容器化服务——一个 Flask API、一个 Postgres 数据库、一个 Nginx 反向代理、一个 Redis 缓存和一个后台 worker。每隔一周，就会有一个崩溃。日志杂乱无章。错误不够明显。我浪费大量时间去调试那些本可以在几秒钟内自动检测并修复的问题。

所以我构建了一个更好的方案：一个 Python 代理，它可以实时监控你的容器，发现错误，使用 Claude 找出问题根源，并在不吵醒你的情况下修复它们。我称之为“容器医生”。这不是什么魔法，而是 Docker API + LLM 推理 + 一些自动化粘合代码的组合。下面我将详细介绍我是如何构建它的，过程中出现了哪些问题，以及我原本会怎么做。

目录

1. 为什么不直接用 Prometheus？
2. 架构
3. 项目设置
4. 监控脚本——逐行详解
5. Claude 诊断提示（以及结构为何重要）
6. 自动修复逻辑——故意保守
7. 添加 Slack 通知
8. 健康检查端点
9. 限制 Claude 调用频次
10. Docker Compose——完整设置
11. 我在生产中捕获的真实错误
12. 成本分析——实际花销
13. 安全考量
14. 我原本会做的改进
15. 下一步计划

为什么不直接用 Prometheus？

问得好。Prometheus、Grafana、DataDog——它们都很棒。但对于我的环境来说，它们有些大材小用了。我在一台每月 20 美元的 Linode 上运行着 5 个容器。搭建 Prometheus 意味着要部署一个指标服务器，为每个服务配置导出器，构建 Grafana 仪表盘，并编写告警规则。这本身就是一个为了监控一个副项目而启动的副项目。

即便如此，这些工具也只会告诉你发生了什么。它们会显示内存飙升或 500 错误率，但不会告诉你为什么。你仍然需要一个人类去查看日志，找出根本原因，并决定如何处理。

这正是我想填补的空白。我不需要另一个仪表盘。我需要一个能读取堆栈跟踪、理解上下文、并直接修复或在你醒来时准确告知你该做什么的东西。Claude 在这方面的表现出奇地好。它读取 Python 回溯并指出问题的速度，比大多数初级开发人员（说实在的，甚至比一些高级开发人员）还要快。

架构

各部分是如何组合在一起的：

┌─────────────────────────────────────────────┐
│                 Docker 主机                   │
│                                               │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │   web    │  │   api    │  │    db    │   │
│  │ (nginx)  │  │ (flask)  │  │(postgres)│   │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘   │
│       │              │              │         │
│       └──────────────┼──────────────┘         │
│                      │                         │
│                 Docker 套接字                   │
│                      │                         │
│            ┌─────────┴─────────┐              │
│            │  Container Doctor │              │
│            │   (Python 代理)    │              │
│            └─────────┬─────────┘              │
│                      │                         │
└──────────────────────┼─────────────────────────┘
                       │
              ┌────────┴────────┐
              │   Claude API    │
              │   (诊断)        │
              └────────┬────────┘
                       │
              ┌────────┴────────┐
              │  Slack Webhook  │
              │   (告警)        │
              └─────────────────┘

工作流程如下：

1. Container Doctor 在自己的容器中运行，并挂载了 Docker 套接字。
2. 每隔 10 秒，它会从每个目标容器拉取最近 50 行日志。
3. 它扫描错误模式（关键词如 “error”、“exception”、“traceback”、“fatal”）。
4. 当发现错误时，它会将日志连同结构化的提示发送给 Claude。
5. Claude 返回一个 JSON 格式的诊断结果：根本原因、严重程度、建议修复方案以及是否可以安全地自动重启。
6. 如果严重性高且自动重启安全，脚本将重启容器。
7. 无论是否重启，它都会通过 Slack 发送包含完整诊断结果的通知。
8. 一个简单的健康检查端点可以让你查看医生自身状态。

关键点在于：脚本本身并不试图进行智能诊断。它将所有思考任务外包给了 Claude。脚本的工作只是管道工作：收集日志，将其路由到 Claude，然后执行响应结果。

项目设置

创建你的项目目录：

mkdir container-doctor && cd container-doctor

这是你的 requirements.txt：

docker==7.0.0
anthropic>=0.28.0
python-dotenv==1.0.0
flask==3.0.0
requests==2.31.0

在本地安装以进行测试：pip install -r requirements.txt

创建一个 .env 文件：

ANTHROPIC_API_KEY=sk-ant-...
TARGET_CONTAINERS=web,api,db
CHECK_INTERVAL=10
LOG_LINES=50
AUTO_FIX=true
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
POSTGRES_USER=user
POSTGRES_PASSWORD=changeme
POSTGRES_DB=mydb
MAX_DIAGNOSES_PER_HOUR=20

关于 CHECK_INTERVAL 的一点说明：10 秒太频繁了。对于生产环境，我会把它提高到 30-60 秒。开发时我设得很低以便更快看到结果，老实说后来忘了改。我的 API 账单提醒了我。

监控脚本——逐行详解

以下是完整的 container_doctor.py。之后我会讲解重要部分：

import docker
import json
import time
import logging
import os
import requests
from datetime import datetime, timedelta
from collections import defaultdict
from threading import Thread
from flask import Flask, jsonify
from anthropic import Anthropic

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

client = Anthropic()
docker_client = None

# --- 配置 ---
TARGET_CONTAINERS = os.getenv("TARGET_CONTAINERS", "").split(",")
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "10"))
LOG_LINES = int(os.getenv("LOG_LINES", "50"))
AUTO_FIX = os.getenv("AUTO_FIX", "true").lower() == "true"
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL", "")
MAX_DIAGNOSES = int(os.getenv("MAX_DIAGNOSES_PER_HOUR", "20"))

# --- 状态跟踪 ---
diagnosis_history = []
fix_history = defaultdict(list)
last_error_seen = {}
rate_limit_counter = defaultdict(int)
rate_limit_reset = datetime.now() + timedelta(hours=1)

app = Flask(__name__)

def get_docker_client():
    """延迟初始化 Docker 客户端。"""
    global docker_client
    if docker_client is None:
        docker_client = docker.from_env()
    return docker_client

def get_container_logs(container_name):
    """从容器获取最后 N 行日志。"""
    try:
        container = get_docker_client().containers.get(container_name)
        logs = container.logs(
            tail=LOG_LINES,
            timestamps=True
        ).decode("utf-8")
        return logs
    except docker.errors.NotFound:
        logger.warning(f"容器 '{container_name}' 未找到。跳过。")
        return None
    except docker.errors.APIError as e:
        logger.error(f"Docker API 错误 {container_name}: {e}")
        return None
    except Exception as e:
        logger.error(f"获取 {container_name} 日志时出现意外错误: {e}")
        return None

def detect_errors(logs):
    """检查日志是否包含错误模式。"""
    error_patterns = [
        "error", "exception", "traceback", "failed", "crash",
        "fatal", "panic", "segmentation fault", "out of memory",
        "killed", "oomkiller", "connection refused", "timeout",
        "permission denied", "no such file", "errno"
    ]
    logs_lower = logs.lower()
    found = []
    for pattern in error_patterns:
        if pattern in logs_lower:
            found.append(pattern)
    return found

def is_new_error(container_name, logs):
    """检查这是否是一个新的错误，还是我们之前已经诊断过的相同错误。"""
    log_hash = hash(logs[-200:])  # 哈希最后 200 个字符
    if last_error_seen.get(container_name) == log_hash:
        return False
    last_error_seen[container_name] = log_hash
    return True

def check_rate_limit():
    """确保我们不会用过多请求骚扰 Claude。"""
    global rate_limit_counter, rate_limit_reset

    now = datetime.now()
    if now > rate_limit_reset:
        rate_limit_counter.clear()
        rate_limit_reset = now + timedelta(hours=1)

    total = sum(rate_limit_counter.values())
    if total >= MAX_DIAGNOSES:
        logger.warning(f"已达到速率限制 ({total}/{MAX_DIAGNOSES} 每小时)。跳过诊断。")
        return False
    return True

def diagnose_with_claude(container_name, logs, error_patterns):
    """将日志发送给 Claude 进行诊断。"""
    if not check_rate_limit():
        return None

    rate_limit_counter[container_name] += 1

    prompt = f"""你是一位 DevOps 专家，正在分析容器日志。

容器: {container_name}
时间戳: {datetime.now().isoformat()}
检测到的模式: {', '.join(error_patterns)}

最近的日志:
---
{logs}
---

分析这些日志，并仅以有效的 JSON 格式回复（不要使用 markdown，不要解释）：
{{
    "root_cause": "一句话准确解释出了什么问题",
    "severity": "low|medium|high",
    "suggested_fix": "操作员应应用的分步修复方案",
    "auto_restart_safe": true 或 false,
    "config_suggestions": ["ENV_VAR=value", "..."],
    "likely_recurring": true 或 false,
    "estimated_impact": "如果不修复，什么会崩溃"
}}
"""

    try:
        message = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=600,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        return message.content[0].text
    except Exception as e:
        logger.error(f"Claude API 错误: {e}")
        return None

def parse_diagnosis(diagnosis_text):
    """从 Claude 的响应中提取 JSON。"""
    if not diagnosis_text:
        return None
    try:
        start = diagnosis_text.find("{")
        end = diagnosis_text.rfind("}") + 1
        if start >= 0 and end > start:
            json_str = diagnosis_text[start:end]
            return json.loads(json_str)
    except json.JSONDecodeError as e:
        logger.error(f"JSON 解析错误: {e}")
        logger.debug(f"原始响应: {diagnosis_text}")
    except Exception as e:
        logger.error(f"解析诊断结果失败: {e}")
    return None

def apply_fix(container_name, diagnosis):
    """如果安全，则应用自动修复。"""
    if not AUTO_FIX:
        logger.info(f"全局禁用自动修复。跳过 {container_name}。")
        return False

    if not diagnosis.get("auto_restart_safe"):
        logger.info(f"Claude 表示对 {container_name} 进行重启不安全。跳过。")
        return False

    # 每小时不重启同一容器超过 3 次
    recent_fixes = [
        t for t in fix_history[container_name]
        if t > datetime.now() - timedelta(hours=1)
    ]
    if len(recent_fixes) >= 3:
        logger.warning(
            f"容器 {container_name} 本小时内已重启 {len(recent_fixes)} 次。情况更严重。跳过。"
        )
        send_slack_alert(
            container_name, diagnosis,
            extra="重复失败：此容器在过去一小时内已重启 3 次以上。需要人工干预。"
        )
        return False

    try:
        container = get_docker_client().containers.get(container_name)
        logger.info(f"正在重启容器 {container_name}...")
        container.restart(timeout=30)
        fix_history[container_name].append(datetime.now())
        logger.info(f"容器 {container_name} 成功重启")

        # 重启后确认它是否真的在运行
        time.sleep(5)
        container.reload()
        if container.status != "running":
            logger.error(f"容器 {container_name} 重启后启动失败")
            return False

        return True
    except Exception as e:
        logger.error(f"重启 {container_name} 失败: {e}")
        return False

def send_slack_alert(container_name, diagnosis, extra=""):
    """将诊断结果发送到 Slack。"""
    if not SLACK_WEBHOOK:
        return

    severity_emoji = {
        "low": "🟡",
        "medium": "🟠",
        "high": "🔴"
    }

    s```