AIOps 实践与智能运维
一句话：AIOps 不是”AI 取代运维”，而是用算法处理人脑处理不了的海量监控数据——异常检测、告警降噪、根因推断、容量预测——让告警更准、定位更快、容量更合理。
概述
SRE Agent 侧重”Agent 架构 + 工具串联”，AIOps 侧重”算法 + 数据分析”。两者互补：
SRE Agent = 用 LLM 做推理和工具调用（侧重"怎么执行"）
AIOps = 用 ML/统计算法做异常检测和模式识别（侧重"怎么发现"）
AIOps 能力全景
graph TD
subgraph 数据层
METRICS[指标流]
LOGS[日志流]
TRACES[追踪流]
EVENTS[变更事件]
end
subgraph 算法层
AD[异常检测<br/>时序分析 / 统计模型]
CD[告警降噪<br/>聚类 / 关联 / 抑制]
RCA[根因推断<br/>拓扑关联 / 因果推断]
CP[容量预测<br/>趋势预测 / 容量推演]
end
subgraph 输出层
SMART_ALERT[智能告警]
RCA_REPORT[根因报告]
CAPACITY_PLAN[容量建议]
end
METRICS --> AD
LOGS --> AD
TRACES --> RCA
EVENTS --> RCA
AD --> CD
CD --> SMART_ALERT
RCA --> RCA_REPORT
METRICS --> CP
CP --> CAPACITY_PLAN
能力一：智能异常检测
传统告警的问题
传统阈值告警的困境：
- 静态阈值：白天 1000 QPS 正常，凌晨 100 QPS 也正常 → 固定阈值误报多
- 窗口固定：周末 vs 工作日、节假日 vs 日常 → 季节性模式被忽略
- 单维度：只看 CPU 或只看 QPS → 多维关联异常无法发现
- 噪音大：阈值敏感 → 误报多 → 告警疲劳 → 真正故障被忽略
时序异常检测算法
| 算法 | 类型 | 适用场景 | 优点 | 缺点 |
|---|
| 3-Sigma / Z-Score | 统计 | 正态分布数据 | 简单快速 | 不适合非正态分布 |
| IQR | 统计 | 非正态分布 | 鲁棒性强 | 灵敏度低 |
| EWMA | 平滑 | 渐变趋势 | 响应快 | 滞后 |
| Holt-Winters | 分解 | 季节性数据 | 适合周期模式 | 需调参 |
| Prophet | 分解 | 复杂季节性 | 自动调参 | 较重 |
| Isolation Forest | ML | 多维异常 | 无需分布假设 | 需训练 |
| LSTM-AE | 深度学习 | 复杂时序 | 捕捉长期依赖 | 需大量数据 |
Python 实现示例
import numpy as np
from collections import deque
class MultiStrategyAnomalyDetector:
"""多策略异常检测器"""
def __init__(self, strategy="auto"):
self.strategy = strategy
def detect(self, series: list[float],
timestamps: list[float]) -> list[dict]:
"""检测时序异常"""
if self.strategy == "auto":
# 根据数据特征自动选择策略
strategy = self._select_strategy(series)
else:
strategy = self.strategy
if strategy == "zscore":
return self._zscore_detect(series, timestamps)
elif strategy == "iqr":
return self._iqr_detect(series, timestamps)
elif strategy == "ewma":
return self._ewma_detect(series, timestamps)
elif strategy == "seasonal":
return self._seasonal_detect(series, timestamps)
def _select_strategy(self, series):
"""自动选择策略"""
# 检查是否有季节性
if self._has_seasonality(series):
return "seasonal"
# 检查是否正态分布
if self._is_normal(series):
return "zscore"
return "iqr"
def _zscore_detect(self, series, timestamps, threshold=3):
"""Z-Score 检测"""
mean = np.mean(series)
std = np.std(series)
anomalies = []
for i, (val, ts) in enumerate(zip(series, timestamps)):
z = (val - mean) / std if std > 0 else 0
if abs(z) > threshold:
anomalies.append({
"timestamp": ts,
"value": val,
"zscore": z,
"expected": mean,
"deviation": f"{abs(z):.1f} sigma",
})
return anomalies
def _ewma_detect(self, series, timestamps, alpha=0.3, threshold=3):
"""EWMA 检测：适合渐变趋势"""
ewma = [series[0]]
anomalies = []
for i in range(1, len(series)):
new_ewma = alpha * series[i] + (1 - alpha) * ewma[-1]
ewma.append(new_ewma)
# 用 EWMA 的方差作为动态阈值
ewma_std = np.std(ewma[max(0, i-20):i+1]) # 滑动窗口
if ewma_std > 0:
z = (series[i] - new_ewma) / ewma_std
if abs(z) > threshold:
anomalies.append({
"timestamp": timestamps[i],
"value": series[i],
"expected": new_ewma,
"zscore": z,
})
return anomalies
def _has_seasonality(self, series):
"""检查季节性（简化版：自相关）"""
if len(series) < 48: # 至少 2 天的小时数据
return False
lag = 24 # 假设日周期
if len(series) <= lag:
return False
corr = np.corrcoef(series[:-lag], series[lag:])[0, 1]
return corr > 0.5
def _is_normal(self, series):
"""检查是否近似正态分布"""
if len(series) < 30:
return True # 数据少时假设正态
# 用 Shapiro-Wilk 检验（简化版）
mean = np.mean(series)
std = np.std(series)
if std == 0:
return True
skewness = np.mean(((np.array(series) - mean) / std) ** 3)
return abs(skewness) < 1 # 偏度 < 1
Prometheus 异常检测规则
# 基于 predict_linear 的异常检测
- alert: AnomalyDetected
expr: |
# 实际值偏离预测值超过 3 倍标准差
abs(
rate(http_requests_total[5m])
- predict_linear(rate(http_requests_total[7d])[7d:1h], 0)
) > 3 * stddev_over_time(rate(http_requests_total[7d])[7d:1h])
for: 10m
labels:
severity: warning
type: anomaly
annotations:
summary: "{{ $labels.service }} 流量异常检测触发"
# 基于历史同比的异常检测
- alert: DailyComparisonAnomaly
expr: |
# 当前 1h QPS vs 昨天同时段 1h QPS，偏差 > 50%
abs(
sum(rate(http_requests_total[1h])) by (service)
- sum(rate(http_requests_total[1h] offset 1d)) by (service)
) / sum(rate(http_requests_total[1h] offset 1d)) by (service) > 0.5
for: 15m
labels:
severity: warning
type: anomaly
能力二：告警智能降噪
问题
告警风暴场景：
14:00: 节点 A NotReady
14:00:01 节点 A 上 Pod-1 down（×10 个 Pod）
14:00:02 节点 A 上 Pod-2 down
14:00:03 服务 A 依赖 Pod-1，成功率下降告警
14:00:04 服务 B 依赖 Pod-2，延迟升高告警
14:00:05 服务 C 依赖服务 A，级联告警
...
→ On-Call 收到 50+ 条告警，根本看不过来
降噪策略
| 策略 | 做法 | 效果 |
|---|
| 告警分组 | 同一服务/节点的告警合并为一条 | 50 条 → 5 条 |
| 因果关联 | 识别因果链，只通知根因告警 | 5 条 → 1 条 |
| 抑制（Inhibit） | 高严重度告警抑制低严重度告警 | 减少噪音 |
| 频率限制 | 同一告警 N 分钟内只通知一次 | 防告警风暴 |
| 智能路由 | 按根因分类路由到不同 On-Call | 减少误叫 |
Alertmanager 降噪配置
route:
group_by: ['service', 'node'] # 按服务和节点分组
group_wait: 30s # 等 30 秒收集同组告警
group_interval: 5m # 同组告警间隔
repeat_interval: 4h # 重复通知间隔
receiver: 'default'
routes:
- match:
severity: critical
receiver: 'pagerduty'
group_wait: 10s # 紧急告警快速通知
- match:
severity: warning
receiver: 'ticket'
group_wait: 60s # 非紧急等 1 分钟
inhibit_rules:
# 节点 NotReady 时抑制该节点上的所有 Pod 告警
- source_match:
alertname: NodeNotReady
target_match_re:
alertname: PodDown|PodCrashLoop
equal: ['node']
# 服务 critical 告警时抑制 warning 告警
- source_match:
severity: critical
target_match:
severity: warning
equal: ['service']
智能根因关联
class AlertCorrelator:
"""告警关联分析：识别因果链"""
def __init__(self):
self.dependency_graph = {} # 服务依赖图
self.alert_history = deque(maxlen=1000)
def correlate(self, new_alert: dict) -> dict:
"""分析新告警与历史告警的关联"""
# Step 1: 找时间窗口内的告警（前 5 分钟）
recent = [
a for a in self.alert_history
if (new_alert["timestamp"] - a["timestamp"]).total_seconds() < 300
]
# Step 2: 按依赖关系构建因果链
causal_chain = self._find_causal_chain(new_alert, recent)
# Step 3: 判断是否是根因
if not causal_chain:
# 没有上游告警 → 可能是根因
self.alert_history.append(new_alert)
return {
"alert": new_alert,
"is_root_cause": True,
"suppressed_alerts": [],
}
else:
# 有上游告警 → 是级联告警，抑制
self.alert_history.append(new_alert)
return {
"alert": new_alert,
"is_root_cause": False,
"caused_by": causal_chain[0],
"suppressed_alerts": [new_alert["alert_id"]],
"root_cause_alert": causal_chain[-1],
}
def _find_causal_chain(self, alert, recent):
"""基于依赖图找因果链"""
chain = []
for r in recent:
# 如果告警的服务依赖了上游告警的服务
if r["service"] in self.dependency_graph.get(alert["service"], []):
chain.append(r)
return chain
能力三：根因自动推断
拓扑关联分析
class RootCauseInferencer:
"""根因推断：基于服务拓扑 + 时间窗口"""
def __init__(self, topology):
"""
topology: 服务调用拓扑
{
"api-gateway": ["order-svc", "user-svc"],
"order-svc": ["mysql", "redis"],
"user-svc": ["mysql", "kafka"],
}
"""
self.topology = topology
def infer(self, anomalies: list[dict]) -> dict:
"""
输入一批同时段异常，推断最可能的根因
anomalies = [
{"service": "api-gateway", "metric": "error_rate", "value": 0.15, "time": T},
{"service": "order-svc", "metric": "error_rate", "value": 0.30, "time": T-10},
{"service": "mysql", "metric": "connections", "value": 0.95, "time": T-30},
]
"""
# Step 1: 按时间排序（最早的异常最可能是根因）
sorted_anomalies = sorted(anomalies, key=lambda x: x["time"])
# Step 2: 构建异常服务集合
anomaly_services = {a["service"] for a in sorted_anomalies}
# Step 3: 找最深的根（没有上游异常的服务）
candidates = []
for svc in anomaly_services:
upstream = self.topology.get(svc, [])
upstream_anomalies = set(upstream) & anomaly_services
if not upstream_anomalies:
# 没有上游异常 → 最可能的根因
candidates.append(svc)
# Step 4: 如果有多个候选，取最早的
if candidates:
earliest = min(
candidates,
key=lambda s: next(
a["time"] for a in sorted_anomalies if a["service"] == s
)
)
return {
"root_cause_service": earliest,
"confidence": "high" if len(candidates) == 1 else "medium",
"anomaly_chain": self._build_chain(earliest, anomaly_services),
"all_anomalies": sorted_anomalies,
}
return {"root_cause_service": None, "confidence": "low"}
def _build_chain(self, root, anomaly_services):
"""从根因开始构建影响链"""
chain = [root]
visited = {root}
# BFS 找下游被影响的服务
queue = [root]
while queue:
svc = queue.pop(0)
for downstream, deps in self.topology.items():
if svc in deps and downstream in anomaly_services and downstream not in visited:
chain.append(downstream)
visited.add(downstream)
queue.append(downstream)
return chain
能力四：容量智能预测
class CapacityPredictor:
"""容量预测：基于历史趋势预测资源需求"""
def predict(self, history: list[float], days_ahead: int = 7) -> dict:
"""
基于历史数据预测未来容量需求
"""
from numpy.polynomial import polynomial as P
x = np.arange(len(history))
y = np.array(history)
# 线性回归
coeffs = np.polyfit(x, y, 1)
slope, intercept = coeffs
# 预测
future_x = np.arange(len(history), len(history) + days_ahead * 24)
prediction = slope * future_x + intercept
# 计算何时达到容量上限
capacity_limit = self._get_capacity_limit()
if slope > 0:
days_until_full = int((capacity_limit - intercept) / slope / 24 - len(history) / 24)
else:
days_until_full = -1 # 不增长
return {
"current_value": history[-1],
"predicted_7d": prediction[-1],
"predicted_30d": slope * (len(history) + 30 * 24) + intercept,
"capacity_limit": capacity_limit,
"days_until_full": days_until_full,
"growth_rate_daily": slope * 24,
"recommendation": (
f"URGENT: {days_until_full} 天后容量耗尽" if 0 < days_until_full < 14
else f"PLAN: {days_until_full} 天后需扩容" if days_until_full < 30
else "OK: 容量充足"
),
}
AIOps 成熟度
| 级别 | 特征 | 技术栈 |
|---|
| L1 | 静态阈值告警 | Prometheus + Alertmanager |
| L2 | 动态阈值 + 基于同比 | predict_linear + 历史对比 |
| L3 | 异常检测算法 | 统计模型 + 时序分解 |
| L4 | 智能降噪 + 根因推断 | 拓扑分析 + 因果推断 |
| L5 | 预测性运维 + 自愈 | ML 预测 + 自动修复 |
与 SRE Agent 的协同
告警触发 → AIOps 降噪（识别根因告警）
↓
SRE Agent 接管 → 告警分析（LLM 归因 + RAG 检索）
↓
SRE Agent → 故障诊断（工具链串联排查）
↓
AIOps → 根因推断验证（拓扑关联确认）
↓
SRE Agent → 执行修复（受限自动操作 or 人工审批）
↓
AIOps → 验证恢复（异常消失确认）
常见问题 / 坑点
| 问题 | 原因 | 解决方案 |
|---|
| 异常检测误报多 | 算法不适应季节性 | 用季节性分解 + 动态基线 |
| 告警降噪过度 | 关联规则太激进 | 保守策略：宁可多报不漏报 |
| 根因推断不准 | 拓扑图过时 | 定期更新服务依赖图 |
| ML 模型漂移 | 业务变化后模型失效 | 定期重训练 + 模型监控 |
| 算法太重 | 生产环境扛不住 | 轻量统计算法优先，ML 做离线分析 |
关联知识
参考资源
状态