AI 异常检测与根因分析：云原生可观测性的智能化，从阈值告警到模式发现

一、静态阈值的检测盲区：正常波动与真实异常的混淆

传统监控依赖静态阈值告警——CPU 超过 80% 就告警、RT 超过 500ms 就告警。但实际运行中，"正常"是动态的：工作日白天的 QPS 是凌晨的 10 倍，大促期间基线会上移 3—5 倍。静态阈值要么对正常峰值误报（工作日白天频繁告警），要么对异常低谷漏报（凌晨流量骤降可能是故障信号）。

AI 异常检测的核心思路是"动态基线"——基于历史数据学习正常模式的边界，只有偏离动态基线的行为才被标记为异常。这比静态阈值更精准，但也引入了新的复杂性：模型训练、特征选择、误报控制。

二、AI 异常检测的架构与算法选择

flowchart TB
    A[时序指标数据] --> B[特征工程]
    B --> C[周期性特征]
    B --> D[趋势特征]
    B --> E[统计特征]
    C --> F[异常检测模型]
    D --> F
    E --> F
    F --> G{异常评分}
    G -->|超过阈值| H[异常事件]
    G -->|正常范围| I[更新基线]
    H --> J[根因分析]
    J --> K[关联指标查找]
    J --> L[拓扑路径追踪]
    K --> M[根因报告]
    L --> M

异常检测的主流算法：3-Sigma（简单但仅适用正态分布）、IQR（鲁棒但灵敏度低）、Isolation Forest（无监督，适合多维异常）、LSTM-AE（时序建模，适合复杂周期）。生产环境推荐 Isolation Forest——无需标注数据、训练快、对多维异常效果好。

三、生产级实现：异常检测与根因分析引擎

# anomaly_detector.py — AI 异常检测与根因分析引擎
# 设计意图：基于 Isolation Forest 检测指标异常，
# 结合拓扑关系和关联分析推断根因

import numpy as np
from sklearn.ensemble import IsolationForest
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta

@dataclass
class MetricPoint:
    name: str
    timestamp: datetime
    value: float
    labels: dict

@dataclass
class AnomalyEvent:
    metric_name: str
    timestamp: datetime
    value: float
    expected_range: tuple  # (lower, upper)
    anomaly_score: float   # -1 到 1，越负越异常
    severity: str
    root_cause_candidates: List[dict]

class AnomalyDetector:
    """AI 异常检测引擎"""

    def __init__(self, contamination: float = 0.05):
        # contamination: 预期异常比例，影响检测灵敏度
        self.contamination = contamination
        self.models: Dict[str, IsolationForest] = {}
        self.baselines: Dict[str, dict] = {}

    def train(self, metric_name: str, history: List[MetricPoint]):
        """训练异常检测模型"""
        if len(history) < 100:
            return  # 数据不足，跳过训练

        # 特征工程：将时序数据转换为特征向量
        features = self._extract_features(history)

        # 训练 Isolation Forest
        model = IsolationForest(
            contamination=self.contamination,
            n_estimators=100,
            random_state=42,
        )
        model.fit(features)

        self.models[metric_name] = model

        # 计算动态基线（用于异常报告中的期望范围）
        values = [p.value for p in history]
        self.baselines[metric_name] = {
            'mean': np.mean(values),
            'std': np.std(values),
            'p5': np.percentile(values, 5),
            'p95': np.percentile(values, 95),
        }

    def detect(self, metric_name: str, point: MetricPoint) -> Optional[AnomalyEvent]:
        """检测单点异常"""
        if metric_name not in self.models:
            return None

        model = self.models[metric_name]
        feature = self._extract_single_feature(point)

        # Isolation Forest 预测
        prediction = model.predict([feature])[0]
        score = model.score_samples([feature])[0]

        if prediction == -1:  # 异常
            baseline = self.baselines.get(metric_name, {})
            expected_range = (
                baseline.get('p5', 0),
                baseline.get('p95', 0),
            )

            severity = self._classify_severity(score, point.value, baseline)

            return AnomalyEvent(
                metric_name=metric_name,
                timestamp=point.timestamp,
                value=point.value,
                expected_range=expected_range,
                anomaly_score=score,
                severity=severity,
                root_cause_candidates=[],
            )

        return None

    def _extract_features(self, history: List[MetricPoint]) -> np.ndarray:
        """从时序数据中提取特征向量"""
        # 设计意图：原始值不足以判断异常，
        # 需要结合时间特征和统计特征
        features = []

        for i in range(len(history)):
            point = history[i]
            feature = [
                point.value,
                point.timestamp.hour,           # 小时（捕获日周期）
                point.timestamp.weekday(),       # 星期（捕获周周期）
                self._rolling_mean(history, i, 10),   # 10 点滚动均值
                self._rolling_std(history, i, 10),    # 10 点滚动标准差
            ]
            features.append(feature)

        return np.array(features)

    def _extract_single_feature(self, point: MetricPoint) -> list:
        """提取单点特征（简化版，无滚动统计）"""
        return [
            point.value,
            point.timestamp.hour,
            point.timestamp.weekday(),
            point.value,  # 无历史数据时用当前值替代
            0.0,          # 无历史数据时标准差为 0
        ]

    def _rolling_mean(self, history: List[MetricPoint], idx: int, window: int) -> float:
        """计算滚动均值"""
        start = max(0, idx - window)
        values = [history[j].value for j in range(start, idx + 1)]
        return np.mean(values) if values else 0.0

    def _rolling_std(self, history: List[MetricPoint], idx: int, window: int) -> float:
        """计算滚动标准差"""
        start = max(0, idx - window)
        values = [history[j].value for j in range(start, idx + 1)]
        return np.std(values) if len(values) > 1 else 0.0

    def _classify_severity(self, score: float, value: float, baseline: dict) -> str:
        """分类异常严重度"""
        mean = baseline.get('mean', 0)
        std = baseline.get('std', 1)

        # 偏离程度：偏离均值几个标准差
        deviation = abs(value - mean) / max(std, 0.001)

        if deviation > 5 or score < -0.5:
            return 'critical'
        elif deviation > 3 or score < -0.3:
            return 'warning'
        else:
            return 'info'


class RootCauseAnalyzer:
    """根因分析引擎"""

    def __init__(self, topology: dict):
        self.topology = topology  # 服务依赖拓扑

    def analyze(self, anomaly: AnomalyEvent,
                concurrent_anomalies: List[AnomalyEvent]) -> List[dict]:
        """
        根因分析：在并发异常中查找最可能的根因
        设计意图：多个指标同时异常时，通过拓扑关系和时序分析
        推断哪个是最上游的根因
        """
        candidates = []

        for other in concurrent_anomalies:
            if other.metric_name == anomaly.metric_name:
                continue

            # 计算关联得分
            relation_score = self._compute_relation_score(
                anomaly, other
            )

            if relation_score > 0.5:
                candidates.append({
                    'metric': other.metric_name,
                    'value': other.value,
                    'severity': other.severity,
                    'relation_score': relation_score,
                    'reason': self._explain_relation(anomaly, other),
                })

        # 按关联得分排序
        candidates.sort(key=lambda c: c['relation_score'], reverse=True)

        return candidates[:5]  # 返回 Top 5 候选根因

    def _compute_relation_score(self, anomaly_a: AnomalyEvent,
                                 anomaly_b: AnomalyEvent) -> float:
        """计算两个异常事件之间的关联得分"""
        score = 0.0

        # 维度 1：时间接近度
        time_diff = abs((anomaly_a.timestamp - anomaly_b.timestamp).total_seconds())
        if time_diff < 60:
            score += 0.4  # 1 分钟内，强关联
        elif time_diff < 300:
            score += 0.2  # 5 分钟内，中等关联

        # 维度 2：拓扑关系
        if self._are_upstream_downstream(anomaly_a.metric_name, anomaly_b.metric_name):
            score += 0.4

        # 维度 3：标签重叠
        common_labels = set(anomaly_a.metric_name.split('_')) & set(anomaly_b.metric_name.split('_'))
        if common_labels:
            score += 0.2

        return min(score, 1.0)

    def _are_upstream_downstream(self, metric_a: str, metric_b: str) -> bool:
        """判断两个指标是否属于上下游服务"""
        # 简化实现：从指标名提取服务名，检查拓扑关系
        service_a = metric_a.split('_')[0]
        service_b = metric_b.split('_')[0]
        return service_b in self.topology.get(service_a, set())

    def _explain_relation(self, anomaly_a: AnomalyEvent,
                           anomaly_b: AnomalyEvent) -> str:
        """生成关联解释"""
        if self._are_upstream_downstream(anomaly_a.metric_name, anomaly_b.metric_name):
            return f"{anomaly_b.metric_name} 是 {anomaly_a.metric_name} 的上游依赖"
        time_diff = abs((anomaly_a.timestamp - anomaly_b.timestamp).total_seconds())
        return f"时间接近（相差 {time_diff:.0f}s），可能同源"

四、Trade-offs：AI 异常检测的精度与实用性平衡

模型训练的数据需求。 Isolation Forest 至少需要 100 个数据点才能训练出有效模型，新上线的服务在初始阶段无法使用 AI 检测。建议新服务先用静态阈值，积累足够数据后切换到 AI 检测。

冷启动与热更新。 模型训练是离线批处理，无法实时更新。当服务架构变更（如新增依赖）后，旧模型可能产生误报。建议每日重新训练模型，并在架构变更后触发增量训练。

多维异常的复杂性。 单指标异常检测相对简单，但真正的故障通常表现为多维异常（CPU、RT、错误率同时异常）。多维 Isolation Forest 可以处理，但特征工程更复杂，训练数据需求更大。

根因推断的不确定性。 根因分析给出的是"候选根因"而非"确定根因"，需要人工确认。过度依赖自动根因推断可能导致错误定位。建议将根因分析结果作为"排查起点"而非"最终结论"。

五、总结

AI 异常检测将监控从"静态阈值"升级为"动态基线"，是云原生可观测性的核心能力。落地路径：第一步，对核心指标积累历史数据，训练 Isolation Forest 模型；第二步，将 AI 检测与静态阈值并行运行，验证检测效果；第三步，引入根因分析，在异常事件中标注候选根因；第四步，建立模型效果评估体系，持续优化特征和参数。核心原则：AI 检测是辅助工具，运维人员的经验判断仍是故障定位的最终决策者。