【模型】Informer模型详解

Informer 是一种针对长时间序列预测任务设计的深度学习模型，特别适用于解决序列数据的高效建模与预测问题。Informer 提出了许多创新的机制，尤其是在计算效率方面，能够显著提高长时间序列预测的准确性和速度。以下是对该模型的详细介绍。

1. 模型架构

Informer的核心思想是通过结合自注意力机制和稀疏化的操作来提高效率，并使用长短期记忆网络（LSTM）结合注意力机制进行建模。它主要由以下几部分组成：

1. 输入编码层：对输入的时间序列数据进行预处理。通常，数据会经过标准化、嵌入等步骤，并转换为适合自注意力机制处理的形式。
2. Encoder-Decoder结构：Informer采用类似于Transformer的Encoder-Decoder架构。
   • Encoder：由多个自注意力层（Self-Attention）和前馈神经网络（Feed-Forward Networks）堆叠组成。对于每个输入序列，使用了全局和局部自注意力机制的组合。
   • Decoder：与传统的Transformer类似，解码器通过对未来时间步的预测，结合编码器的输出生成最终结果。
3. 稀疏化注意力（Sparse Attention）：
   • ProbSparse Attention：Informer引入了一种新的注意力机制——ProbSparse Attention，其通过只关注时间序列中的重要时间点（而非所有时间点）来减少计算复杂度。通过对输入序列中的关键部分进行采样和选择，它在保证准确性的同时，大幅提高了计算效率。
4. 细节：
   • 自注意力计算过程：传统的自注意力机制计算复杂度为 O(N2)O(N^2)，其中 NN 为序列长度。Informer通过采用稀疏注意力的方法，使得复杂度降到 O(Nlog⁡N)O(N\log{N})。
   • 多层叠加：Encoder部分通常由多个自注意力层和前馈神经网络层叠加，Decoder则接收Encoder的输出，预测未来时间步的目标序列。

2. 算法使用PyTorch实现

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import DataLoader, TensorDataset
from sklearn.metrics import mean_squared_error
import matplotlib.pyplot as plt
import numpy as np

# 定义Informer模型类
class AttentionLayer(nn.Module):
    def __init__(self, embed_size, heads, dropout=0.1):
        super(AttentionLayer, self).__init__()
        self.attn = nn.MultiheadAttention(embed_size, heads, dropout=dropout)
    
    def forward(self, query, key, value):
        # query, key, value的shape是 (batch_size, seq_len, embed_size)
        output, _ = self.attn(query, key, value)
        return output

class InformerEncoderLayer(nn.Module):
    def __init__(self, embed_size, heads, ff_hid_dim, dropout=0.1):
        super(InformerEncoderLayer, self).__init__()
        self.attn = AttentionLayer(embed_size, heads, dropout)
        self.ffn = nn.Sequential(
            nn.Linear(embed_size, ff_hid_dim),
            nn.ReLU(),
            nn.Linear(ff_hid_dim, embed_size)
        )
        self.layernorm1 = nn.LayerNorm(embed_size)
        self.layernorm2 = nn.LayerNorm(embed_size)
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, x):
        attn_out = self.attn(x, x, x)
        x = self.layernorm1(x + self.dropout(attn_out))  # Add & Norm
        
        ffn_out = self.ffn(x)
        x = self.layernorm2(x + self.dropout(ffn_out))  # Add & Norm
        
        return x

class Informer(nn.Module):
    def __init__(self,input_size , embed_size, hidden_size, num_layers, num_heads, forecast_horizon, ff_hid_dim=2048, dropout=0.1):
        super(Informer, self).__init__()
        
        # 1. 输入嵌入层
        self.input_size = input_size
        self.embed_size = embed_size
        self.fc_input = nn.Linear(input_size, embed_size)

        # 2. 定义编码器层
        self.encoder_layers = nn.ModuleList([
            InformerEncoderLayer(embed_size, num_heads, ff_hid_dim, dropout) for _ in range(num_layers)
        ])
        
        # 3. 输出层
        self.fc_output = nn.Linear(embed_size, forecast_horizon)

    def forward(self, x):
        # x 的形状是 (batch_size, seq_len, input_size)
        x = self.fc_input(x)  # 将输入数据嵌入到较高维度

        # 逐层编码
        for encoder_layer in self.encoder_layers:
            x = encoder_layer(x)
        
        # 只取最后时间步的输出
        out = x[:, -1, :]  # 获取最后一个时间步的编码表示
        
        # 输出层
        out = self.fc_output(out)
        
        return out

# 设置设备，使用GPU（如果可用）
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f'Using device: {device}')

# 假设你已经准备好了数据，X_train, X_test, y_train, y_test等
# 将数据转换为torch tensors，并转移到设备（GPU/CPU）
X_train_tensor = torch.Tensor(X_train).to(device)
X_test_tensor = torch.Tensor(X_test).to(device)
y_train_tensor = torch.Tensor(y_train).squeeze(-1).to(device)  # 确保y_train是正确形状
y_test_tensor = torch.Tensor(y_test).squeeze(-1).to(device)

# 创建训练数据和测试数据集
train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
test_dataset = TensorDataset(X_test_tensor, y_test_tensor)

# 定义 DataLoader
batch_size = 512
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# 初始化Informer模型
input_size = X_train.shape[2]  # 特征数量
embed_size = 64  # 嵌入维度
hidden_size = 64  # 隐藏层神经元数量
num_layers = 2  # 编码器层数
num_heads = 4  # 自注意力的头数
forecast_horizon = 5  # 预测的目标步长

model = Informer(input_size, embed_size, hidden_size, num_layers, num_heads, forecast_horizon).to(device)

# 定义训练函数
def train_model_with_dataloader(model, train_loader, test_loader, epochs=50, lr=0.001):
    criterion = nn.MSELoss()
    optimizer = optim.Adam(model.parameters(), lr=lr)
    
    train_loss = []
    val_loss = []
    
    for epoch in range(epochs):
        model.train()
        epoch_train_loss = 0
        for X_batch, y_batch in train_loader:
            optimizer.zero_grad()
            output = model(X_batch)
            loss = criterion(output, y_batch)
            loss.backward()
            optimizer.step()
            epoch_train_loss += loss.item()
        
        train_loss.append(epoch_train_loss / len(train_loader))
        
        model.eval()
        epoch_val_loss = 0
        with torch.no_grad():
            for X_batch, y_batch in test_loader:
                output = model(X_batch)
                loss = criterion(output, y_batch)
                epoch_val_loss += loss.item()
        
        val_loss.append(epoch_val_loss / len(test_loader))
        
        if (epoch + 1) % 10 == 0:  # 每10个epoch输出一次结果
            print(f'Epoch [{epoch+1}/{epochs}], Train Loss: {train_loss[-1]:.4f}, Validation Loss: {val_loss[-1]:.4f}')
    
    plt.plot(train_loss, label='Train Loss')
    plt.plot(val_loss, label='Validation Loss')
    plt.title('Loss vs Epochs')
    plt.xlabel('Epochs')
    plt.ylabel('Loss')
    plt.legend()
    plt.show()

# 训练模型
train_model_with_dataloader(model, train_loader, test_loader, epochs=100)

# 评估模型
def evaluate_model_with_dataloader(model, test_loader):
    model.eval()
    y_pred_list = []
    y_test_list = []
    
    with torch.no_grad():
        for X_batch, y_batch in test_loader:
            y_pred = model(X_batch)
            y_pred_list.append(y_pred.cpu().numpy())
            y_test_list.append(y_batch.cpu().numpy())
    
    y_pred_rescaled = np.concatenate(y_pred_list, axis=0)
    y_test_rescaled = np.concatenate(y_test_list, axis=0)
    
    mse = mean_squared_error(y_test_rescaled, y_pred_rescaled)
    print(f'Mean Squared Error: {mse:.4f}')
    return y_pred_rescaled, y_test_rescaled

# 评估模型性能
y_pred_rescaled, y_test_rescaled = evaluate_model_with_dataloader(model, test_loader)

# 保存模型
def save_model(model, path='./model_files/multisteps_informer_model.pth'):
    torch.save(model.state_dict(), path)
    print(f'Model saved to {path}')

save_model(model)

2.1代码解析

（1）Informer和Transformer代码实现的不同之处

总结：

• Informer的设计强调通过稀疏注意力机制降低计算和内存开销，特别适合处理长时间序列的预测任务。其通过Top-k Attention和稀疏化的方式，有效捕捉序列中的长时间依赖，同时避免了标准Transformer中计算复杂度的急剧增长。
• Transformer采用全连接的自注意力机制，能够处理全局依赖，但在长序列任务中，由于计算复杂度为 O(N2)O(N^2)，在处理长序列时会面临计算和内存瓶颈。

因此，Informer相较于Transformer在长时间序列预测任务中更具优势，能够高效地捕捉长时间依赖并降低计算负担，而Transformer更适合处理传统的序列到序列任务，尤其是在序列长度相对较短的情况下。

3. 模型的优缺点

优点：

• 高效的计算：通过使用稀疏注意力，Informer能有效减少计算复杂度，尤其适合长时间序列数据的建模。
• 准确性：由于加入了对重要时间点的重点关注，Informer能够更准确地捕捉序列中的重要模式。
• 扩展性：由于基于Transformer架构，Informer能够非常容易地与其他类型的模型（例如LSTM、GRU等）结合，进一步提升预测性能。

缺点：

• 对长序列的内存要求：尽管计算复杂度减少，但仍然需要较大的内存来存储模型的权重和计算梯度，可能不适用于非常长的序列。
• 对超参数敏感：Informer对于网络架构、学习率等超参数较为敏感，需要细致调优。
• 训练时间较长：尽管计算上有优化，但由于模型结构复杂，训练时间仍然较长。

4. 模型算法变种

1. Informer-XL：一种变种，它通过引入长距离依赖建模方法，能够捕获跨时间步长的长期依赖性，提升了模型在超长序列上的表现。
2. Time Series Transformer (TST)：类似于Informer，但在处理时间序列时对时间步长进行特殊处理，优化了在时间序列任务中的效果。

5. 模型特点

• 稀疏化自注意力：Informer通过稀疏化自注意力机制降低了计算开销，尤其适用于长时间序列数据。
• 效率提升：在长时间序列数据上的训练效率较高，能够处理传统Transformer处理起来较慢的问题。
• 长短期依赖建模：能够捕捉长时间的依赖关系，适用于许多时间序列预测任务。

6. 应用场景

• 金融时间序列预测：例如股票价格预测、汇率预测等，长时间序列的预测问题非常适合使用Informer。
• 气象预测：天气预报、气候变化预测等，需要高效捕捉长时段的气候数据。
• 交通流量预测：通过交通流量的历史数据来预测未来的交通状况。
• 能源消耗预测：根据历史能源使用情况预测未来的电力需求或能源消耗。
• 健康医疗数据预测：使用长时间的健康监测数据预测疾病的发生或患者的康复过程。