云计算百科
云计算领域专业知识百科平台

《Transformer模型PyTorch实现全攻略:架构拆解、代码示例与优化技巧》

本篇技术博文摘要 🌟

  • 文章开篇即阐明使用PyTorch框架实现Transformer的实际价值与重要意义,指出其在自然语言处理等领域的核心地位。
  • 随后,文章以清晰的步骤化结构展开:首先指导读者导入必要的PyTorch库与模块,奠定开发基础;
  • 接着深入拆解Transformer的核心组件,包括多头注意力机制、位置前馈网络与位置编码的原理与实现,并逐一提供可运行的代码示例。
  • 在此基础上,文章详细讲解了如何将这些基础模块组合为编码器层与解码器层,进而构建完整的Transformer模型架构,同时提供了详细的模型参数初始化说明。
  • 文章不仅停留在模型构建层面,更进一步演示了模型的训练流程与评估方法,包括损失函数选择、优化器配置以及性能评估指标,最后对代码优化与模型评估结果进行了系统性汇总,真正实现了从理论到实践、从构建到评估的全链路贯通。

引言 📘

  • 在这个变幻莫测、快速发展的技术时代,与时俱进是每个IT工程师的必修课。
  • 我是盛透侧视攻城狮,一名什么都会一丢丢的网络安全工程师,也是众多技术社区的活跃成员以及多家大厂官方认可人员,希望能够与各位在此共同成长。

上节回顾

目录

本篇技术博文摘要 🌟

引言 📘

上节回顾

1.PyTorch 构建 Transformer 模型意义

2.具体使用 PyTorch 构建 Transformer 模型步骤

2.1导入必要的库和模块

2.2定义基本构建块:

2.2.1多头注意力

多头注意力具体示例:

2.2.2位置前馈网络(Position-wise Feed-Forward Network)

位置前馈网络具体示例:

2.2.3位置编码

位置编码具体示例:

2.3构建编码器块(Encoder Layer)

2.3.1编码器层

编码器层具体示例:

2.4构建解码器模块

2.4.1解码器层

解码器层具体示例:

3.如何构建完整的 Transformer 模型

3.1如何构建完整的 Transformer 模型具体示例:

3.2模型初始化参数说明:

4.训练 PyTorch Transformer 模型

4.1训练 PyTorch Transformer 模型示例:

5.模型评估性能

5.1模型评估性能具体示例:

6.PyTorch 构建+评估 Transformer 模型代码优化汇总

6.1模型性能评估结果如下:​

欢迎各位彦祖与热巴畅游本人专栏与技术博客

你的三连是我最大的动力

点击➡️指向的专栏名即可闪现


1.PyTorch 构建 Transformer 模型意义

  • Transformer 是现代机器学习中最强大的模型之一。

  • Transformer 模型是一种基于自注意力机制(Self-Attention) 的深度学习架构,它彻底改变了自然语言处理(NLP)领域,并成为现代深度学习模型(如 BERT、GPT 等)的基础。

  • Transformer 是现代 NLP 领域的核心架构,凭借其强大的长距离依赖建模能力和高效的并行计算优势,在语言翻译和文本摘要等任务中超越了传统的 长短期记忆 (LSTM) 网络。

2.具体使用 PyTorch 构建 Transformer 模型步骤

2.1导入必要的库和模块

  • 导入 PyTorch 核心库、神经网络模块、优化器模块、数据处理工具,以及数学和对象复制模块,为定义模型架构、管理数据和训练过程提供支持。

import torch
import torch.nn as nn
import torch.optim as optim
import torch.utils.data as data
import math
import copy

  • 说明:

    • torch:PyTorch 的核心库,用于张量操作和自动求导。

    • torch.nn:PyTorch 的神经网络模块,包含各种层和损失函数。

    • torch.optim:优化算法模块,如 Adam、SGD 等。

    • math:数学函数库,用于计算平方根等。

    • copy:用于深度复制对象。

2.2定义基本构建块:

2.2.1多头注意力

  • 通过多个"注意力头"计算序列中每对位置之间的关系,能够捕捉输入序列的不同特征和模式。

  • MultiHeadAttention 类封装了 Transformer 模型中常用的多头注意力机制,负责将输入拆分成多个注意力头,对每个注意力头施加注意力,然后将结果组合起来,这样模型就可以在不同尺度上捕捉输入数据中的各种关系,提高模型的表达能力。
多头注意力具体示例:

class MultiHeadAttention(nn.Module):
def __init__(self, d_model, num_heads):
super(MultiHeadAttention, self).__init__()
assert d_model % num_heads == 0, "d_model必须能被num_heads整除"

self.d_model = d_model # 模型维度(如512)
self.num_heads = num_heads # 注意力头数(如8)
self.d_k = d_model // num_heads # 每个头的维度(如64)

# 定义线性变换层(无需偏置)
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):
"""
计算缩放点积注意力
输入形状:
Q: (batch_size, num_heads, seq_length, d_k)
K, V: 同Q
输出形状: (batch_size, num_heads, seq_length, d_k)
"""
# 计算注意力分数(Q和K的点积)
attn_scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)

# 应用掩码(如填充掩码或未来信息掩码)
if mask is not None:
attn_scores = attn_scores.masked_fill(mask == 0, -1e9)

# 计算注意力权重(softmax归一化)
attn_probs = torch.softmax(attn_scores, dim=-1)

# 对值向量加权求和
output = torch.matmul(attn_probs, V)
return output

def split_heads(self, x):
"""
将输入张量分割为多个头
输入形状: (batch_size, seq_length, d_model)
输出形状: (batch_size, num_heads, seq_length, d_k)
"""
batch_size, seq_length, d_model = x.size()
return x.view(batch_size, seq_length, self.num_heads, self.d_k).transpose(1, 2)

def combine_heads(self, x):
"""
将多个头的输出合并回原始形状
输入形状: (batch_size, num_heads, seq_length, d_k)
输出形状: (batch_size, seq_length, d_model)
"""
batch_size, _, seq_length, d_k = x.size()
return x.transpose(1, 2).contiguous().view(batch_size, seq_length, self.d_model)

def forward(self, Q, K, V, mask=None):
"""
前向传播
输入形状: Q/K/V: (batch_size, seq_length, d_model)
输出形状: (batch_size, seq_length, d_model)
"""
# 线性变换并分割多头
Q = self.split_heads(self.W_q(Q)) # (batch, heads, seq_len, d_k)
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

  • 补充说明:
    • 多头注意力机制:将输入分割成多个头,每个头独立计算注意力,最后将结果合并。

    • 缩放点积注意力:计算查询和键的点积,缩放后使用 softmax 计算注意力权重,最后对值进行加权求和。

    • 掩码:用于屏蔽无效位置(如填充部分)。

2.2.2位置前馈网络(Position-wise Feed-Forward Network)

  • 由两个全连接层和一个 ReLU 激活函数组成,用于进一步处理注意力机制的输出。
位置前馈网络具体示例:

class PositionWiseFeedForward(nn.Module):
def __init__(self, d_model, d_ff):
super(PositionWiseFeedForward, self).__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)))

2.2.3位置编码

  • 位置编码用于注入输入序列中每个 token 的位置信息。

  • 使用不同频率的正弦和余弦函数来生成位置编码。

位置编码具体示例:

class PositionalEncoding(nn.Module):
def __init__(self, d_model, max_seq_length):
super(PositionalEncoding, self).__init__()
pe = torch.zeros(max_seq_length, d_model) # 初始化位置编码矩阵
position = torch.arange(0, max_seq_length, 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) # 奇数位置使用余弦函数
self.register_buffer('pe', pe.unsqueeze(0)) # 注册为缓冲区

def forward(self, x):
# 将位置编码添加到输入中
return x + self.pe[:, :x.size(1)]

2.3构建编码器块(Encoder Layer)

2.3.1编码器层

  • 包含一个自注意力机制和一个前馈网络,每个子层后接残差连接和层归一化。
编码器层具体示例:

class EncoderLayer(nn.Module):
def __init__(self, d_model, num_heads, d_ff, dropout):
super(EncoderLayer, self).__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

2.4构建解码器模块

2.4.1解码器层

  • 包含一个自注意力机制、一个交叉注意力机制和一个前馈网络,每个子层后接残差连接和层归一化。
解码器层具体示例:

class DecoderLayer(nn.Module):
def __init__(self, d_model, num_heads, d_ff, dropout):
super(DecoderLayer, self).__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, enc_output, src_mask, tgt_mask):
# 自注意力机制
attn_output = self.self_attn(x, x, x, tgt_mask)
x = self.norm1(x + self.dropout(attn_output)) # 残差连接和层归一化

# 交叉注意力机制
attn_output = self.cross_attn(x, enc_output, enc_output, src_mask)
x = self.norm2(x + self.dropout(attn_output)) # 残差连接和层归一化

# 前馈网络
ff_output = self.feed_forward(x)
x = self.norm3(x + self.dropout(ff_output)) # 残差连接和层归一化
return x

3.如何构建完整的 Transformer 模型

3.1如何构建完整的 Transformer 模型具体示例:

class Transformer(nn.Module):
def __init__(self, src_vocab_size, tgt_vocab_size, d_model, num_heads, num_layers, d_ff, max_seq_length, dropout):
super(Transformer, self).__init__()
self.encoder_embedding = nn.Embedding(src_vocab_size, d_model) # 编码器词嵌入
self.decoder_embedding = nn.Embedding(tgt_vocab_size, d_model) # 解码器词嵌入
self.positional_encoding = PositionalEncoding(d_model, max_seq_length) # 位置编码

# 编码器和解码器层
self.encoder_layers = nn.ModuleList([EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)])
self.decoder_layers = nn.ModuleList([DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)])

self.fc = nn.Linear(d_model, tgt_vocab_size) # 最终的全连接层
self.dropout = nn.Dropout(dropout) # Dropout

def generate_mask(self, src, tgt):
# 源掩码:屏蔽填充符(假设填充符索引为0)
# 形状:(batch_size, 1, 1, seq_length)
src_mask = (src != 0).unsqueeze(1).unsqueeze(2)

# 目标掩码:屏蔽填充符和未来信息
# 形状:(batch_size, 1, seq_length, 1)
tgt_mask = (tgt != 0).unsqueeze(1).unsqueeze(3)
seq_length = tgt.size(1)
# 生成上三角矩阵掩码,防止解码时看到未来信息
nopeak_mask = (1 – torch.triu(torch.ones(1, seq_length, seq_length), diagonal=1)).bool()
tgt_mask = tgt_mask & nopeak_mask # 合并填充掩码和未来信息掩码
return src_mask, tgt_mask

def forward(self, src, tgt):
# 生成掩码
src_mask, tgt_mask = self.generate_mask(src, tgt)

# 编码器部分
src_embedded = self.dropout(self.positional_encoding(self.encoder_embedding(src)))
enc_output = src_embedded
for enc_layer in self.encoder_layers:
enc_output = enc_layer(enc_output, src_mask)

# 解码器部分
tgt_embedded = self.dropout(self.positional_encoding(self.decoder_embedding(tgt)))
dec_output = tgt_embedded
for dec_layer in self.decoder_layers:
dec_output = dec_layer(dec_output, enc_output, src_mask, tgt_mask)

# 最终输出
output = self.fc(dec_output)
return output

  • 说明:
    • Transformer 模型:包含编码器和解码器部分,每个部分由多个层堆叠而成。

    • 掩码生成:用于屏蔽无效位置和未来信息。

    • 前向传播:依次通过编码器和解码器,最后通过全连接层输出。

3.2模型初始化参数说明:

class Transformer(nn.Module):
def __init__(
self,
src_vocab_size, # 源语言词汇表大小(如英文单词数)
tgt_vocab_size, # 目标语言词汇表大小(如中文单词数)
d_model=512, # 模型维度(每个词向量的长度)
num_heads=8, # 多头注意力的头数
num_layers=6, # 编码器/解码器的堆叠层数
d_ff=2048, # 前馈网络隐藏层维度
max_seq_length=100, # 最大序列长度(用于位置编码)
dropout=0.1 # Dropout概率
):

4.训练 PyTorch Transformer 模型

  • 使用随机数据训练模型,计算损失并更新参数。

4.1训练 PyTorch Transformer 模型示例:

# 超参数
src_vocab_size = 5000 # 源词汇表大小
tgt_vocab_size = 5000 # 目标词汇表大小
d_model = 512 # 模型维度
num_heads = 8 # 注意力头数量
num_layers = 6 # 编码器和解码器层数
d_ff = 2048 # 前馈网络内层维度
max_seq_length = 100 # 最大序列长度
dropout = 0.1 # Dropout 概率

# 初始化模型
transformer = Transformer(src_vocab_size, tgt_vocab_size, d_model, num_heads, num_layers, d_ff, max_seq_length, dropout)

# 生成随机数据
src_data = torch.randint(1, src_vocab_size, (64, max_seq_length)) # 源序列
tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length)) # 目标序列

# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss(ignore_index=0) # 忽略填充部分的损失
optimizer = optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)

# 训练循环
transformer.train()
for epoch in range(100):
optimizer.zero_grad() # 清空梯度,防止累积

# 输入目标序列时去掉最后一个词(用于预测下一个词)
output = transformer(src_data, tgt_data[:, :-1])

# 计算损失时,目标序列从第二个词开始(即预测下一个词)
# output形状: (batch_size, seq_length-1, tgt_vocab_size)
# 目标形状: (batch_size, seq_length-1)
loss = criterion(
output.contiguous().view(-1, tgt_vocab_size),
tgt_data[:, 1:].contiguous().view(-1)
)

loss.backward() # 反向传播
optimizer.step() # 更新参数
print(f"Epoch: {epoch+1}, Loss: {loss.item()}")

5.模型评估性能

  • 在验证数据上计算损失,评估模型性能

5.1模型评估性能具体示例:

transformer.eval()
# 生成验证数据
val_src_data = torch.randint(1, src_vocab_size, (64, max_seq_length))
val_tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length))
# 假设输入为一批英文和对应的中文翻译(已转换为索引)
# 示例数据:
# src_data: [[3, 14, 25, …, 0, 0], …] # 英文句子(0为填充符)
# tgt_data: [[5, 20, 36, …, 0, 0], …] # 中文翻译(0为填充符)
# 注意:实际应用中需对文本进行分词、编码、填充等预处理
with torch.no_grad():
val_output = transformer(val_src_data, val_tgt_data[:, :-1])
val_loss = criterion(val_output.contiguous().view(-1, tgt_vocab_size), val_tgt_data[:, 1:].contiguous().view(-1))
print(f"Validation Loss: {val_loss.item()}")

6.PyTorch 构建+评估 Transformer 模型代码优化汇总

import torch
import torch.nn as nn
import torch.optim as optim
import math

class PositionalEncoding(nn.Module):
"""位置编码层"""
def __init__(self, d_model, max_seq_length=100):
super(PositionalEncoding, self).__init__()

pe = torch.zeros(max_seq_length, d_model)
position = torch.arange(0, max_seq_length, 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) # [1, max_seq_length, d_model]
self.register_buffer('pe', pe)

def forward(self, x):
# x: [batch_size, seq_length, d_model]
return x + self.pe[:, :x.size(1), :]

class PositionwiseFeedForward(nn.Module):
"""位置前馈网络"""
def __init__(self, d_model, d_ff):
super(PositionwiseFeedForward, self).__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 MultiHeadAttention(nn.Module):
def __init__(self, model_dim, n_heads):
super(MultiHeadAttention, self).__init__()
assert model_dim % n_heads == 0, "d_model必须能被num_heads整除"

self.model_dim = model_dim # 模型维度(如512)
self.n_heads = n_heads # 注意力头数(如8)
self.d_k = model_dim // n_heads # 每个头的维度(如64)

# 定义线性变换层(无需偏置)
self.w_q = nn.Linear(model_dim, model_dim) # 查询变换
self.w_k = nn.Linear(model_dim, model_dim) # 键变换
self.w_v = nn.Linear(model_dim, model_dim) # 值变换
self.w_o = nn.Linear(model_dim, model_dim) # 输出变换

def scaled_dot_product_attention(self, query, key, value, mask=None):
"""
计算缩放点积注意力
输入形状:
query: (batch_size, num_heads, seq_length, d_k)
key, value: 同query
输出形状: (batch_size, num_heads, seq_length, d_k)
"""
# 计算注意力分数(query和key的点积)
attention_scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(self.d_k)

# 应用掩码(如填充掩码或未来信息掩码)
if mask is not None:
attention_scores = attention_scores.masked_fill(mask == 0, -1e9)

# 计算注意力权重(softmax归一化)
attention_probs = torch.softmax(attention_scores, dim=-1)

# 对值向量加权求和
output = torch.matmul(attention_probs, value)
return output

def split_heads(self, x):
"""
将输入张量分割为多个头
输入形状: (batch_size, seq_length, model_dim)
输出形状: (batch_size, num_heads, seq_length, d_k)
"""
batch_size, seq_length, model_dim = x.size()
return x.view(batch_size, seq_length, self.n_heads, self.d_k).transpose(1, 2)

def combine_heads(self, x):
"""
将多个头的输出合并回原始形状
输入形状: (batch_size, num_heads, seq_length, d_k)
输出形状: (batch_size, seq_length, model_dim)
"""
batch_size, _, seq_length, d_k = x.size()
return x.transpose(1, 2).contiguous().view(batch_size, seq_length, self.model_dim)

def forward(self, query, key, value, mask=None):
"""
前向传播
输入形状: query/key/value: (batch_size, seq_length, model_dim)
输出形状: (batch_size, seq_length, model_dim)
"""
# 线性变换并分割多头
query = self.split_heads(self.w_q(query)) # (batch, heads, seq_len, d_k)
key = self.split_heads(self.w_k(key))
value = self.split_heads(self.w_v(value))

# 计算注意力
attention_output = self.scaled_dot_product_attention(query, key, value, mask)

# 合并多头并输出变换
output = self.w_o(self.combine_heads(attention_output))
return output

class EncoderLayer(nn.Module):
def __init__(self, model_dim, n_heads, ff_dim, dropout_rate):
super(EncoderLayer, self).__init__()
self.self_attention = MultiHeadAttention(model_dim, n_heads) # 自注意力机制
self.feed_forward = PositionwiseFeedForward(model_dim, ff_dim) # 前馈网络
self.norm1 = nn.LayerNorm(model_dim) # 层归一化
self.norm2 = nn.LayerNorm(model_dim)
self.dropout = nn.Dropout(dropout_rate) # Dropout

def forward(self, x, mask):
# 自注意力机制
attention_output = self.self_attention(x, x, x, mask)
x = self.norm1(x + self.dropout(attention_output)) # 残差连接和层归一化

# 前馈网络
ff_output = self.feed_forward(x)
x = self.norm2(x + self.dropout(ff_output)) # 残差连接和层归一化
return x

class DecoderLayer(nn.Module):
def __init__(self, model_dim, n_heads, ff_dim, dropout_rate):
super(DecoderLayer, self).__init__()
self.self_attention = MultiHeadAttention(model_dim, n_heads) # 自注意力机制
self.cross_attention = MultiHeadAttention(model_dim, n_heads) # 交叉注意力机制
self.feed_forward = PositionwiseFeedForward(model_dim, ff_dim) # 前馈网络
self.norm1 = nn.LayerNorm(model_dim) # 层归一化
self.norm2 = nn.LayerNorm(model_dim)
self.norm3 = nn.LayerNorm(model_dim)
self.dropout = nn.Dropout(dropout_rate) # Dropout

def forward(self, x, encoder_output, source_mask, target_mask):
# 自注意力机制
attention_output = self.self_attention(x, x, x, target_mask)
x = self.norm1(x + self.dropout(attention_output)) # 残差连接和层归一化

# 交叉注意力机制
cross_attention_output = self.cross_attention(x, encoder_output, encoder_output, source_mask)
x = self.norm2(x + self.dropout(cross_attention_output)) # 残差连接和层归一化

# 前馈网络
ff_output = self.feed_forward(x)
x = self.norm3(x + self.dropout(ff_output)) # 残差连接和层归一化
return x

class Transformer(nn.Module):
def __init__(self, src_vocab_size, tgt_vocab_size, model_dim, n_heads, n_layers, ff_dim, max_seq_length, dropout_rate):
super(Transformer, self).__init__()
self.encoder_embedding = nn.Embedding(src_vocab_size, model_dim) # 编码器词嵌入
self.decoder_embedding = nn.Embedding(tgt_vocab_size, model_dim) # 解码器词嵌入
self.positional_encoding = PositionalEncoding(model_dim, max_seq_length) # 位置编码

# 编码器和解码器层
self.encoder_layers = nn.ModuleList(
[EncoderLayer(model_dim, n_heads, ff_dim, dropout_rate) for _ in range(n_layers)])
self.decoder_layers = nn.ModuleList(
[DecoderLayer(model_dim, n_heads, ff_dim, dropout_rate) for _ in range(n_layers)])

self.fc = nn.Linear(model_dim, tgt_vocab_size) # 最终的全连接层
self.dropout = nn.Dropout(dropout_rate) # Dropout

@staticmethod
def generate_mask(source, target):
# 源掩码:屏蔽填充符(假设填充符索引为0)
# 形状:(batch_size, 1, 1, seq_length)
source_mask = (source != 0).unsqueeze(1).unsqueeze(2)

# 目标掩码:屏蔽填充符和未来信息
# 形状:(batch_size, 1, seq_length, 1)
target_mask = (target != 0).unsqueeze(1).unsqueeze(3)
seq_length = target.size(1)
# 生成上三角矩阵掩码,防止解码时看到未来信息
no_peak_mask = (1 – torch.triu(torch.ones(1, seq_length, seq_length), diagonal=1)).bool()
target_mask = target_mask & no_peak_mask # 合并填充掩码和未来信息掩码
return source_mask, target_mask

def forward(self, source, target):
# 生成掩码
source_mask, target_mask = self.generate_mask(source, target)

# 编码器部分
source_embedded = self.dropout(self.positional_encoding(self.encoder_embedding(source)))
encoder_output = source_embedded
for encoder_layer in self.encoder_layers:
encoder_output = encoder_layer(encoder_output, source_mask)

# 解码器部分
target_embedded = self.dropout(self.positional_encoding(self.decoder_embedding(target)))
decoder_output = target_embedded
for decoder_layer in self.decoder_layers:
decoder_output = decoder_layer(decoder_output, encoder_output, source_mask, target_mask)

# 最终输出
output = self.fc(decoder_output)
return output

# 超参数
src_vocab_size = 5000 # 源词汇表大小
tgt_vocab_size = 5000 # 目标词汇表大小
model_dim = 512 # 模型维度
n_heads = 8 # 注意力头数量
n_layers = 6 # 编码器和解码器层数
ff_dim = 2048 # 前馈网络内层维度
max_seq_length = 100 # 最大序列长度
dropout_rate = 0.1 # Dropout 概率

# 初始化模型
transformer = Transformer(src_vocab_size, tgt_vocab_size, model_dim, n_heads,
n_layers, ff_dim, max_seq_length, dropout_rate)

# 生成随机数据
src_data = torch.randint(1, src_vocab_size, (64, max_seq_length)) # 源序列
tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length)) # 目标序列

# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss(ignore_index=0) # 忽略填充部分的损失
optimizer = optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)

# 训练循环
transformer.train()
for epoch in range(10):
optimizer.zero_grad() # 清空梯度,防止累积

# 输入目标序列时去掉最后一个词(用于预测下一个词)
output = transformer(src_data, tgt_data[:, :-1])

# 计算损失时,目标序列从第二个词开始(即预测下一个词)
# output形状: (batch_size, seq_length-1, tgt_vocab_size)
# 目标形状: (batch_size, seq_length-1)
loss = criterion(
output.contiguous().view(-1, tgt_vocab_size),
tgt_data[:, 1:].contiguous().view(-1)
)

loss.backward() # 反向传播
optimizer.step() # 更新参数
print(f"Epoch: {epoch + 1}, Loss: {loss.item()}")

# 验证模式
transformer.eval()
# 生成验证数据
val_src_data = torch.randint(1, src_vocab_size, (64, max_seq_length))
val_tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length))

with torch.no_grad():
val_output = transformer(val_src_data, val_tgt_data[:, :-1])
val_loss = criterion(val_output.contiguous().view(-1, tgt_vocab_size),
val_tgt_data[:, 1:].contiguous().view(-1))
print(f"Validation Loss: {val_loss.item()}")

6.1模型性能评估结果如下:

欢迎各位彦祖与热巴畅游本人专栏与技术博客

你的三连是我最大的动力

点击➡️指向的专栏名即可闪现

➡️计算机组成原理

➡️操作系统

➡️渗透终极之红队攻击行动 

➡️动画可视化数据结构与算法

➡️ 永恒之心蓝队联纵合横防御

➡️华为高级网络工程师

➡️华为高级防火墙防御集成部署

 ➡️ 未授权访问漏洞横向渗透利用

 ➡️逆向软件破解工程

➡️MYSQL REDIS 进阶实操

➡️红帽高级工程师​

➡️红帽系统管理员

 ➡️HVV 全国各地面试题汇总

赞(0)
未经允许不得转载:网硕互联帮助中心 » 《Transformer模型PyTorch实现全攻略:架构拆解、代码示例与优化技巧》
分享到: 更多 (0)

评论 抢沙发

评论前必须登录!