告警不只是通知：让Grafana大盘通过LLM自动生成故障上下文分析报告

一、告警通知的"信息赤字"：为什么收到告警后还要打开三个系统

值班工程师收到一条PagerDuty告警时，通常看到的信息只有：告警名称、触发值、时间戳。接下来需要打开Grafana看仪表盘曲线、打开ELK查相关日志、打开Tracing看调用链。这三个动作加起来可能需要3-5分钟——这恰恰是MTTD（Mean Time to Detect）中最容易被忽视的"上下文拼接时间"。

设想另一种体验：收到告警的同时，附带一段200字的分析摘要——"order-svc的p99延迟从20ms升至450ms，时间窗口与Redis实例cache-prod-03的CPU使用率突增重合。建议优先排查Redis慢查询，最近5分钟有3个KEYS *全量扫描操作。"

这个能力不是科幻。Grafana面板的截图数据 + Prometheus指标上下文 + LLM的推理能力，三者结合即可实现。

sequenceDiagram participant AM as AlertManager participant WH as Webhook服务 participant GF as Grafana API participant PM as Prometheus participant LLM as LLM推理 AM->>WH: 告警触发通知 WH->>GF: 获取相关面板截图<br/>GET /api/render/d/xxx GF-->>WH: PNG截图 WH->>PM: 查询相关指标上下文<br/>(5分钟窗口) PM-->>WH: 时序数据JSON WH->>LLM: 发送分析请求<br/>(截图+数据+提示词) LLM-->>WH: 生成分析报告 WH->>AM: 附加分析报告的增强告警 WH->>Chat: 推送到企业微信/飞书

二、架构拆解：从告警触发到报告生成的全链路

整个链路分为四个环节，架构设计遵循"松耦合、异步化"原则：

环节1：告警网关。AlertManager的Webhook Receiver接收原始告警，提取AlertName、Labels、Annotations中的关键信息。这个环节不需要改造，用AlertManager自带的webhook配置即可。

环节2：上下文采集器。根据告警Labels（如service=order-svc,instance=10.0.1.47:8080），并发查询三个数据源：Prometheus查询相关指标的时间序列（告警前后各5分钟），Grafana API渲染相关面板截图，ELK查询同时间段内的ERROR/WARN日志摘要。

环节3：LLM推理引擎。将上下文采集结果和告警原始信息打包为提示词，发送给LLM。提示词工程是这个环节的核心质量决定因素。

环节4：增强告警分发。将LLM分析报告附加到告警通知中，通过企业微信或飞书机器人推送到值班群。

#!/usr/bin/env python3 """ Grafana + LLM 智能告警解读服务 接收AlertManager告警 → 采集上下文 → LLM生成分析报告 → 分发增强告警 """ import asyncio import json import os from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from datetime import datetime, timedelta @dataclass class AlertContext: """告警上下文""" alert_name: str severity: str service: str instance: str trigger_value: float trigger_time: datetime prometheus_metrics: Dict[str, Any] = field(default_factory=dict) grafana_screenshots: List[str] = field(default_factory=list) elk_logs: List[str] = field(default_factory=list) class AlertContextCollector: """告警上下文采集器""" def __init__( self, prometheus_url: str, grafana_url: str, grafana_api_key: str, elk_url: str ): self.prom_url = prometheus_url self.grafana_url = grafana_url self.grafana_key = grafana_api_key self.elk_url = elk_url async def query_prometheus( self, query: str, start: datetime, end: datetime ) -> Dict[str, Any]: """ 查询Prometheus指标 Args: query: PromQL查询语句 start: 起始时间 end: 结束时间 Returns: 时序数据字典 """ import aiohttp url = f"{self.prom_url}/api/v1/query_range" params = { "query": query, "start": start.timestamp(), "end": end.timestamp(), "step": "15s" } try: async with aiohttp.ClientSession() as session: async with session.get( url, params=params, timeout=10 ) as resp: if resp.status == 200: return await resp.json() else: print(f"Prometheus查询失败: HTTP {resp.status}") return {"status": "error", "code": resp.status} except Exception as e: print(f"Prometheus连接异常: {e}") return {"status": "error", "error": str(e)} async def capture_grafana_panel( self, dashboard_uid: str, panel_id: int, time_range: str = "10m" ) -> Optional[bytes]: """ 渲染Grafana面板截图 Args: dashboard_uid: 仪表盘UID panel_id: 面板ID time_range: 时间范围 Returns: PNG格式的截图字节数据 """ import aiohttp # Grafana 8.0+ 的渲染API render_url = ( f"{self.grafana_url}/render/d-solo/{dashboard_uid}" ) params = { "panelId": panel_id, "width": 1000, "height": 500, "tz": "Asia/Shanghai", "from": f"now-{time_range}", "to": "now" } headers = { "Authorization": f"Bearer {self.grafana_key}" } try: async with aiohttp.ClientSession() as session: async with session.get( render_url, params=params, headers=headers, timeout=30 ) as resp: if resp.status == 200: return await resp.read() else: print(f"Grafana截图失败: HTTP {resp.status}") return None except Exception as e: print(f"Grafana连接异常: {e}") return None async def collect_context(self, alert: Dict) -> AlertContext: """采集告警的完整上下文""" ctx = AlertContext( alert_name=alert.get("labels", {}).get("alertname", "unknown"), severity=alert.get("labels", {}).get("severity", "warning"), service=alert.get("labels", {}).get("service", ""), instance=alert.get("labels", {}).get("instance", ""), trigger_value=float( alert.get("annotations", {}).get("value", "0") ), trigger_time=datetime.now() ) now = datetime.now() window_start = now - timedelta(minutes=10) # 并发采集多个数据源 tasks = [] # Prometheus查询 prom_queries = [ f'rate(http_request_duration_seconds_sum{{service="{ctx.service}"}}[5m])', f'container_memory_working_set_bytes{{pod=~"{ctx.service}.*"}}', f'rate(http_requests_total{{service="{ctx.service}",status=~"5.."}}[5m])', ] for q in prom_queries: tasks.append( self.query_prometheus(q, window_start, now) ) # Grafana截图 dashboard_map = { "api": ("api-overview", 2), "jvm": ("jvm-metrics", 1), } for ds_uid, pid in dashboard_map.values(): tasks.append( self.capture_grafana_panel(ds_uid, pid) ) results = await asyncio.gather(*tasks, return_exceptions=True) # 解析结果 prom_count = len(prom_queries) for i in range(prom_count): if not isinstance(results[i], Exception): ctx.prometheus_metrics[f"query_{i}"] = results[i] return ctx class LLMAnalyzer: """基于LLM的告警分析引擎""" def __init__(self, endpoint: str, model: str = "qwen2.5-7b-instruct"): self.endpoint = endpoint self.model = model def _build_analysis_prompt(self, ctx: AlertContext) -> str: """构建告警分析提示词""" prompt = f"""你是SRE运维专家。请分析以下告警并生成简明报告。 ## 告警信息 - 告警名称: {ctx.alert_name} - 严重程度: {ctx.severity} - 服务: {ctx.service} - 实例: {ctx.instance} - 触发值: {ctx.trigger_value} - 触发时间: {ctx.trigger_time} ## 要求 1. 判断根因方向（代码bug/资源不足/依赖故障/配置变更/网络问题） 2. 给出影响范围评估（单实例/单服务/级联） 3. 列出3条优先级从高到低的排查步骤 4. 如果有明确匹配的Runbook，引用其编号 请用Markdown格式回复，总字数不超过300字。""" return prompt async def analyze(self, ctx: AlertContext) -> str: """调用LLM生成分析报告""" import aiohttp prompt = self._build_analysis_prompt(ctx) payload = { "model": self.model, "messages": [ {"role": "system", "content": "你是SRE专家，回复简洁专业。"}, {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 512 } try: async with aiohttp.ClientSession() as session: async with session.post( self.endpoint, json=payload, timeout=aiohttp.ClientTimeout(total=15) ) as resp: if resp.status == 200: data = await resp.json() return data["choices"][0]["message"]["content"] return f"LLM分析失败: HTTP {resp.status}" except Exception as e: return f"LLM分析异常: {e}" class AlertEnhancer: """告警增强服务主控""" def __init__( self, collector: AlertContextCollector, analyzer: LLMAnalyzer ): self.collector = collector self.analyzer = analyzer async def process_alert(self, alert: Dict) -> Dict: """ 处理一条告警，返回增强后的告警 Args: alert: AlertManager格式的告警JSON Returns: 附加了llm_analysis字段的增强告警 """ # 1. 采集上下文 ctx = await self.collector.collect_context(alert) # 2. LLM分析 analysis = await self.analyzer.analyze(ctx) # 3. 增强告警 enhanced = dict(alert) enhanced["annotations"] = enhanced.get("annotations", {}) enhanced["annotations"]["llm_analysis"] = analysis enhanced["annotations"]["analyzed_at"] = ( datetime.now().isoformat() ) return enhanced # Flask Webhook入口（生产部署示例） async def handle_webhook(request): """AlertManager Webhook处理函数""" try: alerts = await request.json() except json.JSONDecodeError: return {"status": "error", "message": "Invalid JSON"}, 400 collector = AlertContextCollector( prometheus_url=os.getenv("PROMETHEUS_URL", "http://prom:9090"), grafana_url=os.getenv("GRAFANA_URL", "http://grafana:3000"), grafana_api_key=os.getenv("GRAFANA_API_KEY", ""), elk_url=os.getenv("ELK_URL", "http://elastic:9200") ) analyzer = LLMAnalyzer( endpoint=os.getenv( "LLM_ENDPOINT", "http://llm-svc:8000/v1/chat/completions" ) ) enhancer = AlertEnhancer(collector, analyzer) enhanced_alerts = [] for alert in alerts.get("alerts", []): enhanced = await enhancer.process_alert(alert) enhanced_alerts.append(enhanced) return {"status": "ok", "processed": len(enhanced_alerts)}, 200 if __name__ == "__main__": print("告警增强服务就绪") print("支持的数据源: Prometheus + Grafana + ELK") print("LLM推理端点配置: LLM_ENDPOINT环境变量")

三、提示词工程：让LLM产出合格的分析报告

提示词的质量直接决定LLM输出的可用性。经过多轮迭代，有效的提示词需要包含三个要素：

角色锚定。"你是SRE运维专家"这种角色声明能让LLM收敛到正确的知识域。模糊的角色如"你是一个AI助手"会产生泛化、无用的回复。

精准约束。必须明确域定回答范围——"判断根因方向（代码bug/资源不足/依赖故障/配置变更/网络问题）"将LLM的推理空间从开放域压缩到运维领域的5个分类，大幅降低幻觉率。

输出格式。指定"Markdown格式，总字数不超过300字"能防止LLM输出过于冗长的分析。告警分析需要的是"快速诊断"而非"学术论文"。

⚠️ 一个重要的避坑点：不要在提示词中要求LLM执行PromQL查询或调用API。LLM不具备实时系统访问能力，它的角色是"基于已有上下文做推断"，不是"替代监控系统"。

四、延迟与成本的现实约束

方案的两个现实约束需要在落地前评估：

端到端延迟。从AlertManager发出告警到LLM完成分析，完整链路耗时在5-10秒。主要瓶颈是Grafana截图渲染（2-4秒）和LLM推理（1-2秒）。在P0级故障场景中，5秒的延迟是可接受的——前提是告警通知本身就已经在第一时间送达，LLM分析是"附加增强"而非"阻塞条件"。

单告警分析成本。使用7B模型本地部署时，单次推理的GPU时间成本约$0.0001-0.0005。按每天200条告警计算，日成本不到$0.1。但如果使用GPT-4 API，单次推理成本约$0.01-0.03，日成本$2-6。对告警密集的生产环境，本地部署7B模型是最经济的选择。

五、总结

Grafana + LLM的智能告警解读不是替代传统监控，而是在传统监控的"告警→通知"之间插入一个"上下文拼接"环节。这个环节把值班工程师打开多个系统的时间从3-5分钟压缩到零，让告警通知从"告诉你出问题了"升级为"告诉你问题出在哪、怎么排查"。

方案投入产出的关键是在提示词中定义清晰的"根因方向分类"。分类不必求全，5个方向（代码bug/资源不足/依赖故障/配置变更/网络问题）覆盖了90%以上的生产告警场景。如果LLM无法确定具体方向而返回"建议人工排查"，这本身就是有价值的——至少排除了AI误判的风险。