基于 LSTM 的 GPU 利用率异常检测——时序预测在推理集群运维中的落地实践

一、GPU 利用率的"假正常"——为什么 75% 不代表一切正常

线上 8 卡 A100 推理集群的 GPU 利用率面板上，各卡数值在 65%～82% 之间波动，所有指标"在正常范围内"。但用户反馈偶发性"卡顿"——一个月内出现 7 次，每次持续 3～8 分钟。查遍告警规则，没有任何一个触发。

深入分析后发现，问题出在 GPU 利用率的时序模式上：正常情况下，GPU 利用率在 70%～85% 之间呈连续的、缓慢波动的模式。但在问题发生前 2～3 分钟，GPU 利用率出现了高频震荡（±15%，周期 5～10 秒），绝对值仍然在"正常范围"内（最低 55%），但波动模式完全异常——这是 KV Cache 频繁 Preemption 导致的 GPU kernel 启动/停止抖动。

静态阈值的监控看不到波动模式。只有基于时间序列预测的异常检测，才能从历史模式中学习"什么是正常"，进而识别模式偏离。

二、LSTM 时序预测的架构与异常判定逻辑

2.1 为什么 LSTM 而非统计方法

时序异常检测有三条技术路线：

方法	原理	优点	缺点
统计阈值（3-sigma）	偏离均值 3σ 判定异常	简单、可解释	无法识别模式异常
孤立森林	无监督异常隔离	无需标注、多特征	不利用时间依赖
LSTM 预测	学习时序模式，预测 vs 实际偏差	捕获时间依赖、识别模式异常	需要历史数据训练、计算成本高

GPU 利用率的异常往往不是数值异常而是模式异常（如震荡、趋势反转），LSTM 的时序依赖学习能力比统计方法更适合。

2.2 异常检测架构

flowchart LR
    subgraph Data[数据采集]
        D1[GPU Metrics<br/>30s 采样]
        D2[请求 QPS]
        D3[显存使用]
    end

    subgraph Preprocess[预处理]
        P1[缺失值填充<br/>前向填充]
        P2[滑动窗口<br/>window=64 点]
        P3[标准化<br/>MinMaxScaler]
    end

    subgraph Model[LSTM 模型]
        M1[Encoder<br/>2层LSTM-64]
        M2[Decoder<br/>1层LSTM-32]
        M3[预测未来<br/>16点/8分钟]
    end

    subgraph Detection[异常判定]
        E1[预测误差计算<br/>MAE]
        E2[动态阈值<br/>训练集误差的P99]
        E3[连续超标<br/>≥5点触发告警]
    end

    D1 --> P1
    D2 --> P1
    D3 --> P1
    P1 --> P2
    P2 --> P3
    P3 --> M1
    M1 --> M2
    M2 --> M3
    M3 --> E1
    E1 --> E2
    E2 --> E3

2.3 关键设计决策

预测未来 16 个点（8 分钟）：GPU 异常检测的窗口需要足够提前量。16 个点（30s 采样 × 16 = 8 分钟）的预测窗口既能让运维有时间响应，又不会因为预测步长过大导致误差爆炸。

动态阈值：不同时间段的 GPU 利用率基线不同（白天高、凌晨低）。使用训练集预测误差的 P99 作为动态阈值，比固定 MAE 阈值更适应基线变化。

连续超标确认：同一天内业务负载的自然波动可能导致单点预测误差偏大（如突发流量）。要求连续 5 个点超标才触发告警。

三、LSTM 模型的生产级实现

3.1 数据处理与窗口构建

# data_loader.py —— GPU 时序数据加载与窗口化
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from typing import Tuple
import torch
from torch.utils.data import Dataset, DataLoader


class GPUTimeSeriesDataset(Dataset):
    """
    GPU 时间序列数据集

    将原始的多维 GPU 指标转换为 LSTM 可用的
    (输入窗口, 预测目标) 样本对。
    """

    def __init__(
        self,
        df: pd.DataFrame,
        input_len: int = 64,   # 输入窗口：64 个采样点（32 分钟）
        pred_len: int = 16,    # 预测长度：16 个采样点（8 分钟）
        feature_cols: list = None,
    ):
        """
        Args:
            df: 包含时间序列的 DataFrame
            input_len: 输入序列长度
            pred_len: 预测序列长度
            feature_cols: 特征列名列表，默认使用 GPU 利用率相关列
        """
        if feature_cols is None:
            feature_cols = [
                "gpu_utilization",
                "gpu_memory_used",
                "request_qps",
                "kv_cache_hit_rate",
            ]

        # 1. 提取特征，填充缺失值（前向填充）
        data = df[feature_cols].fillna(method="ffill").values
        # 对于开头无法前向填充的 NaN，用 0 填充
        data = np.nan_to_num(data, nan=0.0)

        # 2. 归一化：每个特征独立缩放到 [0, 1]
        self.scaler = MinMaxScaler(feature_range=(0, 1))
        data = self.scaler.fit_transform(data)

        # 3. 构建 (输入, 目标) 样本对
        self.samples = []
        total_len = input_len + pred_len
        for i in range(len(data) - total_len + 1):
            x = data[i:i + input_len]         # 输入窗口
            y = data[i + input_len:           # 目标窗口（仅 GPU 利用率）
                    i + total_len, 0]          # 第 0 列 = GPU 利用率
            self.samples.append((
                torch.FloatTensor(x),
                torch.FloatTensor(y),
            ))

        self.input_len = input_len
        self.pred_len = pred_len
        self.n_features = len(feature_cols)

    def __len__(self):
        return len(self.samples)

    def __getitem__(self, idx):
        return self.samples[idx]

    def inverse_transform_gpu_util(
        self, normalized_value: np.ndarray
    ) -> np.ndarray:
        """
        将归一化的 GPU 利用率反变换回原始值

        注意：scaler 对所有特征统一缩放，
        反变换 GPU 利用率（第 0 列）时需要构建全特征向量
        """
        dummy = np.zeros((len(normalized_value), self.n_features))
        dummy[:, 0] = normalized_value
        return self.scaler.inverse_transform(dummy)[:, 0]

3.2 LSTM Seq2Seq 模型

# model.py —— LSTM 时序预测模型
import torch
import torch.nn as nn
import pytorch_lightning as pl
from torch.optim.lr_scheduler import ReduceLROnPlateau


class GPUPredictor(pl.LightningModule):
    """
    基于 LSTM Seq2Seq 的 GPU 利用率预测模型

    架构：Encoder(2层LSTM) → Decoder(1层LSTM) → 全连接输出

    设计考量：
    - 2层 Encoder 捕获多层次时序模式
    - 1层 Decoder 减少过拟合风险
    - 预测 GPU 利用率，输入包含多维协变量（QPS、显存等）
    """

    def __init__(
        self,
        input_dim: int = 4,       # 输入特征维度
        hidden_dim: int = 64,     # LSTM 隐藏层维度
        encoder_layers: int = 2,  # Encoder LSTM 层数
        decoder_layers: int = 1,  # Decoder LSTM 层数
        pred_len: int = 16,       # 预测长度
        dropout: float = 0.2,     # Dropout 比例（防过拟合）
        learning_rate: float = 0.001,
    ):
        super().__init__()
        self.save_hyperparameters()

        # ---- Encoder ----
        self.encoder = nn.LSTM(
            input_size=input_dim,
            hidden_size=hidden_dim,
            num_layers=encoder_layers,
            dropout=dropout if encoder_layers > 1 else 0,
            batch_first=True,  # (batch, seq, feature)
        )

        # ---- Decoder ----
        self.decoder = nn.LSTM(
            input_size=1,  # 自回归：用上一时刻预测作为下一时刻输入
            hidden_size=hidden_dim,
            num_layers=decoder_layers,
            dropout=dropout if decoder_layers > 1 else 0,
            batch_first=True,
        )

        # ---- 输出层 ----
        self.fc = nn.Linear(hidden_dim, 1)

        self.pred_len = pred_len
        self.learning_rate = learning_rate
        self.criterion = nn.MSELoss()

    def forward(
        self, x: torch.Tensor
    ) -> torch.Tensor:
        """
        Args:
            x: (batch, input_len, input_dim)

        Returns:
            pred: (batch, pred_len) GPU 利用率预测序列
        """
        batch_size = x.size(0)

        # 1. Encoder: 编码输入序列
        #    outputs: (batch, input_len, hidden_dim)
        #    (h_n, c_n): (num_layers, batch, hidden_dim)
        _, (h_n, c_n) = self.encoder(x)

        # 2. Decoder: 逐步预测
        #    初始输入为 0（表示从 Encoder 上下文开始预测）
        decoder_input = torch.zeros(
            batch_size, 1, device=x.device)
        # 取 Encoder 最后一层状态作为 Decoder 初始状态
        h_decoder = h_n[-1:].repeat(
            self.decoder.num_layers, 1, 1)
        c_decoder = c_n[-1:].repeat(
            self.decoder.num_layers, 1, 1)

        outputs = []
        for _ in range(self.pred_len):
            # decoder_input: (batch, 1, 1)
            out, (h_decoder, c_decoder) = self.decoder(
                decoder_input.unsqueeze(-1),
                (h_decoder, c_decoder),
            )
            # out: (batch, 1, hidden_dim)
            pred = self.fc(out.squeeze(1))  # (batch, 1)
            outputs.append(pred)
            # 自回归：用当前预测作为下一步输入
            decoder_input = pred

        return torch.cat(outputs, dim=1)  # (batch, pred_len)

    def training_step(self, batch, batch_idx):
        x, y = batch
        y_pred = self(x)
        loss = self.criterion(y_pred, y)
        self.log("train_loss", loss, on_epoch=True)
        return loss

    def validation_step(self, batch, batch_idx):
        x, y = batch
        y_pred = self(x)
        loss = self.criterion(y_pred, y)
        # 记录 MAE（更直观的误差指标）
        mae = nn.L1Loss()(y_pred, y)
        self.log("val_loss", loss, on_epoch=True)
        self.log("val_mae", mae, on_epoch=True)
        return loss

    def configure_optimizers(self):
        optimizer = torch.optim.AdamW(
            self.parameters(),
            lr=self.learning_rate,
            weight_decay=1e-5,  # L2 正则化
        )
        scheduler = ReduceLROnPlateau(
            optimizer,
            mode="min",
            factor=0.5,     # loss 不下降时学习率减半
            patience=5,     # 5 个 epoch 不下降触发
            min_lr=1e-6,
        )
        return {
            "optimizer": optimizer,
            "lr_scheduler": {
                "scheduler": scheduler,
                "monitor": "val_loss",
            },
        }

3.3 异常检测逻辑

# detector.py —— 基于预测误差的异常检测
import numpy as np
from typing import List, Tuple


class GPUAnomalyDetector:
    """
    基于 LSTM 预测误差的 GPU 异常检测器

    核心理念：
    - 在历史数据上训练 LSTM 学习"正常模式"
    - 对新的数据点进行预测，计算预测误差
    - 预测误差超过训练集 P99 误差 → 模式偏离"正常" → 触发告警
    """

    def __init__(
        self,
        model: GPUPredictor,
        dataset: GPUTimeSeriesDataset,
        threshold_percentile: float = 99.0,
        consecutive_points: int = 5,
    ):
        """
        Args:
            model: 已训练的 LSTM 预测模型
            dataset: 用于计算阈值的数据集
            threshold_percentile: 阈值分位数（默认 P99）
            consecutive_points: 连续异常点数（触发告警阈值）
        """
        self.model = model
        self.dataset = dataset
        self.consecutive_points = consecutive_points

        # 计算异常阈值
        self.threshold = self._compute_threshold(
            threshold_percentile)

    def _compute_threshold(
        self, percentile: float
    ) -> float:
        """
        从训练数据计算异常判定阈值

        原理：用训练好的模型对训练集做预测，
        取预测误差（MAE）的 P99 作为阈值。
        任何超过此误差的新数据点被认为是异常模式。
        """
        self.model.eval()
        loader = DataLoader(
            self.dataset, batch_size=64, shuffle=False)

        errors = []
        with torch.no_grad():
            for x, y in loader:
                y_pred = self.model(x)
                # 逐样本 MAE
                mae = torch.abs(
                    y_pred - y).mean(dim=1).cpu().numpy()
                errors.extend(mae)

        threshold = float(np.percentile(errors, percentile))
        print(f"阈值: P{percentile} = {threshold:.6f}")
        print(f"误差范围: [{np.min(errors):.6f}, "
              f"{np.max(errors):.6f}]")
        return threshold

    def detect(
        self, x: torch.Tensor, actual: torch.Tensor
    ) -> List[Tuple[int, float, float]]:
        """
        检测单批次数据中的异常点

        Args:
            x: 输入窗口序列，shape (n, input_len, n_features)
            actual: 实际观测值，shape (n, pred_len)

        Returns:
            List[(样本索引, 预测误差, 异常严重程度)]
        """
        self.model.eval()
        with torch.no_grad():
            y_pred = self.model(x)

        errors = torch.abs(
            y_pred - actual).mean(dim=1).cpu().numpy()

        # 找异常点
        anomalies = []
        consecutive = 0  # 连续异常点计数
        for i, error in enumerate(errors):
            if error > self.threshold:
                consecutive += 1
                if consecutive >= self.consecutive_points:
                    # 连续超标达到阈值，加入告警列表
                    severity = error / self.threshold
                    anomalies.append((i, error, severity))
            else:
                consecutive = 0  # 中断，重置计数

        return anomalies

四、时序预测的工程落地挑战

4.1 模型漂移与定期重训

GPU 利用率的时间模式不是静态的。随着用户群增长、推理流量变化、模型版本更新，"正常模式"的定义在持续漂移。7 天前训练的 LSTM 对当前数据的预测误差可能系统性地偏高。

推荐策略：每周全量重训 + 每日增量微调。全量重训用最近 30 天数据，增量微调用最近 3 天数据仅训练 2～3 个 epoch。

4.2 冷启动问题

新上线的推理集群没有足够的历史数据训练 LSTM。在数据积累的前 7 天，需要回退到统计方法的阈值告警（如 3-sigma），作为 LSTM 的过渡方案。

4.3 GPU 训练成本

LSTM 模型本身在 CPU 上训练（数据量小，约 10 万采样点），不需要 GPU。单次全量重训约 3 分钟（16 核 CPU）。但推理集群的 GPU 资源已经紧张，检测模型的推理（预测）必须在 CPU 上运行，不能占用宝贵的 GPU 显存。

4.4 可解释性不足

LSTM 告诉你"这个模式异常"，但不告诉你"为什么异常"。运维收到告警后仍然需要打开 Grafana 手动排查。实践中，可以结合特征重要性分析（如对最后一层做积分梯度）给出"GPU 利用率振荡是主要异常信号"一类的辅助信息。

五、总结

将 LSTM 时序预测引入 GPU 监控，改变了对"异常"的定义——从"数值异常"升级到"模式异常"：

1. 多维特征建模：不仅仅预测 GPU 利用率数值，还输入 QPS、显存、KV Cache 命中率等协变量，提高对"假正常"的识别能力。
2. 预测误差驱动检测：利用 LSTM 对历史模式的学习能力，当新模式显著偏离预测值时触发告警，早于静态阈值 2～8 分钟发现问题。
3. 动态阈值校准：基于训练集 P99 误差的阈值比固定值更适应不同时间段的基线变化。
4. 工程配套：模型漂移 → 每周重训，冷启动 → 统计回退，计算成本 → CPU 推理不占 GPU，可解释性 → 后续结合特征归因。