2. 多头注意力机制实现
任务说明
本案例实现标准多头注意力模块(Multi-Head Attention),核心目标是将输入序列投影到多个注意力头,分别计算缩放点积注意力后拼接输出,并确保与PyTorch官方nn.MultiheadAttention实现一致。输入为形状(batch, seq_len, d_model)的张量,输出形状不变。
核心实现
import torch
import torch.nn as nn
import torch.nn.functional as F
import math
class MultiHeadAttention(nn.Module):
"""标准多头注意力实现(无mask,无bias)"""
def __init__(self, d_model, num_heads):
super().__init__()
assert d_model % num_heads == 0, "d_model必须能被num_heads整除"
self.d_model = d_model
self.num_heads = num_heads
self.head_dim = d_model // num_heads
# 定义投影矩阵(合并W_Q, W_K, W_V为一个参数块)
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)
# 缩放因子
self.scale = 1.0 / math.sqrt(self.head_dim)
def forward(self, x):
"""
x: (batch, seq_len, d_model)
返回: (batch, seq_len, d_model)
"""
batch, seq_len, _ = x.shape
# 1. 线性投影并拆分为多头
# Q, K, V: (batch, num_heads, seq_len, head_dim)
Q = self.W_Q(x).view(batch, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
K = self.W_K(x).view(batch, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
V = self.W_V(x).view(batch, seq_len, self.num_heads, self.head_dim).transpose(1, 2)
# 2. 缩放点积注意力
# scores: (batch, num_heads, seq_len, seq_len)
scores = torch.matmul(Q, K.transpose(-2, -1)) * self.scale
attn_weights = F.softmax(scores, dim=-1)
# 3. 加权求和
# context: (batch, num_heads, seq_len, head_dim)
context = torch.matmul(attn_weights, V)
# 4. 拼接多头并输出投影
# 先转回 (batch, seq_len, num_heads, head_dim) 再展平
context = context.transpose(1, 2).contiguous().view(batch, seq_len, self.d_model)
output = self.W_O(context)
return output
与标准实现一致性验证
为验证自定义实现与PyTorch官方一致,需在相同初始化条件下对比输出。由于官方nn.MultiheadAttention默认使用_reset_parameters初始化,我们采用固定随机种子并手动同步权重:
def test_consistency():
torch.manual_seed(42)
d_model, num_heads, batch, seq_len = 8, 2, 2, 5
# 固定输入
x = torch.randn(batch, seq_len, d_model)
# 自定义实现
custom = MultiHeadAttention(d_model, num_heads)
# 官方实现(batch_first=True,无bias)
official = nn.MultiheadAttention(d_model, num_heads, batch_first=True, bias=False)
# 手动同步权重(官方实现内部参数命名不同)
with torch.no_grad():
# 官方权重形状: in_proj_weight (3*d_model, d_model), out_proj.weight (d_model, d_model)
official.in_proj_weight[0:d_model] = custom.W_Q.weight.data
official.in_proj_weight[d_model:2*d_model] = custom.W_K.weight.data
official.in_proj_weight[2*d_model:3*d_model] = custom.W_V.weight.data
official.out_proj.weight.data = custom.W_O.weight.data
# 前向计算
custom_out = custom(x)
official_out, _ = official(x, x, x) # 自注意力
# 比较输出
diff = (custom_out - official_out).abs().max().item()
print(f"最大误差: {diff:.2e}")
assert diff < 1e-5, "自定义实现与官方实现不一致"
print("一致性验证通过")
if __name__ == "__main__":
test_consistency()
运行与验证
将上述两个代码块依次保存为mha.py,在Python 3.8+、PyTorch 1.8+环境中运行:
python mha.py
预期输出:
最大误差: 0.00e+00
一致性验证通过
判定标准:最大误差小于1e-5即视为与标准实现一致。若误差较大,检查权重同步是否正确或初始化是否对齐。
关键说明
- 本实现未包含mask和bias,与验证代码中官方配置(
bias=False,无mask)严格对应 - 缩放因子使用
1/√head_dim,其中head_dim = d_model / num_heads - 若需使用mask,可在
scores上添加-inf后再做softmax,但本案例验证范围不含mask