AIOps 落地实战:别搞“PPT运维”!三步走教你用 Python 搭建智能告警引擎

import numpy as npfrom datetime import datetimeclass DynamicAlertEngine: def __init__(self, history_data): ””” history_data: 过去7天同一时段的历史指标数据（如CPU使用率） ””” self.history_data = np.array(history_data)# 计算历史数据的 IQR（四分位距） self.q1 = np.percentile(self.history_data, 25) self.q3 = np.percentile(self.history_data, 75) self.iqr = self.q3 - self.q1 self.median = np.median(self.history_data)# 设定异常边界：超过上下界 1.5 倍 IQR 视为异常 self.lower_bound = self.q1 - 1.5 * self.iqr self.upper_bound = self.q3 + 1.5 * self.iqr def check_anomaly(self, current_value, metric_name=”CPU”): ””” 实时检测当前值是否异常 ””” timestamp = datetime.now().strftime(”%Y-%m-%d %H:%M:%S”) if current_value > self.upper_bound: msg = (f”🚨 [{timestamp}] {metric_name} 异常飙升！\n” f”当前值: {current_value:.2f}% | 历史中位数: {self.median:.2f}%\n” f”动态上限: {self.upper_bound:.2f}% | 偏离度: {current_value - self.upper_bound:.2f}%”) return True, msg elif current_value < self.lower_bound: msg = (f”⚠️ [{timestamp}] {metric_name} 异常跌落！\n” f”当前值: {current_value:.2f}% | 历史中位数: {self.median:.2f}%\n” f”动态下限: {self.lower_bound:.2f}%”) return True, msg else: return False, f”✅ [{timestamp}] {metric_name} 运行在正常动态基线内 ({current_value:.2f}%)”# --- 模拟实战测试 ---# 假设过去7天中午12点的CPU历史数据是：[60, 62, 65, 61, 63, 64, 60]history_cpu = [60, 62, 65, 61, 63, 64, 60]engine = DynamicAlertEngine(history_cpu)# 场景1：正常业务高峰，CPU 68%，传统固定阈值(80%)会报警，但 AI 知道这是正常的is_alert, msg1 = engine.check_anomaly(68)print(msg1) # 场景2：突发异常，CPU 飙到 85%，AI 精准捕获并给出偏离度is_alert, msg2 = engine.check_anomaly(85)print(msg2)

# 伪代码：AI 拓扑根因推断def find_root_cause(anomaly_service, topology_graph):# 1. 获取该服务的所有上下游依赖 dependencies = topology_graph.get_dependencies(anomaly_service)# 2. 并行检查这些依赖是否也发生了异常 for dep in dependencies: if dep.status == ”anomaly” and dep.anomaly_time <= anomaly_service.anomaly_time:# 3. 发现上游更早发生异常，判定为根因 return f”根因定位：上游服务 {dep.name} 发生 {dep.error_type}，导致 {anomaly_service} 级联超时。” return ”未发现上游异常，请排查本服务内部逻辑或数据库锁。”