【NLP 78、手搓Transformer模型结构】
你以为走不出的淤泥，也迟早会云淡风轻
—— 25.5.31
引言 ——《Attention is all you need》
《Attention is all you need》这篇论文可以说是自然语言处理领域的一座里程碑，它提出的 Transformer 结构带来了一场技术革命。
研究背景与目标
在 Transformer 出现之前，自然语言处理领域主要依赖循环神经网络（RNN）及其变体，如长短期记忆网络（LSTM）和门控循环单元（GRU），还有卷积神经网络（CNN）来处理序列数据。然而，RNN 存在顺序计算的限制，难以并行化，而且随着序列长度增加，信息容易丢失，LSTM 等虽有所改进，但对于长期依赖问题仍不够理想；CNN 在处理长序列数据时，不同位置信号关联的操作数会随距离增长，学习远距离依赖较困难。因此，论文旨在寻找一种新的模型架构，既能有效处理长序列数据中的长期依赖关系，又能实现高度并行化，提高训练效率。
核心创新点
- 抛弃传统架构，引入注意力机制：Transformer 完全基于注意力机制，摒弃了 RNN 和 CNN。注意力机制就像一个聪明的 “筛选器”，能让模型自动关注输入序列中不同位置的重要信息，无论它们距离多远，比如在处理 “The dog chased the cat” 这样的句子时，能理解 “chased” 和 “dog”“cat” 的关系，而不受距离影响。
- 自注意力机制（self - attention）：这是 Transformer 的核心组件。每个单词会生成 Query、Key 和 Value 三个向量，通过计算 Query 与所有 Key 的相似度得分，对 Value 进行加权求和，得到该单词的上下文相关表示。例如，在分析 “它是一只可爱的猫，它喜欢玩耍” 时，能确定两个 “它” 都指代 “猫”。
- 多头注意力机制（Multi - Head Attention）：相当于多个不同的自注意力机制集成，每个头关注数据的不同方面，能捕捉更丰富的语义信息。可以想象成多个不同的 “小眼睛”，从不同角度观察数据，然后把结果综合起来。
- 编码器 - 解码器架构（Encoder - Decoder）：编码器由多个堆叠的自注意力层和前馈神经网络组成，负责将输入序列编码成一个固定长度的向量表示；解码器除了自注意力层和前馈神经网络外，还有一个编码器 - 解码器注意力层，用于在解码时关注编码器的输出。就像一个 “翻译官”，编码器把源语言 “理解” 成一种中间表示，解码器再根据这个中间表示生成目标语言。
- 位置编码（Positional Encoding）：由于 Transformer 本身没有内置的顺序信息，位置编码被引入来给单词添加位置信息，使模型能区分不同位置的单词，比如 “我爱你” 和 “你爱我”，通过位置编码能让模型理解其中的差异。
二、手搓Transformer模型
1.导入相应包
torch：导入Pytorch包，PyTorch 是一个开源的深度学习框架，提供了张量操作和自动微分功能操作和自动微分功能，是代码的核心依赖。
- 核心功能：
- 定义和操作多维张量（如矩阵、向量），支持 CPU 和 GPU 加速。
- 构建神经网络模型，管理模型参数和计算图。
- 提供优化器（如 SGD、Adam）和损失函数（如交叉熵）。
- 在代码中的使用场景：
- 定义模型层（如nn.Linear、nn.Embedding）。
- 实现张量运算（如矩阵乘法torch.matmul）。
- 执行前向传播和反向传播。
torch.nn：Pytorch神经网络模块，nn 是 PyTorch 中构建神经网络的高层接口，封装了常用的层（Layer）和模块（Module）。
- 核心功能：
- 定义神经网络层，如全连接层（nn.Linear）、卷积层（nn.Conv2d）、归一化层（nn.LayerNorm）等。
- 提供损失函数（如nn.CrossEntropyLoss）。
- 支持自定义神经网络模块（通过继承nn.Module）。
- 在代码中的使用场景：
- 定义 Transformer 的各个组件，如MultiHeadAttention、EncoderLayer等（均继承自nn.Module）。
- 使用nn.Embedding实现词嵌入层，nn.LayerNorm实现层归一化。
torch.nn.functional：PyTorch 神经网络函数（torch.nn.functional），包含了神经网络中常用的函数式接口（非类接口），通常用于定义层的操作（如激活函数、池化等）。
- 核心功能：
- 激活函数：如 ReLU（F.relu）、Softmax（F.softmax）。
- 张量操作：如 dropout（F.dropout）、归一化（F.layer_norm）。
- 损失函数：如均方误差（F.mse_loss）。
- 在代码中的使用场景：
- 在scaled_dot_product_attention方法中使用F.softmax计算注意力权重。
- 在位置前馈网络（PositionwiseFeedForward）中使用F.relu作为激活函数。
math：Python 标准库math，提供数学运算函数，用于实现一些数值计算逻辑。
- 核心功能：
- 基础数学函数：如平方根（math.sqrt）、指数（math.exp）、对数（math.log）。
- 常数：如圆周率math.pi。
- 在代码中的使用场景：
- 在位置编码（PositionalEncoding）中计算频率因子div_term时，使用math.log计算对数。
- 在缩放点积注意力中，使用math.sqrt计算缩放因子（1/sqrt(d_k)）。
import torch import torch.nn as nn import torch.nn.functional as F import math
2.位置编码层实现
Ⅰ、运行流程
① 类继承与父类初始化 ——> ② 创建位置编码矩阵 pe ——> ③ 生成位置索引 position
④ 生成频率因子 div_term ——> ⑤ 填充正弦和余弦值 ——> ⑥ 添加批次维度
⑦ 注册为缓冲区 ——> ⑦ 输入张量 x ——> ⑧ 获取位置编码 ——> ⑨ 输出结果
输入: d_model, max_len ↓ ┌───────────────────────────────────────────┐ │ 初始化位置编码矩阵和位置索引 │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 计算频率因子 (div_term) │ │ div_term = exp(-log(10000) * 2i/d_model) │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 生成位置编码矩阵 │ │ pe[:, 0::2] = sin(position * div_term) │ │ pe[:, 1::2] = cos(position * div_term) │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 添加批次维度并注册缓冲区 │ │ pe = pe.unsqueeze(0) │ │ self.register_buffer('pe', pe) │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 前向传播方法 │ │ x = x + self.pe[:, :x.size(1)] │ └───────────────────────────────────────────┘ ↓ 输出: x + 位置编码Ⅱ、初始化
d_model：模型维度（即词嵌入维度），决定位置编码的列数（每个位置的特征维度）。
max_len：模型支持的最大序列长度，决定位置编码的行数（最多编码max_len个位置）。
pe：创建一个形状为 (max_len, d_model) 的全零张量，用于存储位置编码。初始化位置编码矩阵，后续通过正弦 / 余弦函数填充具体值。
torch.zeros()：创建指定形状的全零张量。
参数名 类型 描述 size int 或 tuple 张量形状，如 (3, 5) 表示 3 行 5 列的矩阵。 out=None Tensor 输出张量（可选），结果将写入此张量。 dtype=None torch.dtype 数据类型（如 torch.float32），默认与当前张量库一致。 layout=torch.strided torch.layout 内存布局（如 torch.sparse_coo 表示稀疏张量）。 device=None torch.device 设备（如 'cuda' 或 'cpu'）。 requires_grad=False bool 是否需要梯度（用于自动微分）。 position：为每个位置生成唯一的索引值，用于计算位置编码的频率和相位。
- torch.arange(0, max_len)：生成从 0 到max_len-1的一维张量，表示序列中的位置索引（如第 0 个位置，第 1 个位置，…）。
- .unsqueeze(1)：将一维张量转换为形状 (max_len, 1) 的二维张量（添加列维度）。
div_term：通过控制不同频率的正弦曲线，为不同位置生成唯一且有区分度的编码向量，从而让模型能够学习到序列中的相对位置和绝对位置信息。
- torch.arange(0, d_model, 2)：生成偶数索引序列 [0, 2, 4, ..., d_model-2]（若d_model为偶数），对应位置编码的偶数维度。
- math.log(10000.0)：以 10 为底的对数，用于计算频率的缩放因子。
- 最终结果是形状为 (d_model//2,) 的张量，每个元素代表对应维度的频率参数。
torch.arange()：生成指定范围的连续整数张量（类似 Python 的 range，但返回张量）。
参数名 类型 描述 start int 或 float 起始值（包含），默认 0。 end int 或 float 结束值（不包含）。 step=1 int 或 float 步长（间隔），默认 1。 out=None Tensor 输出张量（可选）。 dtype=None torch.dtype 数据类型，默认与当前张量库一致。 layout=torch.strided torch.layout 内存布局。 device=None torch.device 设备。 requires_grad=False bool 是否需要梯度。 torch.exp()：对输入张量的每个元素计算自然指数 e^x。
参数名 类型 描述 input Tensor 输入张量。 out=None Tensor 输出张量（可选）。 math.log()：计算自然对数 \(\ln(x)\) 或指定底数的对数（Python 标准库函数，非 PyTorch）。
参数名 类型 描述 x float 输入值（必须大于 0）。 base=math.e float 对数底数（可选），默认 e（自然对数）。若指定 base=10，则计算常用对数 log_{10}(x)。 torch.sin()：对输入张量中的每个元素计算正弦值（按元素操作），结果为弧度制的正弦值。
参数名 类型 说明 input Tensor 输入张量，数据类型需为浮点型（如 float32、float64）。 out Tensor, 可选 输出张量，用于存储计算结果，形状需与 input 一致。 torch.cos()：对输入张量中的每个元素计算余弦值（按元素操作），结果为弧度制的余弦值。
参数名 类型 说明 input Tensor 输入张量，数据类型需为浮点型（如 float32、float64）。 out Tensor, 可选 输出张量，用于存储计算结果，形状需与 input 一致。 unsqueeze()：在指定维度上为张量添加一个大小为 1 的维度（即增加维度数），常用于调整张量形状以适配模型输入或广播机制。
参数名 类型 说明 dim int 要插入维度的位置（索引从 0 开始）。 若为负数，表示从最后一维向前数（如 -1 表示最后一维）。
self.register_buffer()： 在 PyTorch 模型中注册一个缓冲区（Buffer），用于存储非训练参数（如位置编码、均值 / 方差统计量等）。特点：
- 不参与梯度更新；
- 随模型一起保存和加载；
- 当模型调用 .to(device) 时，缓冲区会自动移动到对应设备（如 GPU）。
# 使用正弦和余弦函数生成不同频率的编码，使模型能够学习序列中的相对位置关系。 class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() # 创建位置编码矩阵 (max_len, d_model) pe = torch.zeros(max_len, d_model) # 生成位置索引 (max_len, 1) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) # 生成频率因子 (d_model//2,) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) # 偶数位置使用正弦函数： # 【:】：选取所有行 【0：：2】：每行中取偶数索引的列（从0索引处出发，每隔两个列） pe[:, 0::2] = torch.sin(position * div_term) # 奇数位置使用余弦函数： # 【:】：选取所有行 【1：：2】：每行中取奇数索引的列（从1索引处出发，每隔两个列） pe[:, 1::2] = torch.cos(position * div_term) # 使位置编码能够适配批次输入并作为模型的静态参数使用。 # 添加批次维度 (1, max_len, d_model)：适配 PyTorch 模型处理批量数据的要求。 pe = pe.unsqueeze(0) # 注册为缓冲区（非训练参数）：位置编码是预计算的固定值，不需要在训练过程中更新，因此不应该作为模型参数（如 nn.Parameter）。 # 设备一致性：将 pe 注册为缓冲区后，当模型调用 .to(device) 时，pe 会自动跟随模型移动到指定设备（如 GPU），确保与输入数据的设备一致性。 # 模型保存与加载：缓冲区会随模型一起保存和加载，确保推理时位置编码可用。 self.register_buffer('pe', pe)Ⅲ、前向计算
x：输入序列的嵌入表示：通常是词向量（Word Embedding）或其他特征的张量。[batch_size, seq_len, d_model]
self.pe[:, : x.size(1)]：预计算的位置编码矩阵：通过正弦 / 余弦函数生成，用于为模型提供序列的位置信息。[1, max_len, d_model]
def forward(self, x): # 截取与输入序列等长的位置编码 x = x + self.pe[:, :x.size(1)] return xⅣ、完整代码
# 使用正弦和余弦函数生成不同频率的编码，使模型能够学习序列中的相对位置关系。 class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() # 创建位置编码矩阵 (max_len, d_model) pe = torch.zeros(max_len, d_model) # 生成位置索引 (max_len, 1) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) # 生成频率因子 (d_model//2,) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) # 偶数位置使用正弦函数： # 【:】：选取所有行 【0：：2】：每行中取偶数索引的列（从0索引处出发，每隔两个列） pe[:, 0::2] = torch.sin(position * div_term) # 奇数位置使用余弦函数： # 【:】：选取所有行 【1：：2】：每行中取奇数索引的列（从1索引处出发，每隔两个列） pe[:, 1::2] = torch.cos(position * div_term) # 使位置编码能够适配批次输入并作为模型的静态参数使用。 # 添加批次维度 (1, max_len, d_model)：适配 PyTorch 模型处理批量数据的要求。 pe = pe.unsqueeze(0) # 注册为缓冲区（非训练参数）：位置编码是预计算的固定值，不需要在训练过程中更新，因此不应该作为模型参数（如 nn.Parameter）。 # 设备一致性：将 pe 注册为缓冲区后，当模型调用 .to(device) 时，pe 会自动跟随模型移动到指定设备（如 GPU），确保与输入数据的设备一致性。 # 模型保存与加载：缓冲区会随模型一起保存和加载，确保推理时位置编码可用。 self.register_buffer('pe', pe) def forward(self, x): # 截取与输入序列等长的位置编码 x = x + self.pe[:, :x.size(1)] return x3.多头注意力机制实现
Ⅰ、运行流程
① 初始化 ——> ② 线性投影变换 + 多头分割 ——> ③ 计算缩放点积注意力 ——>
④ 合并多头 + 最终投影
输入: Q, K, V, mask(可选) ↓ ┌───────────────────────────────────────────┐ │ 初始化参数 │ │ d_k = d_model // num_heads │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 线性变换 │ │ Q = W_q(Q), K = W_k(K), V = W_v(V) │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 多头分割 │ │ split_heads: [batch, seq_len, d_model] │ │ → [batch, num_heads, seq_len, d_k] │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 缩放点积注意力计算 │ │ 1. scores = Q·K^T / √d_k │ │ 2. 应用掩码 (masked_fill) │ │ 3. attention = softmax(scores) │ │ 4. output = attention·V │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 合并多头 │ │ combine_heads: [batch, num_heads, seq_len, d_k] │ │ → [batch, seq_len, d_model] │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 最终线性变换 │ │ output = W_o(output) │ └───────────────────────────────────────────┘ ↓ 输出: 注意力输出
Ⅱ、初始化
d_model：模型的总维度（embedding 维度），表示输入和输出的特征向量长度。
num_heads：注意力头的数量，将整个注意力机制分成多个 “头” 并行计算。
super().__init__()：调用父类的初始化方法，避免重复编写代码。
assert：在代码中检查某个条件是否为真。如果条件为假，assert 会抛出 AssertionError 异常，帮助开发者快速定位问题。
- condition：必须为真的表达式。
- error_message（可选）：断言失败时显示的自定义提示信息。
- 作用：
- 参数校验：确保d_model可被num_heads整除，保证每个头的维度d_k为整数（如d_model=512, num_heads=8时，d_k=64）。
- 线性层定义：创建 4 个线性层用于投影输入到 Q/K/V 空间及输出空间，权重可学习。
self.W_q：将输入的查询（Query）投影到 d_model 维度。
self.W_k：将输入的键（Key）分别投影到 d_model 维度。
self.W_v：将输入的值（Value）分别投影到 d_model 维度。
self.W_o：将多个注意力头的输出合并后，再投影回 d_model 维度。
nn.Linear()：实现全连接层（线性变换），公式为 \(y = xA^T + b\)，用于将输入特征映射到输出特征。
参数 描述 in_features 输入特征维度（必需）。 out_features 输出特征维度（必需）。 bias 是否添加偏置项（默认 True）。 class MultiHeadAttention(nn.Module): 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.d_k = d_model // num_heads # 定义线性变换层：Q, K, V投影矩阵和输出投影矩阵 self.W_q = nn.Linear(d_model, d_model) # Q投影 self.W_k = nn.Linear(d_model, d_model) # K投影 self.W_v = nn.Linear(d_model, d_model) # V投影 self.W_o = nn.Linear(d_model, d_model) # 输出投影Ⅲ、分割头
- 作用：
- 维度重塑：将输入张量（形状为[batch, seq_len, d_model]）拆分为多个头，重塑为[batch, seq_len, num_heads, d_k]。
- 维度转置：通过transpose(1, 2)将头维度提前，变为[batch, num_heads, seq_len, d_k]，便于并行计算每个头的注意力。
- 例子：
- batch_size = 2（两个样本）
- seq_len = 4（每个样本有 4 个 token）
- d_model = 8（每个 token 用 8 维向量表示）
- num_heads = 4（4 个头）
- d_k = 2（每个头的维度是 2）
- 那么输入 x 的形状是 [2, 4, 8]，经过view后变成 [2, 4, 4, 2]，再经过transpose后形状变为 [2, 4, 4, 2]。最终输出的形状是 [batch_size, num_heads, seq_len, d_k]，这正是多头注意力机制后续计算所需的格式。
x：输入张量，通常为编码器或解码器的特征向量。
batch_size：一次训练或推理中处理的样本数量。
seq_len：序列长度，即输入序列中 Token 的数量（如句子中的单词数）。
d_model：模型维度，是输入和输出的特征维度。
size()：返回张量的形状
参数 描述 dim 可选参数，指定要返回的维度大小（若指定，则返回标量；否则返回元组）。 view()：重塑张量的形状，返回的新张量与原张量共享内存。
参数 描述 input 输入张量（必需）。 *shape 新形状（必需，元组或多个整数参数）。 transpose()：交换张量的两个维度。
参数 描述 input 输入张量（必需）。 dim0 第一个要交换的维度索引（必需）。 dim1 第二个要交换的维度索引（必需）。 def split_heads(self, x): # 将输入分割成多个头 batch_size, seq_len, d_model = x.size() return x.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)Ⅳ、缩放点积注意力
- 作用：
- 相似度计算：通过矩阵乘法Q·K^T计算查询与键的相似度，除以√d_k防止梯度消失（缩放点积注意力的核心）。
- 掩码处理：若存在掩码（如填充掩码或前瞻掩码），将无效位置的分数设为-1e9，经 softmax 后权重趋近于 0。
- 加权聚合：用注意力权重对值向量V加权求和，得到头的输出。
Q：查询矩阵，形状通常为 [batch_size, num_heads, seq_len_q, d_k]。
K：键矩阵，形状通常为 [batch_size, num_heads, seq_len_k, d_k]。
V：值矩阵，形状通常为 [batch_size, num_heads, seq_len_k, d_k]。
mask：可选的布尔掩码，用于屏蔽某些位置的注意力计算（如填充位置或未来 token）。
scores：计算 Q 与 K 的转置的矩阵乘法，除以缩放因子，防止梯度消失（当d_k较大时，点积结果可能方差较大），得到注意力分数矩阵。
torch.matmul()：张量矩阵乘法，支持多种维度组合（如向量点积、矩阵乘、批量矩阵乘）
参数 描述 input 第一个输入张量（必需）。 other 第二个输入张量（必需）。 out 输出张量（可选）。 transpose()：交换张量的两个维度。
参数 描述 input 输入张量（必需）。 dim0 第一个要交换的维度索引（必需）。 dim1 第二个要交换的维度索引（必需）。 math.sqrt()：计算数值的平方根
参数 描述 x 输入数值（必需，需为正数）。 masked_fill()：根据掩码（mask）对张量元素进行替换。
参数 描述 input 输入张量（必需）。 mask 布尔掩码张量（必需，与input形状相同）。 value 替换值（必需）。 attention：将注意力分数归一化为概率分布（值在 [0,1] 且和为 1）。
softmax()：将输入张量在指定维度上归一化为概率分布（值在 [0,1] 且和为 1）。
参数 描述 input 输入张量（必需）。 dim 要计算 softmax 的维度（必需）。 dtype 输出数据类型（可选）。 output：根据注意力权重对 V 进行加权求和。
torch.matmul()：张量矩阵乘法，支持多种维度组合（如向量点积、矩阵乘、批量矩阵乘）。
def scaled_dot_product_attention(self, Q, K, V, mask=None): # 计算注意力分数，缩放以稳定梯度 scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k) # 应用掩码：将掩码位置的分数设为负无穷，softmax后趋近于0 if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) # 应用softmax获取注意力权重 attention = F.softmax(scores, dim=-1) # 加权求和得到输出 output = torch.matmul(attention, V) return outputⅤ、合并头
x：输入张量，通常为编码器或解码器的特征向量。
batch_size：一次训练或推理中处理的样本数量。
num_heads：注意力头的数量，用于将模型维度拆分为多个子空间进行并行计算。
seq_len：序列长度，即输入序列中 Token 的数量（如句子中的单词数）。
d_k：每个注意力头的维度，通过d_model除以num_heads得到。
self.d_model：模型维度，是输入和输出的特征维度。
size()：返回张量的形状
参数 描述 dim 可选参数，指定要返回的维度大小（若指定，则返回标量；否则返回元组）。 transpose()：交换张量的两个维度。
参数 描述 input 输入张量（必需）。 dim0 第一个要交换的维度索引（必需）。 dim1 第二个要交换的维度索引（必需）。 contiguous()：重新排列张量的内存布局，使其在内存中连续存储（某些操作要求输入必须连续）。
参数 描述 input 输入张量（必需）。 view()：重塑张量的形状，返回的新张量与原张量共享内存。
参数 描述 input 输入张量（必需）。 *shape 新形状（必需，元组或多个整数参数）。 def combine_heads(self, x): # 将多个头的输出合并 batch_size, num_heads, seq_len, d_k = x.size() return x.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model)Ⅵ、前向传播
Q：查询输入，形状通常为 [batch_size, nums_head, seq_len_q, d_model]。
K：键输入，形状通常为 [batch_size, nums_head, seq_len_k, d_model]。
V：值输入，形状通常为 [batch_size, nums_head, seq_len_k, d_model]。
mask：可选的布尔掩码，用于屏蔽某些位置的注意力计算（如填充或未来 token）。
self.split_heads()：将张量分割为多个头，实现多头并行计算。
attn_output：缩放点积注意力（Scaled Dot-Product Attention）的输出，形状为 (batch_size, num_heads, seq_len, d_k)。
self.scaled_dot_product_attention()：实现缩放点积注意力的核心计算逻辑。
- 计算注意力分数：通过 Q·K^T 计算 Token 间的相关性，除以 √d_k 防止梯度消失。
- 应用掩码：通过 mask 屏蔽无效位置（如填充或未来 Token）。
- Softmax 归一化：将分数转化为概率分布（注意力权重）。
- 加权求和：用注意力权重对 Value 向量加权，得到注意力输出。
output：多头注意力的最终输出，形状为 (batch_size, seq_len, d_model)。
self.W_o()：多头注意力的输出投影层，是一个线性变换矩阵。
self.combine_heads()：将多头分离的特征（(batch_size, num_heads, seq_len, d_k)）合并为单张量。
def forward(self, Q, K, V, mask=None): # 线性变换 + 多头分割 Q = self.split_heads(self.W_q(Q)) K = self.split_heads(self.W_k(K)) V = self.split_heads(self.W_v(V)) # 计算注意力 attn_output = self.scaled_dot_product_attention(Q, K, V, mask) # 合并头并应用最终线性变换 output = self.W_o(self.combine_heads(attn_output)) return outputⅦ、完整代码
class MultiHeadAttention(nn.Module): 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.d_k = d_model // num_heads # 定义线性变换层：Q, K, 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) def scaled_dot_product_attention(self, Q, K, V, mask=None): # 计算注意力分数，缩放以稳定梯度 scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k) # 应用掩码：将掩码位置的分数设为负无穷，softmax后趋近于0 if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) # 应用softmax获取注意力权重 attention = F.softmax(scores, dim=-1) # 加权求和得到输出 output = torch.matmul(attention, V) return output def split_heads(self, x): # 将输入分割成多个头 batch_size, seq_len, d_model = x.size() return x.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2) def combine_heads(self, x): # 将多个头的输出合并 batch_size, num_heads, seq_len, d_k = x.size() return x.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model) def forward(self, Q, K, V, mask=None): # 线性变换 + 多头分割 Q = self.split_heads(self.W_q(Q)) K = self.split_heads(self.W_k(K)) V = self.split_heads(self.W_v(V)) # 计算注意力 attn_output = self.scaled_dot_product_attention(Q, K, V, mask) # 合并头并应用最终线性变换 output = self.W_o(self.combine_heads(attn_output)) return output4.编码层
Ⅰ、运行流程
① 初始化阶段 ——> ② 自注意力计算 ——> ③ 残差连接 ——> ④ Dropout与层归一化
——> ⑤ 前馈网络计算 ——> ⑥ 残差链接 ——> ⑦ Dropout与层归一化
输入 x [batch, seq_len, d_model] ↓ ┌───────────────────────────────────┐ │ 步骤1：自注意力 + 残差 + 层归一化 │ └───────────────────────────────────┘ ↓ 1. 自注意力计算： x → MultiHeadAttention(x, x, x, mask) → attn_output ↓ 2. 残差连接： x + attn_output → residual_1 ↓ 3. Dropout与层归一化： LayerNorm(x + Dropout(residual_1)) → x_1 ↓ ┌───────────────────────────────────┐ │ 步骤2：前馈网络 + 残差 + 层归一化 │ └───────────────────────────────────┘ ↓ 1. 前馈网络计算： x_1 → PositionwiseFeedForward(x_1) → ff_output ↓ 2. 残差连接： x_1 + ff_output → residual_2 ↓ 3. Dropout与层归一化： LayerNorm(x_1 + Dropout(residual_2)) → 输出 x
Ⅱ、初始化
d_model：模型维度。
num_heads：注意力头数量。
d_ff：前馈网络隐藏层维度。
dropout：Dropout 概率，用于防止过拟合。
d_model：模型的总维度（embedding 维度），表示输入和输出的特征向量长度。
num_heads：注意力头的数量，将整个注意力机制分成多个 “头” 并行计算。
super().__init__()：调用父类的初始化方法，避免重复编写代码。
MultiHeadAttention()：将输入特征分解到多个子空间（头），并行计算注意力，最后合并结果。
PositionwiseFeedForward()：对每个 Token 的特征进行逐位置的前馈变换。
nn.LayerNorm()：对输入的特征维度进行归一化，使数据在训练过程中保持稳定分布，加速收敛并提高模型泛化能力。
其中，μ 和 σ 是当前样本的均值和方差，γ 和 β 是可学习的缩放和平移参数。
参数 描述 normalized_shape 输入特征的形状（如整数或元组），指定需要归一化的维度。例如：512 或 (512,)。 eps 数值稳定性的小常数（默认 1e-5），防止分母为零。 elementwise_affine 是否学习可训练的仿射参数 \(\gamma\) 和 \(\beta\)（默认 True）。 nn.Dropout()：在训练过程中随机将输入的部分元素置为 0，减少神经元对特定特征的依赖，从而防止过拟合。
参数 描述 p 元素被置零的概率（默认 0.5），取值范围 [0, 1]。 inplace 是否直接在原张量上操作（默认 False）。 # 编码层 class EncoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 自注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.dropout = nn.Dropout(dropout) # Dropout层Ⅲ、前向传播
x：编码层的输入和经过每层操作后的中间输出，形状为(batch_size, src_len, d_model) 。
mask：源序列的填充掩码，形状为(batch_size, 1, 1, src_len) ，用于屏蔽填充位置。
attn_output：自注意力模块的输出。
self.self_attn()：实现自注意力机制（Self-Attention），计算输入序列中各 Token 之间的依赖关系。
self.norm1()：对自注意力的输出进行层归一化（Layer Normalization）。
self.dropout()：在自注意力输出后应用随机失活（Dropout），按概率丢弃部分神经元激活值。
ff_output：前馈网络模块的输出。
self.feed_forward()：实现位置前馈网络（Positionwise Feed-Forward Network），对每个 Token 的特征进行非线性变换。
self.norm2()：对前馈网络的输出进行层归一化。
self.dropout()：在自注意力输出后应用随机失活（Dropout），按概率丢弃部分神经元激活值。
def forward(self, x, mask): # 自注意力 + 残差连接 + 层归一化 attn_output = self.self_attn(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Ⅳ、完整代码
# 编码层 class EncoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 自注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, x, mask): # 自注意力 + 残差连接 + 层归一化 attn_output = self.self_attn(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 x5.解码层
Ⅰ、运行流程
① 初始化 ——> ② 掩蔽自注意力计算 ——> ③ 残差链接与层归一化 ——> ④ 跨注意力计算 ——> ⑤ 残差链接与层归一化 ——> ⑥ 前馈网络计算 ——> ⑦ 残差链接与层归一化
输入： x [batch, tgt_len, d_model] ← 解码器输入（目标序列） encoder_output [batch, src_len, d_model] ← 编码器输出（源序列） src_mask [batch, 1, src_len] ← 源序列填充掩码 tgt_mask [batch, tgt_len, tgt_len] ← 目标序列因果掩码（三角矩阵） ↓ ┌───────────────────────────────────────────┐ │ 步骤1：掩蔽自注意力 + 残差 + 层归一化 │ └───────────────────────────────────────────┘ ↓ 1. 掩蔽自注意力： x → MultiHeadAttention(x, x, x, tgt_mask) → attn_output1 ↓ 2. 残差与归一化： x + Dropout(attn_output1) → LayerNorm → x_1 ↓ ┌───────────────────────────────────────────┐ │ 步骤2：编码器-解码器注意力 + 残差 + 层归一化 │ └───────────────────────────────────────────┘ ↓ 1. 跨注意力： x_1 → MultiHeadAttention(x_1, encoder_output, encoder_output, src_mask) → attn_output2 ↓ 2. 残差与归一化： x_1 + Dropout(attn_output2) → LayerNorm → x_2 ↓ ┌───────────────────────────────────────────┐ │ 步骤3：前馈网络 + 残差 + 层归一化 │ └───────────────────────────────────────────┘ ↓ 1. 前馈网络： x_2 → PositionwiseFeedForward(x_2) → ff_output ↓ 2. 残差与归一化： x_2 + Dropout(ff_output) → LayerNorm → 输出 x
Ⅱ、初始化
d_model：模型维度
num_heads：注意力头数量。
d_ff：前馈网络隐藏层维度。
dropout：Dropout 概率，用于防止过拟合。
super().__init__()：调用父类的初始化方法，避免重复编写代码。
MultiHeadAttention()：将输入特征投影到多个子空间，并行计算多个注意力头，然后合并结果。
PositionwiseFeedForward()：捕获不同子空间的语义关系，增强模型对复杂模式的表达能力。
nn.LayerNorm()：对输入的特征维度进行归一化，使数据在训练过程中保持稳定分布，加速收敛并提高模型泛化能力。
其中，μ 和 σ 是当前样本的均值和方差，γ 和 β 是可学习的缩放和平移参数。
参数 描述 normalized_shape 输入特征的形状（如整数或元组），指定需要归一化的维度。例如：512 或 (512,)。 eps 数值稳定性的小常数（默认 1e-5），防止分母为零。 elementwise_affine 是否学习可训练的仿射参数 \(\gamma\) 和 \(\beta\)（默认 True）。 nn.Dropout()：在训练过程中随机将输入的部分元素置为 0，减少神经元对特定特征的依赖，从而防止过拟合。
参数 描述 p 元素被置零的概率（默认 0.5），取值范围 [0, 1]。 inplace 是否直接在原张量上操作（默认 False）。 class DecoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 掩蔽自注意力 self.cross_attn = MultiHeadAttention(d_model, num_heads) # 编码器-解码器注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.norm3 = nn.LayerNorm(d_model) # 第三层归一化 self.dropout = nn.Dropout(dropout) # Dropout层Ⅲ、前向传播
x：解码层的输入和经过每层操作后的中间输出，形状为(batch_size, tgt_len, d_model) 。
encoder_output：编码器的输出，形状为(batch_size, src_len, d_model) ，作为解码器中编码器 - 解码器注意力的输入。
src_mask：源序列的填充掩码，形状为(batch_size, 1, 1, src_len) 。
tgl_mask：目标序列的掩码，形状为(batch_size, 1, tgt_len, tgt_len) ，组合了填充掩码和三角掩码。
attn_output1：编码器的输出，形状为(batch_size, src_len, d_model) ，作为解码器中编码器 - 解码器注意力的输入。
self.self_attn()：实现掩码自注意力（Masked Self-Attention），处理目标序列内部的依赖关系。
self.norm1()：对自注意力的输出进行层归一化（Layer Normalization）。
self.dropout()：在自注意力后随机丢弃部分神经元输出，防止过拟合。
attn_output2：编码器 - 解码器注意力（Cross-Attention）的输出，融合目标序列与编码器输出的信息。
self.cross_attn()：实现编码器 - 解码器注意力，查询（Query）来自解码器，键（Key）和值（Value）来自编码器。
self.norm2()：对编码器 - 解码器注意力的输出进行层归一化。
self.dropout()：在交叉注意力后应用 Dropout，防止过拟合。
ff_output：位置前馈网络（Positionwise Feed-Forward Network）的输出，对特征进行非线性变换。
self.feed_forward()：实现两层线性变换（通常是ReLU激活），对每个位置的特征独立处理。
self.norm3()：对前馈网络的输出进行层归一化。
self.dropout()：在前馈网络后应用 Dropout。
def forward(self, x, encoder_output, src_mask, tgt_mask): # 掩蔽自注意力：只关注已生成的token，通过三角掩码防止解码器看到未来位置 attn_output1 = self.self_attn(x, x, x, tgt_mask) x = self.norm1(x + self.dropout(attn_output1)) # 编码器-解码器注意力：关注编码器输出，连接编码和解码过程。 attn_output2 = self.cross_attn(x, encoder_output, encoder_output, src_mask) x = self.norm2(x + self.dropout(attn_output2)) # 前馈网络 ff_output = self.feed_forward(x) x = self.norm3(x + self.dropout(ff_output)) return xⅣ、完整代码
# 解码层 class DecoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 掩蔽自注意力 self.cross_attn = MultiHeadAttention(d_model, num_heads) # 编码器-解码器注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.norm3 = nn.LayerNorm(d_model) # 第三层归一化 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, x, encoder_output, src_mask, tgt_mask): # 掩蔽自注意力：只关注已生成的token，通过三角掩码防止解码器看到未来位置 attn_output1 = self.self_attn(x, x, x, tgt_mask) x = self.norm1(x + self.dropout(attn_output1)) # 编码器-解码器注意力：关注编码器输出，连接编码和解码过程。 attn_output2 = self.cross_attn(x, encoder_output, encoder_output, src_mask) x = self.norm2(x + self.dropout(attn_output2)) # 前馈网络 ff_output = self.feed_forward(x) x = self.norm3(x + self.dropout(ff_output)) return x6.编码器
Ⅰ、运行流程
① 初始化 ——> ② 词嵌入 + 位置编码 ——> ③ 逐层通过编码器层 ——> ④ 输出结果
输入： src [batch, src_len] ← 输入序列token ID src_mask [batch, 1, src_len] ← 填充掩码（屏蔽填充token） ↓ ┌─────────────────────────────────────┐ │ 步骤1：词嵌入 + 位置编码 + Dropout │ └─────────────────────────────────────┘ ↓ 1. 词嵌入与缩放： src → Embedding → 向量 × √d_model → [batch, src_len, d_model] ↓ 2. 位置编码： 添加位置信息 → [batch, src_len, d_model] ↓ 3. Dropout： 随机丢弃部分元素 → src ↓ ┌─────────────────────────────────────┐ │ 步骤2：逐层通过编码器层 │ └─────────────────────────────────────┘ ↓ for 每层编码器层: ┌─────────────────────────────────┐ │ 1. 自注意力 + 残差 + 层归一化 │ └─────────────────────────────────┘ ↓ ┌─────────────────────────────────┐ │ 2. 前馈网络 + 残差 + 层归一化 │ └─────────────────────────────────┘ ↓ 输出： src [batch, src_len, d_model] ← 编码后的序列表示Ⅱ、初始化
num_layers：编码器层的数量。
d_model：模型维度。
num_heads：注意力头数量。
d_ff：前馈网络隐藏层维度。
input_vocab_size：输入序列的词汇表大小，用于初始化词嵌入层。
super().__init__()：调用父类的初始化方法，避免重复编写代码。
nn.Embedding()：将离散的 token ID 映射为连续的向量表示。
参数 类型 描述 num_embeddings int 词汇表大小。 embedding_dim int 嵌入向量维度。 padding_idx int (可选) 填充标记的索引。 PositionalEncoding()：实现位置编码的具体类，通过特定的算法（如基于三角函数的方法）生成位置编码向量，并添加到词嵌入向量上。
EncoderLayer()：Transformer 编码器的核心组成单元，负责对输入序列进行单层的特征变换。通过堆叠多个 EncoderLayer，编码器能够逐层提取更复杂的语义表示，捕捉序列中的长距离依赖关系。
nn.ModuleList()：存储子模块的列表，用于动态构建神经网络。
参数 类型 描述 modules iterable (可选) 初始化模块列表。 nn.Dropout()：在训练过程中随机将输入的部分元素置为 0，减少神经元对特定特征的依赖，从而防止过拟合。
参数 描述 p 元素被置零的概率（默认 0.5），取值范围 [0, 1]。 inplace 是否直接在原张量上操作（默认 False）。 # 编码器：处理输入序列，提取特征表示。 class Encoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, input_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(input_vocab_size, d_model) # 词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个编码器层 self.dropout = nn.Dropout(dropout) # Dropout层Ⅲ、前向传播
src：编码器的输入序列，形状为(batch_size, src_len) ，元素为 token ID。
src_mask：源序列的填充掩码，形状为(batch_size, 1, 1, src_len) 。
self.embedding()：词嵌入层，将输入的 token ID 映射为d_model维度的向量。
math.sqrt()：计算一个数的平方根（Python 内置函数）。
参数 类型 描述 x float 要计算平方根的数值。 self.pos_encoding()：对经过词嵌入后的目标序列添加位置编码信息。在 Transformer 架构中，原始的词嵌入本身不包含位置信息，而位置编码可以让模型感知到单词在序列中的位置，这对于捕捉序列顺序关系至关重要 。
self.dropout()：按照设定的概率对输入进行随机失活（将神经元输出置为 0）操作。在训练过程中，它通过随机丢弃部分神经元的输出，来防止模型过拟合 。
self.layers：由多个EncoderLayer组成的模块列表，堆叠形成编码器。
layer：每个单独的EncoderLayer层
layer()：Transformer 编码器的核心处理单元，通过堆叠多个 EncoderLayer，模型能够逐层提取更抽象的语义表示，为解码器提供丰富的上下文信息。
def forward(self, src, src_mask): # 词嵌入并缩放（乘以√d_model） src = self.embedding(src) * math.sqrt(self.d_model) # 添加位置编码 src = self.pos_encoding(src) src = self.dropout(src) # 依次通过各编码器层 for layer in self.layers: src = layer(src, src_mask) return srcⅣ、完整代码
# 编码器：处理输入序列，提取特征表示。 class Encoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, input_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(input_vocab_size, d_model) # 词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个编码器层 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, src, src_mask): # 词嵌入并缩放（乘以√d_model） src = self.embedding(src) * math.sqrt(self.d_model) # 添加位置编码 src = self.pos_encoding(src) src = self.dropout(src) # 依次通过各编码器层 for layer in self.layers: src = layer(src, src_mask) return src7.解码器
Ⅰ、运行流程
① 初始化 ——> ② 词嵌入与位置编码 ——> ③ 逐层通过解码器层 ——> ④ 输出结果
输入： tgt [batch, tgt_len] ← 目标序列token ID encoder_output [batch, src_len, d_model] ← 编码器输出 src_mask [batch, 1, src_len] ← 源序列填充掩码 tgt_mask [batch, tgt_len, tgt_len] ← 目标序列因果掩码 ↓ ┌─────────────────────────────────────┐ │ 步骤1：词嵌入 + 位置编码 + Dropout │ └─────────────────────────────────────┘ ↓ 1. 词嵌入与缩放： tgt → Embedding → 向量 × √d_model → [batch, tgt_len, d_model] ↓ 2. 位置编码： 添加位置信息 → [batch, tgt_len, d_model] ↓ 3. Dropout： 随机丢弃部分元素 → tgt ↓ ┌─────────────────────────────────────┐ │ 步骤2：逐层通过解码器层 │ └─────────────────────────────────────┘ ↓ for 每层解码器层: ┌─────────────────────────────────┐ │ 1. 掩蔽自注意力 + 残差 + 层归一化 │ └─────────────────────────────────┘ ↓ ┌─────────────────────────────────┐ │ 2. 编码器-解码器注意力 + 残差 + 层归一化 │ └─────────────────────────────────┘ ↓ ┌─────────────────────────────────┐ │ 3. 前馈网络 + 残差 + 层归一化 │ └─────────────────────────────────┘ ↓ 输出： tgt [batch, tgt_len, d_model] ← 解码后的目标序列表示Ⅱ、初始化
num_layers：解码器层的数量。
d_model：模型维度。
num_heads：注意力头数量。
d_ff：前馈网络隐藏层维度。
target_vocab_size：目标序列的词汇表大小，用于初始化目标词嵌入层。
dropout：Dropout 概率，用于防止过拟合。
super().__init__()：调用父类的初始化方法，避免重复编写代码。
nn.Embedding()：将离散的 token ID 映射为连续的向量表示。
参数 类型 描述 num_embeddings int 词汇表大小。 embedding_dim int 嵌入向量维度。 padding_idx int (可选) 填充标记的索引。 PositionalEncoding()：实现位置编码的具体类，通过特定的算法（如基于三角函数的方法）生成位置编码向量，并添加到词嵌入向量上。
DecorderLayer()：定义了解码器中的单个层结构，通常包含自注意力机制（用于处理目标序列自身的依赖关系）、编码器 - 解码器注意力机制（用于结合编码器的输出信息）以及前馈神经网络等组件。它对输入进行一系列的变换操作，更新目标序列的特征表示。
nn.ModuleList()：存储子模块的列表，用于动态构建神经网络。
参数 类型 描述 modules iterable (可选) 初始化模块列表。 nn.Dropout()：在训练过程中随机将输入的部分元素置为 0，减少神经元对特定特征的依赖，从而防止过拟合。
参数 描述 p 元素被置零的概率（默认 0.5），取值范围 [0, 1]。 inplace 是否直接在原张量上操作（默认 False）。 class Decoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, target_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(target_vocab_size, d_model) # 目标词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个解码器层 self.dropout = nn.Dropout(dropout) # Dropout层Ⅲ、前向传播
tgt：目标序列，形状为(batch_size, tgt_len) 。
encoder_output：编码器的输出，形状为(batch_size, src_len, d_model) 。
src_mask：源序列的填充掩码，通过make_src_mask方法生成。
tgt_mask：目标序列的掩码，通过make_tgt_mask方法生成。
self.embedding()：将目标序列中的离散词（通常是词的索引）映射为连续的向量表示，即词嵌入向量。这些向量能够捕捉单词的语义信息。
self.pos_encoding()：对经过词嵌入后的目标序列添加位置编码信息。
math.sqrt()：计算一个数的平方根（Python 内置函数）。
参数 类型 描述 x float 要计算平方根的数值。 self.dropout()：按照设定的概率对输入进行随机失活（将神经元输出置为 0）操作。
self.layers：nn.ModuleList类型的容器，存储了多个堆叠的解码器层（DecoderLayer）。这些解码器层按顺序依次处理输入数据，逐步对目标序列进行特征提取和转换。
layer：每个单独的DecoderLayer编码器层
layer()：DecoderLayer类的实例对象的调用。它会对输入（目标序列特征、编码器输出、源序列掩码、目标序列掩码等）进行处理，通过内部的多头注意力机制和前馈网络等组件，对目标序列的特征进行更新和转换。
def forward(self, tgt, encoder_output, src_mask, tgt_mask): # 目标序列词嵌入并缩放 tgt = self.embedding(tgt) * math.sqrt(self.d_model) # 添加位置编码 tgt = self.pos_encoding(tgt) tgt = self.dropout(tgt) # 依次通过各解码器层 for layer in self.layers: tgt = layer(tgt, encoder_output, src_mask, tgt_mask) return tgtⅣ、完整代码
# 解码器：根据编码器输出和已生成的输出，逐步生成目标序列。 class Decoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, target_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(target_vocab_size, d_model) # 目标词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个解码器层 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, tgt, encoder_output, src_mask, tgt_mask): # 目标序列词嵌入并缩放 tgt = self.embedding(tgt) * math.sqrt(self.d_model) # 添加位置编码 tgt = self.pos_encoding(tgt) tgt = self.dropout(tgt) # 依次通过各解码器层 for layer in self.layers: tgt = layer(tgt, encoder_output, src_mask, tgt_mask) return tgt8.Transformer模型
Ⅰ、运行流程
① 初始化 ——> ② 创建掩码 ——> ③ 编码器处理 ——> ④ 解码器处理 ——> ⑤ 输出层映射
输入： src [batch, src_len] ← 源序列token ID tgt [batch, tgt_len] ← 目标序列token ID（训练时为已生成部分） ↓ ┌───────────────────────────┐ │ 步骤1：创建掩码 │ └───────────────────────────┘ ↓ src_mask = make_src_mask(src) → [batch, 1, 1, src_len] tgt_mask = make_tgt_mask(tgt) → [batch, 1, tgt_len, tgt_len] ↓ ┌───────────────────────────┐ │ 步骤2：编码器处理 │ └───────────────────────────┘ ↓ src → 词嵌入 + 位置编码 → 多层编码器层 → encoder_output ↓ ┌───────────────────────────┐ │ 步骤3：解码器处理 │ └───────────────────────────┘ ↓ tgt → 词嵌入 + 位置编码 → 多层解码器层（每层包含： ├─ 掩蔽自注意力（使用tgt_mask） └─ 编码器-解码器注意力（使用src_mask） ) → decoder_output ↓ ┌───────────────────────────┐ │ 步骤4：输出层映射 │ └───────────────────────────┘ ↓ decoder_output → 线性层 → output [batch, tgt_len, target_vocab_size]
Ⅱ、初始化
nn.Linear()：实现全连接层（线性变换），将输入特征映射到输出特征。
参数 类型 描述 in_features int 输入特征维度。 out_features int 输出特征维度。 bias bool (可选) 是否添加偏置项（默认True）。 encoder：编码器模块，处理输入序列，将其转换为特征表示。
decoder：解码器模块，基于编码器输出和已生成的部分目标序列生成后续内容。
src_pad_idx：源语言序列中填充标记（如 ）的索引，用于创建掩码。
tgt_pad_idx：目标语言序列中填充标记的索引，用于创建掩码。
fc_out：线性输出层，将解码器输出（维度为 encoder.d_model）映射到目标词汇表大小，以便预测下一个 token。
# Transformer模型：连接编码器和解码器，输出预测序列。 class Transformer(nn.Module): def __init__(self, encoder, decoder, src_pad_idx, tgt_pad_idx): super().__init__() self.encoder = encoder # 编码器 self.decoder = decoder # 解码器 self.src_pad_idx = src_pad_idx # 源序列填充标记索引 self.tgt_pad_idx = tgt_pad_idx # 目标序列填充标记索引 # 输出层：将解码器输出映射到目标词汇表大小 self.fc_out = nn.Linear(encoder.d_model, decoder.embedding.num_embeddings)Ⅲ、源序列掩码（src_mask）
src：输入的源序列张量，形状为 (batch_size, src_seq_len)。
src_mask：源序列的填充掩码，形状为 (batch_size, 1, 1, src_seq_len)。值为 True 表示非填充位置，False 表示填充位置。
.unsqueeze()：在指定维度上插入一个新的维度，扩展张量的形状。
参数 类型 描述 dim int 要插入新维度的位置。
def make_src_mask(self, src): # 创建源序列的填充掩码 (batch_size, 1, 1, seq_len) src_mask = (src != self.src_pad_idx).unsqueeze(1).unsqueeze(2) return src_maskⅣ、目标序列掩码（tgt_mask）
tgt：目标序列，形状为(batch_size, tgt_len) 。
tgt_pad_mask：目标序列的填充掩码，形状为 (batch_size, 1, 1, tgt_seq_len)。
seq_len：目标序列中填充标记对应的索引。
.size()：返回张量的形状，等同于 shape 属性。
参数 类型 描述 dim int (可选) 指定要返回的维度大小。 tgt_sub_mask：三角掩码（下三角矩阵），确保解码器只关注已生成的 token。形状为 (tgt_seq_len, tgt_seq_len)。
torch.trill()：返回矩阵的下三角部分，其余元素置零。
参数 类型 描述 input Tensor 输入矩阵。 diagonal int (可选) 对角线偏移量（默认 0）。 torch.ones()：创建全为 1 的张量。
参数 类型 描述 size tuple 张量形状。 dtype torch.dtype (可选) 数据类型。 device torch.device (可选) 设备（CPU/GPU）。 tgt_mask：组合后的掩码，同时考虑填充和未来信息屏蔽，形状为 (batch_size, 1, tgt_seq_len, tgt_seq_len)。
def make_tgt_mask(self, tgt): # 创建目标序列的填充掩码 (batch_size, 1, 1, seq_len) tgt_pad_mask = (tgt != self.tgt_pad_idx).unsqueeze(1).unsqueeze(2) # 创建三角掩码（下三角为1，上三角为0） seq_len = tgt.size(1) tgt_sub_mask = torch.tril(torch.ones((seq_len, seq_len), device=tgt.device)).bool() # 组合填充掩码和三角掩码 tgt_mask = tgt_pad_mask & tgt_sub_mask return tgt_maskⅤ、前向传播
src：源序列，形状为(batch_size, src_len) 。
tgt：目标序列，形状为(batch_size, tgt_len) 。
src_mask：源序列的填充掩码，通过make_src_mask方法生成。
self.make_src_mask()：生成源序列的填充掩码（Padding Mask），用于在自注意力计算中忽略填充标记（如）的影响。防止编码器关注无意义的 padding 位置。
tgt_mask：目标序列的掩码，通过make_tgt_mask方法生成。
self.make_tgt_mask()：生成目标序列的组合掩码，同时处理填充和未来信息屏蔽（确保解码器仅关注已生成的 token）。
def forward(self, src, tgt): # 创建掩码 src_mask = self.make_src_mask(src) tgt_mask = self.make_tgt_mask(tgt) # 通过编码器 encoder_output = self.encoder(src, src_mask) # 通过解码器 decoder_output = self.decoder(tgt, encoder_output, src_mask, tgt_mask) # 输出层：映射到词汇表大小 output = self.fc_out(decoder_output) return outputⅥ、完整代码
# Transformer模型：连接编码器和解码器，输出预测序列。 class Transformer(nn.Module): def __init__(self, encoder, decoder, src_pad_idx, tgt_pad_idx): super().__init__() self.encoder = encoder # 编码器 self.decoder = decoder # 解码器 self.src_pad_idx = src_pad_idx # 源序列填充标记索引 self.tgt_pad_idx = tgt_pad_idx # 目标序列填充标记索引 # 输出层：将解码器输出映射到目标词汇表大小 self.fc_out = nn.Linear(encoder.d_model, decoder.embedding.num_embeddings) def make_src_mask(self, src): # 创建源序列的填充掩码 (batch_size, 1, 1, seq_len) src_mask = (src != self.src_pad_idx).unsqueeze(1).unsqueeze(2) return src_mask def make_tgt_mask(self, tgt): # 创建目标序列的填充掩码 (batch_size, 1, 1, seq_len) tgt_pad_mask = (tgt != self.tgt_pad_idx).unsqueeze(1).unsqueeze(2) # 创建三角掩码（下三角为1，上三角为0） seq_len = tgt.size(1) tgt_sub_mask = torch.tril(torch.ones((seq_len, seq_len), device=tgt.device)).bool() # 组合填充掩码和三角掩码 tgt_mask = tgt_pad_mask & tgt_sub_mask return tgt_mask def forward(self, src, tgt): # 创建掩码 src_mask = self.make_src_mask(src) tgt_mask = self.make_tgt_mask(tgt) # 通过编码器 encoder_output = self.encoder(src, src_mask) # 通过解码器 decoder_output = self.decoder(tgt, encoder_output, src_mask, tgt_mask) # 输出层：映射到词汇表大小 output = self.fc_out(decoder_output) return output9.完整模型
import torch import torch.nn as nn import torch.nn.functional as F import math # 使用正弦和余弦函数生成不同频率的编码，使模型能够学习序列中的相对位置关系。 class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() # 创建位置编码矩阵 (max_len, d_model) pe = torch.zeros(max_len, d_model) # 生成位置索引 (max_len, 1) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) # 生成频率因子 (d_model//2,) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) # 偶数位置使用正弦函数： # 【:】：选取所有行 【0：：2】：每行中取偶数索引的列（从0索引处出发，每隔两个列） pe[:, 0::2] = torch.sin(position * div_term) # 奇数位置使用余弦函数： # 【:】：选取所有行 【1：：2】：每行中取奇数索引的列（从1索引处出发，每隔两个列） pe[:, 1::2] = torch.cos(position * div_term) # 使位置编码能够适配批次输入并作为模型的静态参数使用。 # 添加批次维度 (1, max_len, d_model)：适配 PyTorch 模型处理批量数据的要求。 pe = pe.unsqueeze(0) # 注册为缓冲区（非训练参数）：位置编码是预计算的固定值，不需要在训练过程中更新，因此不应该作为模型参数（如 nn.Parameter）。 # 设备一致性：将 pe 注册为缓冲区后，当模型调用 .to(device) 时，pe 会自动跟随模型移动到指定设备（如 GPU），确保与输入数据的设备一致性。 # 模型保存与加载：缓冲区会随模型一起保存和加载，确保推理时位置编码可用。 self.register_buffer('pe', pe) def forward(self, x): # 截取与输入序列等长的位置编码 x = x + self.pe[:, :x.size(1)] return x class MultiHeadAttention(nn.Module): 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.d_k = d_model // num_heads # 定义线性变换层：Q, K, 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) def scaled_dot_product_attention(self, Q, K, V, mask=None): # 计算注意力分数，缩放以稳定梯度 scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k) # 应用掩码：将掩码位置的分数设为负无穷，softmax后趋近于0 if mask is not None: scores = scores.masked_fill(mask == 0, -1e9) # 应用softmax获取注意力权重 attention = F.softmax(scores, dim=-1) # 加权求和得到输出 output = torch.matmul(attention, V) return output def split_heads(self, x): # 将输入分割成多个头 batch_size, seq_len, d_model = x.size() return x.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2) def combine_heads(self, x): # 将多个头的输出合并 batch_size, num_heads, seq_len, d_k = x.size() return x.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model) def forward(self, Q, K, V, mask=None): # 线性变换 + 多头分割 Q = self.split_heads(self.W_q(Q)) K = self.split_heads(self.W_k(K)) V = self.split_heads(self.W_v(V)) # 计算注意力 attn_output = self.scaled_dot_product_attention(Q, K, V, mask) # 合并头并应用最终线性变换 output = self.W_o(self.combine_heads(attn_output)) return output # 位置前馈网络：对每个位置的特征进行非线性变换，增加模型表达能力。 class PositionwiseFeedForward(nn.Module): def __init__(self, d_model, d_ff): super().__init__() self.fc1 = nn.Linear(d_model, d_ff) # 第一层线性变换 self.fc2 = nn.Linear(d_ff, d_model) # 第二层线性变换 self.relu = nn.ReLU() # 非线性激活函数 def forward(self, x): return self.fc2(self.relu(self.fc1(x))) # 编码层 class EncoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 自注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, x, mask): # 自注意力 + 残差连接 + 层归一化 attn_output = self.self_attn(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 # 解码层 class DecoderLayer(nn.Module): def __init__(self, d_model, num_heads, d_ff, dropout): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # 掩蔽自注意力 self.cross_attn = MultiHeadAttention(d_model, num_heads) # 编码器-解码器注意力 self.feed_forward = PositionwiseFeedForward(d_model, d_ff) # 前馈网络 self.norm1 = nn.LayerNorm(d_model) # 第一层归一化 self.norm2 = nn.LayerNorm(d_model) # 第二层归一化 self.norm3 = nn.LayerNorm(d_model) # 第三层归一化 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, x, encoder_output, src_mask, tgt_mask): # 掩蔽自注意力：只关注已生成的token，通过三角掩码防止解码器看到未来位置 attn_output1 = self.self_attn(x, x, x, tgt_mask) x = self.norm1(x + self.dropout(attn_output1)) # 编码器-解码器注意力：关注编码器输出，连接编码和解码过程。 attn_output2 = self.cross_attn(x, encoder_output, encoder_output, src_mask) x = self.norm2(x + self.dropout(attn_output2)) # 前馈网络 ff_output = self.feed_forward(x) x = self.norm3(x + self.dropout(ff_output)) return x # 编码器：处理输入序列，提取特征表示。 class Encoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, input_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(input_vocab_size, d_model) # 词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个编码器层 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, src, src_mask): # 词嵌入并缩放（乘以√d_model） src = self.embedding(src) * math.sqrt(self.d_model) # 添加位置编码 src = self.pos_encoding(src) src = self.dropout(src) # 依次通过各编码器层 for layer in self.layers: src = layer(src, src_mask) return src # 解码器：根据编码器输出和已生成的输出，逐步生成目标序列。 class Decoder(nn.Module): def __init__(self, num_layers, d_model, num_heads, d_ff, target_vocab_size, dropout): super().__init__() self.d_model = d_model self.embedding = nn.Embedding(target_vocab_size, d_model) # 目标词嵌入层 self.pos_encoding = PositionalEncoding(d_model) # 位置编码 self.layers = nn.ModuleList([ DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers) ]) # 堆叠多个解码器层 self.dropout = nn.Dropout(dropout) # Dropout层 def forward(self, tgt, encoder_output, src_mask, tgt_mask): # 目标序列词嵌入并缩放 tgt = self.embedding(tgt) * math.sqrt(self.d_model) # 添加位置编码 tgt = self.pos_encoding(tgt) tgt = self.dropout(tgt) # 依次通过各解码器层 for layer in self.layers: tgt = layer(tgt, encoder_output, src_mask, tgt_mask) return tgt # Transformer模型：连接编码器和解码器，输出预测序列。 class Transformer(nn.Module): def __init__(self, encoder, decoder, src_pad_idx, tgt_pad_idx): super().__init__() self.encoder = encoder # 编码器 self.decoder = decoder # 解码器 self.src_pad_idx = src_pad_idx # 源序列填充标记索引 self.tgt_pad_idx = tgt_pad_idx # 目标序列填充标记索引 # 输出层：将解码器输出映射到目标词汇表大小 self.fc_out = nn.Linear(encoder.d_model, decoder.embedding.num_embeddings) def make_src_mask(self, src): # 创建源序列的填充掩码 (batch_size, 1, 1, seq_len) src_mask = (src != self.src_pad_idx).unsqueeze(1).unsqueeze(2) return src_mask def make_tgt_mask(self, tgt): # 创建目标序列的填充掩码 (batch_size, 1, 1, seq_len) tgt_pad_mask = (tgt != self.tgt_pad_idx).unsqueeze(1).unsqueeze(2) # 创建三角掩码（下三角为1，上三角为0） seq_len = tgt.size(1) tgt_sub_mask = torch.tril(torch.ones((seq_len, seq_len), device=tgt.device)).bool() # 组合填充掩码和三角掩码 tgt_mask = tgt_pad_mask & tgt_sub_mask return tgt_mask def forward(self, src, tgt): # 创建掩码 src_mask = self.make_src_mask(src) tgt_mask = self.make_tgt_mask(tgt) # 通过编码器 encoder_output = self.encoder(src, src_mask) # 通过解码器 decoder_output = self.decoder(tgt, encoder_output, src_mask, tgt_mask) # 输出层：映射到词汇表大小 output = self.fc_out(decoder_output) return output三、模型创建
代码运行流程
输入参数 → ┌───────────────────────────────────────────┐ │ 创建编码器 │ │ Encoder( │ num_layers, d_model, num_heads, d_ff, │ src_vocab_size, dropout │ ) → encoder │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 创建解码器 │ │ Decoder( │ num_layers, d_model, num_heads, d_ff, │ tgt_vocab_size, dropout │ ) → decoder │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 创建完整Transformer模型 │ │ Transformer(encoder, decoder, │ src_pad_idx, tgt_pad_idx) → model │ └───────────────────────────────────────────┘ ↓ ┌───────────────────────────────────────────┐ │ 参数初始化 │ │ 遍历model.parameters(): │ │ ├─ 若参数维度 > 1（如权重矩阵）: │ │ │ ↓ │ │ │ 使用Xavier均匀分布初始化 │ │ │ nn.init.xavier_uniform_(p) │ │ └─ 否则（如偏置项）: 跳过 │ └───────────────────────────────────────────┘ ↓ 输出: model（初始化后的Transformer模型）
变量名 类型 说明 src_vocab_size int 源语言（如英文）词汇表的大小，决定编码器词嵌入层的维度。 tgt_vocab_size int 目标语言（如中文）词汇表的大小，决定解码器词嵌入层的维度。 src_pad_idx int 源语言中填充标记（）的索引，用于生成源序列的填充掩码。 tgt_pad_idx int 目标语言中填充标记（）的索引，用于生成目标序列的填充掩码。 d_model int Transformer 模型的隐藏层维度（词嵌入维度），默认值为 512。 num_heads int 多头注意力机制中的头数，默认值为 8。要求 d_model 能被 num_heads 整除。 num_layers int 编码器和解码器中堆叠的层数，默认值为 6。 d_ff int 位置前馈网络（FFN）的中间层维度，通常为 d_model 的 4 倍，默认值为 2048。 dropout float dropout 概率，用于防止过拟合，默认值为 0.1。 encoder Encoder 类实例 创建的编码器对象，包含词嵌入、位置编码和多层编码器层。 decoder Decoder 类实例 创建的解码器对象，包含词嵌入、位置编码和多层解码器层（含跨注意力机制）。 model Transformer 类实例 组装后的完整 Transformer 模型，包含编码器、解码器、掩码生成和输出层。 p torch.Tensor 模型参数张量（如权重矩阵、偏置项），遍历 model.parameters() 得到。 Encoder()：创建 Transformer 的编码器模块。
Decoder()：创建 Transformer 的解码器模块。
Transformer()：将编码器和解码器组合成完整的 Transformer 模型。
model.parameters()：返回模型中所有可训练参数的迭代器。
.dim()：返回张量的维度数（即秩）。
nn.init_xavier_uniform_()：使用 Xavier 均匀分布初始化参数。
Xavier 均匀化分布（Xavier Uniform Distribution）是一种用于神经网络权重初始化的方法，由论文《Understanding the difficulty of training deep feedforward neural networks》（2010）提出，旨在解决深层网络训练中因权重初始化不当导致的梯度消失或爆炸问题。其核心思想是让权重的初始分布满足输入和输出的方差保持一致，从而使信号在网络中更稳定地传播。
假设模型中有一个线性层nn.Linear(10, 20)，其参数包括：
- weight：形状为[20, 10]的二维张量（需要 Xavier 初始化）
- bias：形状为[20]的一维张量（应保持默认初始化）
在 Transformer 模型中，对参数进行维度判断后再应用 Xavier 初始化的核心目的是：只对需要的权重矩阵进行初始化，避免对偏置和其他特殊参数造成干扰，从而确保模型训练的稳定性和有效性。
参数 类型 描述 tensor Tensor 要初始化的张量。 gain float (可选) 缩放因子（默认 1） # 模型创建函数 def create_transformer_model(src_vocab_size, tgt_vocab_size, src_pad_idx, tgt_pad_idx, d_model=512, num_heads=8, num_layers=6, d_ff=2048, dropout=0.1): # 创建编码器和解码器 encoder = Encoder(num_layers, d_model, num_heads, d_ff, src_vocab_size, dropout) decoder = Decoder(num_layers, d_model, num_heads, d_ff, tgt_vocab_size, dropout) # 创建完整的Transformer模型 model = Transformer(encoder, decoder, src_pad_idx, tgt_pad_idx) # 使用Xavier均匀分布初始化参数 # 确保参数矩阵的输入和输出的方差保持一致，从而让信号在神经网络中更稳定地传播。 for p in model.parameters(): if p.dim() > 1: # 对输入的参数张量（通常是权重矩阵）使用Xavier 均匀分布进行原地初始化（_后缀表示原地修改）。 nn.init.xavier_uniform_(p) return model四、模型文件及完整代码
通过网盘分享的文件：Transformer源代码和模型文件
链接: https://pan.baidu.com/s/147A4-T57M3NowRoIuv8Ziw?pwd=xwni 提取码: xwni
--来自百度网盘超级会员v3的分享
- 作用：
- 作用：
- 作用：
- 核心功能：
- 核心功能：
- 核心功能：
- 核心功能：
