AI 驱动的 VPS 自动运维：用 LLM 智能分析日志、预测故障与自动修复

# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合运维分析的轻量模型（2GB VPS 也能跑）
ollama pull qwen2.5:7b

# 或更轻量的选择（1GB VPS）
# ollama pull phi:mini

ollama ps
# 应该显示已运行的模型

mkdir -p /opt/ai-ops
vim /opt/ai-ops/collect_data.sh

#!/bin/bash
# /opt/ai-ops/collect_data.sh - 系统数据采集

TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
OUTPUT_DIR="/tmp/ai-ops-data"
mkdir -p "$OUTPUT_DIR"

echo "=== 系统概况 [$TIMESTAMP] ===" > "$OUTPUT_DIR/summary.txt"
echo "CPU 负载:" >> "$OUTPUT_DIR/summary.txt"
uptime >> "$OUTPUT_DIR/summary.txt"

echo -e "\n内存使用:" >> "$OUTPUT_DIR/summary.txt"
free -h >> "$OUTPUT_DIR/summary.txt"

echo -e "\n磁盘使用:" >> "$OUTPUT_DIR/summary.txt"
df -h / /var /tmp --output=source,pcent,used,avail,target 2>/dev/null >> "$OUTPUT_DIR/summary.txt"

echo -e "\nTOP 5 内存进程:" >> "$OUTPUT_DIR/summary.txt"
ps aux --sort=-%mem | head -6 >> "$OUTPUT_DIR/summary.txt"

echo -e "\n系统日志(最近5分钟):" >> "$OUTPUT_DIR/summary.txt"
journalctl --since "5 min ago" --no-pager -n 50 >> "$OUTPUT_DIR/summary.txt" 2>&1

echo -e "\nDocker 容器状态:" >> "$OUTPUT_DIR/summary.txt"
docker ps -a 2>/dev/null >> "$OUTPUT_DIR/summary.txt"

echo "Data collected: $TIMESTAMP"

chmod +x /opt/ai-ops/collect_data.sh
# 每 5 分钟采集一次
(crontab -l 2>/dev/null; echo "*/5 * * * * /opt/ai-ops/collect_data.sh") | crontab -

vim /opt/ai-ops/analyze.py

#!/usr/bin/env python3
"""
AI 运维分析引擎
读取采集数据 → 调用本地 LLM → 输出诊断结果和修复建议
"""
import json
import subprocess
import time
from pathlib import Path

DATA_DIR = Path("/tmp/ai-ops-data")
OLLAMA_URL = "http://localhost:11434"
MODEL = "qwen2.5:7b"

def read_system_data():
    """读取采集的系统数据"""
    summary_file = DATA_DIR / "summary.txt"
    if not summary_file.exists():
        return "暂无数据"
    return summary_file.read_text(encoding="utf-8")

def analyze_with_llm(system_data: str) -> dict:
    """调用本地 LLM 分析系统状态"""
    prompt = f"""你是一个 VPS 运维专家。分析以下系统数据，判断是否存在异常或潜在风险。

系统数据：
{system_data}

请按以下 JSON 格式输出分析结果（不要包含其他内容）：
{{
  "status": "normal|warning|critical",
  "issues": [
    {{
      "severity": "low|medium|high",
      "type": "问题类型（如：磁盘不足、内存压力、服务宕机等）",
      "detail": "具体问题描述",
      "recommendation": "建议的修复方案"
    }}
  ],
  "summary": "一句话总结系统状态"
}}
"""
    try:
        result = subprocess.run(
            ["curl", "-s", f"{OLLAMA_URL}/api/generate",
             "-d", json.dumps({
                 "model": MODEL,
                 "prompt": prompt,
                 "stream": False,
                 "options": {"temperature": 0.1}
             })],
            capture_output=True, text=True, timeout=120
        )
        response = json.loads(result.stdout)
        # 从响应中提取 JSON
        text = response.get("response", "{}")
        # 清理可能的 markdown 代码块
        if "```json" in text:
            text = text.split("```json")[1].split("```")[0]
        elif "```" in text:
            text = text.split("```")[1].split("```")[0]
        return json.loads(text.strip())
    except Exception as e:
        return {"status": "error", "summary": f"分析失败: {str(e)}"}

def execute_remediation(action: str):
    """执行修复命令（谨慎使用）"""
    safe_commands = {
        "clean_docker": "docker system prune -f --volumes",
        "restart_service": "systemctl restart docker",  # 示例
        "clear_logs": "journalctl --vacuum-time=3d",
        "clean_cache": "apt-get clean && rm -rf /tmp/*",
    }
    if action in safe_commands:
        subprocess.run(safe_commands[action], shell=True)
        return f"已执行: {action}"
    return f"跳过危险操作: {action}（需人工确认）"

def main():
    system_data = read_system_data()
    result = analyze_with_llm(system_data)

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] AI 运维状态: {result.get('status', 'unknown')}")
    print(f"总结: {result.get('summary', 'N/A')}")

    issues = result.get("issues", [])
    for issue in issues:
        print(f"  [{issue.get('severity', 'info').upper()}] {issue.get('type', '')}")
        print(f"    详情: {issue.get('detail', '')}")
        print(f"    建议: {issue.get('recommendation', '')}")
        # 高危问题自动修复
        if issue.get("severity") == "high":
            rec = issue.get("recommendation", "").lower()
            if "清理" in rec or "docker" in rec or "日志" in rec:
                # 尝试自动修复
                pass

    # 保存分析结果
    output = DATA_DIR / f"analysis_{int(time.time())}.json"
    output.write_text(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

chmod +x /opt/ai-ops/analyze.py
# 每 10 分钟分析一次
(crontab -l 2>/dev/null; echo "*/10 * * * * /usr/bin/python3 /opt/ai-ops/analyze.py >> /var/log/ai-ops.log 2>&1") | crontab -

vim /opt/ai-ops/notify.py

#!/usr/bin/env python3
"""发送 AI 运维告警到 Telegram"""
import json
import requests
from pathlib import Path

# 从环境变量读取
BOT_TOKEN = "YOUR_BOT_TOKEN"  # @BotFather 创建
CHAT_ID = "YOUR_CHAT_ID"      # 你的 Telegram ID
DATA_DIR = Path("/tmp/ai-ops-data")

def send_alert(message: str):
    url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": CHAT_ID,
        "text": message,
        "parse_mode": "Markdown"
    })

def check_and_notify():
    """检查最新分析结果，如有严重问题则发送告警"""
    analysis_files = sorted(DATA_DIR.glob("analysis_*.json"))
    if not analysis_files:
        return
    latest = json.loads(analysis_files[-1].read_text())
    status = latest.get("status", "normal")
    if status not in ("warning", "critical"):
        return  # 一切正常，不打扰

    # 读取系统数据补充上下文
    summary = DATA_DIR / "summary.txt"
    sys_info = summary.read_text().split("\n")[0] if summary.exists() else ""

    msg = f"🚨 *AI 运维告警*\n"
    msg += f"状态: {'⚠️ Warning' if status == 'warning' else '🔥 Critical'}\n"
    msg += f"总结: {latest.get('summary', 'N/A')}\n\n"

    for issue in latest.get("issues", []):
        emoji = {"low": "ℹ️", "medium": "⚡", "high": "🔴"}
        msg += f"{emoji.get(issue.get('severity',''), '•')} *{issue.get('type','')}*\n"
        msg += f"  {issue.get('detail','')}\n"
        msg += f"  💡 {issue.get('recommendation','')}\n\n"

    send_alert(msg)
    print(f"Alert sent: {status}")

if __name__ == "__main__":
    check_and_notify()

vim /opt/ai-ops/deploy.sh

#!/bin/bash
# AI 运维助手一键部署脚本
set -e

echo "🚀 部署 AI 运维助手..."

# 1. 安装 Ollama
if ! command -v ollama &>/dev/null; then
    echo "📦 安装 Ollama..."
    curl -fsSL https://ollama.com/install.sh | sh
fi

# 2. 拉取模型
echo "🤖 拉取 AI 模型..."
ollama pull qwen2.5:7b

# 3. 创建脚本目录
mkdir -p /opt/ai-ops /tmp/ai-ops-data

# 4. 设置采集任务（每5分钟）
echo "⏰ 设置定时采集..."
(crontab -l 2>/dev/null | grep -v "collect_data.sh"; echo "*/5 * * * * /opt/ai-ops/collect_data.sh") | crontab -

# 5. 设置分析任务（每10分钟）
echo "🔍 设置定时分析..."
(crontab -l 2>/dev/null | grep -v "analyze.py"; echo "*/10 * * * * /usr/bin/python3 /opt/ai-ops/analyze.py >> /var/log/ai-ops.log 2>&1") | crontab -

# 6. 安装 Python 依赖
pip3 install requests -q

echo "✅ 部署完成！AI 运维助手正在运行。"
echo "   - 每 5 分钟采集系统数据"
echo "   - 每 10 分钟 AI 分析一次"
echo "   - 严重问题时 Telegram 通知你"
echo ""
echo "📋 查看日志: tail -f /var/log/ai-ops.log"

{
  "status": "warning",
  "issues": [
    {
      "severity": "high",
      "type": "磁盘空间不足",
      "detail": "/var 分区使用率 94%，预计 2-3 天内写满。主要占用为 Docker 容器日志和 journald 系统日志。",
      "recommendation": "执行 journalctl --vacuum-time=3d 清理历史日志，然后 docker system prune -f 清理无用镜像和容器。"
    }
  ],
  "summary": "/var 分区磁盘空间告警，需立即清理日志和 Docker 资源。"
}

# 手动修复
journalctl --vacuum-time=3d
docker system prune -f

# 设置自动日志清理（防止再次发生）
echo "MaxUse=100M" >> /etc/systemd/journald.conf
systemctl restart systemd-journald

{
  "status": "critical",
  "issues": [
    {
      "severity": "high",
      "type": "容器服务宕机",
      "detail": "Nginx 容器 (nginx-proxy) 在 3 分钟前异常退出，退出码 137（OOM Kill）。当前内存使用 85%，OOM Killer 可能误杀了 Nginx。",
      "recommendation": "重启容器并增加内存限制：docker restart nginx-proxy；或在 docker-compose.yml 中设置 mem_limit: 256m 并调整 swappiness。"
    }
  ],
  "summary": "Nginx 容器被 OOM Killer 误杀，需要调整内存配置。"
}

pip3 install flask
vim /opt/ai-ops/dashboard.py

#!/usr/bin/env python3
"""AI 运维历史面板 - Flask Web 界面"""
import json
from flask import Flask, jsonify, render_template_string
from pathlib import Path

app = Flask(__name__)
DATA_DIR = Path("/tmp/ai-ops-data")

HTML = """
<!DOCTYPE html>
<html>
<head><title>AI 运维面板</title>
<style>
  body { font-family: sans-serif; max-width: 800px; margin: auto; padding: 20px; background: #0f172a; color: #e2e8f0; }
  .card { background: #1e293b; border-radius: 8px; padding: 16px; margin: 12px 0; }
  .critical { border-left: 4px solid #ef4444; }
  .warning { border-left: 4px solid #f59e0b; }
  .normal { border-left: 4px solid #22c55e; }
  .high { color: #ef4444; } .medium { color: #f59e0b; } .low { color: #94a3b8; }
  pre { background: #0f172a; padding: 8px; border-radius: 4px; overflow-x: auto; }
</style></head>
<body>
  <h1>🤖 AI 运维面板</h1>
  <p>最新分析时间: {{ time }}</p>
  {% for r in records %}
  <div class="card {{ r.status }}">
    <strong>状态: {{ r.status }}</strong>
    <p>{{ r.summary }}</p>
    {% for issue in r.issues %}
      <div>
        <span class="{{ issue.severity }}">[{{ issue.severity.upper() }}]</span>
        <strong>{{ issue.type }}</strong>
        <p>{{ issue.detail }}</p>
        <pre>{{ issue.recommendation }}</pre>
      </div>
    {% endfor %}
  </div>
  {% endfor %}
</body></html>
"""

@app.route("/")
def dashboard():
    records = []
    for f in sorted(DATA_DIR.glob("analysis_*.json"), reverse=True)[:20]:
        records.append(json.loads(f.read_text()))
    return render_template_string(
        HTML, records=records,
        time=records[0]["_time"] if records else "N/A"
    )

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9090)

python3 /opt/ai-ops/dashboard.py &
# 通过 http://你的VPS-IP:9090 访问