VPS 智能巡检：AI Agent 自动发现安全隐患与性能瓶颈

引言

你管理着几台甚至十几台 VPS，跑着网站、API、数据库、Docker 容器。日常运维中，你是否经历过这样的场景？

某天发现服务器被暴力破解了，但日志里早就有异常登录记录；
网站突然变慢，排查半天才发现是某个进程占满了 CPU；
月底账单来了，才发现某台 VPS 资源利用率不到 5%，白白浪费钱；
SSL 证书过期导致服务中断，因为日历提醒被你忽略了。

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

本文将带你从零开始，构建一套基于 AI Agent 的 VPS 智能巡检系统。这套系统每天自动执行一次全面健康检查，包括：

安全扫描：检测异常登录、开放端口、安全漏洞
性能分析：识别资源瓶颈、慢查询、内存泄漏
容量预测：基于历史数据预测磁盘和带宽使用趋势
自动修复：对常见问题自动生成修复脚本或执行安全操作
智能报告：用自然语言生成可读性强的巡检报告

全部使用开源工具和本地运行的 LLM，总成本为零。

架构设计

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

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

组件详解

组件	作用	技术选型
采集层	收集系统指标、日志、配置	Node Exporter, systemd journal, SSH
分析层	识别异常模式、性能瓶颈	Prometheus Query, LLM 推理
决策层	评估风险、制定修复方案	策略库 + LLM 生成
执行层	自动执行安全操作	Ansible, Shell 脚本

第一步：搭建数据采集层

我们需要在每台 VPS 上安装轻量级采集器，定期收集关键指标。

1.1 安装 Node Exporter

Node Exporter 负责采集系统级指标：CPU、内存、磁盘、网络等。

# 下载 Node Exporter
wget https://github.com/prometheus/node_exporter/releases/download/v1.8.2/node_exporter-1.8.2.linux-amd64.tar.gz
tar xzf node_exporter-1.8.2.linux-amd64.tar.gz
sudo cp node_exporter-1.8.2.linux-amd64/node_exporter /usr/local/bin/

# 创建 systemd 服务
sudo tee /etc/systemd/system/node_exporter.service << 'EOF'
[Unit]
Description=Node Exporter
Wants=network-online.target
After=network-online.target

[Service]
User=node_exporter
Group=node_exporter
Type=simple
ExecStart=/usr/local/bin/node_exporter \
  --web.listen-address=:9100 \
  --collector.diskstats \
  --collector.filesystem \
  --collector.meminfo \
  --collector.netdev \
  --collector.loadavg

[Install]
WantedBy=multi-user.target
EOF

# 启动服务
sudo systemctl daemon-reload
sudo systemctl enable --now node_exporter

验证安装：

curl http://localhost:9100/metrics | head -20

1.2 配置 SSH 远程采集

对于多台 VPS，我们使用 SSH 密钥认证进行远程采集。

# 在管理中心生成密钥
ssh-keygen -t ed25519 -f ~/.ssh/vps_inspector -N ""

# 将公钥复制到所有 VPS
for host in vps1.example.com vps2.example.com; do
    ssh-copy-id -i ~/.ssh/vps_inspector.pub $host
done

在 VPS 上限制 SSH 密钥权限：

# 编辑 ~/.ssh/authorized_keys
from="10.0.0.0/8",command="/usr/local/bin/inspector.sh" ssh-ed25519 AAAA...

第二步：构建 AI Agent 编排框架

我们用 Python 编写一个编排框架，负责调度采集任务、调用 LLM 分析、生成报告。

2.1 项目结构

mkdir -p ~/vps-inspector/{config,scripts,logs,reports}
cd ~/vps-inspector

vps-inspector/
├── config/
│   ├── vps_list.yaml      # VPS 清单
│   ├── rules.yaml         # 告警规则
│   └── llm_config.yaml    # LLM 配置
├── scripts/
│   ├── collect_metrics.py # 采集指标
│   ├── analyze.py         # AI 分析
│   └── report.py          # 生成报告
├── requirements.txt       # Python 依赖
└── run_inspection.sh      # 主入口

2.2 定义 VPS 清单

# config/vps_list.yaml
vps_list:
  - name: production-web
    ip: 10.0.1.10
    user: deploy
    role: web
    priority: high
    
  - name: production-db
    ip: 10.0.1.11
    user: deploy
    role: database
    priority: critical
    
  - name: staging-api
    ip: 10.0.2.10
    user: deploy
    role: api
    priority: medium

2.3 采集指标脚本

#!/usr/bin/env python3
"""Collect system metrics from remote VPS."""

import subprocess
import json
from datetime import datetime

def collect_ssh_metrics(host, user, command):
    """Execute command on remote VPS via SSH."""
    cmd = f"ssh -o StrictHostKeyChecking=no -i ~/.ssh/vps_inspector {user}@{host} '{command}'"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    return result.stdout if result.returncode == 0 else ""

def collect_system_metrics(host, user):
    """Collect basic system metrics."""
    metrics = {
        "timestamp": datetime.now().isoformat(),
        "host": host,
        "cpu_usage": 0,
        "memory_usage": 0,
        "disk_usage": {},
        "load_average": [],
        "top_processes": []
    }
    
    # CPU usage
    cpu_output = collect_ssh_metrics(host, user, "top -bn1 | grep 'Cpu(s)'")
    if cpu_output:
        metrics["cpu_usage"] = float(cpu_output.split(',')[0].split(':')[1].strip())
    
    # Memory usage
    mem_output = collect_ssh_metrics(host, user, "free | grep Mem")
    if mem_output:
        parts = mem_output.split()
        total = int(parts[1])
        used = int(parts[2])
        metrics["memory_usage"] = round(used / total * 100, 2)
    
    # Disk usage
    disk_output = collect_ssh_metrics(host, user, "df -h /")
    if disk_output:
        lines = disk_output.strip().split('\n')
        if len(lines) > 1:
            parts = lines[1].split()
            metrics["disk_usage"]["root"] = {
                "total": parts[1],
                "used": parts[2],
                "available": parts[3],
                "usage_percent": parts[4]
            }
    
    # Load average
    load_output = collect_ssh_metrics(host, user, "uptime")
    if load_output:
        parts = load_output.split('load average:')
        if len(parts) > 1:
            metrics["load_average"] = [float(x.strip()) for x in parts[1].split(',')]
    
    # Top processes by CPU
    top_output = collect_ssh_metrics(host, user, "ps aux --sort=-%cpu | head -6")
    if top_output:
        lines = top_output.strip().split('\n')[1:]  # Skip header
        metrics["top_processes"] = [line.split(None, 10) for line in lines]
    
    return metrics

def collect_security_metrics(host, user):
    """Collect security-related metrics."""
    metrics = {
        "failed_logins": [],
        "open_ports": [],
        "recent_updates": [],
        "firewall_status": ""
    }
    
    # Failed login attempts
    fail_output = collect_ssh_metrics(host, user, "journalctl -u sshd --since '24 hours ago' | grep 'Failed password' | tail -20")
    if fail_output:
        metrics["failed_logins"] = fail_output.strip().split('\n')
    
    # Open ports
    port_output = collect_ssh_metrics(host, user, "ss -tuln | grep LISTEN")
    if port_output:
        metrics["open_ports"] = port_output.strip().split('\n')
    
    # Firewall status
    fw_output = collect_ssh_metrics(host, user, "ufw status 2>/dev/null || iptables -L -n | head -20")
    if fw_output:
        metrics["firewall_status"] = fw_output
    
    return metrics

if __name__ == "__main__":
    import yaml
    
    with open("config/vps_list.yaml") as f:
        config = yaml.safe_load(f)
    
    all_metrics = []
    
    for vps in config["vps_list"]:
        print(f"Collecting metrics from {vps['name']}...")
        
        system_metrics = collect_system_metrics(vps["ip"], vps["user"])
        security_metrics = collect_security_metrics(vps["ip"], vps["user"])
        
        all_metrics.append({
            "name": vps["name"],
            "role": vps["role"],
            "priority": vps["priority"],
            "system": system_metrics,
            "security": security_metrics
        })
    
    # Save to JSON
    output_file = f"logs/inspection_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    with open(output_file, "w") as f:
        json.dump(all_metrics, f, indent=2, ensure_ascii=False)
    
    print(f"Metrics saved to {output_file}")

第三步：AI 分析引擎

这是系统的核心——让 LLM 理解采集到的数据，识别异常模式。

3.1 提示词设计

# scripts/analyze.py

SYSTEM_PROMPT = """你是一位资深 VPS 运维专家。你的任务是分析系统巡检数据，识别潜在的安全隐患和性能瓶颈，并给出可执行的修复建议。

请按照以下格式输出：
1. 【整体评分】0-100 分，100 分为最佳
2. 【安全风险】列出发现的安全问题，按严重程度排序
3. 【性能瓶颈】列出资源使用异常的情况
4. 【容量趋势】基于历史数据预测未来 7 天的使用情况
5. 【修复建议】给出具体的命令或配置修改方案
6. 【优先级】标记哪些需要立即处理，哪些可以等待"""

USER_PROMPT_TEMPLATE = """请分析以下 VPS 巡检数据：

VPS 名称：{name}
角色：{role}
优先级：{priority}

系统指标：
{system_metrics}

安全指标：
{security_metrics}

历史数据（过去 7 天）：
{historical_data}

请给出详细的分析报告。"""

3.2 调用 LLM 分析

import openai
from dotenv import load_dotenv

load_dotenv()

def analyze_with_llm(vps_metrics, historical_data=None):
    """Use LLM to analyze VPS metrics."""
    
    prompt = USER_PROMPT_TEMPLATE.format(
        name=vps_metrics["name"],
        role=vps_metrics["role"],
        priority=vps_metrics["priority"],
        system_metrics=json.dumps(vps_metrics["system"], indent=2),
        security_metrics=json.dumps(vps_metrics["security"], indent=2),
        historical_data=json.dumps(historical_data or {}, indent=2)
    )
    
    response = openai.chat.completions.create(
        model="gpt-4o-mini",  # 或使用本地 Ollama
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,
        max_tokens=2000
    )
    
    return response.choices[0].message.content

3.3 异常检测规则

除了 LLM 分析，我们还设置一些硬性规则来检测明显异常：

# config/rules.yaml
rules:
  - name: "high_cpu_usage"
    condition: "cpu_usage > 80"
    severity: "warning"
    action: "alert"
    
  - name: "critical_memory_usage"
    condition: "memory_usage > 90"
    severity: "critical"
    action: "alert_and_restart"
    
  - name: "disk_full_warning"
    condition: "disk_usage.root.usage_percent > 85"
    severity: "warning"
    action: "alert"
    
  - name: "failed_login_spike"
    condition: "len(failed_logins) > 10"
    severity: "critical"
    action: "block_ip"
    
  - name: "unusual_port_open"
    condition: "port not in allowed_ports"
    severity: "warning"
    action: "alert"

def check_rules(vps_metrics, rules):
    """Check metrics against predefined rules."""
    violations = []
    
    for rule in rules:
        if rule["name"] == "high_cpu_usage":
            if vps_metrics["system"]["cpu_usage"] > 80:
                violations.append({
                    "rule": rule["name"],
                    "severity": rule["severity"],
                    "value": vps_metrics["system"]["cpu_usage"],
                    "threshold": 80
                })
        
        elif rule["name"] == "failed_login_spike":
            if len(vps_metrics["security"]["failed_logins"]) > 10:
                violations.append({
                    "rule": rule["name"],
                    "severity": rule["severity"],
                    "count": len(vps_metrics["security"]["failed_logins"]),
                    "threshold": 10
                })
    
    return violations

第四步：自动生成修复方案

AI Agent 不仅能发现问题，还能自动生成修复脚本。

4.1 安全修复示例

def generate_security_fixes(violations, vps_metrics):
    """Generate fix scripts for security issues."""
    
    fixes = []
    
    for violation in violations:
        if violation["rule"] == "failed_login_spike":
            # 找出攻击 IP 并加入黑名单
            attack_ips = extract_attack_ips(vps_metrics["security"]["failed_logins"])
            
            fix_script = f"""#!/bin/bash
# Block attacking IPs
ATTACK_IPS={','.join(attack_ips)}

for ip in ${{ATTACK_IPS}}; do
    ufw deny from $ip
    echo "Blocked $ip"
done

# Restart Fail2Ban
systemctl restart fail2ban
"""
            fixes.append({
                "type": "security",
                "description": "Block attacking IPs and restart Fail2Ban",
                "script": fix_script
            })
        
        elif violation["rule"] == "unusual_port_open":
            # 关闭不必要的端口
            fix_script = f"""#!/bin/bash
# Close unnecessary ports
ufw deny 2375/tcp  # Docker API
ufw deny 6379/tcp  # Redis
ufw deny 27017/tcp # MongoDB

# Reload firewall
ufw reload
"""
            fixes.append({
                "type": "security",
                "description": "Close unnecessary exposed ports",
                "script": fix_script
            })
    
    return fixes

4.2 性能优化示例

def generate_performance_fixes(vps_metrics):
    """Generate fix scripts for performance issues."""
    
    fixes = []
    
    if vps_metrics["system"]["memory_usage"] > 90:
        # 清理缓存
        fix_script = """#!/bin/bash
# Clear system cache
sync
echo 3 | sudo tee /proc/sys/vm/drop_caches

# Restart memory-hungry services
systemctl restart docker
"""
        fixes.append({
            "type": "performance",
            "description": "Clear cache and restart Docker",
            "script": fix_script
        })
    
    if vps_metrics["system"]["disk_usage"]["root"]["usage_percent"] > 85:
        # 清理日志和临时文件
        fix_script = """#!/bin/bash
# Clean old logs
find /var/log -name "*.gz" -mtime +7 -delete
find /var/log -name "*.log" -size +100M -exec truncate -s 0 {} \\;

# Clean package cache
apt clean
apt autoremove -y

# Clean temp files
rm -rf /tmp/*
rm -rf /var/tmp/*
"""
        fixes.append({
            "type": "performance",
            "description": "Clean disk space",
            "script": fix_script
        })
    
    return fixes

第五步：生成智能报告

最后，将分析结果和修复建议生成一份易读的 HTML 报告。

5.1 报告模板

REPORT_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>VPS 智能巡检报告 - {{ date }}</title>
    <style>
        body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; margin: 0; padding: 20px; background: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; background: white; border-radius: 8px; padding: 30px; box-shadow: 0 2px 8px rgba(0,0,0,0.1); }
        h1 { color: #333; border-bottom: 2px solid #007bff; padding-bottom: 10px; }
        .score { font-size: 48px; font-weight: bold; color: {{ score_color }}; text-align: center; margin: 20px 0; }
        .section { margin: 30px 0; }
        .section h2 { color: #555; border-left: 4px solid #007bff; padding-left: 15px; }
        .violation { background: #fff3cd; border-left: 4px solid #ffc107; padding: 15px; margin: 10px 0; border-radius: 4px; }
        .violation.critical { background: #f8d7da; border-left-color: #dc3545; }
        .fix-script { background: #1e1e1e; color: #d4d4d4; padding: 15px; border-radius: 4px; overflow-x: auto; font-family: 'Courier New', monospace; }
        table { width: 100%; border-collapse: collapse; margin: 20px 0; }
        th, td { padding: 12px; text-align: left; border-bottom: 1px solid #ddd; }
        th { background: #f8f9fa; font-weight: 600; }
        .badge { display: inline-block; padding: 4px 12px; border-radius: 12px; font-size: 12px; font-weight: 600; }
        .badge-critical { background: #dc3545; color: white; }
        .badge-warning { background: #ffc107; color: #333; }
        .badge-info { background: #17a2b8; color: white; }
    </style>
</head>
<body>
    <div class="container">
        <h1>🔍 VPS 智能巡检报告</h1>
        <p>生成时间：{{ date }}</p>
        
        <div class="score">
            整体评分：<span style="color: {{ score_color }}">{{ overall_score }}/100</span>
        </div>
        
        <div class="section">
            <h2>📊 VPS 概览</h2>
            <table>
                <tr>
                    <th>VPS 名称</th>
                    <th>角色</th>
                    <th>CPU</th>
                    <th>内存</th>
                    <th>磁盘</th>
                    <th>安全评分</th>
                </tr>
                {{#each vps_list}}
                <tr>
                    <td>{{this.name}}</td>
                    <td>{{this.role}}</td>
                    <td>{{this.system.cpu_usage}}%</td>
                    <td>{{this.system.memory_usage}}%</td>
                    <td>{{this.system.disk_usage.root.usage_percent}}%</td>
                    <td><span class="badge badge-{{this.security_score_class}}">{{this.security_score}}</span></td>
                </tr>
                {{/each}}
            </table>
        </div>
        
        <div class="section">
            <h2>⚠️ 发现的问题</h2>
            {{#each violations}}
            <div class="violation {{this.severity}}">
                <strong>{{this.rule}}</strong>
                <p>{{this.description}}</p>
                <p>当前值：{{this.value}} | 阈值：{{this.threshold}}</p>
            </div>
            {{/each}}
        </div>
        
        <div class="section">
            <h2>🛠️ 修复建议</h2>
            {{#each fixes}}
            <div class="section">
                <h3>{{this.description}}</h3>
                <pre class="fix-script">{{this.script}}</pre>
            </div>
            {{/each}}
        </div>
        
        <div class="section">
            <h2>🤖 AI 分析摘要</h2>
            <p>{{llm_summary}}</p>
        </div>
    </div>
</body>
</html>
"""

5.2 运行巡检

#!/bin/bash
# run_inspection.sh

cd ~/vps-inspector

echo "Starting VPS inspection..."

# Step 1: Collect metrics
python3 scripts/collect_metrics.py

# Step 2: Analyze with LLM
python3 scripts/analyze.py

# Step 3: Generate report
python3 scripts/report.py

# Step 4: Send notification (optional)
# curl -X POST https://your-webhook-url/notify -d '{"message": "Inspection complete"}'

echo "Inspection complete! Report saved to reports/latest.html"

第六步：定时执行与告警通知

6.1 配置 Cron 定时任务

# 每天凌晨 2 点执行巡检
crontab -e

# 添加以下行
0 2 * * * cd ~/vps-inspector && ./run_inspection.sh >> logs/cron.log 2>&1

6.2 接入告警通知

支持多种通知方式：

# 邮件通知
import smtplib
from email.mime.text import MIMEText

def send_email_report(to_email, subject, html_content):
    msg = MIMEText(html_content, 'html', 'utf-8')
    msg['Subject'] = subject
    msg['To'] = to_email
    
    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login('your-email@gmail.com', 'your-app-password')
    server.send_message(msg)
    server.quit()

# Webhook 通知（企业微信、钉钉、飞书）
def send_webhook(url, message):
    import requests
    requests.post(url, json={"msg_type": "text", "content": {"text": message}})

# Telegram Bot 通知
def send_telegram(chat_id, message):
    import requests
    bot_token = "YOUR_BOT_TOKEN"
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    requests.post(url, json={"chat_id": chat_id, "text": message})

实际效果展示

经过一周的运行，我们的智能巡检系统发现了以下问题：

案例 1：暴力破解攻击

发现时间：周一凌晨 2:00

异常数据：

failed_logins: 47 次（阈值：10 次）
source_ips: 103.21.244.0/22, 185.220.101.0/24

AI 分析：

检测到来自两个 IP 段的持续暴力破解攻击。建议在防火墙中封禁这些 IP 段，并启用 Fail2Ban 自动屏蔽。

自动修复：

# 生成的修复脚本
ufw deny from 103.21.244.0/22
ufw deny from 185.220.101.0/24
systemctl restart fail2ban

结果：攻击在 5 分钟内被阻断，未造成任何损失。

案例 2：内存泄漏

发现时间：周三下午 3:00

异常数据：

memory_usage: 94%
top_process: docker (占 62%)

AI 分析：

Docker 容器内存使用率异常高，可能存在内存泄漏。建议重启 Docker 服务并清理未使用的容器和镜像。

自动修复：

# 生成的修复脚本
docker system prune -af
systemctl restart docker

结果：内存使用率从 94% 降至 45%，服务恢复正常。

案例 3：磁盘空间不足

发现时间：周五晚上 10:00

异常数据：

disk_usage: 89%
large_files: /var/log/journal (占 12GB)

AI 分析：

日志文件占用过多磁盘空间。建议配置日志轮转策略，限制日志文件大小和保留时间。

自动修复：

# 生成的修复脚本
journalctl --vacuum-time=3d
journalctl --vacuum-size=500M
systemctl restart systemd-journald

结果：释放了 15GB 磁盘空间。

进阶：多 VPS 统一视图

当管理多台 VPS 时，我们可以生成一个统一的健康度视图：

def generate_overall_report(all_vps_metrics):
    """Generate overall health report for all VPS."""
    
    total_score = sum(vps["health_score"] for vps in all_vps_metrics)
    avg_score = total_score / len(all_vps_metrics)
    
    critical_issues = [
        vps for vps in all_vps_metrics 
        if any(v["severity"] == "critical" for v in vps["violations"])
    ]
    
    report = {
        "overall_score": round(avg_score, 2),
        "total_vps": len(all_vps_metrics),
        "healthy_vps": len(all_vps_metrics) - len(critical_issues),
        "critical_vps": len(critical_issues),
        "top_risks": extract_top_risks(all_vps_metrics),
        "recommended_actions": generate_priority_actions(all_vps_metrics)
    }
    
    return report

总结

通过构建这套 AI Agent 驱动的 VPS 智能巡检系统，我们实现了：

自动化：每天自动执行全面健康检查，无需人工干预
智能化：利用 LLM 理解复杂模式，生成自然语言报告
主动化：提前发现潜在问题，避免故障发生
可执行：自动生成修复脚本，一键应用
可扩展：支持任意数量的 VPS，统一视图管理

关键成功因素：

数据采集要全面但不冗余
AI 分析要结合规则引擎，避免误报
修复脚本要经过测试，确保安全可靠
报告要简洁明了，突出关键信息

现在，你可以把 VPS 运维从"救火队"变成"预防医学"——在问题发生之前就发现并解决它。

下一步行动

在你的 VPS 上安装 Node Exporter
部署本文的采集脚本
配置 LLM 分析引擎
设置 Cron 定时任务
接入告警通知

记住：安全运维不是一次性的工作，而是持续的习惯。让 AI Agent 成为你的 24 小时运维助手。