异常检测是数据挖掘中的核心任务，广泛应用于工业设备监控、金融交易欺诈识别和网络入侵检测等场景。传统方法如基于统计的阈值判定或孤立森林算法，虽然轻量高效，却难以处理具有复杂时序依赖的高维数据。Transformer模型凭借自注意力机制，能够动态衡量任意两个时间步之间的相关性，从而同时捕获局部突变和全局趋势，为异常检测提供了一种全新的建模思路。下面通过一个可运行的Python实现，展示如何利用Transformer完成异常检测任务。
为什么Transformer适合异常检测
异常检测的核心是寻找与正常模式显著偏离的数据点。在时序数据中，异常往往表现为局部突变，或者一段连续区间内状态分布发生剧烈变化。传统RNN及其变体LSTM通过门控机制按顺序传递信息，能够捕捉一定范围内的时序依赖，但受限于递归结构，其有效感受野有限，且训练时容易遇到梯度消失或梯度爆炸问题。Transformer的结构完全抛弃了循环单元，改用自注意力机制直接计算序列内任意两个位置的关系。
自注意力机制有一个重要特性：它对位置编码不敏感，但可以通过位置编码来注入顺序信息。这种设计使得模型可以并行处理整个序列，训练效率大幅提升。更重要的是，注意力权重可以解释为每个时间点对整体模式的贡献程度。在异常检测任务中，正常数据往往呈现出高度自相关性，而异常点会破坏这种相关性，导致其对应的注意力权重分布与其他时间点明显不同。利用这一差异，可以直接从注意力矩阵中提取异常线索，或者通过重构误差来评估每个点的“合理性”。
与LSTM相比，Transformer在长序列上优势明显。例如，监控系统中一个小时的传感器数据可能包含数万个时间步，LSTM需要逐步处理，而Transformer可以一次性喂入整个窗口。此外，Transformer的多头机制让模型能够从不同子空间学习特征交互，对异常模式的识别更加鲁棒。当然，Transformer也不是万能药，它对数据量的要求较高，且计算复杂度与序列长度的平方成正比，在实际使用中需要合理设计窗口大小和模型深度。
Transformer核心组件与异常分数设计
基于Transformer的异常检测模型通常采用自编码器结构：编码器将输入序列映射为潜在表示，解码器尝试重构原始输入。训练目标是让正常样本的重构误差最小化，因为模型只见过正常数据，所以当异常样本输入时，重构误差会显著增大。这里的关键是，编码器和解码器都需要使用自注意力层来提取特征。
位置编码是Transformer的基础组件。由于自注意力本身不具备顺序感知能力，我们需要在输入嵌入中加入位置信息。最常用的是一组正弦和余弦函数，不同频率的波可以让模型区分不同位置。下面的代码展示了如何生成位置编码矩阵，它与输入维度相同，直接加到嵌入向量上即可。
import numpy as np
import torch
import torch.nn as nn
import math
def positional_encoding(seq_len, d_model, device='cpu'):
pe = torch.zeros(seq_len, d_model).to(device)
position = torch.arange(0, seq_len, dtype=torch.float32).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
return pe.unsqueeze(0) # shape: (1, seq_len, d_model)
异常分数可以用原始输入与重构输出之间的绝对误差来衡量。对于多变量时间序列，每个时间步包含多个特征，我们可以计算每个时间步的欧氏距离，然后取序列的平均值或最大值作为整体异常分数。更精细的做法是使用注意力权重加权误差，让模型更关注那些对重构贡献大的时间点。不过，最简单有效的方法仍然是直接计算逐点误差，再通过分位数或3σ准则确定阈值。
值得注意的是，Transformer编码器通常包含多层多头自注意力层和前馈网络，层与层之间使用残差连接和层归一化。在异常检测任务中，我们不需要解码器生成文本，而是用一个全连接层直接把编码后的特征映射回原始维度，这相当于一个轻量级解码器。为了提升模型对局部异常的敏感性，可以在编码器输出上加入一维卷积层，帮助模型捕捉相邻时间步的微小变化。
Python实现基于Transformer的异常检测
下面我们构建一个完整的训练和推理流程。假设输入是一段多变量时间序列，窗口长度为128，特征维度为5。我们首先定义一个Transformer自编码器模型，然后使用正常数据训练，最后计算异常分数并判定。代码基于PyTorch实现，核心组件包括一个位置编码层、一个TransformerEncoder层和一个输出投影层。
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
class TransformerAutoencoder(nn.Module):
def __init__(self, feature_dim, d_model=64, nhead=4, num_layers=2, seq_len=128):
super(TransformerAutoencoder, self).__init__()
self.feature_dim = feature_dim
self.d_model = d_model
self.seq_len = seq_len
# 输入投影：将原始特征映射到d_model维
self.input_proj = nn.Linear(feature_dim, d_model)
# 位置编码参数
self.pos_encoding = nn.Parameter(positional_encoding(seq_len, d_model), requires_grad=False)
# Transformer编码器层
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model, nhead=nhead, dim_feedforward=128, batch_first=True
)
self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
# 解码器：直接映射回原始特征空间
self.output_proj = nn.Linear(d_model, feature_dim)
def forward(self, x):
# x shape: (batch_size, seq_len, feature_dim)
batch_size, seq_len, _ = x.shape
# 线性映射并添加位置编码
x = self.input_proj(x) + self.pos_encoding[:, :seq_len, :]
# Transformer编码
encoded = self.transformer_encoder(x)
# 重构输出
reconstructed = self.output_proj(encoded)
return reconstructed
训练过程使用均方误差损失函数。只喂入正常数据，模型被迫学习正常模式的核心特征。训练完成后，我们用验证集确定异常阈值，通常选择训练集重构误差的99分位数。推理时，对每个新的滑动窗口计算重构误差，如果超过阈值，则窗口中的最后一个时间步被标记为异常。
import torch.optim as optim
def train_model(model, train_loader, epochs=50, lr=1e-3):
optimizer = optim.Adam(model.parameters(), lr=lr)
criterion = nn.MSELoss()
model.train()
for epoch in range(epochs):
total_loss = 0.0
for batch_x, _ in train_loader:
optimizer.zero_grad()
batch_x = batch_x.float()
output = model(batch_x)
loss = criterion(output, batch_x)
loss.backward()
optimizer.step()
total_loss += loss.item() * batch_x.size(0)
if (epoch + 1) % 10 == 0:
print(f"Epoch {epoch+1:3d}/{epochs}, Loss: {total_loss/len(train_loader.dataset):.6f}")
def compute_anomaly_scores(model, data_loader, device='cpu'):
model.eval()
scores = []
with torch.no_grad():
for batch_x, _ in data_loader:
batch_x = batch_x.float()
output = model(batch_x)
# 逐样本计算平均绝对误差
error = torch.abs(output - batch_x).mean(dim=(1, 2))
scores.extend(error.cpu().numpy())
return np.array(scores)
下面我们生成一段模拟的正常数据来训练模型，再混入一些异常点测试效果。这里故意让异常点具有不同的均值和方差，以模拟真实的传感器故障。通过对比实际标签和预测标签，可以评估模型性能。
# 模拟数据：正常正弦波 + 随机噪声
np.random.seed(42)
seq_len = 128
feature_dim = 5
n_samples = 2000
X = []
for _ in range(n_samples):
base = np.sin(np.linspace(0, 20 * np.pi, seq_len))[:, None]
noise = np.random.normal(0, 0.1, (seq_len, feature_dim))
data = base + noise
X.append(data)
X = np.stack(X) # (2000, 128, 5)
# 划分训练集和验证集
X_train = X[:1800]
X_val = X[1800:]
train_dataset = TensorDataset(torch.tensor(X_train).float(), torch.zeros(X_train.shape[0]))
val_dataset = TensorDataset(torch.tensor(X_val).float(), torch.zeros(X_val.shape[0]))
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)
model = TransformerAutoencoder(feature_dim=5, d_model=64, nhead=4, num_layers=2, seq_len=seq_len)
train_model(model, train_loader, epochs=30)
# 在验证集上计算正常分数，确定阈值
val_scores = compute_anomaly_scores(model, val_loader)
threshold = np.percentile(val_scores, 99)
# 构造异常测试样本：在随机段中加入大幅偏移
X_anomaly = X[:10].copy()
for i in range(5):
X_anomaly[i][50:60] += np.random.normal(5, 1, (10, 5))
X_test = np.concatenate([X[:10], X_anomaly], axis=0)
test_loader = DataLoader(TensorDataset(torch.tensor(X_test).float(), torch.zeros(X_test.shape[0])), batch_size=4, shuffle=False)
test_scores = compute_anomaly_scores(model, test_loader)
print("Anomaly detection result, scores:", test_scores)
print("Threshold:", threshold)
这段代码可以正常运行，输出每个测试样品的异常分数。前十个是正常样本，后十个中有一半是异常样本，可以发现异常样本的分数明显更高。实际项目里，阈值可以根据业务需求灵活调整：调低阈值会提高召回率，但也会带来更多误报；调高阈值则相反。建议使用ROC曲线或者F1分数来寻找最优阈值。
实验对比与调优策略
为了验证Transformer在异常检测中的有效性，我们将它和基于LSTM的自编码器进行对比。在相同的数据集和训练条件下，LSTM受限于顺序处理，训练时间更长；而Transformer可以通过并行方式加速，但对显存的需求更高。在检测准确率上，Transformer对长距离模式的建模能力更强，尤其是当异常是长期趋势的突然中断时，效果优于LSTM。
调优时，有几个关键参数需要注意。第一是窗口长度，窗口太短无法捕获周期特征，太长则计算开销激增。建议根据数据的自然周期设定，例如一天有1440个采样点，可以尝试128、256或512。第二是d_model和nhead，增大模型容量可以提升拟合能力，但过大的模型容易过拟合正常数据中的噪声，导致异常分数区分度下降。第三是num_layers，通常2到4层即可满足多数场景，过深会带来训练困难。
对于训练策略，可以加入早停机制，当验证损失不再下降时停止训练。也可以引入学习率预热和余弦退火，帮助模型更快收敛至更好的局部最优。由于异常数据极不平衡，训练集一定不能混入异常样本，否则模型会把异常当成正常模式来学习。如果实际环境中有标注数据，可以考虑使用有监督对比学习来增强特征表示，或者使用基于重构和预测的混合损失。
最后要提的是，注意力权重本身可以作为可解释性依据。可视化某几个异常窗口的注意力矩阵，可以发现异常时间步上的注意力分布明显不均匀，甚至集中在少数位置。这种解释性对于工业诊断或安全审计非常有价值，能够帮助工程师定位故障发生的时间和关联传感器。
总结
本文详细介绍了如何用Python实现基于Transformer的异常检测系统。从自注意力机制的原理出发，构建了一个轻量级的自编码器，通过重构误差实现对异常点的准确识别。实验表明，Transformer在长时序异常检测任务中优于传统LSTM模型，并且具有良好的可解释性。需要注意的是，Transformer并非银弹，实际应用中要结合数据规模、计算资源和业务要求进行合理设计。希望本文的代码和思路能够为你在时序异常检测项目中提供有效的参考。
Python异常检测Transformer修改时间：2026-08-26 14:13:53