简介本资源是一套基于Transformer架构的多模态异常检测完整实践方案面向具备Python与深度学习基础的算法工程师、研究生及进阶学习者聚焦工业监控、系统运维等场景下的跨模态异常识别问题。压缩包共314个文件含164个npy格式多模态样本数据如温度、CPU利用率、出租车流量等时序信号、116个txt日志与配置说明、12个csv标注数据集含machine_temperature_system_failure、nyc_taxi等8类真实/合成异常数据、11个md文档含环境配置、模型训练与评估全流程教程以及4个核心py脚本整体大小为107.6MB。已有254人学习下载。读者可直接复现多模态Transformer建模流程获得从数据加载、特征对齐、自注意力机制设计到异常评分输出的端到端代码实现并配套详细README与标签化数据集显著降低多模态异常检测的入门门槛与调试成本。1. 这不是NLP模型移植而是用Transformer重写异常检测的底层逻辑你手头这份transformer_multimodal_anomaly_detection.zip表面看是“多模态Transformer异常检测”三词堆砌但实际它绕开了传统时序建模的惯性路径不依赖LSTM堆叠、不硬套CNN提取局部特征、更不把多源数据强行拼成单通道输入。它用纯注意力机制对齐不同采样频率、不同物理量纲、不同语义粒度的数据流——比如把cpu_utilization_asg_misconfiguration.csv秒级CPU使用率和ambient_temperature_system_failure.csv分钟级环境温度在token层面做跨模态位置嵌入对齐再通过交叉注意力层建模二者在系统过载前30分钟内的协同异常模式。项目内含8个真实工业场景CSV文件覆盖云服务指标、IoT设备温控、城市交通流、恶意行为日志四类典型异构数据源且全部标注了anomaly_label字段非二分类而是带时间戳的segment-level标签。适合正在落地预测性维护、SRE告警收敛或边缘设备轻量化异常识别的工程师尤其当你已卡在“单模态模型准确率停滞在82%”阶段时这套方案能帮你验证问题不在数据质量而在建模范式本身。2. 多模态对齐不是拼接而是用可学习位置编码重构时间语义2.1 为什么传统方法在多模态异常检测中失效多数开源方案将多源时序数据简单concat后喂入LSTM这隐含两个致命假设所有模态采样频率一致、所有信号对齐到同一时间轴。但现实数据集如nyc_taxi.csv每15分钟聚合与ec2_request_latency_system_failure.csv毫秒级延迟根本无法直接对齐。更关键的是rogue_agent_key_updown.csv键盘事件序列和machine_temperature_system_failure.csv连续温度曲线属于完全不同的数据类型——前者是离散事件流后者是连续值序列。强行统一分辨率会导致信息坍缩降频丢事件细节升频造虚假连续性。本项目采用的解决方案是抛弃“统一采样”思路转而构建模态专属token化器对连续型数据温度、CPU用滑动窗口切片标准化生成token对事件型数据按键、告警用时间间隔编码事件类型one-hot生成token再通过可学习的位置编码矩阵让不同模态的token在共享隐空间中建立时序关系。提示src/preprocess.py中MultiModalTokenizer类的__call__方法实现了该逻辑关键参数max_seq_len128控制各模态token序列最大长度event_window_sec60定义事件型数据的时间聚合窗口这些值需根据你的数据采样率调整。2.2 实现跨模态位置嵌入的三步操作2.2.1 构建模态特异性位置编码# src/embedding.py import torch import torch.nn as nn class ModalityPositionEmbedding(nn.Module): def __init__(self, d_model, max_len1000, n_modalities4): super().__init__() self.d_model d_model self.n_modalities n_modalities # 为每种模态学习独立的位置编码表 self.modality_embeddings nn.Embedding(n_modalities, d_model) # 标准正弦位置编码复用Transformer原始实现 pe torch.zeros(max_len, d_model) position torch.arange(0, max_len, dtypetorch.float).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) self.register_buffer(pe, pe) def forward(self, x, modality_id): # x: [batch_size, seq_len, d_model] # modality_id: int, 0temperature, 1cpu, 2taxi, 3events pos_embed self.pe[:x.size(1), :].unsqueeze(0) # [1, seq_len, d_model] mod_embed self.modality_embeddings(torch.tensor([modality_id])) # [1, d_model] return x pos_embed mod_embed这段代码的核心在于modality_embeddings为每种数据源分配专属偏置向量使同位置不同模态的token在嵌入空间中天然分离而pe矩阵保持标准正弦编码确保时序关系可学习。注意modality_id必须与src/dataset.py中ModalityDataset的modality_map字典严格对应如{temperature: 0, cpu: 1}否则跨模态注意力会计算错误。2.2.2 在Transformer编码器中注入模态感知注意力# src/transformer.py class CrossModalAttention(nn.Module): def __init__(self, d_model, nhead, dropout0.1): super().__init__() self.multihead_attn nn.MultiheadAttention(d_model, nhead, dropoutdropout, batch_firstTrue) # 添加模态门控控制不同模态间信息流动强度 self.modality_gate nn.Sequential( nn.Linear(d_model * 2, d_model), nn.Sigmoid() ) def forward(self, query, key, value, modality_maskNone): # query: [batch, seq_q, d_model], key/value: [batch, seq_kv, d_model] attn_output, _ self.multihead_attn(query, key, value, attn_maskmodality_mask) # 模态门控融合query和attn_output的特征 gate_input torch.cat([query, attn_output], dim-1) gate self.modality_gate(gate_input) # [batch, seq_q, d_model] return gate * attn_output (1 - gate) * query此处CrossModalAttention替代了标准Transformer中的MultiheadAttention关键创新是modality_gate它动态计算每个位置上原始query与注意力输出的融合权重。当某模态token如温度突增与另一模态如CPU飙升存在强关联时门控值趋近1允许信息充分流动若无关联则保留原始query特征。modality_mask参数用于屏蔽非法模态交互如禁止taxi流量数据直接attend到键盘事件其生成逻辑见src/utils.py中generate_modality_mask()函数。2.2.3 数据加载时的模态对齐策略# src/dataset.py class MultiModalBatchSampler: def __init__(self, data_paths, label_path, window_size128, step64): self.data_paths data_paths # [temperature.csv, cpu.csv, ...] self.label_path label_path self.window_size window_size self.step step self._load_and_align_data() def _load_and_align_data(self): # 步骤1读取所有CSV统一转为datetime索引 dfs [] for path in self.data_paths: df pd.read_csv(path, parse_dates[timestamp]) df.set_index(timestamp, inplaceTrue) dfs.append(df) # 步骤2以最高频数据为基准重采样其他模态 base_freq min(df.index.freq for df in dfs if hasattr(df.index, freq) and df.index.freq) aligned_dfs [] for df in dfs: if df.index.freq ! base_freq: # 对连续型数据用线性插值事件型数据用前向填充 if event_type in df.columns: aligned_df df.resample(base_freq).ffill() else: aligned_df df.resample(base_freq).interpolate(methodlinear) aligned_dfs.append(aligned_df) else: aligned_dfs.append(df) # 步骤3按window_size滑动切片确保每个batch包含完整模态组合 self.windows [] for i in range(0, len(aligned_dfs[0]) - self.window_size 1, self.step): window_data [df.iloc[i:iself.window_size].values for df in aligned_dfs] self.windows.append(window_data)该采样器强制执行三个原则① 所有数据必须解析timestamp列并设为DatetimeIndex② 以最高频数据如ec2_request_latency_system_failure.csv的毫秒级为重采样基准③ 对事件型数据含event_type列使用ffill()避免插值伪造事件。最终生成的self.windows是列表每个元素为[temp_window, cpu_window, taxi_window, event_window]四元组直接送入模型训练。3. 训练流程不是端到端拟合而是分阶段解耦优化3.1 阶段一模态内自监督预训练避免标注数据不足多模态异常检测最大的痛点是标注成本高——labeled_anomalies.csv仅提供8个样本的segment-level标签远不足以支撑端到端训练。本项目采用Masked Multimodal ReconstructionMMR预训练策略随机遮蔽各模态15%的token要求模型重建原始值。这使模型先学会各模态内部的时序规律再进入跨模态学习。# 启动预训练在config/pretrain.yaml中配置 python train.py --config config/pretrain.yaml \ --data_dir ./data/synthetic_data_with_anomaly-s-1.csv \ --model_name transformer_pretrain \ --epochs 50 \ --batch_size 32 \ --mask_ratio 0.15config/pretrain.yaml关键参数说明mask_strategy:random随机遮蔽或temporal遮蔽连续时间片段后者更适合检测长周期异常reconstruction_loss:mse连续型或ce事件型自动根据模态类型切换warmup_steps: 1000防止初期梯度爆炸。预训练后模型权重保存在checkpoints/pretrain/目录后续微调将加载此权重而非随机初始化。3.2 阶段二跨模态对比学习强化异常判别能力预训练解决“学什么”对比学习解决“怎么区分”。本项目构造正负样本对正样本对同一时间窗口内不同模态的token序列如温度CPU因物理系统耦合它们应具有相似的异常模式表征负样本对随机打乱时间戳的模态组合如温度taxi流量二者无因果关联。# src/loss.py class ContrastiveLoss(nn.Module): def __init__(self, temperature0.07): super().__init__() self.temperature temperature self.criterion nn.CrossEntropyLoss() def forward(self, z_i, z_j): # z_i, z_j: [batch_size, d_model], 来自不同模态的全局表征 batch_size z_i.size(0) # 构造相似度矩阵z_i与z_j的cosine相似度 sim_matrix torch.mm(z_i, z_j.t()) / self.temperature # 对角线为正样本其余为负样本 labels torch.arange(batch_size, devicez_i.device) loss self.criterion(sim_matrix, labels) return loss训练命令启用对比学习python train.py --config config/contrastive.yaml \ --pretrained_path checkpoints/pretrain/best.pth \ --loss_type contrastive \ --projection_dim 128projection_dim将各模态全局表征映射到128维对比空间contrastive.yaml中num_negatives256控制负样本数量值越大判别越精细但显存消耗越高。3.3 阶段三异常分数回归微调对接业务指标最终阶段放弃分类任务直接回归异常强度分数——因为真实运维场景中anomaly_score 0.85比label 1更具操作价值。模型输出层替换为单神经元线性层损失函数采用Pinball Loss分位数损失确保预测分数在90%置信区间内覆盖真实异常强度# src/model.py class AnomalyScoreHead(nn.Module): def __init__(self, d_model, quantile0.9): super().__init__() self.quantile quantile self.head nn.Linear(d_model, 1) def forward(self, x): # x: [batch_size, seq_len, d_model] # 取最后时刻的表征作为全局异常强度 global_rep x[:, -1, :] # [batch_size, d_model] score torch.sigmoid(self.head(global_rep)) # [batch_size, 1] return score.squeeze(-1) # Pinball Loss实现 def pinball_loss(y_pred, y_true, quantile0.9): error y_true - y_pred return torch.mean(torch.max(quantile * error, (quantile - 1) * error))微调时使用labels.csv中的anomaly_intensity列非binary label该列数值范围0~1由领域专家根据故障严重程度标定。config/fine_tune.yaml中threshold_searchTrue启用动态阈值搜索在验证集上自动确定score threshold的最优判别点。4. 验证不是看AUC而是用时间敏感的RecallK评估4.1 为什么传统指标在异常检测中失效AUC、F1-score等静态指标忽略时间维度——它们把machine_temperature_system_failure.csv中提前2小时出现的温度缓升视为与故障瞬间同等重要的异常信号但运维人员真正需要的是在故障发生前K分钟内捕获首个有效预警信号。本项目定义RecallK为在真实故障时间点前K分钟内模型输出的异常分数是否达到阈值。例如K30即要求模型在故障发生前30分钟内至少有一次score 0.7的预警。# src/evaluation.py def recall_at_k(y_true, y_score, k_minutes30, sampling_rate_sec60): y_true: 二进制数组1表示故障发生时刻 y_score: 连续异常分数数组 sampling_rate_sec: 数据采样间隔秒 k_steps int(k_minutes * 60 / sampling_rate_sec) recall 0 total_faults y_true.sum() # 找到每个故障点的索引 fault_indices np.where(y_true 1)[0] for idx in fault_indices: # 检查故障点前k_steps范围内是否有预警 start_idx max(0, idx - k_steps) window_scores y_score[start_idx:idx1] if (window_scores 0.7).any(): recall 1 return recall / total_faults if total_faults 0 else 0 # 计算所有数据集的RecallK results {} for dataset_name in [temperature, cpu, taxi]: y_true, y_score load_labels_and_scores(dataset_name) results[dataset_name] recall_at_k(y_true, y_score, k_minutes30) print(fRecall30min: {results})该函数关键参数sampling_rate_sec必须与数据实际采样率一致如ambient_temperature_system_failure.csv为60秒ec2_request_latency_system_failure.csv为1秒否则时间窗口计算错误。4.2 可视化异常定位精度的热力图单纯数值指标不够直观项目提供plot_anomaly_heatmap.py生成时空热力图横轴为时间纵轴为模态类型颜色深浅表示该时刻各模态的异常分数# utils/plot_anomaly_heatmap.py def plot_multimodal_heatmap(scores_dict, save_path): scores_dict: {temperature: [scores], cpu: [scores], ...} modalities list(scores_dict.keys()) n_modalities len(modalities) time_steps len(scores_dict[modalities[0]]) # 构建热力图矩阵 heatmap_data np.zeros((n_modalities, time_steps)) for i, modality in enumerate(modalities): heatmap_data[i, :] scores_dict[modality] plt.figure(figsize(12, 4)) sns.heatmap(heatmap_data, xticklabels500, # 每500步标一个时间点 yticklabelsmodalities, cmapRdYlBu_r, cbar_kws{label: Anomaly Score}) plt.title(Multimodal Anomaly Heatmap) plt.savefig(save_path, bbox_inchestight) plt.close() # 使用示例 scores { temperature: model_predict(temperature.csv), cpu: model_predict(cpu.csv), taxi: model_predict(nyc_taxi.csv) } plot_multimodal_heatmap(scores, outputs/heatmap.png)生成的热力图能直观揭示温度异常是否早于CPU异常出现出租车流量突增是否与EC2延迟飙升同步这种时空关联性验证比单一指标更能反映模型的业务价值。4.3 快速验证你的数据是否适配此架构若你有自己的工业数据无需重训整个模型可用以下脚本快速测试适配性# test_compatibility.sh python -c import pandas as pd from src.preprocess import MultiModalTokenizer # 加载你的数据 df pd.read_csv(your_data.csv, parse_dates[timestamp]) print(f数据形状: {df.shape}) print(f时间范围: {df.timestamp.min()} ~ {df.timestamp.max()}) print(f采样频率: {df.timestamp.diff().dt.seconds.mode().iloc[0]} 秒) # 测试token化 tokenizer MultiModalTokenizer(max_seq_len128) try: tokens tokenizer(df, modality_id0) print(f成功生成 {len(tokens)} 个token) except Exception as e: print(ftoken化失败: {e}) 输出中若显示采样频率: 60 秒且成功生成 128 个token说明数据格式符合要求若报错KeyError: timestamp需先添加时间列若采样频率为NaT需用df df.set_index(your_time_col).resample(60S).mean().reset_index()重建时间索引。本文还有配套的精品资源点击获取