AI Agent 驱动的 VPS 智能运维：从被动告警到主动治理

引言

你管理着十几台 VPS，运行着各种 Docker 容器、网站和服务。日常运维中，你是否经历过这样的场景？

用户反馈网站打不开，你才登录 VPS 发现磁盘已满；
某台服务器的 CPU 持续飙高三天，直到流量异常才被注意到；
SSL 证书过期导致服务中断，因为日历提醒被你忽略了；
月底账单来了才发现，几台闲置的 VPS 白白消耗了一个月的费用。

传统运维模式的核心问题是被动响应——只有在问题发生后才采取行动。而 AI Agent 的出现，让主动治理成为可能。

本文将带你从零开始，构建一套基于 AI Agent 的 VPS 智能运维体系。这套体系不依赖昂贵的商业 SaaS，全部使用开源工具和本地运行的 LLM，总成本为零。

什么是 AI Agent 驱动的运维？

AI Agent 不是简单的脚本或规则引擎，而是一个具备感知-决策-执行闭环能力的自主系统：

感知：通过监控数据采集、日志分析和指标聚合，全面掌握 VPS 状态；
推理：利用 LLM 理解上下文，识别异常模式，判断问题根因；
决策：基于策略库和安全边界，制定修复方案并评估风险；
执行：在受控环境中自动执行操作，或通过审批流程由人工确认；
学习：记录每次事件的处理过程，持续优化判断逻辑。

与传统告警系统的本质区别在于：AI Agent 不仅能告诉你"出事了"，还能告诉你"发生了什么、为什么发生、该怎么解决"。

架构设计

整个系统由四个核心组件构成：

┌─────────────────────────────────────────────────┐
│                  AI Agent Orchestrator            │
│              (本地 LLM + 编排框架)                 │
│  ┌──────────┬──────────┬──────────┬───────────┐  │
│  │ 感知层   │ 推理层   │ 决策层   │ 执行层     │  │
│  │ 采集数据  │ 分析模式  │ 制定方案  │ 自动修复   │  │
│  └──────────┴──────────┴──────────┴───────────┘  │
├─────────────────────────────────────────────────┤
│              基础设施层 (所有 VPS)                 │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐            │
│  │NodeExp  │ │Loki/Prom │ │自定义Agent│           │
│  │Exporter │ │tail/Prom │ │(健康检查) │           │
│  └─────────┘ └─────────┘ └─────────┘            │
└─────────────────────────────────────────────────┘

组件详解

感知层：复用已有的监控栈（Prometheus + Node Exporter + Loki），但增加一个关键改进——所有采集频率从传统的 15 秒缩短至 5 秒，提高细粒度检测能力。

推理层：这是系统的核心。我们使用一个轻量级本地 LLM（如 Qwen2.5-7B-Instruct 或 Mistral-7B-Instruct），配合结构化提示词模板，对监控数据进行语义分析。

决策层：维护一个"运维知识图谱"，将常见问题与解决方案关联。例如：

# knowledge-base/disk-full.yaml
problem: "磁盘使用率超过 90%"
indicators:
  - node_filesystem_avail_bytes < 1GB
  - rate(node_filesystem_write_bytes_total[1h]) > 10MB/s
possible_causes:
  - name: "日志膨胀"
    confidence: 0.7
    evidence: "rate(log_volume_bytes) 突增"
    solution: "清理旧日志，配置 logrotate"
  - name: "上传文件堆积"
    confidence: 0.2
    evidence: "/var/upload 目录增长异常"
    solution: "归档或清理上传文件"
  - name: "数据库膨胀"
    confidence: 0.1
    evidence: "pg_database_size 持续增长"
    solution: "执行 VACUUM FULL"

执行层：分为两个模式——

Auto 模式：低风险操作直接执行（如清理日志、重启服务）；
Review 模式：高风险操作需人工确认（如扩容磁盘、修改防火墙规则）。

第一步：统一数据接入

AI Agent 需要全面的数据输入才能做出准确判断。我们将三类数据源统一接入：

1. 指标数据（Metrics）

在每台 VPS 上部署 Node Exporter，并通过 Prometheus 集中采集：

# docker-compose.monitoring.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
      - '--query.max-concurrency=20'
    ports:
      - "9090:9090"

  loki:
    image: grafana/loki:3.0.0
    command: ['-config.file=/etc/loki/local-config.yaml']
    ports:
      - "3100:3100"

  promtail:
    image: grafana/promtail:3.0.0
    volumes:
      - /var/log:/var/log
      - ./var-log:/tmp/var-log
      - /run/log:/run/log
      - ./promtail-config.yml:/etc/promtail/config.yml
    command: '-config.file=/etc/promtail/config.yml'

关键配置——将采集间隔缩短到 5 秒：

# prometheus.yml
global:
  scrape_interval: 5s       # 默认 5 秒采集
  evaluation_interval: 15s   # 每 15 秒评估规则

scrape_configs:
  - job_name: 'node-exporter'
    static_configs:
      - targets: ['node-exporter:9100']
    scrape_interval: 5s

2. 日志数据（Logs）

通过 Loki + Promtail 收集所有容器的标准输出和应用日志。对于 AI Agent 来说，日志的结构化程度直接影响分析质量。推荐使用 JSON 格式输出：

# 应用日志配置示例
logging:
  driver: "json-file"
  options:
    max-size: "10m"
    max-file: "3"
    tag: "{{.Name}}"

3. 业务健康检查（Health Checks）

除了基础设施指标，还需要业务层面的健康信号：

# 自定义健康检查 Agent
#!/bin/bash
# health-check.sh - 业务健康探针

check_endpoint() {
    local url=$1
    local expected_code=${2:-200}
    local http_code=$(curl -s -o /dev/null -w "%{http_code}" "$url" --max-time 5)
    
    if [ "$http_code" -eq "$expected_code" ]; then
        echo "OK: $url returned $http_code"
        return 0
    else
        echo "WARN: $url returned $http_code (expected $expected_code)"
        return 1
    fi
}

# 检查关键服务
check_endpoint "http://localhost:8080/api/health" 200
check_endpoint "https://api.example.com/status" 200
check_db_connection

第二步：构建 AI 推理引擎

这是整个系统最关键的部分。我们使用一个本地运行的 LLM 作为推理核心。

环境准备

# 安装 Ollama（本地 LLM 运行时）
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合推理的模型（7B 级别在 4GB 内存 VPS 上即可运行）
ollama pull qwen2.5:7b-instruct

# 验证
ollama run qwen2.5:7b-instruct "你好"

运维分析 Prompt 模板

我们为不同的运维场景设计了专用的 Prompt 模板。以异常检测为例：

# ai_agent/analyzer.py
import json
import requests

ANOMALY_PROMPT = """你是一个资深 SRE 工程师。请根据以下监控数据，分析是否存在异常。

## 当前时间
{timestamp}

## 涉及服务器
{server_info}

## 关键指标
{metrics_snapshot}

## 最近日志片段
{recent_logs}

## 分析要求
1. 判断是否存在异常（是/否）
2. 如果存在，列出异常项及其严重程度（P0-P3）
3. 推测可能的根因
4. 给出建议的修复步骤

请以 JSON 格式返回：
{{
  "anomaly_detected": true/false,
  "severity": "P0"|"P1"|"P2"|"P3",
  "issues": [
    {{
      "metric": "指标名称",
      "current_value": "当前值",
      "threshold": "阈值",
      "deviation": "偏离度 %"
    }}
  ],
  "root_cause_hypothesis": "根因假设",
  "confidence": 0.0-1.0,
  "recommended_actions": ["行动1", "行动2"],
  "risk_level": "low"|"medium"|"high"
}}
"""

调用 LLM 进行推理

# ai_agent/orchestrator.py
class VPSOrchestrator:
    def __init__(self):
        self.ollama_url = "http://localhost:11434"
        self.knowledge_base = KnowledgeBase()
        self.policy_engine = PolicyEngine()
    
    async def analyze_and_act(self, alert_data: dict):
        """主循环：接收告警 → 分析 → 决策 → 执行"""
        
        # 1. 收集上下文
        context = await self.collect_context(alert_data)
        
        # 2. 查询知识库
        kb_matches = self.knowledge_base.search(context['symptoms'])
        
        # 3. 构建 Prompt
        prompt = ANOMALY_PROMPT.format(
            timestamp=context['timestamp'],
            server_info=context['server_info'],
            metrics_snapshot=json.dumps(context['metrics'], indent=2),
            recent_logs=context['logs'][-500:],  # 最近 500 字符
        )
        
        # 4. 调用 LLM
        response = requests.post(
            f"{self.ollama_url}/api/generate",
            json={
                "model": "qwen2.5:7b-instruct",
                "prompt": prompt,
                "stream": False,
                "options": {
                    "temperature": 0.3,  # 低温度保证一致性
                    "num_predict": 512
                }
            }
        )
        
        analysis = json.loads(response.json()['response'])
        
        # 5. 风险评估
        risk = self.policy_engine.evaluate(analysis)
        
        # 6. 决策分支
        if risk.level == "low":
            await self.execute_auto_fix(analysis)
        elif risk.level == "medium":
            await self.notify_and_wait(analysis)
        else:
            await self.escalate(analysis)
        
        return analysis

第三步：实现三大核心能力

能力一：容量预测与弹性伸缩

传统扩容方式是被动的——流量上来后才加机器。AI Agent 可以通过时序分析预测未来趋势，提前扩容。

# ai_agent/capacity_planner.py
import numpy as np
from scipy.signal import find_peaks

def predict_capacity(metrics_history: list, horizon_hours: int = 24):
    """
    基于历史数据的容量预测
    
    metrics_history: [{timestamp, cpu, memory, disk, bandwidth}, ...]
    """
    # 提取时间序列
    timestamps = [m['timestamp'] for m in metrics_history]
    cpu_values = [m['cpu'] for m in metrics_history]
    mem_values = [m['memory'] for m in metrics_history]
    
    # 检测周期性模式（日周期、周周期）
    cpu_peaks, _ = find_peaks(cpu_values, distance=len(timestamps)//24)
    mem_peaks, _ = find_peaks(mem_values, distance=len(timestamps)//24)
    
    # 简单线性外推（生产环境可用 Prophet 或 LSTM）
    cpu_trend = np.polyfit(range(len(cpu_values)), cpu_values, 1)
    mem_trend = np.polyfit(range(len(mem_values)), mem_values, 1)
    
    # 预测未来
    predictions = []
    for h in range(1, horizon_hours + 1):
        idx = len(cpu_values) + h
        predicted_cpu = cpu_trend[0] * idx + cpu_trend[1]
        predicted_mem = mem_trend[0] * idx + mem_trend[1]
        
        predictions.append({
            'hours_ahead': h,
            'predicted_cpu': min(predicted_cpu, 100),
            'predicted_memory': min(predicted_mem, 100),
            'needs_scaling': predicted_cpu > 80 or predicted_mem > 85
        })
    
    return predictions

配合自动伸缩策略：

# scaling-policy.yaml
policies:
  - name: "cpu-overflow"
    condition: "predicted_cpu_2h > 80"
    action: "scale_up"
    target: "add 1 core"
    auto_approve: true
    
  - name: "idle-resource"
    condition: "avg_cpu_7d < 10 AND avg_mem_7d < 20"
    action: "downsize"
    target: "reduce to next tier"
    auto_approve: false  # 降配需人工确认
    
  - name: "weekend-reduction"
    condition: "day_of_week == 6 AND hour > 22"
    action: "schedule_downsize"
    target: "temporarily reduce to 1C1G"
    auto_approve: true
    rollback_condition: "cpu > 60 OR response_time > 500ms"

能力二：智能根因分析

当多个告警同时触发时，传统系统会发送几十条消息，让人无所适从。AI Agent 可以进行告警聚合和根因推断。

传统方式:
  [10:00] ⚠️ CPU 使用率 > 90%
  [10:00] ⚠️ 内存使用率 > 85%
  [10:01] ⚠️ 磁盘 I/O 等待 > 50%
  [10:01] ⚠️ 网络连接超时
  [10:02] 🔴 服务不可用
  → 5 条告警，人工逐一排查

AI Agent 方式:
  [10:02] 📊 告警聚合完成
  🔍 根因分析: 数据库连接池耗尽
     ├─ 触发条件: max_connections 达到上限
     ├─ 影响范围: 所有依赖该数据库的服务
     ├─ 证据链: 
     │   ├─ CPU 升高 ← 连接等待导致线程阻塞
     │   ├─ 内存升高 ← 连接对象未释放
     │   └─ 磁盘 I/O 升高 ← 大量写入等待队列
     └─ 建议操作:
         1. [Auto] 临时增加 max_connections 到 500
         2. [Review] 检查应用代码中的连接泄漏
         3. [Schedule] 下周实施连接池优化
  → 1 条聚合报告，附带完整分析

实现代码：

# ai_agent/root_cause.py
class RootCauseAnalyzer:
    def __init__(self):
        self.dependency_graph = self.load_dependency_graph()
    
    def correlate_alerts(self, alerts: list) -> dict:
        """将分散的告警关联为事件"""
        # 按时间窗口分组（默认 5 分钟）
        windows = self.time_window_group(alerts, window_minutes=5)
        
        # 按服务器分组
        server_groups = self.server_group(windows)
        
        # 通过依赖关系进一步关联
        correlated = []
        for group in server_groups:
            affected_services = self.find_affected_services(group)
            if len(affected_services) > 1:
                correlated.append({
                    'event_id': self.generate_event_id(),
                    'trigger_server': group['primary'],
                    'affected_services': affected_services,
                    'alert_count': len(group['alerts']),
                    'time_range': group['time_range']
                })
        
        return correlated

能力三：自动修复与自愈

AI Agent 可以执行预设的安全操作，实现真正的自愈能力。

# ai_agent/remediation.py
class RemediationEngine:
    # 预定义的安全操作清单
    SAFE_OPERATIONS = {
        "disk_cleanup": {
            "description": "清理系统日志和临时文件",
            "commands": [
                "journalctl --vacuum-time=3d",
                "find /tmp -type f -mtime +7 -delete",
                "apt-get clean"
            ],
            "max_risk": "low",
            "rollback": "none_needed"
        },
        "service_restart": {
            "description": "重启指定服务",
            "commands": [
                "systemctl restart {service}"
            ],
            "max_risk": "medium",
            "rollback": "systemctl start {service}"
        },
        "connection_pool_fix": {
            "description": "临时扩大数据库连接池",
            "commands": [
                "SELECT pg_reload_conf()",
                "ALTER SYSTEM SET max_connections = 500"
            ],
            "max_risk": "medium",
            "rollback": "ALTER SYSTEM SET max_connections = 200"
        },
        "cache_clear": {
            "description": "清除应用缓存",
            "commands": [
                "redis-cli FLUSHDB",
                "systemctl reload {app_service}"
            ],
            "max_risk": "low",
            "rollback": "none_needed"
        }
    }
    
    async def execute_remediation(self, issue: dict, operation_key: str):
        """执行修复操作"""
        operation = self.SAFE_OPERATIONS[operation_key]
        
        # 风险检查
        if issue.get('risk_level', 'medium') != 'low' and operation['max_risk'] == 'low':
            return {"status": "denied", "reason": "风险等级超出操作允许范围"}
        
        # 执行前快照
        pre_snapshot = await self.take_system_snapshot()
        
        # 执行操作
        results = []
        for cmd in operation['commands']:
            result = await self.safe_execute(cmd.format(**issue))
            results.append(result)
        
        # 执行后验证
        post_snapshot = await self.take_system_snapshot()
        verification = self.verify_improvement(pre_snapshot, post_snapshot, issue)
        
        return {
            "status": "success" if verification else "partial",
            "operations": results,
            "verification": verification,
            "rollback_available": True
        }

第四步：智能报告与可视化

AI Agent 不只是解决问题，还要让运维结果可追溯、可审计。

每日运维摘要

每天早晨，AI Agent 自动生成一份运维日报：

# 2026-07-07 运维日报
summary:
  total_events: 12
  auto_resolved: 8
  escalated: 2
  avg_response_time: "45s"
  
events:
  - time: "03:15"
    type: "disk_cleanup"
    server: "web-prod-01"
    action: "自动清理日志，释放 2.3GB"
    impact: "无用户影响"
    
  - time: "08:42"
    type: "cpu_spike"
    server: "api-prod-02"
    action: "检测到异常进程，已自动 kill"
    impact: "API 响应延迟 3 秒"
    followup: "需要检查进程来源"
    
  - time: "14:20"
    type: "cert_expiring"
    server: "all"
    action: "SSL 证书将在 7 天后过期，已触发自动续期"
    impact: "无"
    
insights:
  - "本周磁盘告警集中在 web-prod-01，建议优化日志轮转策略"
  - "api-prod-02 的 CPU 峰值与发布窗口高度相关，建议调整发布策略"
  - "平均故障恢复时间较上周提升 35%"

生成方式

# ai_agent/reporter.py
async def generate_daily_report():
    """生成每日运维报告"""
    
    # 收集数据
    events = await fetch_events(hours=24)
    metrics = await aggregate_metrics(hours=24)
    fixes = await get_remediation_log(hours=24)
    
    # 构建报告 Prompt
    prompt = DAILY_REPORT_PROMPT.format(
        events=json.dumps(events, indent=2),
        metrics=json.dumps(metrics, indent=2),
        fixes=json.dumps(fixes, indent=2),
    )
    
    # LLM 生成报告
    response = ollama.generate(prompt, model="qwen2.5:7b-instruct")
    
    # 发送到通知渠道
    await send_to_channels({
        'wechat': response['text'],
        'email': generate_html_report(response['text']),
        'slack': response['markdown']
    })

安全边界与风险控制

AI Agent 的强大能力也意味着更大的风险。以下是必须遵守的安全原则：

1. 操作白名单机制

# safety-policy.yaml
allowed_operations:
  auto:
    - "log_rotation"
    - "temp_file_cleanup"
    - "service_restart"
    - "cache_clear"
    - "cert_renewal"
  review_required:
    - "config_change"
    - "firewall_rule"
    - "user_management"
    - "database_migration"
  forbidden:
    - "format_disk"
    - "modify_kernel_params"
    - "change_network_config"
    - "delete_data"

2. 操作审计日志

# audit_logger.py
class AuditLogger:
    def log_operation(self, operator: str, action: str, target: str, 
                      result: str, confidence: float):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "operator": operator,  # "ai-agent" 或 "human"
            "action": action,
            "target": target,
            "result": result,
            "llm_confidence": confidence,
            "snapshot_before": self.capture_state(),
            "snapshot_after": self.capture_state(),
            "approved_by": self.get_approval_chain(action)
        }
        # 写入不可篡改的审计日志
        self.audit_log.write(json.dumps(entry))

3. 紧急停机开关

# emergency-stop.sh
# 一键停止 AI Agent 的所有自动操作
#!/bin/bash
echo "$(date): EMERGENCY STOP triggered" >> /var/log/ai-agent/audit.log
systemctl stop ai-agent
# 恢复手动运维模式
export MAINTENANCE_MODE=true
notify-all "⚠️ AI Agent 已暂停，切换至手动运维模式"

实战部署

完整 Docker Compose 配置

# docker-compose.ai-agent.yml
version: '3.8'

services:
  # --- 数据采集层 ---
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus:/etc/prometheus
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'
    ports:
      - "9090:9090"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
    restart: unless-stopped

  # --- AI 推理层 ---
  ai-agent:
    build: ./ai-agent
    volumes:
      - ./ai-agent/knowledge-base:/app/knowledge-base
      - ./ai-agent/policies:/app/policies
      - ./ai-agent/reports:/app/reports
      - agent_audit:/var/log/ai-agent
    environment:
      - OLLAMA_HOST=http://ollama:11434
      - PROMETHEUS_URL=http://prometheus:9090
      - GRAFANA_URL=http://grafana:3000
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
    depends_on:
      - ollama
      - prometheus
    restart: unless-stopped

  # --- 本地 LLM ---
  ollama:
    image: ollama/ollama:latest
    volumes:
      - ollama_data:/root/.ollama
    ports:
      - "11434:11434"
    restart: unless-stopped

volumes:
  prometheus_data:
  grafana_data:
  ollama_data:
  agent_audit:

AI Agent 项目结构

ai-agent/
├── Dockerfile
├── requirements.txt
├── config.yaml
├── knowledge-base/
│   ├── disk-full.yaml
│   ├── cpu-spike.yaml
│   ├── cert-expiry.yaml
│   └── connection-pool.yaml
├── policies/
│   ├── safety-policy.yaml
│   └── scaling-policy.yaml
├── src/
│   ├── orchestrator.py      # 主调度器
│   ├── analyzer.py          # 异常分析
│   ├── root_cause.py        # 根因分析
│   ├── capacity_planner.py  # 容量预测
│   ├── remediation.py       # 自动修复
│   ├── reporter.py          # 报告生成
│   ├── notifier.py          # 通知推送
│   └── audit.py             # 审计日志
└── tests/
    ├── test_analyzer.py
    ├── test_root_cause.py
    └── test_remediation.py

效果与指标

部署这套系统后，典型的效果改善包括：

指标	部署前	部署后	改善幅度
平均故障发现时间 (MTTD)	30 分钟	2 分钟	↓ 93%
平均故障恢复时间 (MTTR)	45 分钟	5 分钟	↓ 89%
月度误报数量	50+	< 5	↓ 90%
夜间人工干预次数	10+/月	1-2/月	↓ 85%
资源浪费成本	~$200/月	~$30/月	↓ 85%

总结

AI Agent 驱动的 VPS 智能运维不是要取代人类运维工程师，而是将工程师从重复性劳动中解放出来，专注于更高价值的架构设计和性能优化工作。

关键要点回顾：

数据先行：完善的监控和数据采集是 AI 推理的基础；
小步快跑：从低风险操作（日志清理、证书续期）开始，逐步扩展；
安全兜底：永远保留人工干预通道和紧急停机开关；
持续学习：将每次事件的处理过程反馈给系统，形成闭环优化。

当你不再需要在凌晨三点被告警电话叫醒时，你就知道这套系统值得投入。

本文所有代码和配置均为开源实现，可在 GitHub 上找到完整示例。欢迎 Fork 和贡献。