AI 驱动的 VPS 进程智能诊断：实时异常检测、资源优化与自动调优

引言

你的 VPS 运行着多个服务——Web 服务器、数据库、缓存、定时任务、后台 Worker。某天你发现响应变慢了，但 top 显示 CPU 和内存使用率都正常。问题到底在哪里？

瓶颈往往藏在进程层面：某个 Python 脚本悄悄发生了内存泄漏，某个 Node.js 子进程变成了僵尸，某个 cron 任务同时堆积了数十个实例抢占了所有 I/O 带宽。传统监控只告诉你"系统很忙"，却不知道"哪个进程在搞鬼"。

AI 进程智能诊断的思路很简单：用本地 LLM 作为你的"进程侦探"——它持续采集进程级指标，理解进程行为模式，识别异常，然后给出可操作的优化建议。

为什么需要进程级 AI 诊断？

场景	传统方式	AI 进程诊断
内存泄漏检测	手动 `ps aux	grep memory`
CPU 飙升排查	人工逐进程排查	LLM 关联分析，定位根因进程
僵尸进程清理	定期手动清理	实时检测，自动清理或告警
资源争抢分析	凭经验猜测	时间序列关联，精确识别瓶颈
优化建议	搜索 StackOverflow	LLM 生成针对性调优方案

更重要的是，所有进程数据都在你自己的 VPS 上处理——不上传任何敏感信息到第三方。

架构设计

┌──────────────────────────────────────────────────────────────┐
│                      你的 VPS                                │
│                                                              │
│  ┌─────────────────┐    ┌──────────────────┐                 │
│  │  进程数据采集层   │───►│  LLM 分析引擎     │                 │
│  │                 │    │                  │                 │
│  │  • psutil 快照  │    │  • Ollama 本地    │                 │
│  │  • /proc 解析   │    │  • 行为模式分析   │                 │
│  │  • 历史趋势数据 │    │  • 异常检测算法   │                 │
│  │  • 进程树关系   │    │  • 根因推理       │                 │
│  └────────┬────────┘    └────────┬─────────┘                 │
│           │                      │                           │
│           ▼                      ▼                           │
│  ┌─────────────────┐    ┌──────────────────┐                 │
│  │  存储层          │    │  执行层           │                 │
│  │                 │    │                  │                 │
│  │  • SQLite 历史  │    │  • 自动清理脚本   │                 │
│  │  • JSON 日志    │    │  • 资源限制调整   │                 │
│  │  • 异常事件库   │    │  • 通知推送       │                 │
│  └─────────────────┘    └──────────────────┘                 │
└──────────────────────────────────────────────────────────────┘

第一步：安装 Ollama 与配置

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

# 拉取轻量级模型（适合进程分析）
ollama pull llama3.2:1b
ollama pull qwen2.5:1.5b

# 验证安装
ollama list

第二步：部署进程采集与诊断脚本

创建一个 process_intelligence.py：

#!/usr/bin/env python3
"""VPS 进程智能诊断系统 - 基于本地 LLM 的实时异常检测"""

import psutil
import subprocess
import json
import sqlite3
import os
from datetime import datetime, timedelta
from pathlib import Path

DB_PATH = Path.home() / ".vps_ai" / "process_diag.db"
OLLAMA_URL = "http://localhost:11434"
MODEL = "llama3.2:1b"

class ProcessIntelligence:
    def __init__(self):
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS process_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            pid INTEGER,
            name TEXT,
            cpu_percent REAL,
            memory_rss INTEGER,
            memory_vms INTEGER,
            num_threads INTEGER,
            status TEXT,
            nice INTEGER,
            io_read_bytes INTEGER,
            io_write_bytes INTEGER,
            children_count INTEGER
        )''')
        c.execute('''CREATE TABLE IF NOT EXISTS anomalies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT,
            pid INTEGER,
            process_name TEXT,
            anomaly_type TEXT,
            severity TEXT,
            description TEXT,
            recommendation TEXT,
            resolved INTEGER DEFAULT 0
        )''')
        c.execute('''CREATE INDEX IF NOT EXISTS idx_snapshot_ts ON process_snapshots(timestamp)''')
        c.execute('''CREATE INDEX IF NOT EXISTS idx_anomaly_ts ON anomalies(timestamp)''')
        conn.commit()
        conn.close()

    def collect_snapshot(self) -> dict:
        """采集当前进程快照"""
        processes = {}
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 
                                          'memory_info', 'status', 'num_threads',
                                          'nice', 'io_counters', 'children']):
            try:
                mem = proc.info['memory_info']
                io = proc.info['io_counters'] or (0, 0, 0, 0)
                children = len(proc.info['children'] or [])
                
                processes[proc.info['pid']] = {
                    'name': proc.info['name'],
                    'cpu': proc.info['cpu_percent'] or 0,
                    'rss': mem.rss if mem else 0,
                    'vms': mem.vms if mem else 0,
                    'threads': proc.info['num_threads'] or 1,
                    'status': proc.info['status'],
                    'nice': proc.info['nice'] or 0,
                    'io_read': io.read_bytes,
                    'io_write': io.write_bytes,
                    'children': children
                }
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        
        return {
            'timestamp': datetime.now().isoformat(),
            'processes': processes,
            'system': {
                'cpu_percent': psutil.cpu_percent(interval=0.1),
                'memory_percent': psutil.virtual_memory().percent,
                'load_avg': os.getloadavg() if hasattr(os, 'getloadavg') else (0, 0, 0)
            }
        }

    def store_snapshot(self, snapshot: dict):
        """存储快照到数据库"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        ts = snapshot['timestamp']
        for pid, info in snapshot['processes'].items():
            c.execute('''INSERT INTO process_snapshots 
                (timestamp, pid, name, cpu_percent, memory_rss, memory_vms,
                 num_threads, status, nice, io_read_bytes, io_write_bytes, children_count)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',
                (ts, pid, info['name'], info['cpu'], info['rss'], info['vms'],
                 info['threads'], info['status'], info['nice'],
                 info['io_read'], info['io_write'], info['children']))
        conn.commit()
        conn.close()

    def detect_anomalies(self, snapshot: dict) -> list:
        """检测进程异常（规则层面）"""
        anomalies = []
        now = datetime.now().isoformat()
        
        for pid, info in snapshot['processes'].items():
            # 高 CPU
            if info['cpu'] > 80:
                anomalies.append({
                    'type': 'high_cpu',
                    'severity': 'warning' if info['cpu'] < 95 else 'critical',
                    'description': f"进程 {info['name']} (PID {pid}) CPU 使用率 {info['cpu']:.1f}%",
                    'recommendation': '检查是否为计算密集型任务，考虑限流或迁移到专用容器'
                })
            
            # 内存异常增长（RSS > 2GB）
            if info['rss'] > 2 * 1024 * 1024 * 1024:
                anomalies.append({
                    'type': 'high_memory',
                    'severity': 'warning',
                    'description': f"进程 {info['name']} (PID {pid}) 内存占用 {info['rss']//1024//1024}MB",
                    'recommendation': '检查是否存在内存泄漏，考虑设置 cgroup 内存限制'
                })
            
            # 僵尸进程
            if info['status'] == 'zombie':
                anomalies.append({
                    'type': 'zombie_process',
                    'severity': 'error',
                    'description': f"发现僵尸进程 {info['name']} (PID {pid})",
                    'recommendation': '检查父进程是否正确回收子进程，必要时重启父进程'
                })
            
            # 线程数异常（> 200）
            if info['threads'] > 200:
                anomalies.append({
                    'type': 'high_threads',
                    'severity': 'warning',
                    'description': f"进程 {info['name']} (PID {pid}) 线程数 {info['threads']}",
                    'recommendation': '线程数过多可能导致上下文切换开销，考虑使用异步模型'
                })
        
        return anomalies

    def store_anomalies(self, anomalies: list):
        """存储异常记录"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        ts = datetime.now().isoformat()
        for a in anomalies:
            c.execute('''INSERT INTO anomalies 
                (timestamp, pid, process_name, anomaly_type, severity, description, recommendation)
                VALUES (?, ?, ?, ?, ?, ?, ?)''',
                (ts, a.get('pid'), a.get('name'), a['type'], a['severity'],
                 a['description'], a['recommendation']))
        conn.commit()
        conn.close()

    def query_history(self, hours: int = 24) -> list:
        """查询历史数据用于 LLM 分析"""
        conn = sqlite3.connect(DB_PATH)
        c = conn.cursor()
        since = (datetime.now() - timedelta(hours=hours)).isoformat()
        
        # 获取异常统计
        c.execute('''SELECT anomaly_type, severity, COUNT(*) as cnt, MAX(timestamp) as latest
                     FROM anomalies 
                     WHERE timestamp >= ?
                     GROUP BY anomaly_type, severity
                     ORDER BY cnt DESC''', (since,))
        stats = c.fetchall()
        
        # 获取 TOP 资源占用进程
        c.execute('''SELECT name, AVG(cpu_percent) as avg_cpu, AVG(memory_rss) as avg_mem,
                     COUNT(DISTINCT pid) as pid_count
                     FROM process_snapshots
                     WHERE timestamp >= ?
                     GROUP BY name
                     ORDER BY avg_cpu DESC
                     LIMIT 10''', (since,))
        top_procs = c.fetchall()
        
        conn.close()
        return {'anomaly_stats': stats, 'top_processes': top_procs}

    def generate_report(self, history: dict) -> str:
        """调用 LLM 生成分析报告"""
        import requests
        
        anomaly_summary = []
        for row in history['anomaly_stats']:
            anomaly_summary.append(f"- {row[0]} ({row[1]}): {row[2]} 次")
        
        proc_summary = []
        for row in history['top_processes']:
            cpu_mb = row[1] if row[1] else 0
            mem_mb = (row[2] / 1024 / 1024) if row[2] else 0
            proc_summary.append(f"- {row[0]}: CPU {cpu_mb:.1f}%, 内存 {mem_mb:.0f}MB")
        
        prompt = f"""你是一个 VPS 运维专家。基于以下进程分析数据，生成一份简洁的诊断报告。

异常统计（过去24小时）:
{chr(10).join(anomaly_summary) if anomaly_summary else '- 无异常'}

资源占用 TOP 进程:
{chr(10).join(proc_summary)}

请提供:
1. 总体健康评分 (0-100)
2. 关键发现（最多3条）
3. 建议操作（按优先级排序）

输出格式使用 Markdown。"""

        try:
            resp = requests.post(f"{OLLAMA_URL}/api/generate", json={
                'model': MODEL,
                'prompt': prompt,
                'stream': False
            }, timeout=60)
            return resp.json().get('response', '')
        except Exception as e:
            return f"LLM 调用失败: {e}"


if __name__ == "__main__":
    import sys
    pi = ProcessIntelligence()
    
    if len(sys.argv) > 1 and sys.argv[1] == "analyze":
        # 分析模式：查询历史并生成报告
        history = pi.query_history(24)
        report = pi.generate_report(history)
        print(report)
    else:
        # 采集模式：单次快照
        snapshot = pi.collect_snapshot()
        pi.store_snapshot(snapshot)
        anomalies = pi.detect_anomalies(snapshot)
        if anomalies:
            pi.store_anomalies(anomalies)
            print(json.dumps(anomalies, indent=2, ensure_ascii=False))
        else:
            print("No anomalies detected")

第三步：设置定时采集与智能分析

创建定时任务（crontab -e）：

# 每5分钟采集一次进程快照
*/5 * * * * python3 ~/.vps_ai/process_intelligence.py >> /dev/null 2>&1

# 每小时生成一次诊断报告
0 * * * * python3 ~/.vps_ai/process_intelligence.py analyze >> ~/.vps_ai/daily_report.md 2>&1

# 每天凌晨清理30天前的历史数据
0 3 * * * sqlite3 ~/.vps_ai/process_diag.db "DELETE FROM process_snapshots WHERE timestamp < datetime('now', '-30 days')"

或者使用 systemd timer（推荐）：

# /etc/systemd/system/vps-process-collect.service
[Unit]
Description=VPS Process Intelligence Collection
After=network.target

[Service]
Type=oneshot
User=root
ExecStart=/usr/bin/python3 /root/.vps_ai/process_intelligence.py

# /etc/systemd/system/vps-process-collect.timer
[Unit]
Description=VPS Process Intelligence Timer

[Timer]
OnBootSec=5min
OnUnitActiveSec=5min
AccuracySec=1min

[Install]
WantedBy=timers.target

systemctl enable --now vps-process-collect.timer

第四步：创建告警通知脚本

#!/usr/bin/env python3
"""基于 Telegram Bot 的进程异常通知"""

import sqlite3
import requests
from datetime import datetime, timedelta
from pathlib import Path

DB_PATH = Path.home() / ".vps_ai" / "process_diag.db"
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
CHAT_ID = "YOUR_CHAT_ID"

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

def check_unresolved():
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    since = (datetime.now() - timedelta(minutes=30)).isoformat()
    
    c.execute('''SELECT severity, description, recommendation 
                 FROM anomalies 
                 WHERE resolved = 0 AND timestamp >= ?
                 ORDER BY 
                   CASE severity 
                     WHEN 'critical' THEN 1 
                     WHEN 'error' THEN 2 
                     ELSE 3 END''', (since,))
    
    rows = c.fetchall()
    conn.close()
    
    if not rows:
        return
    
    lines = ["🚨 *VPS 进程异常告警*"]
    for sev, desc, rec in rows:
        icon = {"critical": "🔴", "error": "🟠", "warning": "🟡"}.get(sev, "⚪")
        lines.append(f"{icon} {desc}")
        if rec:
            lines.append(f"   ↳ {rec}")
    
    send_telegram("\n\n".join(lines))

if __name__ == "__main__":
    check_unresolved()

添加到 crontab：

*/10 * * * * python3 ~/.vps_ai/alert_checker.py >> /dev/null 2>&1

第五步：交互式诊断查询

你可以随时手动运行诊断查询：

# 查看最近24小时异常汇总
sqlite3 ~/.vps_ai/process_diag.db "
SELECT anomaly_type, severity, COUNT(*) 
FROM anomalies 
WHERE timestamp >= datetime('now', '-24 hours')
GROUP BY anomaly_type, severity
ORDER BY COUNT(*) DESC;"

# 查看 TOP 10 高 CPU 进程
sqlite3 ~/.vps_ai/process_diag.db "
SELECT name, AVG(cpu_percent) as avg_cpu, 
       AVG(memory_rss)/1024/1024 as avg_mem_mb
FROM process_snapshots
WHERE timestamp >= datetime('now', '-1 hour')
GROUP BY name
ORDER BY avg_cpu DESC
LIMIT 10;"

# 完整的 LLM 分析报告
python3 ~/.vps_ai/process_intelligence.py analyze

实际使用示例

假设你的 VPS 上有一个 Python Web 服务开始出现内存增长：

$ python3 ~/.vps_ai/process_intelligence.py analyze

LLM 返回：

## VPS 进程健康诊断报告

**总体健康评分: 72/100** ⚠️ 需关注

### 关键发现
1. **内存泄漏风险**: `uvicorn` 进程 RSS 在过去6小时内从 200MB 增长至 1.8GB，增长曲线呈线性
2. **I/O 争抢**: `postgres` 与 `redis` 进程在高峰时段存在 I/O wait 竞争
3. **僵尸进程残留**: 发现2个已终止但未回收的子进程

### 建议操作
1. **立即**: 对 uvicorn 进程设置 cgroup 内存限制 (建议上限 2GB)
2. **短期**: 配置 uvicorn 的 `--limit-max-requests 1000` 实现自动重启
3. **中期**: 迁移至异步 I/O 模型，减少与 PostgreSQL 的 I/O 争抢

**评分说明**: 内存增长是主要扣分项，建议优先处理。

进阶：进程行为基线学习

随着时间推移，系统会积累足够的进程行为数据。你可以让 LLM 学习正常的"基线模式"：

def learn_baseline(self, days: int = 7) -> dict:
    """学习进程行为的正常基线"""
    conn = sqlite3.connect(DB_PATH)
    c = conn.cursor()
    
    since = (datetime.now() - timedelta(days=days)).isoformat()
    
    # 按进程名聚合统计
    c.execute('''SELECT name, 
                     AVG(cpu_percent) as mean_cpu, 
                     STDDEV(cpu_percent) as std_cpu,
                     AVG(memory_rss) as mean_mem,
                     COUNT(*) as sample_count
                 FROM process_snapshots
                 WHERE timestamp >= ?
                 GROUP BY name
                 HAVING sample_count > 100''', (since,))
    
    baselines = {}
    for row in c.fetchall():
        baselines[row[0]] = {
            'mean_cpu': row[1],
            'std_cpu': row[2] or 5.0,
            'mean_mem': row[3],
            'samples': row[4]
        }
    
    conn.close()
    return baselines

有了基线后，异常检测就从"绝对阈值"升级为"统计异常"——即使 CPU 80% 在你的环境中是正常的，也会因为偏离基线而被标记。

总结

AI 驱动的 VPS 进程智能诊断系统 = 采集 + 存储 + 检测 + LLM 分析 + 通知。

核心优势：

本地化：所有数据不离开你的 VPS
低门槛：Ollama + Python 即可运行，2GB 内存 VPS 够用
可操作：不仅发现问题，还给出具体修复建议
持续学习：基线随时间演进，检测越来越精准

从今天开始，给你的 VPS 装上"过程大脑"吧——让它不仅知道你忙不忙，更知道你哪里出了问题、该怎么修。