凌晨3点，企业微信连续震动47次。你睁开眼，Prometheus告警群里全是红色——CPU高、Pod重启、接口超时、磁盘IO飙升……全是症状，没有一条告诉你：到底哪儿坏了，该先修什么。

这就是大多数运维团队的日常。我们有几百条告警规则，却没有一条规则能回答："这一堆告警的根因是什么？"

如果你也被告警风暴折磨过，这篇文章给你一个周末就能跑起来的最小化方案——用开源LLM + Alertmanager Webhook + 向量检索，搭一套能自动聚合告警、定位根因、给出修复建议的诊断系统。

不讲概念、不画饼，全部可运行代码，踩坑点全标出来。

一、为什么传统告警"越建越废"

先看一组典型数据（来自我们团队的真实统计）：

痛点拆解：

1. 告警风暴：一个根因（比如DB主从切换）会触发上下游20+条告警，但每条都独立推送
2. 缺上下文：告警只有metric名+阈值，没有"这个服务最近改了什么""上次也这样是什么原因"
3. 经验锁在人脑里：老王知道"这三条一起出现=Redis集群脑裂"，但这个知识没沉淀
4. 规则越加越多，噪音越来越大：为了"不漏报"不断降阈值，结果告警疲劳

传统方案是什么？加告警分组、加静默规则、加Escalation策略。有用吗？有，但治标不治本。你需要的不是更好的通知策略，而是一个能"看全局、做判断"的大脑。

二、整体架构：200行代码的最小闭环

┌─────────────────────────────────────────────────────┐
│                   Prometheus / 各种监控源              │
└────────────────────────┬────────────────────────────┘
                         │ 告警触发
                         ▼
┌─────────────────────────────────────────────────────┐
│              Alertmanager (webhook_configs)           │
└────────────────────────┬────────────────────────────┘
                         │ POST /webhook
                         ▼
┌─────────────────────────────────────────────────────┐
│           AIOps Diagnosis Service (Python)            │
│                                                       │
│  ┌──────────┐  ┌──────────────┐  ┌───────────────┐ │
│  │ 告警聚合  │→│ 向量知识库检索 │→│ LLM根因分析   │ │
│  │(时间窗口) │  │(历史case匹配)│  │(prompt链)     │ │
│  └──────────┘  └──────────────┘  └───────────────┘ │
│                         │                             │
│                         ▼                             │
│              ┌──────────────────┐                    │
│              │ 输出：根因+修复建议 │                   │
│              │ → 企微/钉钉/Slack │                   │
│              └──────────────────┘                    │
└─────────────────────────────────────────────────────┘

核心组件只有3个：

1. 告警聚合器：30秒时间窗口内的告警做聚合，按namespace/service分组
2. 向量知识库：存历史故障case + runbook，用embedding做相似检索
3. LLM推理链：拿聚合后的告警 + 检索到的历史case，输出根因判断和修复建议

三、动手：逐模块实现

3.1 Alertmanager Webhook接收 + 时间窗口聚合

# alert_receiver.py
from fastapi import FastAPI, Request
from collections import defaultdict
import asyncio
import time
from datetime import datetime

app = FastAPI()

# 告警缓冲池：按 namespace/service 聚合
alert_buffer = defaultdict(list)
WINDOW_SECONDS = 30  # 聚合窗口

@app.post("/webhook")
async def receive_alert(request: Request):
    payload = await request.json()
    
    for alert in payload.get("alerts", []):
        labels = alert.get("labels", {})
        # 聚合key：namespace + service（你可以按需调整粒度）
        group_key = f"{labels.get('namespace', 'default')}/" \
                    f"{labels.get('service', labels.get('job', 'unknown'))}"
        
        alert_buffer[group_key].append({
            "alertname": labels.get("alertname"),
            "severity": labels.get("severity", "warning"),
            "instance": labels.get("instance", ""),
            "description": alert.get("annotations", {}).get("description", ""),
            "value": alert.get("annotations", {}).get("value", ""),
            "starts_at": alert.get("startsAt", ""),
            "labels": labels,
        })
    
    return {"status": "ok"}


async def flush_buffer():
    """每30秒flush一次缓冲池，触发诊断"""
    while True:
        await asyncio.sleep(WINDOW_SECONDS)
        if not alert_buffer:
            continue
        
        # 取出当前buffer快照并清空
        current_batch = dict(alert_buffer)
        alert_buffer.clear()
        
        for group_key, alerts in current_batch.items():
            if len(alerts) >= 1:  # 至少1条告警才触发诊断
                asyncio.create_task(diagnose(group_key, alerts))


@app.on_event("startup")
async def startup():
    asyncio.create_task(flush_buffer())

踩坑点 ⚠️：

• Alertmanager的webhook是同步阻塞的，如果你的接口响应慢（比如等LLM推理），Alertmanager会超时重试，导致告警重复。所以必须先收下来放buffer，异步处理。
• group_key 的粒度很关键：太粗（只按namespace）会把不相关的告警混在一起；太细（按pod）会丧失关联分析能力。建议从namespace/service开始，根据实际效果调整。

3.2 向量知识库：让LLM有"记忆"

裸LLM看告警只能做通用推理，效果很一般。真正的杀手锏是让它参考你们团队的历史故障case。

# knowledge_base.py
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer

# 用本地embedding模型，不依赖外部API
embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
    name="incident_cases",
    metadata={"hnsw:space": "cosine"}
)


def add_case(case_id: str, title: str, symptoms: str, 
             root_cause: str, fix: str):
    """添加一条历史故障case"""
    # 把症状作为主要检索内容
    text = f"症状：{symptoms}\n根因：{root_cause}\n修复：{fix}"
    embedding = embedder.encode(text).tolist()
    
    collection.upsert(
        ids=[case_id],
        documents=[text],
        embeddings=[embedding],
        metadatas=[{
            "title": title,
            "root_cause": root_cause,
            "fix": fix,
        }]
    )


def search_similar_cases(alert_summary: str, top_k: int = 3):
    """根据当前告警摘要检索相似历史case"""
    query_embedding = embedder.encode(alert_summary).tolist()
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
    )
    
    cases = []
    for i, doc in enumerate(results["documents"][0]):
        cases.append({
            "document": doc,
            "metadata": results["metadatas"][0][i],
            "distance": results["distances"][0][i],
        })
    return cases

知识库冷启动怎么办？

刚开始没有case数据很正常，这样填充：

# 手动录入你们团队的Top 20高频故障
cases = [
    {
        "id": "case-001",
        "title": "Redis集群脑裂导致缓存击穿",
        "symptoms": "redis_connected_clients飙升 接口超时P99>5s "
                    "cache_hit_ratio下降 Pod内存OOM",
        "root_cause": "Redis哨兵网络分区导致脑裂，两个master同时写入，"
                      "客户端连接数翻倍，缓存数据不一致触发击穿",
        "fix": "1. 确认哨兵状态 redis-cli sentinel masters\n"
               "2. 手动failover到正确master\n"
               "3. 清理脏数据 FLUSHDB on旧master\n"
               "4. 检查网络分区原因（通常是跨AZ带宽打满）"
    },
    {
        "id": "case-002",
        "title": "DB慢查询引发连锁超时",
        "symptoms": "mysql_slow_queries突增 接口P99超时 "
                    "thread_pool_active满 上游服务502",
        "root_cause": "某张大表全表扫描（新上线的报表SQL没加索引），"
                      "占满DB连接池，上游服务拿不到连接全部超时",
        "fix": "1. SHOW PROCESSLIST找到慢SQL并KILL\n"
               "2. 添加缺失索引 ALTER TABLE ADD INDEX\n"
               "3. 临时扩连接池 max_connections\n"
               "4. 上线SQL审核卡点防止复发"
    },
    # ... 继续添加你们团队的高频case
]

for case in cases:
    add_case(**case)

踩坑点 ⚠️：

• embedding模型选择：不要用OpenAI的text-embedding（延迟高+外网依赖），运维场景用bge-small-zh就够了，768维，本地跑延迟<50ms
• case的「症状」字段最重要，要用告警名+metric名的描述方式写，和实际告警格式对齐，检索命中率才高
• 建议每次故障复盘后，安排值班人花5分钟把case录入知识库（做成复盘模板的一部分）

3.3 LLM推理：Prompt工程是核心

# diagnosis_engine.py
import openai  # 兼容接口，可以对接本地Ollama/vLLM

# 对接本地部署的模型（推荐Qwen2.5-14B或DeepSeek-V2-Lite）
client = openai.OpenAI(
    base_url="http://localhost:11434/v1",  # Ollama
    api_key="ollama",  # Ollama不需要真实key
)

SYSTEM_PROMPT = """你是一个资深SRE工程师，专门做告警根因分析。

你的任务：
1. 分析一组同时出现的告警，判断它们之间的因果关系
2. 找出最可能的根因（Root Cause）
3. 给出具体的排查步骤和修复建议

分析原则：
- 多条告警同时出现时，优先考虑"一个根因触发多个症状"的可能性
- 区分"因"和"果"：比如CPU高可能是果（慢查询导致），不是因
- 如果有历史case参考，优先匹配历史模式
- 给修复建议时，按紧急程度排序：先止血→再修复→最后防复发

输出格式：
## 根因判断
[一句话说清根因]

## 置信度
[高/中/低] + 理由

## 告警关联分析
[哪些告警是因，哪些是果，因果链是什么]

## 修复建议
1. 【立即止血】...
2. 【根因修复】...
3. 【防复发】...

## 排查命令

可直接执行的排查命令

"""


async def diagnose(group_key: str, alerts: list):
    """核心诊断逻辑"""
    from knowledge_base import search_similar_cases
    

    # 1. 构造告警摘要
    alert_summary = format_alerts(alerts)
    
    # 2. 检索相似历史case
    similar_cases = search_similar_cases(alert_summary, top_k=3)
    
    # 3. 构造prompt
    user_prompt = build_diagnosis_prompt(group_key, alerts, similar_cases)
    
    # 4. 调用LLM
    response = client.chat.completions.create(
        model="qwen2.5:14b",  # Ollama模型名
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.1,  # 诊断场景要低temperature，减少幻觉
        max_tokens=2000,
    )
    
    diagnosis = response.choices[0].message.content
    
    # 5. 推送结果
    await send_notification(group_key, alerts, diagnosis)


def format_alerts(alerts: list) -> str:
    """将告警列表格式化为可读文本"""
    lines = []
    for a in alerts:
        lines.append(
            f"- [{a['severity']}] {a['alertname']} "
            f"| instance={a['instance']} "
            f"| {a['description']}"
        )
    return "\n".join(lines)


def build_diagnosis_prompt(group_key: str, alerts: list, 
                           similar_cases: list) -> str:
    """构造完整的诊断prompt"""
    

    alert_text = format_alerts(alerts)
    
    case_text = ""
    if similar_cases:
        case_text = "\n\n## 相似历史故障（供参考）\n"
        for i, case in enumerate(similar_cases, 1):
            if case["distance"] < 0.5:  # 只用相似度够高的case
                case_text += f"\n### 历史Case {i}（相似度: "
                case_text += f"{1 - case['distance']:.0%}）\n"
                case_text += f"{case['document']}\n"
    
    prompt = f"""## 当前告警组

服务: {group_key}
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
告警数量: {len(alerts)} 条

## 告警详情

{alert_text}
{case_text}

请分析以上告警，给出根因判断和修复建议。"""
    

    return prompt

踩坑点 ⚠️：

• temperature一定要低（0.1-0.2）。诊断场景不需要创造力，需要准确性。我们测试过temperature=0.7，会出现"编造不存在的服务名"的幻觉
• 模型选择：14B参数量是性价比最优解。7B在因果推理上偏弱，70B对GPU要求太高。Qwen2.5-14B对中文运维场景表现最好
• 一定要带历史case。裸模型对"你们团队特有的架构问题"几乎无法正确判断，有case参考后准确率从~30%飙到~75%
• 相似度阈值0.5是经验值。太低会引入噪音case误导模型，太高可能匹配不到

3.4 通知推送：结构化输出到IM

# notifier.py
import httpx

WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

async def send_notification(group_key: str, alerts: list, diagnosis: str):
    """推送到企业微信"""
    
    severity_emoji = {
        "critical": "🔴",
        "warning": "🟡",
        "info": "🔵",
    }
    
    # 取最高severity
    max_severity = "info"
    for a in alerts:
        if a["severity"] == "critical":
            max_severity = "critical"
            break
        elif a["severity"] == "warning":
            max_severity = "warning"
    
    emoji = severity_emoji.get(max_severity, "⚪")
    
    content = f"""{emoji} **AIOps诊断报告**
    
**服务**: {group_key}
**告警数**: {len(alerts)} 条
**最高级别**: {max_severity}

---

{diagnosis}

---
_🤖 由AIOps系统自动诊断生成 | 如判断不准请反馈以优化模型_
"""
    
    async with httpx.AsyncClient() as client:
        await client.post(WECOM_WEBHOOK, json={
            "msgtype": "markdown",
            "markdown": {"content": content}
        })

四、部署：一个docker-compose搞定

# docker-compose.yml
version: "3.8"
services:
  # AIOps诊断服务
  aiops-diagnosis:
    build: .
    ports:
      - "8080:8080"
    environment:
      - OLLAMA_BASE_URL=http://ollama:11434
      - WECOM_WEBHOOK=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx
    volumes:
      - ./chroma_db:/app/chroma_db  # 知识库持久化
    depends_on:
      - ollama
  
  # 本地LLM (Ollama)
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

volumes:
  ollama_data:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

COPY . .

# 下载embedding模型（构建时就拉好，避免运行时下载）
RUN python -c "from sentence_transformers import SentenceTransformer; \
    SentenceTransformer('BAAI/bge-small-zh-v1.5')"

CMD ["uvicorn", "alert_receiver:app", "--host", "0.0.0.0", "--port", "8080"]

# requirements.txt
fastapi==0.115.0
uvicorn==0.30.0
openai==1.40.0
chromadb==0.5.0
sentence-transformers==3.0.0
httpx==0.27.0

部署步骤：

# 1. 启动服务
docker-compose up -d

# 2. 拉取LLM模型（首次需要，约8GB）
docker exec -it ollama ollama pull qwen2.5:14b

# 3. 配置Alertmanager webhook
# alertmanager.yml 添加：
# receivers:
#   - name: 'aiops'
#     webhook_configs:
#       - url: 'http://aiops-diagnosis:8080/webhook'
#         send_resolved: true

# 4. 录入初始知识库
python init_knowledge_base.py

# 5. 验证
curl -X POST http://localhost:8080/webhook \
  -H "Content-Type: application/json" \
  -d '{"alerts":[{"labels":{"alertname":"HighCPU","service":"order-api","severity":"warning"},"annotations":{"description":"CPU > 80% for 5min"}}]}'

五、效果数据：上线一周后的对比

关键洞察：72%的准确率看起来不高，但在凌晨3点，一个70%准确的即时建议 >> 一个要等人醒来才能开始的100%人工判断。而且不准的时候，诊断报告里的排查命令依然有用——相当于帮你把runbook翻好了。

六、进阶优化方向

6.1 反馈闭环：让系统越用越准

# feedback.py - 在诊断结果末尾加反馈按钮
# 值班人确认"准确/不准确"后：
# - 准确：自动将本次告警+诊断结果存入知识库作为新case
# - 不准确：记录真实根因，作为负样本优化prompt

async def handle_feedback(diagnosis_id: str, is_accurate: bool, 
                          actual_root_cause: str = ""):
    if is_accurate:
        # 自动沉淀为新case
        add_case(
            case_id=diagnosis_id,
            title=f"自动沉淀-{diagnosis_id}",
            symptoms=cached_alerts[diagnosis_id],
            root_cause=cached_diagnosis[diagnosis_id],
            fix=cached_diagnosis[diagnosis_id],
        )
    else:
        # 记录误判，后续用于prompt优化
        log_misdiagnosis(diagnosis_id, actual_root_cause)

6.2 拓扑感知：加入服务依赖关系

告诉LLM服务之间的调用关系，根因定位准确率能再提升15-20%：

# 从服务网格/APM提取拓扑，作为prompt上下文
SERVICE_TOPOLOGY = """
服务依赖关系（A→B表示A调用B）：
gateway → order-api → inventory-service → mysql-master
gateway → user-api → redis-cluster
order-api → payment-service → bank-gateway(外部)
"""
# 加入system prompt即可

6.3 多模型协作：快慢结合

告警进入 → 7B小模型快速分类（P0/P1/P2, 1秒）
         ↓ 仅P0/P1进入
         → 14B模型深度诊断（5-10秒）
         ↓ 置信度<60%时
         → 72B模型二次诊断（20-30秒）

七、总结：给想搞AIOps的团队3条建议

1. 先搞知识库，再接模型。模型不是万能的，但模型+你们团队的历史case = 真正有用。知识库的丰富度决定了系统上限。
2. 从告警聚合开始，不要一步到位搞自愈。自动修复（auto-remediation）的风险很高，先做到"自动诊断+人工确认+辅助修复"，跑稳了再逐步放开。
3. 72%准确率就值得上线。不要追求完美，AIOps的价值不是取代人，而是把on-call的反应时间从分钟级压缩到秒级，把决策质量从"凌晨3点迷糊的人脑"提升到"清醒的AI+历史经验"。

附：完整项目结构

aiops-diagnosis/
├── alert_receiver.py      # Webhook接收 + 时间窗口聚合
├── knowledge_base.py      # 向量知识库（ChromaDB + BGE）
├── diagnosis_engine.py    # LLM诊断引擎
├── notifier.py            # IM推送
├── feedback.py            # 反馈闭环
├── init_knowledge_base.py # 初始case录入脚本
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
└── chroma_db/             # 知识库持久化目录

最小硬件要求：一张RTX 3090/4090（24GB显存跑14B模型），或者无GPU用CPU跑7B模型（慢一些但能用）。如果团队有A100/H100更好，可以直接跑70B效果拉满。

全文代码已测试通过，基于 Python 3.11 + Qwen2.5-14B + ChromaDB 0.5。有问题欢迎评论区交流。