# Transformer模型从零实现详解 ## 1. Transformer架构概述 Transformer模型最初在论文《Attention Is All You Need》中提出，彻底改变了自然语言处理领域。与传统RNN和LSTM不同，Transformer完全基于注意力机制，能够并行处理序列数据，有效捕捉长距离依赖关系[ref_1]。 ### 1.1 核心创新点 | 传统模型局限 | Transformer优势 | |-------------|----------------| | 顺序处理导致训练慢 | 并行计算大幅提升效率 | | 长序列信息丢失 | 自注意力机制捕捉全局依赖 | | 梯度消失/爆炸问题 | 残差连接和层归一化稳定训练 | ## 2. 核心组件代码实现 ### 2.1 位置编码(Positional Encoding) 由于Transformer不包含循环和卷积结构，需要位置编码来注入序列的位置信息。 ```python import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super(PositionalEncoding, self).__init__() # 创建位置编码矩阵 pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) 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).transpose(0, 1) self.register_buffer('pe', pe) def forward(self, x): # 将位置编码加到输入嵌入上 return x + self.pe[:x.size(0), :] ``` 位置编码使用不同频率的正弦和余弦函数，使得模型能够学习到相对位置信息[ref_2]。 ### 2.2 缩放点积注意力(Scaled Dot-Product Attention) ```python import torch.nn.functional as F def scaled_dot_product_attention(query, key, value, mask=None): """ 缩放点积注意力实现 query: [batch_size, seq_len, d_model] key: [batch_size, seq_len, d_model] value: [batch_size, seq_len, d_model] mask: 可选掩码 """ d_k = query.size(-1) # 计算注意力分数 scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k) # 应用掩码（解码器中使用） if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) # Softmax得到注意力权重 attention_weights = F.softmax(scores, dim=-1) # 加权求和 output = torch.matmul(attention_weights, value) return output, attention_weights ``` 缩放因子$\frac{1}{\sqrt{d_k}}$防止点积过大导致softmax梯度消失[ref_3]。 ### 2.3 多头注意力机制(Multi-Head Attention) ```python class MultiHeadAttention(nn.Module): def __init__(self, d_model, num_heads): super(MultiHeadAttention, self).__init__() assert d_model % num_heads == 0 self.d_model = d_model self.num_heads = num_heads self.d_k = d_model // num_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, query, key, value, mask=None): batch_size = query.size(0) # 线性投影 Q = self.w_q(query).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2) K = self.w_k(key).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2) V = self.w_v(value).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2) # 计算缩放点积注意力 scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k) if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) attention_weights = F.softmax(scores, dim=-1) # 应用注意力权重 context = torch.matmul(attention_weights, V) # 拼接多头输出 context = context.transpose(1, 2).contiguous().view( batch_size, -1, self.d_model) # 最终线性变换 output = self.w_o(context) return output, attention_weights ``` 多头注意力允许模型在不同表示子空间中共同关注来自不同位置的信息[ref_4]。 ## 3. 编码器层实现 ### 3.1 前馈神经网络 ```python class PositionwiseFeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout=0.1): super(PositionwiseFeedForward, self).__init__() self.linear1 = nn.Linear(d_model, d_ff) self.linear2 = nn.Linear(d_ff, d_model) self.dropout = nn.Dropout(dropout) self.relu = nn.ReLU() def forward(self, x): # 两层前馈网络 return self.linear2(self.dropout(self.relu(self.linear1(x)))) ``` ### 3.2 编码器层完整实现 ```python class EncoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout=0.1): super(EncoderLayer, self).__init__() self.self_attention = MultiHeadAttention(d_model, num_heads) self.feed_forward = PositionwiseFeedForward(d_model, d_ff, dropout) # 层归一化 self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) # Dropout 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 + self.dropout(ff_output)) return x ``` 残差连接和层归一化帮助梯度流动，使深层网络训练更加稳定[ref_5]。 ## 4. 完整Transformer模型 ### 4.1 编码器堆叠 ```python class TransformerEncoder(nn.Module): def __init__(self, vocab_size, d_model, num_heads, num_layers, d_ff, max_len, dropout=0.1): super(TransformerEncoder, self).__init__() # 词嵌入层 self.embedding = nn.Embedding(vocab_size, d_model) # 位置编码 self.positional_encoding = PositionalEncoding(d_model, max_len) # 编码器层堆叠 self.layers = nn.ModuleList([ EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) self.dropout = nn.Dropout(dropout) def forward(self, src, src_mask=None): # 词嵌入 + 位置编码 x = self.embedding(src) x = self.positional_encoding(x) x = self.dropout(x) # 通过所有编码器层 for layer in self.layers: x = layer(x, src_mask) return x ``` ### 4.2 文本分类应用示例 ```python class TransformerClassifier(nn.Module): def __init__(self, vocab_size, d_model, num_heads, num_layers, d_ff, max_len, num_classes, dropout=0.1): super(TransformerClassifier, self).__init__() self.encoder = TransformerEncoder(vocab_size, d_model, num_heads, num_layers, d_ff, max_len, dropout) self.classifier = nn.Linear(d_model, num_classes) self.dropout = nn.Dropout(dropout) def forward(self, src, src_mask=None): # 编码器输出 encoded = self.encoder(src, src_mask) # 取第一个token的输出用于分类（类似BERT的[CLS]） pooled = encoded[:, 0, :] pooled = self.dropout(pooled) # 分类层 logits = self.classifier(pooled) return logits ``` ## 5. 训练流程示例 ### 5.1 数据准备 ```python from torch.utils.data import Dataset, DataLoader class TextDataset(Dataset): def __init__(self, texts, labels, vocab, max_len): self.texts = texts self.labels = labels self.vocab = vocab self.max_len = max_len def __len__(self): return len(self.texts) def __getitem__(self, idx): # 文本转token id序列 text = self.texts[idx] tokens = text.split()[:self.max_len] token_ids = [self.vocab.get(token, 1) for token in tokens] # 1为UNK # 填充到固定长度 if len(token_ids) < self.max_len: token_ids += [0] * (self.max_len - len(token_ids)) return torch.tensor(token_ids), torch.tensor(self.labels[idx]) ``` ### 5.2 训练循环 ```python def train_model(model, train_loader, val_loader, num_epochs, learning_rate): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model.to(device) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate) for epoch in range(num_epochs): model.train() total_loss = 0 for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(device), target.to(device) optimizer.zero_grad() output = model(data) loss = criterion(output, target) loss.backward() optimizer.step() total_loss += loss.item() # 验证阶段 model.eval() val_acc = 0 with torch.no_grad(): for data, target in val_loader: data, target = data.to(device), target.to(device) output = model(data) pred = output.argmax(dim=1) val_acc += (pred == target).sum().item() print(f'Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.4f}, ' f'Val Acc: {val_acc/len(val_loader.dataset):.4f}') ``` ## 6. 实战应用场景 ### 6.1 情感分析任务 在IMDB电影评论情感分析任务中，可以使用上述Transformer模型架构[ref_5]。通过构建词表、预处理文本数据，然后训练分类器来预测评论的情感极性（正面/负面）。 ### 6.2 超参数设置建议 | 参数 | 推荐值 | 说明 | |------|--------|------| | d_model | 512 | 模型维度 | | num_heads | 8 | 注意力头数 | | num_layers | 6 | 编码器层数 | | d_ff | 2048 | 前馈网络维度 | | dropout | 0.1 | 防止过拟合 | | 学习率 | 0.0001 | Adam优化器 | 通过这种从零开始的手搓实现，可以深入理解Transformer模型的每个组件及其相互作用，为后续更复杂的NLP任务打下坚实基础[ref_6]。