AIOps 异常检测的半监督学习与少样本适应：从标注瓶颈到自适应检测

一、异常检测的标注困境：正常数据充足，异常样本稀缺

运维异常检测的核心挑战不是算法选择，而是标注数据。正常运行的监控数据大量且容易获取，但异常样本稀缺——系统大部分时间运行正常，故障发生频率低且持续时间短。更棘手的是，新类型的异常不断出现（系统升级、新功能上线引入新的故障模式），历史异常标签无法覆盖未来场景。

传统监督学习依赖大量标注数据训练分类器，在异常样本稀缺时表现不佳。半监督学习与少样本学习提供了替代方案：利用大量无标注的正常数据建立正常行为基线，仅用少量异常样本微调检测边界，使模型能够识别"偏离正常"的行为，即使从未见过该类型的异常。

二、半监督与少样本异常检测的模型架构

flowchart TD
    A[监控数据] --> B[预训练: 正常数据自编码器]
    B --> C[正常行为基线]
    C --> D[异常分数计算]
    D --> E{分数超过阈值?}
    E -->|是| F[标记为异常]
    E -->|否| G[标记为正常]

    H[少量异常样本] --> I[少样本微调]
    I --> J[调整检测阈值]
    J --> K[优化异常边界]

    subgraph 自编码器架构
        L[编码器: 输入→潜变量]
        M[解码器: 潜变量→重建]
        N[重建误差: 异常分数]
    end

    subgraph 少样本策略
        O[原型网络: 计算异常原型]
        P[度量学习: 距离判断]
        Q[阈值校准: 基于少量样本]
    end

    B --> L
    L --> M
    M --> N
    I --> O
    I --> P
    I --> Q

自编码器（Autoencoder）是半监督异常检测的经典架构：仅用正常数据训练，学习压缩与重建正常模式。异常数据的重建误差显著高于正常数据，重建误差即为异常分数。少样本微调在获得少量异常样本后，调整检测阈值或训练轻量级分类头，优化异常边界。

三、工程实现：半监督异常检测系统

# anomaly_detector.py — 半监督异常检测引擎
import numpy as np
import torch
import torch.nn as nn
from typing import List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class AnomalyResult:
    timestamp: float
    metric_name: str
    score: float          # 异常分数 0-1
    is_anomaly: bool
    threshold: float
    reconstruction_error: float

class AutoencoderAnomalyDetector(nn.Module):
    """基于自编码器的半监督异常检测器"""

    def __init__(self, input_dim: int, latent_dim: int = 16):
        super().__init__()
        # 编码器：逐层压缩
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, latent_dim),
        )
        # 解码器：逐层重建
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 32),
            nn.ReLU(),
            nn.Linear(32, 64),
            nn.ReLU(),
            nn.Linear(64, input_dim),
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        latent = self.encoder(x)
        reconstructed = self.decoder(latent)
        return reconstructed

    def compute_anomaly_score(
        self, x: torch.Tensor
    ) -> torch.Tensor:
        """计算异常分数：重建误差的归一化值"""
        reconstructed = self.forward(x)
        mse = torch.mean((x - reconstructed) ** 2, dim=1)
        return mse

class AnomalyDetectionEngine:
    """异常检测引擎：训练、推理、少样本微调"""

    def __init__(self, input_dim: int):
        self.model = AutoencoderAnomalyDetector(input_dim)
        self.threshold = None  # 动态计算
        self.normal_stats = None  # 正常数据的统计特征

    def train_on_normal_data(
        self,
        normal_data: np.ndarray,
        epochs: int = 50,
        batch_size: int = 64,
        percentile: float = 99.0,
    ):
        """仅用正常数据训练自编码器"""
        X = torch.FloatTensor(normal_data)
        optimizer = torch.optim.Adam(self.model.parameters(), lr=1e-3)
        criterion = nn.MSELoss()

        self.model.train()
        for epoch in range(epochs):
            total_loss = 0
            for i in range(0, len(X), batch_size):
                batch = X[i:i+batch_size]
                reconstructed = self.model(batch)
                loss = criterion(reconstructed, batch)

                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
                total_loss += loss.item()

        # 基于正常数据的重建误差分布设定阈值
        self.model.eval()
        with torch.no_grad():
            scores = self.model.compute_anomaly_score(X).numpy()

        # 取第 percentile 百分位作为阈值
        self.threshold = np.percentile(scores, percentile)
        self.normal_stats = {
            'mean': np.mean(scores),
            'std': np.std(scores),
            'percentile_99': self.threshold,
        }

    def detect(self, data: np.ndarray) -> List[AnomalyResult]:
        """检测异常"""
        if self.threshold is None:
            raise RuntimeError("模型未训练")

        X = torch.FloatTensor(data)
        self.model.eval()

        with torch.no_grad():
            scores = self.model.compute_anomaly_score(X).numpy()

        results = []
        for i, score in enumerate(scores):
            results.append(AnomalyResult(
                timestamp=data[i][0] if data.shape[1] > 0 else i,
                metric_name='unknown',
                score=float(score),
                is_anomaly=score > self.threshold,
                threshold=self.threshold,
                reconstruction_error=float(score),
            ))

        return results

    def few_shot_calibrate(
        self,
        normal_samples: np.ndarray,
        anomaly_samples: np.ndarray,
    ):
        """少样本校准：利用少量异常样本优化检测阈值"""
        X_normal = torch.FloatTensor(normal_samples)
        X_anomaly = torch.FloatTensor(anomaly_samples)

        self.model.eval()
        with torch.no_grad():
            normal_scores = self.model.compute_anomaly_score(
                X_normal
            ).numpy()
            anomaly_scores = self.model.compute_anomaly_score(
                X_anomaly
            ).numpy()

        # 在正常分数与异常分数之间寻找最优阈值
        # 目标：最大化 F1 分数（平衡精确率与召回率）
        all_scores = np.concatenate([normal_scores, anomaly_scores])
        best_f1 = 0
        best_threshold = self.threshold

        for threshold in np.linspace(
            np.min(all_scores), np.max(all_scores), 100
        ):
            tp = np.sum(anomaly_scores > threshold)
            fp = np.sum(normal_scores > threshold)
            fn = np.sum(anomaly_scores <= threshold)

            precision = tp / (tp + fp) if (tp + fp) > 0 else 0
            recall = tp / (tp + fn) if (tp + fn) > 0 else 0
            f1 = 2 * precision * recall / (precision + recall) \
                if (precision + recall) > 0 else 0

            if f1 > best_f1:
                best_f1 = f1
                best_threshold = threshold

        self.threshold = best_threshold
        print(f"少样本校准完成: 阈值 {best_threshold:.4f}, F1 {best_f1:.4f}")

四、半监督异常检测的边界与权衡

概念漂移的挑战：系统正常行为随业务变化（如流量增长、新功能上线），自编码器训练的正常基线可能过时。建议定期用最近的正常数据重训练模型（如每周），并监控正常分数的分布变化，当分布显著偏移时触发重训练。

阈值选择的敏感性：百分位阈值的选择直接影响误报率与漏报率的平衡。99 百分位意味着约 1% 的正常数据被误报，对于大规模监控数据，1% 的误报率仍可能产生大量噪音。建议根据业务容忍度调整百分位，或使用动态阈值（基于滑动窗口的分数分布自适应调整）。

少样本校准的过拟合风险：少量异常样本可能导致阈值过度拟合这些特定样本，对其他类型的异常检测效果反而下降。建议在少样本校准后，在独立的验证集上评估整体检测性能。

多变量相关性：自编码器可以捕获变量间的相关性（如 CPU 与内存的正常联动模式），但当变量数量很多时（>100），重建误差可能被高方差变量主导，掩盖低方差变量的异常。建议对变量分组训练独立的自编码器，或使用注意力机制加权重建误差。

五、总结

半监督异常检测通过仅用正常数据训练自编码器建立正常行为基线，解决了异常样本稀缺的标注困境。少样本校准利用少量异常样本优化检测阈值，提升对已知异常类型的识别精度。工程落地的关键在于：定期重训练应对概念漂移、动态阈值平衡误报与漏报、少样本校准后验证整体性能、变量分组避免高方差主导。半监督检测不是异常检测的终极方案，但在标注数据稀缺的运维场景下，它是最实用的起点。