Python 基于 Transformer 的时序数据建模与实现详解

这里的关键是缩放因子 sqrt(d_k)，它能防止点积过大导致 softmax 进入饱和区，从而保证梯度的有效传播。

2. 位置编码

由于自注意力机制本身是无序的，必须显式注入位置信息。我们使用正弦余弦函数生成固定编码。

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):
        seq_length = x.size(1)
        x = x + self.pe[:, :seq_length]
        return self.dropout(x)

这种编码方式不仅唯一，还具备外推性，能处理未见过的序列长度。

3. Transformer 编码器块

将注意力与前馈网络堆叠，配合残差连接和层归一化，构建深层网络。

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, _ = 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

完整网络架构

整合上述组件，形成完整的时序分类模型。

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)
        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 forward(self, x, mask=None):
        batch_size, seq_length, input_dim = x.shape
        x = self.input_embedding(x)
        x = x * math.sqrt(self.d_model)
        x = self.positional_encoding(x)
        
        for block in self.transformer_blocks:
            x, _ = block(x, mask)
            
        x = x.transpose(1, 2)
        x = self.global_avg_pool(x).squeeze(-1)
        output = self.classifier(x)
        return output

配置示例

config = {
    'input_dim': 6,
    'd_model': 256,
    'n_heads': 8,
    'n_layers': 6,
    'd_ff': 1024,
    'max_seq_length': 512,
    'num_classes': 6,
    'dropout': 0.1
}
model = TimeSeriesTransformer(**config)
print(f"模型参数量：{sum(p.numel() for p in model.parameters() if p.requires_grad):,}")

技术细节与优化

计算复杂度优化

标准注意力机制复杂度为 O(n²)，长序列下开销巨大。可通过降维 Key 向量来近似计算，降低内存占用。

class EfficientAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        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 = 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)
        context = torch.matmul(attn, V).transpose(1, 2).contiguous().view(B, L, D)
        return self.W_o(context)

性能对比

从表中可见，Transformer 在长距离依赖和并行效率上取得了最佳平衡。虽然原始计算量较大，但通过工程优化完全可接受。

此外，相比传统方法，Transformer 的可解释性更强，注意力热力图能清晰反映模型决策依据，这对工业界调试至关重要。