1. 从Transformer到Informer长序列预测的“效率革命”如果你用过Transformer做时间序列预测尤其是那种需要预测未来很长一段时间的任务比如预测接下来一个月的电力负荷、未来半年的商品销量那你大概率被它“折磨”过。我刚开始做这类项目时也是信心满满地套上Transformer结果发现序列稍微长一点比如超过100个时间步训练速度就慢得像蜗牛显存占用更是直接“爆掉”。这感觉就像你想开跑车去越野结果发现底盘太低根本跑不起来。Transformer这个在NLP领域大杀四方的模型为什么在长序列预测上会“水土不服”核心问题就出在它的“心脏”——自注意力机制Self-attention上。这个机制要求序列里的每个元素比如每个时刻的用电量都要和序列里所有其他元素计算一遍关联度。想象一下你有一个长度为L的序列那么计算复杂度就是O(L²)。这意味着如果你的序列长度从100变成1000计算量不是简单地增加10倍而是增加100倍这在实际应用中几乎是不可接受的尤其是对于高频的工业传感器数据或金融数据序列动辄成千上万个点。Informer模型的出现就是为了给Transformer做一次彻底的“减负手术”和“效率升级”。它没有抛弃Transformer强大的全局建模能力而是精准地“动刀”解决了三个最痛的瓶颈自注意力的平方复杂度、多层堆叠的内存瓶颈、以及解码预测时的“挤牙膏”式慢速输出。简单来说Informer的目标就是用更少的计算资源处理更长的历史序列做出更快的多步预测。这对于那些需要根据长时间历史数据比如过去一年的数据来预测未来趋势的场景比如供应链库存预测、能源调度、设备故障预警简直是雪中送炭。接下来我们就掰开揉碎看看Informer是怎么通过三大核心技术——ProbSparse Self-attention、Self-attention Distilling和Generative Style Decoder——来实现这场效率革命的。我会结合我实际调参和使用的经验告诉你它们具体是怎么工作的代码里是怎么实现的以及我们使用时需要注意哪些“坑”。2. ProbSparse Self-attention抓住关键放弃“滥竽充数”的注意力自注意力机制计算量大的根源在于它默认每个查询Query都需要和所有键Key进行“亲密接触”。但你想过没有对于一个预测明天股价的模型来说昨天和前天的数据可能至关重要而三个月前某一天的数据影响力可能微乎其微。传统的自注意力却一视同仁为这些微弱甚至无关的联系付出了巨大的计算代价。Informer的作者提出了一个非常聪明的观点注意力权重矩阵就是那个Q和K点积再softmax后的矩阵通常是稀疏的。也就是说真正重要的注意力连接只占一小部分大部分都是可以忽略的“噪声”。这就好比在一场大型会议上真正有价值的讨论只发生在少数几个关键人物之间让所有人都互相交谈一遍无疑是低效的。那么如何快速找出那些“关键人物”重要的Query呢这就是ProbSparse Self-attention的核心。它的思路不是去精确计算每个注意力权重而是用一种近似的方法快速评估每个Query的“活跃度”或“重要性”然后只对最重要的那部分Query进行精细计算。2.1 如何衡量Query的“重要性”作者从概率的角度来看待注意力。对于第i个Query其与所有Key的注意力分布可以看作一个概率分布。如果一个Query的注意力分布非常“集中”比如只和少数几个Key高度相关那它就是重要的如果它的注意力分布非常“平坦”几乎和所有Key的关联度都差不多那它就是在“滥竽充数”贡献不大。具体怎么量化呢论文里用了一个叫做**“稀疏性度量”**的指标它衡量的是某个Query的注意力分布与均匀分布之间的差异。差异越大说明这个Query越“独特”越值得被关注。计算这个度量的公式本质上是在估计每个Query与所有Key的点积的最大值与其均值之间的差距。这个差距越大稀疏性得分就越高。在实际操作中我们不可能为了评估所有Query而真的去计算它们和所有Key的点积那又回到原点了。Informer采用了一种随机采样的近似方法从所有的Key中随机抽取一小部分比如25个作为样本。对于每一个Query只计算它与这25个采样Key的点积。基于这25个点积值来近似计算该Query的稀疏性度量得分。这个过程计算量很小因为它只涉及所有Query和一小部分Key的交互。2.2 筛选与计算只做“有用功”当我们为所有Query估算出稀疏性得分后就可以进行筛选了。我们只保留得分最高的前u个Queryu是一个远小于序列长度L的数比如u c * ln(L)其中c是常数。这些被选中的“精英”Query才会参与后续完整的注意力计算——即与所有的Key进行点积运算得到精确的注意力权重。对于那些被淘汰的Query它们的注意力输出怎么处理呢总不能直接置零。Informer采用了一个简单的策略用所有Value向量的平均值来填充。这样既保证了输出的完整性又极大地减少了计算量。我实测下来的体验是这个操作的效果非常显著。在序列长度为96的一个实验里我只选取了大约25个最重要的Query模型效果和原来用全部96个Query相差无几但注意力层的计算量却大幅下降。代码实现上关键就在于高效地完成随机采样和Top-k筛选。下面是一个简化版的核心逻辑帮助你理解import torch import torch.nn as nn import torch.nn.functional as F class ProbSparseAttention(nn.Module): def __init__(self, mask_flagTrue, factor5, scaleNone, attention_dropout0.1): super(ProbSparseAttention, self).__init__() self.factor factor # 控制采样数量的因子 self.scale scale self.mask_flag mask_flag self.dropout nn.Dropout(attention_dropout) def _prob_QK(self, Q, K, sample_k, n_top): Q: [batch_size, n_heads, len_q, d_k] K: [batch_size, n_heads, len_k, d_k] 返回筛选后的Q_K矩阵以及被选中的Query的索引 B, H, L_K, E K.shape _, _, L_Q, _ Q.shape # 1. 为每个Query随机采样sample_k个Key # 先扩展K的维度方便索引 K_expand K.unsqueeze(-3).expand(B, H, L_Q, L_K, E) # 随机生成采样索引 [L_Q, sample_k] index_sample torch.randint(L_K, (L_Q, sample_k)) # 取出采样的K K_sample K_expand[:, :, torch.arange(L_Q).unsqueeze(1), index_sample, :] # 2. 计算每个Q与采样K的点积并估算其稀疏性得分M # Q_K_sample: [B, H, L_Q, sample_k] Q_K_sample torch.matmul(Q.unsqueeze(-2), K_sample.transpose(-2, -1)).squeeze(-2) # 计算得分M: max(QK_sample) - mean(QK_sample) M Q_K_sample.max(-1)[0] - Q_K_sample.sum(-1) / sample_k # 3. 根据得分M选出前n_top个最重要的Query M_top M.topk(n_top, sortedFalse)[1] # indices # 4. 使用筛选出的Query计算完整的QK Q_reduce Q[torch.arange(B)[:, None, None], torch.arange(H)[None, :, None], M_top, :] # [B, H, n_top, E] Q_K torch.matmul(Q_reduce, K.transpose(-2, -1)) # [B, H, n_top, L_K] return Q_K, M_top def forward(self, queries, keys, values): B, L_Q, H, D queries.shape _, L_K, _, _ keys.shape queries queries.view(B, L_Q, H, D).transpose(1, 2) keys keys.view(B, L_K, H, D).transpose(1, 2) values values.view(B, L_K, H, D).transpose(1, 2) # 计算需要采样的Key数量和需要保留的Query数量 U_part self.factor * int(torch.ceil(torch.log(torch.tensor(L_K, dtypetorch.float)))).item() u self.factor * int(torch.ceil(torch.log(torch.tensor(L_Q, dtypetorch.float)))).item() U_part U_part if U_part L_K else L_K u u if u L_Q else L_Q # 调用_prob_QK获得筛选后的注意力分数和索引 scores_top, index self._prob_QK(queries, keys, sample_kU_part, n_topu) # 处理注意力分数得到权重并与Value相乘 # ... (此处省略scale, mask, softmax等步骤) # 最终输出只更新了重要Query对应的位置其他位置用Value的均值填充 return output通过这种方式ProbSparse Self-attention成功将注意力计算的时间复杂度从O(L²)降低到了O(L log L)这是一个质的飞跃。它教会我们一个道理在处理复杂问题时识别并聚焦于最关键的信息往往比处理所有信息更有效。3. Self-attention Distilling给特征图“瘦身”提炼精华解决了单层注意力计算效率的问题我们再来看看堆叠多层网络带来的内存瓶颈。传统的Transformer Encoder由N个相同的层堆叠而成每一层都输出一个序列长度不变的特征图。当序列很长时这个特征图会占用大量内存并且层与层之间传递如此庞大的特征计算和通信开销也很大。Informer的应对策略是Self-attention Distilling我更喜欢叫它“注意力蒸馏”或“特征下采样”。它的思想非常直观既然经过ProbSparse Attention之后特征图中已经突出了最重要的信息那么我们就可以在进入下一层之前对特征图进行压缩保留精华丢弃冗余。3.1 如何“蒸馏”具体操作上Informer在相邻的Encoder层之间插入了一个一维卷积Conv1D和最大池化MaxPool的组合。这个操作就像是给特征图做了一次“降维打击”一维卷积作用在特征维度d_model上进行跨通道的信息融合。你可以把它理解为一个智能的过滤器学习如何组合不同特征通道的信息。最大池化作用在序列长度L维度上通常步长为2。这意味着它将序列长度直接减半。例如输入序列长度是96经过池化后就变成了48。为什么最大池化是有效的因为在前一层ProbSparse Attention的输出中不重要的位置已经被平滑处理用Value的均值填充其信息量较低。而重要的位置则具有较高的激活值。最大池化操作会自然地选取每个局部区域内响应最强的特征从而保留了那些关键信息抑制了噪声。这个过程可以用一个简单的公式表示第 (j1) 层的输入 MaxPool( ELU( Conv1d( 第 j 层的输出 ) ) )其中ELU是激活函数。通过这种操作每经过一个蒸馏层序列长度就减半特征维度通过Conv1d可以进行变换或保持。这样越往深层走特征图就越“精炼”占用的内存和计算量也逐层递减。3.2 分而治之的“堆叠蒸馏”策略在实际的Informer Encoder设计中作者采用了一个更鲁棒的策略称为堆叠蒸馏Stacked Distilling。它不是简单的一条路走到黑而是采用了类似“分治”的思想。看一下架构图你会发现Encoder的输入被复制了一份。主分支经过完整的N层每层后接蒸馏而另一个分支的输入序列长度只有主分支的一半相当于从主分支的中间开始它经过较少的层数比如N-1层。最终两个分支输出的特征图在序列长度和特征维度上是对齐的然后将它们拼接Concatenate起来作为Encoder的最终输出。这样设计的好处是什么呢我认为有两点多尺度特征融合主分支处理完整的原始序列保留了全局的、可能比较粗糙的长期模式。副分支处理下采样后的序列感受野相对更大可能捕捉到更宏观的趋势。两者的融合能让模型同时兼顾细节与趋势。增强模型鲁棒性相当于为模型提供了一个正则化路径防止信息在过度的下采样中丢失尤其对于周期性明显或包含短周期模式的序列副分支可能提供更好的表示。在我做风速预测的项目里原始数据是每小时一个点。我尝试过只用主分支和启用堆叠蒸馏。结果发现启用堆叠蒸馏后模型对于短期未来24小时的突变点捕捉能力略有下降因为信息被压缩了但对于中长期未来3-7天的整体趋势预测更加平滑和准确。这提示我们对于不同的预测任务重短期精度还是重长期趋势可以调整蒸馏的强度或分支结构。4. Generative Style Decoder告别“挤牙膏”实现一步到位的预测Transformer在序列生成任务如机器翻译中Decoder采用了一种自回归Autgressive的方式也就是我们常说的“step-by-step”或“挤牙膏”式生成先预测第一个token然后用这个预测结果作为输入的一部分再去预测第二个token如此循环直到生成结束标记。这种方式有两个致命缺点速度慢生成一个长度为T的序列需要串行地运行T次Decoder无法并行。误差累积前面的预测一旦出错错误会像滚雪球一样传递给后面的预测导致结果严重偏离。在长序列时间序列预测中我们往往需要一次性预测未来几十甚至几百个时间点。如果用自回归方式预测耗时将不可接受。Informer提出了Generative Style Decoder生成式解码器彻底摒弃了自回归实现了**“一步预测”**。4.1 解码器的输入“引导段”“目标占位符”生成式解码器的核心思想非常巧妙。它不再一个点一个点地猜而是把整个预测过程当作一个“填充”任务。解码器的输入由两部分拼接而成引导段Start Token Sequence这是一段真实的、已知的历史序列。比如我们要预测未来5天的数据我们可以把紧挨着预测起点之前的10天真实数据作为引导段。这段数据提供了模型开始生成所需的上下文和模式信息。目标占位符Target Placeholder这是一段全为0或一个特殊标记的序列其长度就等于我们想要预测的未来序列长度。它就像一个空白的画布告诉模型“请在这里画出未来5天的预测图”。将这两段拼接起来就构成了解码器的完整输入。例如引导段长度10预测长度5那么解码器输入的总长度就是15。4.2 掩码注意力与并行计算在解码器内部同样使用ProbSparse Self-attention。但这里有一个关键操作必须使用掩码Mask。掩码的作用是确保在计算自注意力时序列中的每个位置只能“看到”它自己以及它之前的位置包括引导段而不能“偷看”它之后的位置。这是为了防止信息泄露即防止模型利用未来的“占位符”信息来预测现在。由于整个输入序列引导段占位符是已知且固定的解码器内部的所有注意力计算都可以并行完成。模型在一次前向传播中就直接输出了整个未来序列的预测值。这带来的速度提升是惊人的。在我对比的实验中预测一个长度为96的未来序列传统自回归Decoder需要串行推理96步而Informer的生成式Decoder只需要1步推理速度提升了数十倍。4.3 丰富的时序嵌入为了让模型更好地理解时间Informer的解码器其实编码器也是使用了非常精细的时序位置嵌入这比原始Transformer的sin/cos位置编码要强大得多。它主要包括三种嵌入的加和标量投影Scalar Projection通过一维卷积将原始的单变量时间序列值映射到高维空间。局部时间戳Local Time Stamp类似Transformer的位置编码表示序列内的相对顺序。全局时间戳Global Time Stamp这是Informer的一大亮点。它将时间序列的绝对时间信息如年、月、日、时、星期几、是否是节假日也编码进来。这对于捕捉日周期、周周期、季节周期等固定模式至关重要。例如预测每日销售额时“星期六”和“星期一”的模式肯定不同预测每小时用电量时“凌晨3点”和“晚上8点”也截然不同。全局时间戳嵌入让模型能够“知道”这些日历信息从而做出更准确的预测。在实际代码中你需要为你的数据构建一个包含所有这些时间特征的DataFrame然后通过一个嵌入层将它们转换为向量。import torch.nn as nn import torch class TokenEmbedding(nn.Module): def __init__(self, c_in, d_model): super(TokenEmbedding, self).__init__() # 用Conv1d代替线性层可以更好地捕捉局部连续性 self.tokenConv nn.Conv1d(in_channelsc_in, out_channelsd_model, kernel_size3, padding1, padding_modecircular) def forward(self, x): # x: [Batch, Sequence Length, Channel] x self.tokenConv(x.permute(0, 2, 1)).transpose(1, 2) return x class TimeFeatureEmbedding(nn.Module): def __init__(self, d_model, freqh): super(TimeFeatureEmbedding, self).__init__() # freq: 时间频率如 h(小时), t(分钟), d(天)等 # 根据频率计算时间特征维度例如小时级数据可能有[month, day, weekday, hour]等 if freq h: self.num_features 4 # 假设我们用了4个时间特征 elif freq t: self.num_features 3 else: self.num_features 1 self.embed nn.Linear(self.num_features, d_model) def forward(self, x): # x: [Batch, Sequence Length, num_time_features] return self.embed(x) class DataEmbedding(nn.Module): def __init__(self, c_in, d_model, freqh, dropout0.1): super(DataEmbedding, self).__init__() self.value_embedding TokenEmbedding(c_inc_in, d_modeld_model) self.position_embedding nn.Embedding.from_pretrained(self._get_sinusoid_encoding_table(), freezeTrue) self.temporal_embedding TimeFeatureEmbedding(d_modeld_model, freqfreq) self.dropout nn.Dropout(pdropout) def forward(self, x, x_mark): # x: 序列值 [B, L, C] # x_mark: 时间特征 [B, L, num_time_features] x self.value_embedding(x) self.position_embedding(x) self.temporal_embedding(x_mark) return self.dropout(x)通过这种生成式解码器和强大的时序嵌入Informer实现了长序列预测的高精度与高效率的统一。它不再需要缓慢的迭代而是像画画一样先勾勒轮廓引导段然后一挥而就完成整幅作品预测序列。5. 实战用Informer预测电力负荷理论说了这么多不如动手跑一跑。这里我以一个经典的公开数据集——电力负荷数据集ETTh1为例带你快速过一遍使用Informer进行多变量长序列预测的流程。我们假设任务是利用过去一周168小时的电力负荷数据预测未来一天24小时的负荷。5.1 数据准备与预处理时间序列预测的第一步也是最重要的一步就是处理好数据。我们需要将原始的时间序列数据转换成模型能接受的“监督学习”格式即一个个历史窗口未来窗口的数据对。import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler def create_dataset(data, lookback, forecast_horizon): 将时间序列数据转换为监督学习格式。 data: 形状为 [总时间步数, 特征数] 的数组 lookback: 历史窗口长度例如 168 forecast_horizon: 预测窗口长度例如 24 X, Y [], [] for i in range(len(data) - lookback - forecast_horizon 1): X.append(data[i:(i lookback), :]) # 历史窗口 Y.append(data[(i lookback):(i lookback forecast_horizon), :]) # 未来窗口 return np.array(X), np.array(Y) # 1. 加载数据 df pd.read_csv(ETTh1.csv) # 假设数据包含‘HUFL’, ‘HULL’, ‘MUFL’, ‘MULL’, ‘LUFL’, ‘LULL’, ‘OT’ 7个特征负荷值 data df[[HUFL, HULL, MUFL, MULL, LUFL, LULL, OT]].values # 2. 划分训练集、验证集、测试集 (按时间顺序) train_ratio, val_ratio 0.7, 0.2 train_size int(len(data) * train_ratio) val_size int(len(data) * val_ratio) train_data data[:train_size] val_data data[train_size:train_sizeval_size] test_data data[train_sizeval_size:] # 3. 标准化 (非常重要) scaler StandardScaler() scaler.fit(train_data) # 只在训练集上拟合scaler train_scaled scaler.transform(train_data) val_scaled scaler.transform(val_data) test_scaled scaler.transform(test_data) # 4. 创建数据对 lookback 168 # 用过去168小时7天预测 horizon 24 # 预测未来24小时 X_train, Y_train create_dataset(train_scaled, lookback, horizon) X_val, Y_val create_dataset(val_scaled, lookback, horizon) X_test, Y_test create_dataset(test_scaled, lookback, horizon) # 5. 转换为PyTorch Tensor import torch X_train_t torch.FloatTensor(X_train).transpose(1, 2) # [B, C, L] Y_train_t torch.FloatTensor(Y_train).transpose(1, 2) # ... 同理处理验证集和测试集5.2 模型构建与关键参数解析接下来我们使用一个开源的Informer实现如informer-pytorch来构建模型。这里我们重点关注几个关键参数from models import Informer model Informer( enc_in7, # 编码器输入特征维度即我们的7个负荷特征 dec_in7, # 解码器输入特征维度同上 c_out7, # 输出特征维度预测7个负荷值 seq_lenlookback, # 输入序列长度历史窗口 label_lenlookback // 2, # 引导段长度通常设为历史窗口的一半 out_lenhorizon, # 预测序列长度未来窗口 factor5, # ProbSparse Attention的采样因子控制稀疏度 d_model512, # 模型隐藏层维度 n_heads8, # 注意力头数 e_layers2, # 编码器层数 d_layers1, # 解码器层数 d_ff2048, # 前馈网络维度 dropout0.1, attnprob, # 使用ProbSparse Attention embedtimeF, # 使用包含时间特征的嵌入 freqh, # 数据频率为小时 activationgelu ).to(device)参数调优经验分享label_len引导段长度这个参数对结果影响很大。太短了模型缺乏足够的启动信息太长了会挤占本应用于学习预测模式的计算资源。我的经验是一般设置为seq_len的1/4到1/2并通过验证集效果来确定。factor稀疏因子控制ProbSparse Attention中保留的Query比例。默认的5是一个不错的起点。如果你的序列周期性很强可以尝试调小如3让模型关注更少的“关键点”如果序列噪声大、模式复杂可以尝试调大如7。d_model和n_heads这是模型容量的关键。对于7维的负荷预测512维和8个头通常够用。如果数据维度更高或序列关系更复杂可以适当增大。e_layers和d_layersInformer的编码器通常不需要像原始Transformer那样堆很多层因为Self-attention Distilling已经让每层的信息很浓缩。2-3层编码器1-2层解码器是常见配置。5.3 训练、预测与结果反标准化训练过程就是标准的深度学习流程使用MSE或MAE作为损失函数。预测完成后切记要将结果反标准化转换回原始数据的量纲才能进行正确的评估和可视化。# 训练循环示例 (简化版) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lr1e-4) model.train() for epoch in range(100): for batch_x, batch_y in train_loader: optimizer.zero_grad() # batch_x: [B, C, L], batch_y: [B, C, S] (S是预测长度) # dec_inp 需要构造包含引导段和占位符 # 通常取batch_x的最后label_len个点作为引导段真实值前面补0作为占位符 outputs model(batch_x, batch_x_mark, dec_inp, dec_inp_mark) loss criterion(outputs, batch_y) loss.backward() optimizer.step() # 预测 model.eval() with torch.no_grad(): preds model(test_x, test_x_mark, test_dec_inp, test_dec_inp_mark) # preds: [B, C, S] # 反标准化 preds_np preds.cpu().numpy().transpose(0, 2, 1) # 变回 [B, S, C] preds_original scaler.inverse_transform(preds_np.reshape(-1, 7)).reshape(preds_np.shape[0], horizon, 7) # 对Y_test同样进行反标准化 Y_test_original scaler.inverse_transform(Y_test.reshape(-1, 7)).reshape(Y_test.shape[0], horizon, 7) # 计算评价指标如MAE, RMSE from sklearn.metrics import mean_absolute_error, mean_squared_error mae mean_absolute_error(Y_test_original.flatten(), preds_original.flatten()) rmse np.sqrt(mean_squared_error(Y_test_original.flatten(), preds_original.flatten())) print(fTest MAE: {mae:.4f}, Test RMSE: {rmse:.4f})踩过几次坑之后我深刻体会到对于Informer这类模型数据预处理和参数初始化比模型本身的结构更重要。确保时间序列是平稳的或做了差分处理处理好缺失值选择合适的历史窗口和预测窗口以及仔细调整label_len和factor这些往往比盲目增加模型层数更能提升效果。