Transformer 时序数据建模与实现详解

Transformer 时序架构设计详解

整体架构概览

TTS-Transformer 采用编码器 - 解码器的设计思路，主要由以下几个核心组件构成：

• 输入嵌入层（Input Embedding）：将时序数据转换为高维特征表示
• 位置编码层（Positional Encoding）：为序列添加位置信息
• 多层 Transformer 编码器（Multi-layer Transformer Encoder）：通过自注意力机制提取时序特征
• 输出层（Output Layer）：根据任务需求进行分类或回归预测

这种模块化设计不仅提高了代码的可维护性，还使得网络结构具有良好的灵活性和可扩展性。

核心组件详细分析

多头自注意力机制（Multi-Head Self-Attention）

多头自注意力机制是 Transformer 的核心创新，负责捕获序列中不同位置之间的依赖关系。这里我们定义一个 PyTorch 模块来实现它：

import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1):
        super(MultiHeadAttention, self).__init__()
        assert d_model % n_heads == 0
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        
        # 线性变换层
        self.W_q = nn.Linear(d_model, d_model, bias=False)
        self.W_k = nn.Linear(d_model, d_model, bias=False)
        self.W_v = nn.Linear(d_model, d_model, bias=False)
        self.W_o = nn.Linear(d_model, d_model)
        self.dropout = nn.Dropout(dropout)
        self.scale = math.sqrt(self.d_k)

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)
        seq_len = query.size(1)
        
        # 线性变换并重塑为多头形式
        Q = self.W_q(query).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_k(key).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_v(value).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        
        # 计算注意力分数 scores = torch.matmul(Q, K.transpose(-2, -1)) / self.scale
        # [batch, heads, seq_len, seq_len]
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        
        # 计算注意力权重
        attention_weights = F.softmax(scores, dim=-1)
        attention_weights = self.dropout(attention_weights)
        
        # 应用注意力权重
        context = torch.matmul(attention_weights, V)
        # [batch, heads, seq_len, d_k]
        
        # 合并多头结果
        context = context.transpose(1, 2).contiguous().view(
            batch_size, seq_len, self.d_model)
        
        # 最终线性变换
        output = self.W_o(context)
        return output, attention_weights

多头自注意力机制的设计体现了以下核心思想：

• 多头并行处理：通过多个注意力头并行计算，捕获不同类型的时序关系
• 缩放点积注意力：使用缩放因子√d_k 避免 softmax 函数进入饱和区域
• 线性变换组合：通过 Query、Key、Value 的线性变换实现特征空间的灵活映射

位置编码（Positional Encoding）

由于自注意力机制本身无法感知序列的顺序信息，位置编码的引入至关重要。我们利用正弦和余弦函数生成唯一的编码：

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_seq_length=5000, dropout=0.1):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(dropout)
        
        # 创建位置编码矩阵
        pe = torch.zeros(max_seq_length, d_model)
        position = torch.arange(0, max_seq_length).unsqueeze(1).float()
        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)
        
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        # x.shape: [batch_size, seq_length, d_model]
        seq_length = x.size(1)
        x = x + self.pe[:, :seq_length]
        return self.dropout(x)

位置编码的设计原理：

• 正弦余弦函数：利用不同频率的正弦余弦函数为每个位置生成唯一的编码
• 相对位置感知：通过数学性质使模型能够学习相对位置关系
• 长度适应性：能够处理训练时未见过长度的序列

Transformer 编码器块（Transformer Encoder Block）

编码器块整合了注意力机制和前馈网络，并加入残差连接与归一化：

class TransformerEncoderBlock(nn.Module):
    def __init__(self, d_model, n_heads, d_ff, dropout=0.1):
        super(TransformerEncoderBlock, self).__init__()
        
        # 多头自注意力层
        self.self_attention = MultiHeadAttention(d_model, n_heads, dropout)
        
        # 前馈神经网络
        self.feed_forward = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model),
            nn.Dropout(dropout)
        )
        
        # 层归一化
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # 多头自注意力 + 残差连接 + 层归一化
        attn_output, attention_weights = self.self_attention(x, x, x, mask)
        x = self.norm1(x + self.dropout(attn_output))
        
        # 前馈网络 + 残差连接 + 层归一化
        ff_output = self.feed_forward(x)
        x = self.norm2(x + ff_output)
        return x, attention_weights

编码器块的关键设计元素：

• 残差连接：解决深层网络的梯度消失问题，促进信息流动
• 层归一化：稳定训练过程，加速收敛
• 前馈网络：增加模型的非线性表达能力

完整的时序 Transformer 网络架构

我们将上述组件组合成一个完整的时序分类模型：

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_dim, d_model, n_heads, n_layers, d_ff, max_seq_length, num_classes, dropout=0.1):
        super(TimeSeriesTransformer, self).__init__()
        self.d_model = d_model
        
        # 输入嵌入层
        self.input_embedding = nn.Linear(input_dim, d_model)
        
        # 位置编码
        self.positional_encoding = PositionalEncoding(d_model, max_seq_length, dropout)
        
        # Transformer 编码器层
        self.transformer_blocks = nn.ModuleList([
            TransformerEncoderBlock(d_model, n_heads, d_ff, dropout)
            for _ in range(n_layers)
        ])
        
        # 全局平均池化
        self.global_avg_pool = nn.AdaptiveAvgPool1d(1)
        
        # 分类头
        self.classifier = nn.Sequential(
            nn.Linear(d_model, d_model // 2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_model // 2, num_classes)
        )
        
        # 参数初始化
        self._init_parameters()

    def _init_parameters(self):
        for module in self.modules():
            if isinstance(module, nn.Linear):
                nn.init.xavier_uniform_(module.weight)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)

    def create_padding_mask(self, x, pad_token=0):
        """创建填充掩码"""
        # 假设 pad_token 用于标识填充位置
        mask = (x != pad_token).unsqueeze(1).unsqueeze(2)
        return mask

    def forward(self, x, mask=None):
        """
        Args:
            x: [batch_size, seq_length, input_dim]
            mask: [batch_size, 1, 1, seq_length] 可选的掩码
        Returns:
            output: [batch_size, num_classes]
            attention_weights: 各层的注意力权重
        """
        batch_size, seq_length, input_dim = x.shape
        
        # 输入嵌入
        x = self.input_embedding(x)
        # [batch_size, seq_length, d_model]
        x = x * math.sqrt(self.d_model)
        
        # 位置编码
        x = self.positional_encoding(x)
        
        # 存储注意力权重
        attention_weights = []
        
        # 通过 Transformer 编码器层
        for transformer_block in self.transformer_blocks:
            x, attn_weights = transformer_block(x, mask)
            attention_weights.append(attn_weights)
        
        # 全局平均池化：[batch_size, seq_length, d_model] -> [batch_size, d_model]
        x = x.transpose(1, 2)
        # [batch_size, d_model, seq_length]
        x = self.global_avg_pool(x).squeeze(-1)
        
        # 分类预测
        output = self.classifier(x)
        # [batch_size, num_classes]
        return output, attention_weights

模型配置与超参数设置

实际部署时，合理的超参数配置至关重要。以下是一个典型的配置示例：

# 模型配置示例
config = {
    'input_dim': 6,      # 输入特征维度（如传感器数据的 6 个维度）
    'd_model': 256,      # 模型隐藏维度
    'n_heads': 8,        # 多头注意力的头数
    'n_layers': 6,       # Transformer 层数
    'd_ff': 1024,        # 前馈网络隐藏维度
    'max_seq_length': 512, # 最大序列长度
    'num_classes': 6,    # 分类类别数
    'dropout': 0.1       # Dropout 概率
}

# 实例化模型
model = TimeSeriesTransformer(**config)

# 打印模型信息
print(f"模型参数量：{sum(p.numel() for p in model.parameters() if p.requires_grad):,}")

技术细节与实现要点

自注意力机制的计算复杂度优化

标准自注意力机制的计算复杂度为 O(n²d)，其中 n 为序列长度，d 为特征维度。对于长序列，这会导致显著的计算和内存开销。我们可以尝试降维策略来优化：

class EfficientAttention(nn.Module):
    """优化版本的注意力机制，适用于长序列"""
    def __init__(self, d_model, n_heads, dropout=0.1, max_seq_length=5000):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        
        # 使用更小的 key 维度进行近似
        self.reduced_dim = min(64, self.d_k)
        self.W_q = nn.Linear(d_model, n_heads * self.reduced_dim, bias=False)
        self.W_k = nn.Linear(d_model, n_heads * self.reduced_dim, bias=False)
        self.W_v = nn.Linear(d_model, d_model, bias=False)
        self.W_o = nn.Linear(d_model, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, query, key, value, mask=None):
        B, L, D = query.shape
        
        # 降维处理 Q 和 K
        Q = self.W_q(query).view(B, L, self.n_heads, self.reduced_dim).transpose(1, 2)
        K = self.W_k(key).view(B, L, self.n_heads, self.reduced_dim).transpose(1, 2)
        V = self.W_v(value).view(B, L, self.n_heads, self.d_k).transpose(1, 2)
        
        # 计算注意力（降维后的复杂度更低）
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.reduced_dim)
        if mask is not None:
            scores.masked_fill_(mask == 0, -1e9)
        
        attn = F.softmax(scores, dim=-1)
        attn = self.dropout(attn)
        
        context = torch.matmul(attn, V).transpose(1, 2).contiguous().view(B, L, D)
        output = self.W_o(context)
        return output, attn

位置编码的改进策略

针对时序数据的特点，可以采用更加灵活的位置编码策略，例如可学习的位置编码或相对位置编码：

class LearnablePositionalEncoding(nn.Module):
    """可学习的位置编码"""
    def __init__(self, d_model, max_seq_length=5000, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(dropout)
        
        # 使用可学习的参数代替固定的正弦余弦编码
        self.pe = nn.Parameter(torch.randn(1, max_seq_length, d_model) * 0.1)

    def forward(self, x):
        seq_len = x.size(1)
        x = x + self.pe[:, :seq_len]
        return self.dropout(x)

class RelativePositionalEncoding(nn.Module):
    """相对位置编码，更适合时序数据"""
    def __init__(self, d_model, max_relative_position=128):
        super().__init__()
        self.d_model = d_model
        self.max_relative_position = max_relative_position
        
        # 相对位置嵌入
        vocab_size = max_relative_position * 2 + 1
        self.relative_position_embeddings = nn.Embedding(vocab_size, d_model)

    def forward(self, length):
        """生成相对位置编码矩阵"""
        range_vec = torch.arange(length)
        distance_mat = range_vec[None, :] - range_vec[:, None]
        distance_mat_clipped = torch.clamp(
            distance_mat, -self.max_relative_position, self.max_relative_position
        )
        final_mat = distance_mat_clipped + self.max_relative_position
        embeddings = self.relative_position_embeddings(final_mat)
        return embeddings

Transformer 与传统方法的性能对比

Transformer 架构在时序数据处理领域取得了显著突破，主要得益于其自注意力机制所带来的长距离依赖建模能力，能够直接捕捉序列中任意位置之间的关系，有效克服了传统 RNN 在处理远程依赖时的局限。同时，Transformer 摒弃了递归结构，实现了高度并行化的计算，大幅提升了模型的训练与推理效率。在多个时序数据集上展现出的优异性能也证明了其出色的泛化能力。此外，通过注意力权重的可视化，Transformer 具备良好的可解释性，有助于深入理解模型的决策逻辑。