用 AI Agent 实现 VPS 智能运维：从零搭建自动化监控与自愈系统

VPS 运维中最让人头疼的事情是什么？

不是配置环境，不是部署应用——是半夜被告警吵醒，爬起来手动修复问题。磁盘满了、进程挂了、内存泄漏、SSL 证书过期……这些问题你大概率都遇到过。

但如果你的 VPS 上跑着一个 AI 运维助手呢？它能 7×24 小时监控服务器状态，发现异常后自动诊断问题、执行修复操作，搞不定的再通知你——而你只需要安心睡觉。

这不是科幻。2026 年的开源 AI Agent 生态已经让这一切变得触手可及。本文带你从零搭建一套 AI Agent 驱动的 VPS 智能运维系统。

核心思路：监控 → 诊断 → 决策 → 行动

传统的 VPS 运维告警系统是"通知型"的——发现问题 → 发告警 → 等人来处理。AI Agent 驱动的智能运维则升级为"闭环型"：

服务器指标 → AI Agent 分析 → 诊断结果 → 制定修复方案 → 执行修复 → 验证效果
     ↑                                                              |
     └────────────────────── 未修复则再次诊断 ──────────────────────┘

这个闭环中，AI Agent 扮演了三个角色：

监控分析师：持续观察 CPU、内存、磁盘、网络等指标，识别异常模式
诊断专家：分析日志和系统状态，推理问题根因
运维工程师：执行修复命令、重启服务、清理资源、调整配置

AI Agent 框架选型

2026 年主流的 AI Agent 框架中，以下三个特别适合 VPS 运维场景：

框架	特点	适合场景
CrewAI	多 Agent 协作，角色分工明确	复杂运维流水线
LangGraph	有向图工作流，状态持久化	需要精细化控制的流程
MCP（Model Context Protocol）	标准化工具调用协议	与现有工具链集成

本文选择 CrewAI + MCP 组合方案——CrewAI 负责 Agent 编排，MCP 负责标准化工具调用接口。

第一步：安装 AI Agent 运维系统

环境准备

我假设你已经在 VPS 上安装了 Python 3.11+ 和 Docker。如果还没有，先跑：

# 更新系统
apt update && apt upgrade -y

# 安装 Python 和 pip
apt install -y python3 python3-pip python3-venv

# 安装 Docker（如果你的 VPS 支持）
curl -fsSL https://get.docker.com | sh

创建项目

mkdir -p ~/ai-vps-ops && cd ~/ai-vps-ops
python3 -m venv venv
source venv/bin/activate

# 安装核心依赖
pip install crewai crewai-tools langchain-openai
pip install psutil requests pyyaml

配置 LLM 后端

AI Agent 需要一个 LLM 来推理和决策。你有三种选择：

方案 A：使用本地部署的模型（推荐） 如果你按之前的文章在 VPS 上部署了 DeepSeek V4 或其他开源模型：

# 使用 Ollama 本地模型
export OPENAI_API_BASE=http://localhost:11434/v1
export OPENAI_MODEL_NAME=deepseek-v4:latest

方案 B：使用 API 服务

export OPENAI_API_KEY=sk-your-api-key
export OPENAI_MODEL_NAME=gpt-4o

第二步：构建运维 Agent 系统

下面是一个完整的 CrewAI 运维系统代码。它包含三个 Agent：

系统架构

┌─────────────────────────────────────────┐
│            Monitor Agent                │
│  角色：系统监控员                        │
│  工具：psutil、Docker stats、磁盘检查    │
└──────────────┬──────────────────────────┘
               │ 上报异常指标
               ▼
┌─────────────────────────────────────────┐
│          Diagnosis Agent                │
│  角色：故障诊断专家                      │
│  工具：日志分析、进程分析、资源分析      │
└──────────────┬──────────────────────────┘
               │ 输出诊断结果 + 修复方案
               ▼
┌─────────────────────────────────────────┐
│          Remediation Agent              │
│  角色：自动化运维工程师                  │
│  工具：Shell 执行、服务管理、磁盘清理    │
└─────────────────────────────────────────┘

完整代码

创建文件 ai_vps_ops.py：

#!/usr/bin/env python3
"""
AI VPS Operations System
CrewAI + MCP based intelligent monitoring and self-healing
"""

import os
import json
import subprocess
from datetime import datetime

import psutil

from crewai import Agent, Task, Crew, Process
from crewai_tools import tool

# ── Tool: 获取系统指标 ──────────────────────────────────────────
@tool("Get System Metrics")
def get_system_metrics():
    """获取服务器的关键性能指标：CPU、内存、磁盘、网络"""
    cpu_percent = psutil.cpu_percent(interval=1)
    memory = psutil.virtual_memory()
    disk = psutil.disk_usage('/')
    net = psutil.net_io_counters()

    metrics = {
        "timestamp": datetime.now().isoformat(),
        "cpu": {
            "percent": cpu_percent,
            "cores": psutil.cpu_count(),
            "load_avg": psutil.getloadavg()
        },
        "memory": {
            "total_gb": round(memory.total / (1024**3), 2),
            "used_gb": round(memory.used / (1024**3), 2),
            "percent": memory.percent
        },
        "disk": {
            "total_gb": round(disk.total / (1024**3), 2),
            "used_gb": round(disk.used / (1024**3), 2),
            "percent": disk.percent
        },
        "network": {
            "bytes_sent_mb": round(net.bytes_sent / (1024**2), 2),
            "bytes_recv_mb": round(net.bytes_recv / (1024**2), 2)
        }
    }
    return json.dumps(metrics, indent=2)


# ── Tool: 获取进程列表 ──────────────────────────────────────────
@tool("Get Top Processes")
def get_top_processes(n: int = 10):
    """获取 CPU/内存占用最高的 N 个进程"""
    processes = []
    for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
        try:
            processes.append(proc.info)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass

    by_cpu = sorted(processes, key=lambda p: p.get('cpu_percent', 0) or 0, reverse=True)[:n]
    by_mem = sorted(processes, key=lambda p: p.get('memory_percent', 0) or 0, reverse=True)[:n]

    return json.dumps({
        "top_by_cpu": by_cpu,
        "top_by_memory": by_mem
    }, indent=2)


# ── Tool: 执行 Shell 命令 ───────────────────────────────────────
@tool("Execute Shell Command")
def run_shell_command(command: str):
    """
    执行 Shell 命令并返回输出。
    用于执行运维修复操作：重启服务、清理磁盘、释放内存等。
    """
    try:
        result = subprocess.run(
            command, shell=True,
            capture_output=True, text=True,
            timeout=30
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "return_code": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "Command timed out after 30 seconds"}
    except Exception as e:
        return {"error": str(e)}


# ── Tool: 检查 Docker 容器状态 ─────────────────────────────────
@tool("Check Docker Status")
def check_docker_status():
    """检查 Docker 容器运行状态"""
    try:
        result = subprocess.run(
            "docker ps -a --format '{{.Names}}|{{.Status}}|{{.Image}}'",
            shell=True, capture_output=True, text=True, timeout=10
        )
        if result.returncode != 0:
            return {"error": "Docker not available", "detail": result.stderr}

        containers = []
        for line in result.stdout.strip().split('\n'):
            if not line:
                continue
            parts = line.split('|')
            containers.append({
                "name": parts[0],
                "status": parts[1],
                "image": parts[2]
            })

        # 统计正常运行和异常的容器
        running = [c for c in containers if 'Up' in c['status']]
        unhealthy = [c for c in containers if 'unhealthy' in c['status'] or 'restarting' in c['status']]

        return json.dumps({
            "total": len(containers),
            "running": len(running),
            "unhealthy": len(unhealthy),
            "containers": containers
        }, indent=2)
    except Exception as e:
        return {"error": str(e)}


# ── Tool: 清理系统 ─────────────────────────────────────────────
@tool("Clean System")
def clean_system(target: str = "all"):
    """
    清理系统资源。
    target 可选值：
    - "docker": 清理未使用的 Docker 资源
    - "apt": 清理 APT 缓存
    - "logs": 清理旧日志文件
    - "all": 全部清理
    """
    results = {}

    if target in ("docker", "all"):
        r = subprocess.run(
            "docker system prune -af --volumes 2>&1 | tail -5",
            shell=True, capture_output=True, text=True, timeout=60
        )
        results["docker"] = r.stdout.strip()

    if target in ("apt", "all"):
        r = subprocess.run(
            "apt clean && apt autoremove -y 2>&1 | tail -3",
            shell=True, capture_output=True, text=True, timeout=120
        )
        results["apt"] = r.stdout.strip()

    if target in ("logs", "all"):
        r = subprocess.run(
            "find /var/log -name '*.log.*' -mtime +7 -delete 2>&1; "
            "journalctl --vacuum-time=3d 2>&1 | tail -3",
            shell=True, capture_output=True, text=True, timeout=60
        )
        results["logs"] = r.stdout.strip()

    return json.dumps(results, indent=2)


# ── Create Agents ────────────────────────────────────────────────

monitor_agent = Agent(
    role="系统监控员",
    goal="持续监控 VPS 系统指标，识别异常状态",
    backstory=(
        "你是 VPS 的 7×24 小时监控专家。你负责采集 CPU、内存、磁盘、网络和 Docker 容器指标，"
        "识别出任何超出正常范围的异常信号。你的报告将交给诊断专家进行根因分析。"
    ),
    tools=[get_system_metrics, get_top_processes, check_docker_status],
    verbose=True,
    allow_delegation=False,
)

diagnosis_agent = Agent(
    role="故障诊断专家",
    goal="分析监控数据，定位问题根因，制定修复方案",
    backstory=(
        "你是资深的系统运维诊断专家。你擅长从系统指标和日志中找出问题的真正原因。"
        "你的诊断报告需要明确：问题是什么、严重程度、根因分析、以及推荐修复步骤。"
    ),
    tools=[get_system_metrics, get_top_processes],
    verbose=True,
    allow_delegation=False,
)

remediation_agent = Agent(
    role="自动化运维工程师",
    goal="执行修复操作，恢复系统健康状态，验证修复效果",
    backstory=(
        "你是高效的自动化运维工程师。你根据诊断专家的修复方案执行操作——"
        "重启服务、清理磁盘、释放内存、重启异常容器等。每次操作后你都会验证效果，"
        "如果修复失败会尝试备用方案或升级到人工处理。"
    ),
    tools=[run_shell_command, clean_system],
    verbose=True,
    allow_delegation=False,
)


# ── Create Tasks ──────────────────────────────────────────────────

monitor_task = Task(
    description="""执行一次完整的系统健康检查，包括：
1. CPU 使用率、负载、温度（如果有）
2. 内存使用情况
3. 磁盘使用情况
4. 网络 I/O
5. Docker 容器状态
6. 按 CPU 和内存排序的前 10 个进程

返回详细的指标报告，并标注所有异常项（如 CPU > 80%、内存 > 85% 等）。""",
    expected_output="一份 JSON 格式的系统指标报告，包含所有指标值和异常标记",
    agent=monitor_agent,
)

diagnosis_task = Task(
    description="""分析监控报告中的异常指标：
1. 对每个异常指标进行根因分析（为什么高？什么进程导致的？）
2. 评估问题的严重程度（严重/警告/信息）
3. 制定具体的修复步骤（step by step）
4. 如果有 Docker 容器异常，分析原因

输出诊断报告和修复方案。""",
    expected_output="诊断报告，包含根因分析、严重度评估和分步修复方案",
    agent=diagnosis_agent,
)

remediation_task = Task(
    description="""根据诊断报告的修复方案执行操作：
1. 按步骤执行修复命令
2. 每步操作后检查效果
3. 如果修复成功，记录操作日志
4. 如果修复失败，尝试备用方案
5. 在所有修复操作完成后，重新执行一次健康检查验证效果

最终输出完整的修复报告。""",
    expected_output="修复执行报告，包含执行的操作、结果和最终系统状态",
    agent=remediation_agent,
)


# ── Create Crew ───────────────────────────────────────────────────

ops_crew = Crew(
    agents=[monitor_agent, diagnosis_agent, remediation_agent],
    tasks=[monitor_task, diagnosis_task, remediation_task],
    process=Process.sequential,
    verbose=True,
)


# ── Main ──────────────────────────────────────────────────────────

if __name__ == "__main__":
    print("=" * 60)
    print("🤖 AI VPS Operations System 启动")
    print(f"📅 {datetime.now().isoformat()}")
    print("=" * 60)

    result = ops_crew.kickoff()

    print("\n" + "=" * 60)
    print("✅ 运维任务完成")
    print("=" * 60)
    print(result)

运行系统

cd ~/ai-vps-ops
source venv/bin/activate
python ai_vps_ops.py

第一次运行时，AI Agent 会依次执行监控 → 诊断 → 修复的完整流程。根据你 VPS 的实际状态，它可能会自动处理一些常见问题。

第三步：配置定时自动化

手动跑脚本不算真正的"智能运维"。我们需要把它设置为定时任务，实现无人值守。

设置 Cron 任务

# 编辑 crontab
crontab -e

添加以下规则：

# 每小时执行一次 AI 运维检查
0 * * * * cd /root/ai-vps-ops && /root/ai-vps-ops/venv/bin/python ai_vps_ops.py --quiet >> /var/log/ai-vps-ops.log 2>&1

# 每天凌晨 3 点执行一次深度清理
0 3 * * * cd /root/ai-vps-ops && /root/ai-vps-ops/venv/bin/python ai_vps_ops.py --deep-clean >> /var/log/ai-vps-ops-deep.log 2>&1

配置通知渠道

当 AI Agent 遇到无法自动修复的问题时，应当通知你。有三种常用的通知方式：

方式一：Telegram Bot 通知

@tool("Send Notification")
def send_notification(message: str):
    """通过 Telegram Bot 发送告警通知"""
    bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")
    if not bot_token or not chat_id:
        return "通知未配置"
    
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    data = {"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}
    requests.post(url, json=data)
    return "通知已发送"

方式二：邮件通知

# 安装 msmtp
apt install -y msmtp msmtp-mta

# 配置 ~/.msmtprc
cat > ~/.msmtprc << 'EOF'
defaults
auth           on
tls            on
tls_trust_file /etc/ssl/certs/ca-certificates.crt

account        default
host           smtp.gmail.com
port           587
from           your-email@gmail.com
user           your-email@gmail.com
password       your-app-password
EOF
chmod 600 ~/.msmtprc

方式三：Webhook（适合已有监控系统）

@tool("Send Webhook")
def send_webhook(payload: dict):
    """发送 Webhook 到外部系统"""
    webhook_url = os.environ.get("WEBHOOK_URL")
    if not webhook_url:
        return "Webhook 未配置"
    requests.post(webhook_url, json=payload)
    return "Webhook 已发送"

第四步：拓展——AI Agent 还能做什么？

基础运维系统跑起来后，你还可以为 AI Agent 添加更多能力：

1. SSL 证书管理

AI Agent 可以定期检查 SSL 证书过期时间，在到期前自动续期：

@tool("Check SSL Certificate")
def check_ssl_certificate(domain: str):
    """检查 SSL 证书过期时间"""
    import ssl, socket
    ctx = ssl.create_default_context()
    with ctx.wrap_socket(socket.socket(), server_hostname=domain) as s:
        s.connect((domain, 443))
        cert = s.getpeercert()
        expiry = cert['notAfter']
        days_left = (datetime.strptime(expiry, '%b %d %H:%M:%S %Y %Z') - datetime.now()).days
        return f"{domain} 证书剩余 {days_left} 天"

2. 日志异常检测

AI Agent 可以分析系统日志，发现安全威胁和异常模式：

@tool("Analyze Auth Logs")
def analyze_auth_logs():
    """分析认证日志中的异常登录尝试"""
    import re
    result = subprocess.run(
        "grep 'Failed password' /var/log/auth.log | tail -100",
        shell=True, capture_output=True, text=True, timeout=10
    )
    failed_attempts = len(result.stdout.strip().split('\n'))
    
    if failed_attempts > 50:
        return f"⚠️ 检测到 {failed_attempts} 次失败登录尝试，可能存在暴力破解攻击"
    return f"过去 {failed_attempts} 次失败登录尝试，在正常范围内"

3. 成本优化建议

AI Agent 可以分析 VPS 的资源使用趋势，给出优化建议：

@tool("Resource Trend Analysis")
def resource_trend_analysis():
    """分析资源使用趋势并给出优化建议"""
    # 收集过去 24 小时的 CPU 平均使用率
    # 如果长期低于 20%，建议降配节省成本
    # 如果长期高于 80%，建议升配避免性能问题
    ...

4. 备份状态检查

@tool("Check Backup Status")
def check_backup_status():
    """检查最近备份的状态和完整性"""
    # 检查备份文件的存在性和大小
    # 验证备份时间是否在预期范围内
    # 异常时触发备份
    ...

实战案例：一次完整的 AI 自愈过程

真实场景演示。假设你的 VPS 上跑着几个 Docker 容器，某天某个容器内存泄漏导致 OOM：

Step 1：监控 Agent 发现问题

[监控 Agent]
⚠️ 异常检测：
- 内存使用率: 92%（阈值: 85%）
- Docker 容器 nginx-proxy 状态: "restarting"
- 异常进程: nginx (内存占用: 1.2GB)

Step 2：诊断 Agent 根因分析

[诊断 Agent]
🔍 根因分析：
- 主因: nginx-proxy 容器内存泄漏，触发 OOM Killer
- 影响范围: 该容器服务中断，其他容器正常
- 严重程度: 中等（单容器故障）
- 推荐方案: 
  1. 重启 nginx-proxy 容器
  2. 在 docker-compose.yml 中设置内存限制
  3. 监控重启后内存使用趋势

Step 3：修复 Agent 执行操作

[修复 Agent]
🛠️ 执行修复：
1. ✅ 重启容器: docker restart nginx-proxy
2. ✅ 设置内存限制: docker update --memory 512M nginx-proxy
3. ✅ 验证: 容器状态 "Up 30s"，健康检查通过
4. ✅ 最终系统状态: 内存降至 65%，所有服务正常

整个过程从发现问题到自动修复，用时不到 2 分钟。而你，正在安心睡觉。

安全注意事项

AI Agent 拥有服务器的 Shell 执行权限，这是强大的能力也是巨大的责任。请务必注意：

权限最小化：不要用 root 运行 Agent。创建一个专用的运维用户，只赋予必要的 sudo 权限
命令白名单：在 run_shell_command 工具中可以增加命令白名单校验
沙箱执行：考虑在 Docker 容器中运行 Agent，限制其文件系统访问
人工确认机制：对于高风险操作（如删除数据、重启生产服务），设置人工确认步骤
操作审计：所有 Agent 执行的命令都应该记录日志，方便事后追溯

# 命令白名单示例
SAFE_COMMANDS = [
    "systemctl", "docker", "journalctl", "df", "free",
    "ps", "top", "kill", "rm", "apt", "find", "du"
]

def validate_command(command: str) -> bool:
    """检查命令是否在白名单中"""
    cmd_base = command.strip().split()[0]
    return cmd_base in SAFE_COMMANDS

总结

AI Agent 驱动的智能运维不是未来——它已经来了。借助 CrewAI、LangGraph 和 MCP 等开源框架，你可以在自己的 VPS 上搭建一套 7×24 小时自动运转的运维系统。

这套系统的价值很直观：

减少告警疲劳：常见问题自动修复，只有搞不定的才通知你
缩短故障恢复时间：从人工处理（分钟到小时级）缩短到 AI 自动处理（秒到分钟级）
降低运维成本：一台 VPS 的月费换来 24/7 的"AI 运维工程师"

下一步，你可以把 Hermes Agent 本身也纳入这套系统——让 AI Agent 管理 AI 运维工具。这大概就是"运维的终极形态"了。

想要了解更多 VPS 智能运维技巧？在评论区留言，我会根据大家的需求继续深入某些话题。