指标异常检测——用 AI 做时序数据的智能监控
指标异常检测——用 AI 做时序数据的智能监控
适读人群：负责系统监控或 SRE 方向的工程师 | 阅读时长：约 13 分钟 | 核心价值：掌握 AI 异常检测在监控系统中的完整集成方案，重点解决误报率控制问题
两年前我们的告警系统是这样工作的：设一个阈值，比如 CPU > 80% 就告警。
有多烂呢？
每个大促前，值班工程师都要手动调高阈值，因为正常的大促流量会触发告警。大促结束之后再调回来。经常有人忘记调回来，导致真正的问题没有及时告警。
还有一类更麻烦的情况：接口响应时间从 100ms 涨到了 150ms，绝对值没超阈值，但这实际上是某个问题的早期信号。静态阈值完全检测不到这种渐进式劣化。
我花了几个月时间，在现有的 Prometheus + Grafana + PagerDuty 体系上，加了一层 AI 异常检测。这篇文章写的是这个方案的完整实现和效果数据。
为什么不直接用 Prometheus 的 AlertManager
Prometheus 的告警规则是静态的，这是根本限制。你可以写复杂的 PromQL 表达式，但本质上还是人工定义"什么算异常"。
时序数据的异常检测需要考虑：
- 周期性模式：周一和周日的流量不一样，凌晨和下午的指标不一样
- 趋势：是绝对值超阈值，还是变化率异常
- 上下文：同样是 CPU 80%，在空载时是问题，在大促期间可能正常
- 多指标联动：单个指标正常，但几个指标同时异常才是真问题
这些都是静态规则很难表达的。
技术方案选型
我调研了几种方案：
方案 A：Prophet（Facebook 开源） 时序预测模型，支持周期性和趋势分解，不需要 LLM。优点是快，缺点是纯统计方法，对于复杂的多指标关联不擅长。
方案 B：Isolation Forest + Z-score 经典的统计异常检测，实现简单，适合单指标检测。
方案 C：LLM 直接分析 把一段时间的指标数据发给 LLM，让它判断是否异常。准确但慢，成本高，不适合高频实时检测。
我最终采用的是混合方案：统计方法做高频实时检测，LLM 做低频的根因分析和告警聚合。
实时指标流 (每分钟)
|
v
[统计异常检测] -- Prophet/Z-score -- 触发候选告警
|
v
[LLM 聚合分析] -- 每 5 分钟批处理 -- 判断是否真实异常
|
v
[告警决策] -- 抑制误报，发送有意义的告警到 PagerDuty统计异常检测实现
import numpy as np
from prophet import Prophet
import pandas as pd
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
@dataclass
class AnomalySignal:
metric_name: str
timestamp: datetime
actual_value: float
expected_value: float
deviation_score: float # 偏差程度，越高越异常
direction: str # UP or DOWN
class StatisticalAnomalyDetector:
def __init__(self, metric_name: str):
self.metric_name = metric_name
self.model = None
self.trained = False
def train(self, historical_data: pd.DataFrame):
"""
historical_data: DataFrame with columns 'ds' (datetime) and 'y' (value)
建议用至少 2 周的历史数据训练，覆盖完整的周期性模式
"""
self.model = Prophet(
changepoint_prior_scale=0.05, # 对趋势变化不要太敏感
seasonality_prior_scale=10, # 允许较强的季节性
daily_seasonality=True,
weekly_seasonality=True,
interval_width=0.95 # 95% 置信区间
)
self.model.fit(historical_data)
self.trained = True
def detect(self, recent_data: pd.DataFrame,
window_minutes: int = 60) -> list[AnomalySignal]:
"""
检测最近 window_minutes 分钟内的异常点
"""
if not self.trained:
raise ValueError("Model not trained yet")
# 生成预测
forecast = self.model.predict(recent_data[['ds']])
anomalies = []
for _, row in recent_data.iterrows():
forecast_row = forecast[forecast['ds'] == row['ds']].iloc[0]
lower = forecast_row['yhat_lower']
upper = forecast_row['yhat_upper']
expected = forecast_row['yhat']
actual = row['y']
if actual < lower or actual > upper:
# 计算偏差分数（超出置信区间多少个标准差）
interval_width = (upper - lower) / 2
deviation = abs(actual - expected) / (interval_width / 1.96)
anomalies.append(AnomalySignal(
metric_name=self.metric_name,
timestamp=row['ds'],
actual_value=actual,
expected_value=expected,
deviation_score=deviation,
direction="UP" if actual > upper else "DOWN"
))
return anomaliesLLM 聚合分析——误报过滤的关键
统计方法的误报率大约是 5-10%（也就是 100 次告警里有 5-10 次是误报）。对于生产告警来说，这个误报率太高了。
LLM 聚合分析的目的是：把 5 分钟内的候选告警批量发给 LLM，让它判断这些告警是否构成真实问题，以及是否需要人工介入。
from openai import OpenAI
import json
def llm_alert_aggregation(
candidate_alerts: list[AnomalySignal],
system_context: str,
recent_events: list[str] # 最近的部署、变更记录
) -> dict:
"""
用 LLM 对候选告警做聚合分析，过滤误报
system_context: 系统的基本信息，如"这是订单服务，正常流量 1000 QPS"
recent_events: 最近发生的事件列表，如部署记录
"""
if not candidate_alerts:
return {"should_alert": False, "reason": "无候选告警"}
# 格式化告警信息
alerts_text = "\n".join([
f"- {a.metric_name}: 实际值={a.actual_value:.2f}, 预期={a.expected_value:.2f}, "
f"偏差分数={a.deviation_score:.1f}, 方向={a.direction}, 时间={a.timestamp}"
for a in candidate_alerts
])
events_text = "\n".join(recent_events) if recent_events else "无"
client = OpenAI()
prompt = f"""你是一个 SRE，正在分析监控告警。
系统信息：
{system_context}
最近发生的事件（部署、配置变更等）：
{events_text}
在过去 5 分钟内，以下指标被统计模型标记为异常：
{alerts_text}
请判断：
1. 这些异常是否构成需要人工介入的真实问题？
2. 如果是，最可能的原因是什么？
3. 建议的紧急程度（P1/P2/P3/无需告警）
注意：
- 如果异常与最近的部署/变更时间吻合，可能是正常的过渡状态
- 如果只有一个指标异常而相关指标正常，误报可能性较高
- 如果多个相关指标同时异常，真实问题可能性高
返回 JSON：
{{
"should_alert": true/false,
"urgency": "P1|P2|P3|NONE",
"summary": "简短摘要（用于告警通知）",
"likely_cause": "最可能的原因",
"recommended_actions": ["建议操作1", "建议操作2"],
"is_likely_false_positive": true/false,
"false_positive_reason": "如果误报，原因是什么"
}}"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.2
)
return json.loads(response.choices[0].message.content)与 PagerDuty 集成
import requests
def send_to_pagerduty(analysis_result: dict, integration_key: str):
"""
根据 LLM 分析结果决定是否发 PagerDuty 告警
"""
if not analysis_result.get("should_alert"):
return # LLM 认为不需要告警，直接返回
urgency = analysis_result.get("urgency", "P3")
# 映射紧急程度到 PagerDuty severity
severity_map = {
"P1": "critical",
"P2": "error",
"P3": "warning"
}
payload = {
"routing_key": integration_key,
"event_action": "trigger",
"payload": {
"summary": analysis_result.get("summary", "AI 检测到异常"),
"severity": severity_map.get(urgency, "warning"),
"source": "AI-Anomaly-Detector",
"custom_details": {
"likely_cause": analysis_result.get("likely_cause", ""),
"recommended_actions": analysis_result.get("recommended_actions", []),
"ai_analysis": analysis_result
}
}
}
response = requests.post(
"https://events.pagerduty.com/v2/enqueue",
json=payload,
headers={"Content-Type": "application/json"}
)
if response.status_code != 202:
print(f"PagerDuty 告警发送失败: {response.status_code} {response.text}")实际效果数据
这个方案运行了大约 4 个月，以下是数据：
告警数量对比：
- 旧方案（静态阈值）：平均每天 23 条告警
- 新方案（AI 过滤后）：平均每天 6 条告警
- 告警减少了 74%
漏报情况：
- 4 个月内真实故障 8 次
- 新方案全部在故障发生 5 分钟内触发告警
- 旧方案有 2 次延迟告警（因为阈值设得保守）
误报率：
- 旧方案误报率：约 40%（10 条告警里有 4 条是正常的）
- 新方案误报率：约 8%（100 条告警里有 8 条是误报）
最让我们有体感的变化是：值班工程师的告警疲劳大幅减少。以前每天处理 20+ 条告警，大多数是"看了一眼没问题关掉"。现在每天 6 条，每一条都需要认真处理，值班工程师开始认真对待每一条告警了。
一个重要的教训
系统上线初期，我们遇到了一次 LLM 漏报：一个缓慢的内存泄漏（内存使用每小时增长 0.5%），统计方法把它标记为候选告警，但 LLM 分析后认为是"正常的渐进式增长，不需要告警"。
结果 6 小时后 OOM 了。
复盘原因：LLM 在分析时只看了 5 分钟的窗口，没有意识到这个增长趋势已经持续了 6 小时。
解决方案：对于渐进式趋势异常，给 LLM 提供更长时间窗口的数据摘要（如最近 1 小时、6 小时的趋势概要），不只是当前 5 分钟的数据点。
AI 异常检测不是万能的。它减少了误报，但不消除漏报风险。关键业务指标依然需要保留一些静态阈值作为最后防线。