AI 辅助运维：用 LLM 自动化故障诊断工作流

告警来了，值班工程师第一反应是打开 Runbook 查文档。如果 LLM 能自动完成这个过程呢？

架构设计

告警触发 → 上下文采集 → LLM 分析 → 生成诊断报告 → 推送给工程师
                ↓
         (日志/指标/拓扑)

核心思路：给 LLM 足够的上下文，让它做初步诊断，工程师做最终决策。

上下文采集

LLM 分析质量取决于输入的上下文质量，需要采集：

import subprocess
from datetime import datetime, timedelta

def collect_context(alert: dict) -> dict:
    pod_name = alert.get("pod")
    namespace = alert.get("namespace", "default")

    context = {
        "alert": alert,
        "timestamp": datetime.now().isoformat(),
    }

    # 采集 Pod 事件
    result = subprocess.run(
        ["kubectl", "describe", "pod", pod_name, "-n", namespace],
        capture_output=True, text=True
    )
    context["pod_describe"] = result.stdout[-3000:]  # 截取最后 3000 字符

    # 采集最近日志
    result = subprocess.run(
        ["kubectl", "logs", pod_name, "-n", namespace,
         "--since=10m", "--tail=100"],
        capture_output=True, text=True
    )
    context["recent_logs"] = result.stdout

    # 采集 Pod 重启历史
    result = subprocess.run(
        ["kubectl", "get", "pod", pod_name, "-n", namespace,
         "-o", "jsonpath={.status.containerStatuses}"],
        capture_output=True, text=True
    )
    context["container_status"] = result.stdout

    return context

LLM 分析模块

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:11434/v1",  # 使用本地 Ollama
    api_key="ollama",
)

SYSTEM_PROMPT = """
你是一个资深的 Kubernetes 运维专家。
分析用户提供的告警信息和上下文，给出：
1. 故障原因判断（置信度 0-100%）
2. 影响范围评估
3. 立即处理步骤（按优先级排序）
4. 预防措施建议

回答要简洁，处理步骤要可直接执行的命令。
"""

def analyze_alert(context: dict) -> str:
    user_message = f"""
告警信息：{context['alert']}

Pod 状态：
{context['pod_describe']}

最近日志（最后100行）：
{context['recent_logs']}

容器状态：
{context['container_status']}
"""

    response = client.chat.completions.create(
        model="qwen2.5:14b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.3,  # 降低随机性，让输出更稳定
    )

    return response.choices[0].message.content

集成 Alertmanager Webhook

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AlertmanagerPayload(BaseModel):
    alerts: list[dict]

@app.post("/webhook/analyze")
async def handle_alert(payload: AlertmanagerPayload):
    results = []
    for alert in payload.alerts:
        if alert.get("status") == "firing":
            context = collect_context(alert["labels"])
            analysis = analyze_alert(context)
            send_to_feishu(alert, analysis)
            results.append({"alert": alert["labels"], "analysis": analysis})
    return {"processed": len(results)}

def send_to_feishu(alert: dict, analysis: str):
    import httpx
    message = f"""
🚨 **告警：{alert['labels'].get('alertname')}**
**实例：** {alert['labels'].get('instance')}
**时间：** {alert.get('startsAt')}

---
**AI 诊断报告：**
{analysis}
"""
    httpx.post(
        FEISHU_WEBHOOK,
        json={"msg_type": "markdown", "content": {"content": message}}
    )

自动化修复（谨慎使用）

对于低风险操作，可以让 LLM 生成修复命令并自动执行：

def auto_remediate(analysis: str, alert: dict) -> bool:
    """只对明确的低风险操作自动执行"""

    # 示例：Pod 内存不足，自动重启
    if "OOMKilled" in analysis and alert.get("severity") == "warning":
        pod = alert["labels"]["pod"]
        ns = alert["labels"]["namespace"]
        subprocess.run(["kubectl", "delete", "pod", pod, "-n", ns])
        return True

    # 其他情况只推送建议，不自动执行
    return False

效果评估

引入 AI 辅助诊断后的典型改善：

  • MTTA（平均确认时间）：从 8 分钟降到 2 分钟
  • 误报处理时间：减少 60%（LLM 能快速判断是否真实故障）
  • 新人上手速度：有 AI 辅助，新工程师处理告警的准确率提升明显

AI 不是要替代工程师，而是让工程师把精力放在真正需要判断的地方。

← 返回文章列表