LLM辅助告警聚合与降噪：从千条告警到3条根因的智能压缩

LLM辅助告警聚合与降噪从千条告警到3条根因的智能压缩一、凌晨三点的告警轰炸磁盘使用率 81%连接数高慢查询主从延迟——全是同一个根因的连锁反应某日凌晨 3:12监控系统在 5 分钟内产生了 247 条告警磁盘使用率超过 80%、连接数接近上限、慢查询暴增、主从延迟 120 秒、Buffer Pool 命中率暴跌……所有告警都是真实的但它们指向同一个根因凌晨 3:00 的一个全表扫描分析任务占满了磁盘 IO引发连锁反应。告警聚合Alert Aggregation和告警降噪Alert Suppression试图解决的核心问题是不是告警太少而是太多不是没有信息而是信息过载。传统的告警聚合依赖人工配置规则如如果 A 和 B 同时告警只发送 A但规则无法覆盖所有组合、维护成本随系统复杂度指数增长。LLM 的引入为告警聚合提供了一个全新的方案让模型理解告警之间的因果关系自动将千条告警压缩到几条根因告警并生成人类可读的故障摘要。二、告警时间线的因果推理与聚合模型flowchart TB A[告警流入br/247条/5分钟] -- B{时间窗口分组br/相邻60秒为一组} B -- C[告警拓扑图构建br/服务/主机/指标三元组] C -- D[因果关系推断br/LLM分析] subgraph LLM[LLM推理] D -- E[识别根因告警br/Time-Sequence分析] E -- F[判断连锁反应链br/A → B → C → D] F -- G[生成降噪决策] end G -- H{输出} H -- I[仅发送根因告警br/3条] H -- J[附带故障摘要br/1段描述] H -- K[压制从属告警br/244条静默] I -- L[(告警归档)] J -- L K -- M[压制窗口br/30分钟内同类不再告警]三层聚合策略时间密集聚合60 秒内的所有告警归为一组拓扑关联聚合同一主机/服务的多指标告警归为一组因果推理聚合LLM 核心分析告警间的因果时序关系定位根因三、告警聚合引擎的实现from typing import List, Dict, Optional, Tuple from datetime import datetime, timedelta from dataclasses import dataclass import json dataclass class Alert: 单条告警的结构化表示 alert_id: str timestamp: datetime host: str service: str metric: str current_value: float threshold: float severity: str # CRITICAL/WARNING/INFO description: str class AlertAggregator: LLM 驱动的告警聚合与降噪 AGGREGATION_WINDOW_SEC 60 # 时间聚合窗口 SUPPRESSION_WINDOW_MIN 30 # 压制窗口 SYSTEM_PROMPT 你是一个运维告警分析专家。请分析以下告警时间线完成告警聚合与降噪。 ## 分析任务 1. 识别根因告警root cause alerts——最先发生、导致其他告警的触发器 2. 识别连锁反应链——哪些告警是根因的连锁效应从属告警 3. 生成故障摘要——用1-2句话解释发生了什么 4. 给出优先级建议 ## 输出格式 { root_causes: [ { alert_id: 根因告警ID, reason: 为什么它是根因, timestamp: 时间 } ], suppressed_alerts: [压制告警ID列表], causal_chains: [ { root_alert_id: 根因ID, cascade: [连锁告警ID1, 连锁告警ID2], chain_description: 连锁反应描述 } ], summary: 故障摘要1-2句话, severity: 整体严重程度, recommended_actions: [建议处理步骤] } def __init__(self, llm_client): self.llm llm_client self.suppression_table: Dict[str, datetime] {} def aggregate(self, alerts: List[Alert]) - Dict: 对告警列表进行聚合与降噪 # 1. 时间窗口分组 grouped self._group_by_time(alerts) # 2. 检查压制表 active_alerts self._filter_suppressed(grouped) if not active_alerts: return { aggregated: True, reason: 所有告警已被压制, sent_alerts: [] } # 3. LLM 因果推理 if len(active_alerts) 3: result self._llm_aggregate(active_alerts) else: # 少于3条告警不需要聚合 result { root_causes: [{alert_id: a.alert_id, reason: 唯一告警} for a in active_alerts], suppressed_alerts: [], summary: f收到 {len(active_alerts)} 条告警, severity: max(a.severity for a in active_alerts), recommended_actions: [] } # 4. 更新压制表 self._update_suppression(active_alerts, result) # 5. 构建发送列表 root_alert_ids {rc[alert_id] for rc in result[root_causes]} sent_alerts [a for a in alerts if a.alert_id in root_alert_ids] return { aggregated: len(result.get(suppressed_alerts, [])) 0, summary: result.get(summary, ), root_cause_count: len(result.get(root_causes, [])), suppressed_count: len(result.get(suppressed_alerts, [])), causal_chains: result.get(causal_chains, []), sent_alerts: sent_alerts, recommended_actions: result.get(recommended_actions, []) } def _group_by_time(self, alerts: List[Alert]) - List[Alert]: 按时间窗口分组取最新窗口的告警 if not alerts: return [] alerts_sorted sorted(alerts, keylambda a: a.timestamp) latest_time alerts_sorted[-1].timestamp window_start latest_time - timedelta(secondsself.AGGREGATION_WINDOW_SEC) return [a for a in alerts_sorted if a.timestamp window_start] def _filter_suppressed(self, alerts: List[Alert]) - List[Alert]: 过滤被压制的告警 now datetime.now() return [ a for a in alerts if self._get_suppression_key(a) not in self.suppression_table or self.suppression_table[self._get_suppression_key(a)] now ] def _llm_aggregate(self, alerts: List[Alert]) - Dict: 使用 LLM 进行因果推理式聚合 # 构建告警时间线描述 alert_descriptions [] for a in sorted(alerts, keylambda x: x.timestamp): alert_descriptions.append( f[{a.alert_id}] {a.timestamp.strftime(%H:%M:%S)} f主机:{a.host} 服务:{a.service} f指标:{a.metric}{a.current_value} (阈值:{a.threshold}) f严重度:{a.severity} ) alert_timeline \n.join(alert_descriptions) # 构建额外的拓扑关联信息 host_groups {} for a in alerts: if a.host not in host_groups: host_groups[a.host] [] host_groups[a.host].append(a.service) topology_hint \n.join( f主机 {h}: {, .join(svcs)} for h, svcs in host_groups.items() ) prompt f告警时间线 (共 {len(alerts)} 条): {alert_timeline} 拓扑关联: {topology_hint} response self.llm.chat( systemself.SYSTEM_PROMPT, user_messageprompt, temperature0.2 ) return self._parse_response(response) def _get_suppression_key(self, alert: Alert) - str: 生成压制键 return f{alert.host}:{alert.service}:{alert.metric} def _update_suppression(self, alerts: List[Alert], result: Dict): 更新压制表 suppressed_ids set(result.get(suppressed_alerts, [])) for a in alerts: if a.alert_id in suppressed_ids: key self._get_suppression_key(a) self.suppression_table[key] ( datetime.now() timedelta(minutesself.SUPPRESSION_WINDOW_MIN) ) def _parse_response(self, response: str) - Dict: 解析 LLM 响应 try: json_match re.search(rjson\s*(.*?), response, re.DOTALL) if json_match: return json.loads(json_match.group(1)) return json.loads(response) except json.JSONDecodeError: return { root_causes: [], suppressed_alerts: [], summary: LLM 解析失败发送全部告警, severity: UNKNOWN } # 使用示例 def handle_alert_burst(alerts: List[Alert], aggregator: AlertAggregator): 处理告警爆发 result aggregator.aggregate(alerts) if result[aggregated]: print(f告警聚合完成:) print(f 输入: {len(alerts)} 条告警) print(f 根因: {result[root_cause_count]} 条) print(f 压制: {result[suppressed_count]} 条) print(f 摘要: {result[summary]}) for chain in result.get(causal_chains, []): print(f 因果链: {chain[chain_description]}) # 仅发送根因告警 for alert in result[sent_alerts]: send_alert(alert, extra_contextresult[summary]) else: # 所有告警直接发送 for alert in alerts: send_alert(alert)四、告警聚合的准确性边界准确率的数据在实际测试中LLM 告警聚合的根因识别准确率约为 82%但存在两种关键错误模式假阴性漏报根因LLM 将一个根因判定为从属告警。这种错误的代价极高——根因告警被静默故障得不到及时处理。假阳性多报告警LLM 将多个关联告警都判定为根因。代价是可以接受的——多几条告警总比漏掉根因好。推荐策略激进降噪 安全网。LLM 可以激进地进行降噪宁可多压也不少压但所有压制操作必须记录日志允许人工一键取消压制。五、总结LLM 辅助告警聚合的本质是用 AI 替代运维工程师的第一反应——看到一堆告警爆发时先找出最先发生的那一条。因果推理是核心价值LLM 理解磁盘满 → 写入失败 → 主从延迟的因果链这是规则引擎做不到的压制表是安全网即使 LLM 判断失误压制也只持续 30 分钟不会永久丢失告警人机协同是最终形态LLM 做初步聚合运维确认后一键执行——AI 提效率人保安全在实际故障场景中这套系统将凌晨 3 点的 247 条告警压缩为 3 条根因告警 1 段故障摘要。值班人员从30 分钟理清告警关系缩短到1 分钟看懂故障摘要——这就是告警降噪的工程价值。