大模型Transformer+源码分析(下)

二、进一步代码实现

1. 1. 上面小模块代码严格遵循原论文《Attention Is All You Need》的设计，采用预归一化（Pre-Norm）结构（先归一化再经过子层），训练稳定性优于原论文的后归一化结构。
2. 2. 所有公式与代码一一对应，符号定义完全一致，便于对照理解。
3. 3. 实际应用需注意：
• • 替换模拟数据为真实数据集（如使用torchtext或Hugging Face Datasets加载）；
• • 增加学习率调度器（如ReduceLROnPlateau）；
• • 完善评估指标（如BLEU分数）；
• • 针对大模型训练，需采用混合精度训练、分布式训练等优化策略。

下面是利用上面封装的小模块进一步实现编码器和解码器，并实现Transformer整体模型：

1. 编码器模块（带详细注释）

# 导入PyTorch的神经网络模块（注：原代码依赖该导入，此处保留以保证代码完整性）import torch.nn as nn# 编码器层（单个）：Transformer编码器的核心组成单元，包含自注意力机制和前馈网络class EncoderLayer(nn.Module):    # 构造函数：初始化编码器层的各个子模块    # 参数说明：    #   d_model: 模型的特征维度（词向量/语义向量的维度，如512）    #   heads: 多头注意力的头数（如8）    #   dropout: dropout的丢弃概率，默认0.1，用于防止过拟合    def __init__(self, d_model, heads, dropout=0.1):        # 调用父类nn.Module的构造函数，必须执行        super().__init__()        # 第一个层归一化模块，用于自注意力子层的预归一化        self.norm1 = Norm(d_model)        # 第二个层归一化模块，用于前馈网络子层的预归一化        self.norm2 = Norm(d_model)        # 多头注意力模块（自注意力使用），传入头数、模型维度和dropout概率        self.attn = MultiHeadAttention(heads, d_model, dropout)        # 位置感知前馈网络模块，传入模型维度和dropout概率        self.ff = FeedForward(d_model, dropout=dropout)        # 第一个dropout层，用于自注意力输出的随机丢弃        self.dropout1 = nn.Dropout(dropout)        # 第二个dropout层，用于前馈网络输出的随机丢弃        self.dropout2 = nn.Dropout(dropout)    # 前向传播函数：定义编码器层的数据流走向    # 参数说明：    #   x: 编码器层的输入张量，形状[batch_size, seq_len, d_model]    #      batch_size: 批次大小（一次训练的样本数）    #      seq_len: 序列长度（输入文本的单词数量）    #      d_model: 模型特征维度    #   mask: 源序列掩码张量，形状[batch_size, 1, seq_len]，用于掩盖填充位（无效字符）    # 返回值：编码器层的输出张量，形状[batch_size, seq_len, d_model]    def forward(self, x, mask):        """        x: [batch_size, seq_len, d_model]        mask: [batch_size, 1, seq_len]        return: [batch_size, seq_len, d_model]        """        # 自注意力子层（采用预归一化策略：先归一化，再做注意力计算）        # 1. 对输入x进行第一层归一化，得到归一化后的特征        x_norm = self.norm1(x)        # 2. 执行多头自注意力计算：Q/K/V均为x_norm（自注意力特性），传入掩码mask        #    返回注意力输出attn_out和注意力权重（权重用_接收，暂不使用）        attn_out, _ = self.attn(x_norm, x_norm, x_norm, mask)        # 3. 残差连接：原始输入x + dropout后的注意力输出（防止深层网络梯度消失）        #    先对attn_out做dropout，再与原始x相加，更新x        x = x + self.dropout1(attn_out)        # FFN前馈网络子层（同样采用预归一化策略）        # 1. 对自注意力子层输出的x进行第二层归一化        x_norm = self.norm2(x)        # 2. 执行前馈网络计算，得到前馈网络输出ff_out        ff_out = self.ff(x_norm)        # 3. 残差连接：自注意力子层的输出x + dropout后的前馈网络输出        #    先对ff_out做dropout，再与当前x相加，更新x        x = x + self.dropout2(ff_out)        # 返回当前编码器层的最终输出，作为下一个编码器层的输入        return x# 编码器（整体）：由N个EncoderLayer堆叠而成，完成输入序列的语义编码class Encoder(nn.Module):    # 构造函数：初始化编码器的各个组件    # 参数说明：    #   vocab_size: 源语言词汇表大小（单词/字符的总数）    #   d_model: 模型的特征维度    #   N: 编码器层（EncoderLayer）的堆叠数量（如6）    #   heads: 多头注意力的头数    #   dropout: dropout的丢弃概率    def __init__(self, vocab_size, d_model, N, heads, dropout):        # 调用父类nn.Module的构造函数        super().__init__()        # 词嵌入模块：将词汇的索引值转换为固定维度的词向量        self.embed = Embedder(vocab_size, d_model)        # 位置编码模块：为词向量添加位置信息，解决Transformer无法感知单词顺序的问题        self.pe = PositionalEncoder(d_model, dropout=dropout)        # 堆叠N个EncoderLayer：通过get_clones函数生成N个相同的编码器层副本        # 避免多个层共享参数，保证每层参数独立训练        self.layers = get_clones(EncoderLayer(d_model, heads, dropout), N)        # 最终层归一化模块：对N个编码器层输出的特征进行归一化，稳定训练        self.norm = Norm(d_model)    # 前向传播函数：定义整个编码器的数据流走向    # 参数说明：    #   src: 源序列输入张量（词汇索引），形状[batch_size, seq_len]    #   src_mask: 源序列掩码张量，形状[batch_size, 1, seq_len]    # 返回值：编码器的最终语义编码输出，形状[batch_size, seq_len, d_model]    def forward(self, src, src_mask):        """        src: [batch_size, seq_len]        src_mask: [batch_size, 1, seq_len]        return: [batch_size, seq_len, d_model]        """        # 1. 词嵌入：将词汇索引src转换为词向量，输出形状[batch_size, seq_len, d_model]        x = self.embed(src)        # 2. 位置编码：为词向量添加位置信息，输出形状保持不变        x = self.pe(x)        # 3. 依次经过N个堆叠的编码器层：逐层对特征进行语义提取和编码        #    每个编码器层的输入是上一层的输出，最终得到深层语义特征        for layer in self.layers:            x = layer(x, src_mask)        # 4. 最终层归一化：对N个编码器层输出的特征进行归一化，返回最终编码结果        return self.norm(x)

2. 解码器模块（带详细注释）

# 导入PyTorch神经网络模块（保证代码依赖完整性，原代码隐含该导入）import torch.nn as nn# 解码器层（单个）：Transformer解码器的核心组成单元，包含掩码自注意力、交叉注意力和前馈网络# 作用：基于编码器的语义输出和已生成的目标序列，完成目标序列的语义建模与生成class DecoderLayer(nn.Module):    # 构造函数：初始化解码器层的所有子模块    # 参数说明：    #   d_model: 模型的特征维度（词向量/语义向量的维度，如512）    #   heads: 多头注意力的头数（如8）    #   dropout: dropout丢弃概率，默认0.1，用于防止模型过拟合    def __init__(self, d_model, heads, dropout=0.1):        # 调用父类nn.Module的构造函数，必须执行的初始化步骤        super().__init__()        # 第一个层归一化模块，用于掩码自注意力子层的预归一化        self.norm1 = Norm(d_model)        # 第二个层归一化模块，用于交叉注意力子层的预归一化        self.norm2 = Norm(d_model)        # 第三个层归一化模块，用于前馈网络子层的预归一化        self.norm3 = Norm(d_model)        # 第一个多头注意力模块：掩码自注意力（防止解码器看到未来位置的信息）        self.attn1 = MultiHeadAttention(heads, d_model, dropout)  # 掩码自注意力        # 第二个多头注意力模块：交叉注意力（融合编码器输出与解码器中间特征）        self.attn2 = MultiHeadAttention(heads, d_model, dropout)  # 交叉注意力        # 位置感知前馈网络模块，对语义特征进行非线性变换与提炼        self.ff = FeedForward(d_model, dropout=dropout)        # 第一个dropout层：用于掩码自注意力输出的随机丢弃        self.dropout1 = nn.Dropout(dropout)        # 第二个dropout层：用于交叉注意力输出的随机丢弃        self.dropout2 = nn.Dropout(dropout)        # 第三个dropout层：用于前馈网络输出的随机丢弃        self.dropout3 = nn.Dropout(dropout)    # 前向传播函数：定义解码器层的数据流走向    # 参数说明：    #   x: 解码器层的输入张量（已生成的目标序列特征），形状[batch_size, trg_len, d_model]    #      batch_size: 批次大小（一次训练的样本数）    #      trg_len: 目标序列长度（已生成的目标文本单词数量）    #      d_model: 模型特征维度    #   e_outputs: 编码器的最终输出张量（源序列语义编码），形状[batch_size, src_len, d_model]    #      src_len: 源序列长度（输入文本的单词数量）    #   src_mask: 源序列掩码张量，形状[batch_size, 1, src_len]，用于掩盖源序列的填充位（无效字符）    #   trg_mask: 目标序列掩码张量，形状[batch_size, trg_len, trg_len]，用于掩盖目标序列填充位和未来位置    # 返回值：解码器层的输出张量，形状[batch_size, trg_len, d_model]    def forward(self, x, e_outputs, src_mask, trg_mask):        """        x: [batch_size, trg_len, d_model]        e_outputs: [batch_size, src_len, d_model]        src_mask: [batch_size, 1, src_len]        trg_mask: [batch_size, trg_len, trg_len]        return: [batch_size, trg_len, d_model]        """        # 掩码自注意力子层（预归一化策略：先归一化，再执行注意力计算）        # 1. 对解码器输入x进行第一层归一化，得到归一化后的目标序列特征        x_norm = self.norm1(x)        # 2. 执行掩码自注意力计算：Q/K/V均为x_norm（自注意力特性），传入trg_mask屏蔽未来位置        #    返回掩码注意力输出attn1_out和注意力权重（权重用_接收，暂不使用）        attn1_out, _ = self.attn1(x_norm, x_norm, x_norm, trg_mask)        # 3. 残差连接：原始输入x + dropout后的掩码自注意力输出，防止深层网络梯度消失        #    先对attn1_out做dropout，再与原始x相加，更新x        x = x + self.dropout1(attn1_out)        # 交叉注意力子层（预归一化策略：融合编码器语义与解码器中间特征）        # 1. 对掩码自注意力输出的x进行第二层归一化        x_norm = self.norm2(x)        # 2. 执行交叉注意力计算：Q为解码器归一化特征x_norm，K/V为编码器输出e_outputs，传入src_mask屏蔽源序列填充位        #    返回交叉注意力输出attn2_out和注意力权重（权重用_接收，暂不使用）        attn2_out, _ = self.attn2(x_norm, e_outputs, e_outputs, src_mask)        # 3. 残差连接：掩码自注意力输出x + dropout后的交叉注意力输出，融合源序列与目标序列语义        #    先对attn2_out做dropout，再与当前x相加，更新x        x = x + self.dropout2(attn2_out)        # FFN前馈网络子层（预归一化策略：对融合后的特征进行非线性提炼）        # 1. 对交叉注意力输出的x进行第三层归一化        x_norm = self.norm3(x)        # 2. 执行前馈网络计算，对归一化后的特征进行非线性变换，得到ff_out        ff_out = self.ff(x_norm)        # 3. 残差连接：交叉注意力输出x + dropout后的前馈网络输出，进一步提炼语义特征        #    先对ff_out做dropout，再与当前x相加，更新x        x = x + self.dropout3(ff_out)        # 返回当前解码器层的最终输出，作为下一个解码器层的输入        return x# 解码器（整体）：由N个DecoderLayer堆叠而成，完成目标序列的语义编码与生成准备class Decoder(nn.Module):    # 构造函数：初始化解码器的所有组件    # 参数说明：    #   vocab_size: 目标语言词汇表大小（目标语言单词/字符的总数）    #   d_model: 模型的特征维度    #   N: 解码器层（DecoderLayer）的堆叠数量（如6）    #   heads: 多头注意力的头数    #   dropout: dropout丢弃概率    def __init__(self, vocab_size, d_model, N, heads, dropout):        # 调用父类nn.Module的构造函数，完成基础初始化        super().__init__()        # 词嵌入模块：将目标语言词汇的索引值转换为固定维度的词向量        self.embed = Embedder(vocab_size, d_model)        # 位置编码模块：为目标语言词向量添加位置信息，解决Transformer无法感知单词顺序的问题        self.pe = PositionalEncoder(d_model, dropout=dropout)        # 堆叠N个DecoderLayer：通过get_clones函数生成N个相同的解码器层副本        # 避免多个层共享参数，保证每层参数独立训练与更新        self.layers = get_clones(DecoderLayer(d_model, heads, dropout), N)        # 最终层归一化模块：对N个解码器层输出的特征进行归一化，稳定训练过程并优化语义表示        self.norm = Norm(d_model)    # 前向传播函数：定义整个解码器的数据流走向    # 参数说明：    #   trg: 目标序列输入张量（词汇索引），形状[batch_size, trg_len]    #   e_outputs: 编码器的最终输出张量（源序列语义编码），形状[batch_size, src_len, d_model]    #   src_mask: 源序列掩码张量，形状[batch_size, 1, src_len]    #   trg_mask: 目标序列掩码张量，形状[batch_size, trg_len, trg_len]    # 返回值：解码器的最终语义编码输出，形状[batch_size, trg_len, d_model]    def forward(self, trg, e_outputs, src_mask, trg_mask):        """        trg: [batch_size, trg_len]        e_outputs: [batch_size, src_len, d_model]        src_mask: [batch_size, 1, src_len]        trg_mask: [batch_size, trg_len, trg_len]        return: [batch_size, trg_len, d_model]        """        # 1. 词嵌入：将目标序列词汇索引trg转换为词向量，输出形状[batch_size, trg_len, d_model]        x = self.embed(trg)        # 2. 位置编码：为目标语言词向量添加位置信息，输出形状保持不变        x = self.pe(x)        # 3. 依次经过N个堆叠的解码器层：逐层对目标序列特征进行语义建模与融合        #    每个解码器层的输入是上一层的输出，最终得到目标序列的深层语义特征        for layer in self.layers:            x = layer(x, e_outputs, src_mask, trg_mask)        # 4. 最终层归一化：对N个解码器层输出的特征进行归一化，返回最终的目标序列语义编码        return self.norm(x)

3. Transformer整体模型（带详细注释）

# 导入PyTorch神经网络模块（保证代码依赖完整性，原代码隐含该导入）import torch.nn as nn# Transformer整体模型：整合编码器（Encoder）和解码器（Decoder），完成端到端的序列转换任务# 作用：实现源序列到目标序列的映射（如机器翻译：英文→德文、文本摘要：长文本→短文本等）class Transformer(nn.Module):    # 构造函数：初始化Transformer的核心组件（编码器、解码器、输出层）    # 参数说明：    #   src_vocab: 源语言词汇表大小（源语言单词/字符的总数）    #   trg_vocab: 目标语言词汇表大小（目标语言单词/字符的总数）    #   d_model: 模型的核心特征维度（所有子模块的统一向量维度，如512）    #   N: 编码器层（EncoderLayer）和解码器层（DecoderLayer）的堆叠数量（如6）    #   heads: 多头注意力机制的头数（如8）    #   dropout: dropout丢弃概率，用于防止模型过拟合    def __init__(self, src_vocab, trg_vocab, d_model, N, heads, dropout):        # 调用父类nn.Module的构造函数，完成基础初始化        super().__init__()        # 初始化编码器：传入源语言词汇表大小等参数，负责对源序列进行语义编码        self.encoder = Encoder(src_vocab, d_model, N, heads, dropout)        # 初始化解码器：传入目标语言词汇表大小等参数，负责基于编码器输出生成目标序列语义        self.decoder = Decoder(trg_vocab, d_model, N, heads, dropout)        # 最终输出层：线性变换层，将解码器输出的d_model维语义向量映射为目标词汇表维度        # 作用：得到每个位置上目标词汇的概率分布，用于后续词汇预测        self.out = nn.Linear(d_model, trg_vocab)  # 最终输出层    # 前向传播函数：定义Transformer整体的数据流走向，完成从源序列到目标序列的完整映射    # 参数说明：    #   src: 源序列输入张量（词汇索引），形状[batch_size, src_len]    #      batch_size: 批次大小（一次训练/推理的样本数）    #      src_len: 源序列长度（源文本的单词数量）    #   trg: 目标序列输入张量（词汇索引），形状[batch_size, trg_len]    #      trg_len: 目标序列长度（目标文本的单词数量）    #   src_mask: 源序列掩码张量，形状[batch_size, 1, src_len]，用于掩盖源序列的填充位（无效字符）    #   trg_mask: 目标序列掩码张量，形状[batch_size, trg_len, trg_len]，用于掩盖目标序列填充位和未来位置    # 返回值：Transformer最终输出张量，形状[batch_size, trg_len, trg_vocab]    #      表示目标序列每个位置上，对应目标词汇表中每个词汇的预测得分（未经过Softmax归一化）    def forward(self, src, trg, src_mask, trg_mask):        """        src: [batch_size, src_len]        trg: [batch_size, trg_len]        src_mask: [batch_size, 1, src_len]        trg_mask: [batch_size, trg_len, trg_len]        return: [batch_size, trg_len, trg_vocab]        """        # 1. 编码器前向传播：对源序列src进行语义编码，得到源序列的深层语义表示e_outputs        #    e_outputs形状：[batch_size, src_len, d_model]        e_outputs = self.encoder(src, src_mask)        # 2. 解码器前向传播：基于目标序列trg和编码器输出e_outputs，生成目标序列的语义表示d_outputs        #    d_outputs形状：[batch_size, trg_len, d_model]        d_outputs = self.decoder(trg, e_outputs, src_mask, trg_mask)        # 3. 输出层线性变换：将解码器输出的d_model维语义向量，映射为trg_vocab维的词汇得分        #    output形状：[batch_size, trg_len, trg_vocab]        output = self.out(d_outputs)        # 返回Transformer最终输出，后续可通过Softmax函数得到词汇概率分布，进行词汇预测        return output

4. 掩码生成与模型训练

def create_masks(src, trg):    """    生成源序列掩码（掩盖填充位）和目标序列掩码（掩盖填充位+未来位置）    src: [batch_size, src_len]    trg: [batch_size, trg_len]    return: src_mask [batch_size, 1, src_len], trg_mask [batch_size, trg_len, trg_len]    """    # 源序列掩码：填充位（0）设为False，有效位设为True    src_mask = (src != 0).unsqueeze(1)  # [bs, 1, src_len]    # 目标序列掩码：下三角掩码 + 填充掩码    trg_len = trg.size(1)    trg_pad_mask = (trg != 0).unsqueeze(1)  # [bs, 1, trg_len]    trg_subseq_mask = torch.tril(torch.ones(trg_len, trg_len, device=src.device)).bool()  # [trg_len, trg_len]    trg_mask = trg_pad_mask & trg_subseq_mask  # [bs, trg_len, trg_len]    return src_mask, trg_mask# 模型初始化与训练示例if __name__ == "__main__":    # 超参数设置（与原论文一致）    d_model = 512    heads = 8    N = 6    dropout = 0.1    epochs = 20    lr = 1e-4    # 模拟数据（实际使用时替换为真实数据集，如WMT翻译数据集）    src_vocab_size = 10000  # 源语言词汇表大小    trg_vocab_size = 10000  # 目标语言词汇表大小    batch_size = 32    src_seq_len = 50    trg_seq_len = 50    # 生成模拟批次数据    src = torch.randint(1, src_vocab_size, (batch_size, src_seq_len))  # 1~vocab_size为有效词，0为填充位    trg = torch.randint(1, trg_vocab_size, (batch_size, trg_seq_len))    # 初始化模型、优化器    model = Transformer(src_vocab_size, trg_vocab_size, d_model, N, heads, dropout)    optimizer = torch.optim.Adam(model.parameters(), lr=lr, betas=(0.9, 0.98), eps=1e-9)    # 模型训练    model.train()    for epoch in range(epochs):        total_loss = 0.0        for batch_idx in range(100):  # 模拟100个批次/epoch            # 生成掩码            src_mask, trg_mask = create_masks(src, trg[:, :-1])  # trg输入去掉最后一个词            # 前向传播            outputs = model(src, trg[:, :-1], src_mask, trg_mask)  # 模型输出：[bs, trg_len-1, trg_vocab]            # 计算损失（目标为trg去掉第一个词）            targets = trg[:, 1:].contiguous().view(-1)  # [bs*(trg_len-1)]            loss = F.cross_entropy(outputs.view(-1, trg_vocab_size), targets, ignore_index=0)            # 反向传播与参数更新            optimizer.zero_grad()            loss.backward()            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)  # 梯度裁剪            optimizer.step()            total_loss += loss.item()        avg_loss = total_loss / 100        print(f"Epoch {epoch+1}/{epochs}, Average Loss: {avg_loss:.4f}")

三、总结

上面的代码仅完成了 Transformer 各模块（小模块、大模块）和整体架构的封装，以及简单的模拟数据测试，并未使用真实公开数据集进行完整的训练、验证与推理试验验证。

下一步，可以基于 WMT14 En-De 机器翻译数据集（英文→德文，经典 Transformer 试验数据集），结合 Hugging Face 生态（datasets加载数据、tokenizers处理文本），完成从数据预处理、模型训练、验证评估到实际推理的全流程试验验证。