从零构建Informer深入解析ProbSparse注意力机制与实战代码实现如果你正在寻找一种能够处理超长序列预测任务的强大模型那么Informer绝对值得你投入时间深入研究。这个基于Transformer改进的模型在AAAI 2021上获得了最佳论文奖它解决了传统Transformer在长序列预测中的三大痛点二次计算复杂度、高内存消耗和逐步解码的低效问题。今天我将带你从零开始一步步搭建Informer的核心组件特别是其革命性的ProbSparse注意力机制。1. 理解Informer的设计哲学在深入代码之前我们需要先理解Informer要解决的核心问题。传统Transformer的自注意力机制虽然强大但在处理长序列时面临严峻挑战。想象一下当序列长度达到1000时自注意力矩阵的大小就是1000×1000计算量和内存消耗都呈平方级增长。Informer的解决方案相当巧妙不是所有注意力权重都同等重要。实际上在大多数情况下只有少数query-key对真正贡献了大部分注意力权重。这个观察是ProbSparse注意力机制的基础。1.1 传统自注意力的效率瓶颈让我们先看看标准自注意力的计算方式import torch import torch.nn as nn import numpy as np class StandardSelfAttention(nn.Module): def __init__(self, d_model, n_heads): super().__init__() self.d_model d_model self.n_heads n_heads self.head_dim d_model // n_heads self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.W_o nn.Linear(d_model, d_model) def forward(self, x): batch_size, seq_len, _ x.shape # 计算Q, K, V Q self.W_q(x) # [batch, seq_len, d_model] K self.W_k(x) # [batch, seq_len, d_model] V self.W_v(x) # [batch, seq_len, d_model] # 多头注意力拆分 Q Q.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2) K K.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2) V V.view(batch_size, seq_len, self.n_heads, self.head_dim).transpose(1, 2) # 计算注意力分数 - 这里就是O(L^2)复杂度的来源 scores torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(self.head_dim) attention torch.softmax(scores, dim-1) # 加权求和 output torch.matmul(attention, V) # 合并多头 output output.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model) output self.W_o(output) return output注意上面的代码展示了标准自注意力的实现其中torch.matmul(Q, K.transpose(-2, -1))这行代码的时间复杂度是O(L²)当序列长度L很大时这会成为性能瓶颈。1.2 ProbSparse注意力的核心思想ProbSparse注意力的关键洞察是我们可以只计算那些重要的query的注意力权重。但如何判断哪些query是重要的呢Informer的作者提出了一个基于KL散度的稀疏性度量方法。对于第i个query其稀疏性度量M(q_i, K)定义为M(q_i, K) max_j {q_i·k_j^T / √d} - 1/L * Σ_{j1}^L (q_i·k_j^T / √d)这个度量的直观理解是如果一个query与所有key的点积分布接近均匀分布那么这个query就是懒惰的反之如果它与某些key的点积特别大与其他的特别小那么这个query就是活跃的应该被选中进行计算。2. 环境配置与数据准备在开始实现之前我们需要搭建合适的开发环境。Informer的官方实现基于PyTorch我推荐使用Python 3.8和PyTorch 1.8版本。2.1 创建虚拟环境# 创建并激活conda环境 conda create -n informer_env python3.8 conda activate informer_env # 安装PyTorch根据你的CUDA版本选择 conda install pytorch1.12.1 torchvision0.13.1 torchaudio0.12.1 cudatoolkit11.3 -c pytorch # 安装其他依赖 pip install numpy pandas scikit-learn matplotlib pip install tqdm tensorboard2.2 数据加载与预处理Informer支持多种时间序列数据集包括ETT电力变压器温度、ECL电力消耗和Weather等。让我们看看如何准备自己的数据。import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler import torch from torch.utils.data import Dataset, DataLoader class TimeSeriesDataset(Dataset): def __init__(self, data_path, seq_len96, label_len48, pred_len24, featuresM, targetOT, scaleTrue, freqh): 初始化时间序列数据集 参数: data_path: 数据文件路径 seq_len: 输入序列长度 label_len: decoder输入中已知序列的长度 pred_len: 预测序列长度 features: 预测模式M为多变量预测多变量S为单变量预测单变量 target: 目标特征列名 scale: 是否进行标准化 freq: 时间频率 self.seq_len seq_len self.label_len label_len self.pred_len pred_len self.features features self.target target self.scale scale self.freq freq # 读取数据 df_raw pd.read_csv(data_path) # 提取时间特征 df_stamp df_raw[[date]] df_stamp[date] pd.to_datetime(df_stamp[date]) df_stamp[month] df_stamp[date].dt.month df_stamp[day] df_stamp[date].dt.day df_stamp[weekday] df_stamp[date].dt.weekday df_stamp[hour] df_stamp[date].dt.hour self.data_stamp df_stamp.drop([date], axis1).values # 提取数值特征 cols list(df_raw.columns) cols.remove(date) if features S: cols [target] df_data df_raw[cols] # 数据标准化 if scale: self.scaler StandardScaler() self.data self.scaler.fit_transform(df_data.values) else: self.data df_data.values self.data_x self.data self.data_y self.data def __len__(self): return len(self.data_x) - self.seq_len - self.pred_len 1 def __getitem__(self, index): s_begin index s_end s_begin self.seq_len r_begin s_end - self.label_len r_end r_begin self.label_len self.pred_len # encoder输入 seq_x self.data_x[s_begin:s_end] seq_x_mark self.data_stamp[s_begin:s_end] # decoder输入前label_len是已知序列后pred_len用0填充 seq_y self.data_y[r_begin:r_end] seq_y_mark self.data_stamp[r_begin:r_end] # 对于decoder输入预测部分用0填充 if self.features MS or self.features M: dec_inp np.zeros((self.label_len self.pred_len, self.data_y.shape[1])) else: dec_inp np.zeros((self.label_len self.pred_len, 1)) dec_inp[:self.label_len] seq_y[:self.label_len] dec_inp_mark seq_y_mark return (torch.FloatTensor(seq_x), torch.FloatTensor(seq_x_mark), torch.FloatTensor(dec_inp), torch.FloatTensor(dec_inp_mark), torch.FloatTensor(seq_y))提示在实际项目中时间特征的处理非常重要。除了基本的年月日时分秒还可以考虑节假日、季节等周期性特征。Informer的官方实现中使用了更复杂的时间编码方式包括正弦位置编码和可学习的时间嵌入。3. ProbSparse注意力机制的完整实现现在让我们进入最核心的部分——实现ProbSparse注意力机制。这是Informer区别于传统Transformer的关键创新。3.1 稀疏性度量的实现首先我们需要实现query稀疏性的度量函数import torch import torch.nn as nn import torch.nn.functional as F import math class ProbSparseAttention(nn.Module): def __init__(self, d_model, n_heads, factor5, dropout0.1): super().__init__() self.d_model d_model self.n_heads n_heads self.head_dim d_model // n_heads self.factor factor # 采样因子控制采样数量 # 线性变换层 self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.W_o nn.Linear(d_model, d_model) self.dropout nn.Dropout(dropout) def _prob_QK(self, Q, K, sample_k, n_top): 计算ProbSparse注意力 参数: Q: query矩阵 [batch, n_heads, seq_len, head_dim] K: key矩阵 [batch, n_heads, seq_len, head_dim] sample_k: 采样的key数量 n_top: 选择的top query数量 返回: Q_K: 稀疏注意力分数 M_top: top query的索引 B, H, L_K, E K.shape _, _, L_Q, _ Q.shape # 计算每个query的稀疏性度量M K_expand K.unsqueeze(-3).expand(B, H, L_Q, L_K, E) # 随机采样部分key index_sample torch.randint(L_K, (L_Q, sample_k)) K_sample K_expand[:, :, torch.arange(L_Q).unsqueeze(1), index_sample, :] # 计算query与采样key的点积 Q_K_sample torch.matmul(Q.unsqueeze(-2), K_sample.transpose(-2, -1)).squeeze() # 计算稀疏性度量M M Q_K_sample.max(-1)[0] - Q_K_sample.sum(-1) / sample_k # 选择top n_top个query M_top M.topk(n_top, sortedFalse)[1] return M_top def _get_initial_context(self, V, L_Q): 初始化上下文向量 参数: V: value矩阵 L_Q: query序列长度 返回: context: 初始化的上下文向量 B, H, L_V, D V.shape if not self.mask_flag: V_sum V.mean(dim-2) contex V_sum.unsqueeze(-2).expand(B, H, L_Q, V_sum.shape[-1]).clone() else: contex V.cumsum(dim-2) return contex def _update_context(self, context_in, V, scores, index, L_Q): 更新选中的query的上下文 参数: context_in: 初始上下文 V: value矩阵 scores: 注意力分数 index: 选中的query索引 L_Q: query序列长度 B, H, L_V, D V.shape if self.mask_flag: attn_mask ProbMask(B, H, L_Q, index, scores, deviceV.device) scores.masked_fill_(attn_mask.mask, -np.inf) attn torch.softmax(scores, dim-1) context_in[torch.arange(B)[:, None, None], torch.arange(H)[None, :, None], index, :] torch.matmul(attn, V).type_as(context_in) return context_in def forward(self, queries, keys, values): B, L_Q, _ queries.shape _, L_K, _ keys.shape H self.n_heads # 线性变换 Q self.W_q(queries).view(B, L_Q, H, -1).transpose(1, 2) K self.W_k(keys).view(B, L_K, H, -1).transpose(1, 2) V self.W_v(values).view(B, L_K, H, -1).transpose(1, 2) # 计算要采样的key数量和要选择的query数量 U_part self.factor * np.ceil(np.log(L_K)).astype(int).item() u self.factor * np.ceil(np.log(L_Q)).astype(int).item() U_part U_part if U_part L_K else L_K u u if u L_Q else L_Q # 计算稀疏性并选择top query scores_top, index self._prob_QK(Q, K, sample_kU_part, n_topu) # 初始化上下文 context self._get_initial_context(V, L_Q) # 更新选中的query的上下文 context self._update_context(context, V, scores_top, index, L_Q) # 合并多头 context context.transpose(1, 2).contiguous().view(B, L_Q, -1) # 输出投影 output self.W_o(context) output self.dropout(output) return output3.2 注意力蒸馏的实现注意力蒸馏是Informer的另一个重要创新它通过下采样减少序列长度从而降低计算复杂度class ConvLayer(nn.Module): 注意力蒸馏层通过卷积和池化减少序列长度 def __init__(self, c_in, c_out, kernel_size3): super().__init__() padding (kernel_size - 1) // 2 self.downConv nn.Conv1d(in_channelsc_in, out_channelsc_out, kernel_sizekernel_size, paddingpadding, padding_modecircular) self.norm nn.BatchNorm1d(c_out) self.activation nn.ELU() self.maxPool nn.MaxPool1d(kernel_size3, stride2, padding1) def forward(self, x): # x: [batch_size, seq_len, d_model] x self.downConv(x.transpose(1, 2)) # 转换为 [batch_size, d_model, seq_len] x self.norm(x) x self.activation(x) x self.maxPool(x) x x.transpose(1, 2) # 转换回 [batch_size, seq_len, d_model] return x class EncoderLayer(nn.Module): Informer编码器层包含ProbSparse注意力和注意力蒸馏 def __init__(self, attention, d_model, d_ffNone, dropout0.1, activationrelu, distilTrue): super().__init__() d_ff d_ff or 4 * d_model self.attention attention self.distil distil # 前馈网络 self.conv1 nn.Conv1d(in_channelsd_model, out_channelsd_ff, kernel_size1) self.conv2 nn.Conv1d(in_channelsd_ff, out_channelsd_model, kernel_size1) # 归一化层 self.norm1 nn.LayerNorm(d_model) self.norm2 nn.LayerNorm(d_model) self.dropout nn.Dropout(dropout) self.activation F.relu if activation relu else F.gelu # 注意力蒸馏层 if distil: self.conv_layer ConvLayer(d_model, d_model) def forward(self, x, attn_maskNone): # 自注意力 残差连接 new_x, attn self.attention(x, x, x, attn_maskattn_mask) x x self.dropout(new_x) x self.norm1(x) # 前馈网络 残差连接 y x.transpose(1, 2) y self.dropout(self.activation(self.conv1(y))) y self.dropout(self.conv2(y)) y y.transpose(1, 2) x x y x self.norm2(x) # 注意力蒸馏 if self.distil: x self.conv_layer(x) return x, attn4. 构建完整的Informer模型现在我们已经有了所有核心组件让我们把它们组合成完整的Informer模型。4.1 编码器实现class Encoder(nn.Module): Informer编码器堆叠多个编码器层每层后接注意力蒸馏 def __init__(self, attn_layers, conv_layersNone, norm_layerNone): super().__init__() self.attn_layers nn.ModuleList(attn_layers) self.conv_layers nn.ModuleList(conv_layers) if conv_layers is not None else None self.norm norm_layer def forward(self, x, attn_maskNone): attns [] if self.conv_layers is not None: # 有注意力蒸馏的情况 for attn_layer, conv_layer in zip(self.attn_layers, self.conv_layers): x, attn attn_layer(x, attn_maskattn_mask) x conv_layer(x) # 应用注意力蒸馏 attns.append(attn) # 最后一层不进行蒸馏 x, attn self.attn_layers[-1](x, attn_maskattn_mask) attns.append(attn) else: # 没有注意力蒸馏的情况 for attn_layer in self.attn_layers: x, attn attn_layer(x, attn_maskattn_mask) attns.append(attn) if self.norm is not None: x self.norm(x) return x, attns class DecoderLayer(nn.Module): Informer解码器层包含自注意力和交叉注意力 def __init__(self, self_attention, cross_attention, d_model, d_ffNone, dropout0.1, activationrelu): super().__init__() d_ff d_ff or 4 * d_model self.self_attention self_attention self.cross_attention cross_attention # 前馈网络 self.conv1 nn.Conv1d(in_channelsd_model, out_channelsd_ff, kernel_size1) self.conv2 nn.Conv1d(in_channelsd_ff, out_channelsd_model, kernel_size1) # 归一化层 self.norm1 nn.LayerNorm(d_model) self.norm2 nn.LayerNorm(d_model) self.norm3 nn.LayerNorm(d_model) self.dropout nn.Dropout(dropout) self.activation F.relu if activation relu else F.gelu def forward(self, x, cross, x_maskNone, cross_maskNone): # 自注意力 残差连接 x x self.dropout(self.self_attention( x, x, x, attn_maskx_mask )[0]) x self.norm1(x) # 交叉注意力 残差连接 x x self.dropout(self.cross_attention( x, cross, cross, attn_maskcross_mask )[0]) x self.norm2(x) # 前馈网络 残差连接 y x.transpose(1, 2) y self.dropout(self.activation(self.conv1(y))) y self.dropout(self.conv2(y)) y y.transpose(1, 2) x x y x self.norm3(x) return x4.2 完整的Informer模型class Informer(nn.Module): 完整的Informer模型 def __init__(self, enc_in, dec_in, c_out, seq_len, label_len, out_len, factor5, d_model512, n_heads8, e_layers3, d_layers2, d_ff2048, dropout0.0, attnprob, embedfixed, freqh, activationgelu, output_attentionFalse, distilTrue, mixTrue, devicetorch.device(cuda:0)): super().__init__() self.pred_len out_len self.attn attn self.output_attention output_attention # 编码 self.enc_embedding DataEmbedding(enc_in, d_model, embed, freq, dropout) # 解码 self.dec_embedding DataEmbedding(dec_in, d_model, embed, freq, dropout) # 注意力机制选择 Attn ProbSparseAttention if attn prob else FullAttention # 编码器 self.encoder Encoder( [ EncoderLayer( Attn(False, factor, attention_dropoutdropout, output_attentionoutput_attention), d_model, d_ff, dropoutdropout, activationactivation, distildistil ) for l in range(e_layers) ], [ ConvLayer( d_model ) for l in range(e_layers-1) ] if distil else None, norm_layertorch.nn.LayerNorm(d_model) ) # 解码器 self.decoder Decoder( [ DecoderLayer( Attn(True, factor, attention_dropoutdropout, output_attentionFalse), Attn(False, factor, attention_dropoutdropout, output_attentionFalse), d_model, d_ff, dropoutdropout, activationactivation, ) for l in range(d_layers) ], norm_layertorch.nn.LayerNorm(d_model) ) # 输出投影 self.projection nn.Linear(d_model, c_out, biasTrue) def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, enc_self_maskNone, dec_self_maskNone, dec_enc_maskNone): # 编码器 enc_out self.enc_embedding(x_enc, x_mark_enc) enc_out, attns self.encoder(enc_out, attn_maskenc_self_mask) # 解码器 dec_out self.dec_embedding(x_dec, x_mark_dec) dec_out self.decoder(dec_out, enc_out, x_maskdec_self_mask, cross_maskdec_enc_mask) # 投影到输出维度 dec_out self.projection(dec_out) if self.output_attention: return dec_out[:, -self.pred_len:, :], attns else: return dec_out[:, -self.pred_len:, :] # [B, L, D]5. 训练策略与调优技巧实现模型只是第一步如何有效地训练Informer同样重要。下面是一些我在实践中总结的训练技巧。5.1 学习率调度与早停class EarlyStopping: 早停机制 def __init__(self, patience7, verboseFalse, delta0): self.patience patience self.verbose verbose self.counter 0 self.best_score None self.early_stop False self.val_loss_min np.Inf self.delta delta def __call__(self, val_loss, model, path): score -val_loss if self.best_score is None: self.best_score score self.save_checkpoint(val_loss, model, path) elif score self.best_score self.delta: self.counter 1 print(fEarlyStopping counter: {self.counter} out of {self.patience}) if self.counter self.patience: self.early_stop True else: self.best_score score self.save_checkpoint(val_loss, model, path) self.counter 0 def save_checkpoint(self, val_loss, model, path): if self.verbose: print(fValidation loss decreased ({self.val_loss_min:.6f} -- {val_loss:.6f}). Saving model...) torch.save(model.state_dict(), path) self.val_loss_min val_loss class AdjustLearningRate: 学习率调整策略 def __init__(self, optimizer, lradjtype1): self.optimizer optimizer self.lradj lradj self.initial_lr [param_group[lr] for param_group in optimizer.param_groups] def step(self, epoch): if self.lradj type1: lr_adjust {epoch: lr * (0.5 ** ((epoch - 1) // 1)) for lr in self.initial_lr} elif self.lradj type2: lr_adjust { 2: 5e-5, 4: 1e-5, 6: 5e-6, 8: 1e-6, 10: 5e-7, 15: 1e-7, 20: 5e-8 } if epoch in lr_adjust.keys(): for param_group, lr in zip(self.optimizer.param_groups, lr_adjust[epoch]): param_group[lr] lr print(fUpdating learning rate to {lr_adjust[epoch]})5.2 训练循环实现def train_epoch(model, train_loader, criterion, optimizer, scheduler, epoch, device): 单个epoch的训练 model.train() total_loss 0 for batch_idx, (batch_x, batch_x_mark, batch_y, batch_y_mark, batch_y_true) in enumerate(train_loader): optimizer.zero_grad() batch_x batch_x.float().to(device) batch_x_mark batch_x_mark.float().to(device) batch_y batch_y.float().to(device) batch_y_mark batch_y_mark.float().to(device) batch_y_true batch_y_true.float().to(device) # decoder输入 dec_inp torch.zeros_like(batch_y[:, -model.pred_len:, :]).float() dec_inp torch.cat([batch_y[:, :model.label_len, :], dec_inp], dim1).float().to(device) # 前向传播 if model.output_attention: outputs, attentions model(batch_x, batch_x_mark, dec_inp, batch_y_mark) else: outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) # 计算损失 loss criterion(outputs, batch_y_true[:, -model.pred_len:, :]) # 反向传播 loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm0.5) optimizer.step() total_loss loss.item() if batch_idx % 100 0: print(fTrain Epoch: {epoch} [{batch_idx * len(batch_x)}/{len(train_loader.dataset)} f({100. * batch_idx / len(train_loader):.0f}%)]\tLoss: {loss.item():.6f}) scheduler.step() return total_loss / len(train_loader) def validate(model, val_loader, criterion, device): 验证 model.eval() total_loss 0 with torch.no_grad(): for batch_x, batch_x_mark, batch_y, batch_y_mark, batch_y_true in val_loader: batch_x batch_x.float().to(device) batch_x_mark batch_x_mark.float().to(device) batch_y batch_y.float().to(device) batch_y_mark batch_y_mark.float().to(device) batch_y_true batch_y_true.float().to(device) dec_inp torch.zeros_like(batch_y[:, -model.pred_len:, :]).float() dec_inp torch.cat([batch_y[:, :model.label_len, :], dec_inp], dim1).float().to(device) if model.output_attention: outputs, _ model(batch_x, batch_x_mark, dec_inp, batch_y_mark) else: outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) loss criterion(outputs, batch_y_true[:, -model.pred_len:, :]) total_loss loss.item() return total_loss / len(val_loader)6. 实战应用与性能优化在实际项目中应用Informer时有几个关键点需要注意6.1 超参数调优根据我的经验以下超参数设置通常能获得不错的效果参数推荐值说明d_model512模型维度影响表示能力n_heads8注意力头数e_layers2-3编码器层数d_layers1-2解码器层数d_ff2048前馈网络维度factor5ProbSparse采样因子dropout0.05-0.1防止过拟合batch_size32-64根据显存调整learning_rate0.0001初始学习率6.2 内存优化技巧处理长序列时内存管理至关重要# 使用混合精度训练 from torch.cuda.amp import autocast, GradScaler scaler GradScaler() def train_with_amp(model, train_loader, criterion, optimizer, device): model.train() for batch in train_loader: optimizer.zero_grad() with autocast(): # 前向传播自动混合精度 outputs model(*batch[:-1]) loss criterion(outputs, batch[-1]) # 反向传播自动缩放梯度 scaler.scale(loss).backward() scaler.step(optimizer) scaler.update() # 梯度累积处理大batch size def train_with_gradient_accumulation(model, train_loader, criterion, optimizer, accumulation_steps4, devicecuda): model.train() optimizer.zero_grad() for i, batch in enumerate(train_loader): # 前向传播 outputs model(*batch[:-1]) loss criterion(outputs, batch[-1]) # 反向传播累积梯度 loss loss / accumulation_steps loss.backward() if (i 1) % accumulation_steps 0: optimizer.step() optimizer.zero_grad()6.3 多GPU训练对于大规模数据集可以使用多GPU加速训练import torch.nn as nn import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP def setup_ddp(): 设置分布式训练 dist.init_process_group(backendnccl) torch.cuda.set_device(int(os.environ[LOCAL_RANK])) def train_ddp(model, train_loader, criterion, optimizer, device): 分布式训练 # 包装模型 model DDP(model, device_ids[device]) for epoch in range(num_epochs): # 设置sampler train_loader.sampler.set_epoch(epoch) for batch in train_loader: # 训练逻辑... pass7. 常见问题与解决方案在实现和训练Informer过程中你可能会遇到以下问题7.1 训练不收敛问题表现损失值波动大或持续不下降。解决方案检查学习率是否合适尝试使用更小的学习率如1e-5增加梯度裁剪的阈值检查数据标准化是否正确尝试不同的优化器AdamW通常效果更好# 使用AdamW优化器 optimizer torch.optim.AdamW(model.parameters(), lr1e-4, weight_decay1e-5) # 使用余弦退火学习率调度 scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max10)7.2 过拟合问题问题表现训练损失持续下降但验证损失开始上升。解决方案增加dropout率使用更早的停止策略增加L2正则化使用数据增强# 数据增强添加噪声 def add_noise(data, noise_level0.01): noise torch.randn_like(data) * noise_level return data noise # 在训练循环中使用 batch_x add_noise(batch_x, noise_level0.01)7.3 内存不足问题表现GPU内存溢出。解决方案减小batch size使用梯度累积使用混合精度训练减少模型大小减小d_model或层数# 检查GPU内存使用情况 import torch print(fGPU内存使用: {torch.cuda.memory_allocated() / 1024**3:.2f} GB) print(fGPU内存缓存: {torch.cuda.memory_cached() / 1024**3:.2f} GB) # 清理缓存 torch.cuda.empty_cache()8. 模型评估与结果分析训练完成后我们需要评估模型的性能。对于时间序列预测常用的评估指标包括MSE、MAE和MAPE。def evaluate_model(model, test_loader, scalerNone, devicecuda): 评估模型性能 model.eval() predictions [] actuals [] with torch.no_grad(): for batch_x, batch_x_mark, batch_y, batch_y_mark, batch_y_true in test_loader: batch_x batch_x.float().to(device) batch_x_mark batch_x_mark.float().to(device) batch_y batch_y.float().to(device) batch_y_mark batch_y_mark.float().to(device) dec_inp torch.zeros_like(batch_y[:, -model.pred_len:, :]).float() dec_inp torch.cat([batch_y[:, :model.label_len, :], dec_inp], dim1).float().to(device) outputs model(batch_x, batch_x_mark, dec_inp, batch_y_mark) # 反标准化 if scaler is not None: outputs scaler.inverse_transform(outputs.cpu().numpy()) batch_y_true scaler.inverse_transform(batch_y_true[:, -model.pred_len:, :].cpu().numpy()) predictions.append(outputs) actuals.append(batch_y_true) predictions np.concatenate(predictions, axis0) actuals np.concatenate(actuals, axis0) # 计算评估指标 mse np.mean((predictions - actuals) ** 2) mae np.mean(np.abs(predictions - actuals)) mape np.mean(np.abs((predictions - actuals) / actuals)) * 100 return { mse: mse, mae: mae, mape: mape, predictions: predictions, actuals: actuals } def plot_predictions(predictions, actuals, feature_idx0, num_samples5): 可视化预测结果 import matplotlib.pyplot as plt fig, axes plt.subplots(num_samples, 1, figsize(15, 3*num_samples)) for i in range(num_samples): ax axes[i] if num_samples 1 else axes ax.plot(actuals[i, :, feature_idx], labelActual, linewidth2) ax.plot(predictions[i, :, feature_idx], labelPredicted, linestyle--, linewidth2) ax.set_title(fSample {i1}) ax.legend() ax.grid(True) plt.tight_layout() plt.show()在实际项目中我发现Informer在电力负荷预测、股票价格预测和气象数据预测等任务上表现优异。特别是在处理具有明显周期性和趋势性的数据时ProbSparse注意力机制能够有效捕捉长期依赖关系而注意力蒸馏则显著降低了计算成本。一个实用的建议是在应用Informer到新领域时先从较小的模型开始如d_model256e_layers2观察训练效果后再逐步增加模型复杂度。同时合理设置seq_len、label_len和pred_len参数对模型性能影响很大需要根据具体任务的数据特性进行调整。最后记住深度学习模型的训练既是科学也是艺术。虽然Informer提供了强大的理论基础和实现框架但在实际应用中数据预处理、特征工程和超参数调优同样重要。多实验、多分析、多调整你就能让Informer在你的特定任务上发挥最大价值。