Construcción de un modelo Transformer con PyTorch
Transformer es uno de los modelos más poderosos del aprendizaje automático moderno.
El modelo Transformer es una arquitectura de aprendizaje profundo basada en el mecanismo de autoatención (Self-Attention) que ha revolucionado el campo del procesamiento del lenguaje natural (NLP) y se ha convertido en la base de modelos modernos de aprendizaje profundo (como BERT, GPT, etc.).
Transformer es la arquitectura central en el ámbito del NLP moderno. Gracias a su potente capacidad para modelar dependencias a larga distancia y sus ventajas de cómputo paralelo eficiente, supera a las redes tradicionales de memoria a largo plazo (LSTM) en tareas como la traducción de idiomas y el resumen de textos.
Si aún no conoces Transformer, puedes consultar:Introducción al modelo Transformer。
Construir un modelo Transformer usando PyTorch
Los pasos para construir el modelo Transformer son los siguientes:
1. Importar las bibliotecas y módulos necesarios
Importa la biblioteca principal de PyTorch, el módulo de redes neuronales, el módulo de optimizadores, herramientas de procesamiento de datos, así como los módulos de matemáticas y copia de objetos, para proporcionar soporte en la definición de la arquitectura del modelo, la gestión de datos y el proceso de entrenamiento.
import torch import torch.nn as nn import torch.optim as optim import torch.utils.data as data import math import copy
Explicación:
torch: Biblioteca principal de PyTorch, utilizada para operaciones con tensores y diferenciación automática.torch.nn: Módulo de redes neuronales de PyTorch, que contiene diversas capas y funciones de pérdida.torch.optim: Módulo de algoritmos de optimización, como Adam, SGD, etc.math: Biblioteca de funciones matemáticas, utilizada para calcular raíces cuadradas, etc.copy: Se utiliza para copiar objetos en profundidad.
Definir los bloques básicos: atención multi-cabeza, red de avance posicional, codificación posicional
Atención multi-cabezaCalcula la relación entre cada par de posiciones en la secuencia mediante múltiples "cabezas de atención", lo que permite capturar diferentes características y patrones de la secuencia de entrada.
La clase MultiHeadAttention encapsula el mecanismo de atención multi-cabeza comúnmente utilizado en el modelo Transformer. Se encarga de dividir la entrada en múltiples cabezas de atención, aplicar atención a cada cabeza y luego combinar los resultados, de modo que el modelo pueda capturar diversas relaciones en los datos de entrada a diferentes escalas, mejorando la capacidad expresiva del modelo.
Ejemplo
def __init__(self, d_model, num_heads):
super(MultiHeadAttention, self).__init__()
assert d_model % num_heads == 0, "d_model debe ser divisible por num_heads"
self.d_model = d_model # Dimensión del modelo (por ejemplo, 512)
self.num_heads = num_heads # Número de cabezas de atención (por ejemplo, 8)
self.d_k = d_model // num_heads # Dimensión de cada cabeza (por ejemplo, 64)
# Definir capas de transformación lineal (sin sesgo)
self.W_q = nn.Linear(d_model, d_model) # Transformación de consultas
self.W_k = nn.Linear(d_model, d_model) # Transformación de claves
self.W_v = nn.Linear(d_model, d_model) # Transformación de valores
self.W_o = nn.Linear(d_model, d_model) # Transformación de salida
def scaled_dot_product_attention(self, Q, K, V, mask=None):
"""
Cálculo de la atención por producto punto escalado
Forma de entrada:
Q: (batch_size, num_heads, seq_length, d_k)
K, V: igual que Q
Forma de salida: (batch_size, num_heads, seq_length, d_k)
"""
# Calcular las puntuaciones de atención (producto punto de Q y K)
attn_scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
# Aplicar máscara (como máscara de relleno o máscara de información futura)
if mask is not None:
attn_scores = attn_scores.masked_fill(mask == 0, -1e9)
# Calcular los pesos de atención (normalización softmax)
attn_probs = torch.softmax(attn_scores, dim=-1)
# Suma ponderada de los vectores de valor
output = torch.matmul(attn_probs, V)
return output
def split_heads(self, x):
"""
Dividir el tensor de entrada en múltiples cabezas
Forma de entrada: (batch_size, seq_length, d_model)
Forma de salida: (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):
"""
Combinar las salidas de múltiples cabezas de vuelta a la forma original
Forma de entrada: (batch_size, num_heads, seq_length, d_k)
Forma de salida: (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):
"""
Propagación hacia adelante
Forma de entrada: Q/K/V: (batch_size, seq_length, d_model)
Forma de salida: (batch_size, seq_length, d_model)
"""
# Transformación lineal y división en múltiples cabezas
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))
# Calcular la atención
attn_output = self.scaled_dot_product_attention(Q, K, V, mask)
# Combinar cabezas y transformación de salida
output = self.W_o(self.combine_heads(attn_output))
return output
Explicación:
Mecanismo de atención multi-cabeza: Divide la entrada en múltiples cabezas, cada cabeza calcula la atención de forma independiente y finalmente se combinan los resultados.
Atención por producto punto escalado: Calcula el producto punto entre consultas y claves, escala, aplica softmax para calcular los pesos de atención y finalmente realiza una suma ponderada de los valores.
Máscara: Se utiliza para enmascarar posiciones no válidas (como las partes de relleno).
Red de avance posicional (Position-wise Feed-Forward Network)
Ejemplo
def __init__(self, d_model, d_ff):
super(PositionWiseFeedForward, self).__init__()
self.fc1 = nn.Linear(d_model, d_ff) # Primera capa completamente conectada
self.fc2 = nn.Linear(d_ff, d_model) # Segunda capa completamente conectada
self.relu = nn.ReLU() # Función de activación
def forward(self, x):
# Cálculo de la red de avance
return self.fc2(self.relu(self.fc1(x)))
Red de avance:Se compone de dos capas completamente conectadas y una función de activación ReLU, utilizada para procesar aún más la salida del mecanismo de atención.
Codificación posicional
La codificación posicional se utiliza para inyectar información de posición de cada token en la secuencia de entrada.
Se utilizan funciones seno y coseno de diferentes frecuencias para generar la codificación posicional.
Ejemplo
def __init__(self, d_model, max_seq_length):
super(PositionalEncoding, self).__init__()
pe = torch.zeros(max_seq_length, d_model) # Inicializar la matriz de codificación posicional
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) # Usar función seno para posiciones pares
pe[:, 1::2] = torch.cos(position * div_term) # Usar función coseno para posiciones impares
self.register_buffer('pe', pe.unsqueeze(0)) # Registrar como buffer
def forward(self, x):
# Agregar la codificación posicional a la entrada
return x + self.pe[:, :x.size(1)]
Construir el bloque del codificador (Encoder Layer)

Capa del codificador:Contiene un mecanismo de autoatención y una red de avance; cada subcapa va seguida de una conexión residual y normalización de capa.
Ejemplo
def __init__(self, d_model, num_heads, d_ff, dropout):
super(EncoderLayer, self).__init__()
self.self_attn = MultiHeadAttention(d_model, num_heads) # Mecanismo de autoatención
self.feed_forward = PositionWiseFeedForward(d_model, d_ff) # Red de avance
self.norm1 = nn.LayerNorm(d_model) # Normalización de capa
self.norm2 = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout) # Dropout
def forward(self, x, mask):
# Mecanismo de autoatención
attn_output = self.self_attn(x, x, x, mask)
x = self.norm1(x + self.dropout(attn_output)) # Conexión residual y normalización de capa
# Red de avance
ff_output = self.feed_forward(x)
x = self.norm2(x + self.dropout(ff_output)) # Conexión residual y normalización de capa
return x
Construir el módulo del decodificador

Capa del decodificador:Contiene un mecanismo de autoatención, un mecanismo de atención cruzada y una red de avance; cada subcapa va seguida de una conexión residual y normalización de capa.
Ejemplo
def __init__(self, d_model, num_heads, d_ff, dropout):
super(DecoderLayer, self).__init__()
self.self_attn = MultiHeadAttention(d_model, num_heads) # Mecanismo de autoatención
self.cross_attn = MultiHeadAttention(d_model, num_heads) # Mecanismo de atención cruzada
self.feed_forward = PositionWiseFeedForward(d_model, d_ff) # Red de avance
self.norm1 = nn.LayerNorm(d_model) # Normalización de capa
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):
# Mecanismo de autoatención
attn_output = self.self_attn(x, x, x, tgt_mask)
x = self.norm1(x + self.dropout(attn_output)) # Conexión residual y normalización de capa
# Mecanismo de atención cruzada
attn_output = self.cross_attn(x, enc_output, enc_output, src_mask)
x = self.norm2(x + self.dropout(attn_output)) # Conexión residual y normalización de capa
# Red de avance
ff_output = self.feed_forward(x)
x = self.norm3(x + self.dropout(ff_output)) # Conexión residual y normalización de capa
return x
Construir el modelo Transformer completo

Ejemplo
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) # Incrustación de palabras del codificador
self.decoder_embedding = nn.Embedding(tgt_vocab_size, d_model) # Incrustación de palabras del decodificador
self.positional_encoding = PositionalEncoding(d_model, max_seq_length) # Codificación posicional
# Capas del codificador y del decodificador
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) # Capa final completamente conectada
self.dropout = nn.Dropout(dropout) # Dropout
def generate_mask(self, src, tgt):
# Máscara de origen: enmascarar los tokens de relleno (suponiendo que el índice del token de relleno es 0)
# Forma: (batch_size, 1, 1, seq_length)
src_mask = (src != 0).unsqueeze(1).unsqueeze(2)
# Máscara de destino: enmascarar los tokens de relleno y la información futura
# Forma: (batch_size, 1, seq_length, 1)
tgt_mask = (tgt != 0).unsqueeze(1).unsqueeze(3)
seq_length = tgt.size(1)
# Generar una máscara de matriz triangular superior para evitar ver información futura durante la decodificación
nopeak_mask = (1 - torch.triu(torch.ones(1, seq_length, seq_length), diagonal=1)).bool()
tgt_mask = tgt_mask & nopeak_mask # Combinar la máscara de relleno y la máscara de información futura
return src_mask, tgt_mask
def forward(self, src, tgt):
# Generar la máscara
src_mask, tgt_mask = self.generate_mask(src, tgt)
# Parte del codificador
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)
# Parte del decodificador
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)
# Salida final
output = self.fc(dec_output)
return output
Explicación:
Modelo Transformer: Consta de las partes de codificador y decodificador, cada una compuesta por múltiples capas apiladas.
Generación de máscaras: Se utiliza para enmascarar posiciones no válidas e información futura.
Propagación hacia adelante: Pasa secuencialmente por el codificador y el decodificador, y finalmente produce la salida a través de la capa totalmente conectada.
Descripción de los parámetros de inicialización del modelo:
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概率
):
Entrenar el modelo Transformer de PyTorch
Entrenar el modelo con datos aleatorios, calcular la pérdida y actualizar los parámetros.
Ejemplo
src_vocab_size = 5000 # Tamaño del vocabulario de origen
tgt_vocab_size = 5000 # Tamaño del vocabulario de destino
d_model = 512 # Dimensión del modelo
num_heads = 8 # Número de cabezas de atención
num_layers = 6 # Número de capas del codificador y del decodificador
d_ff = 2048 # Dimensión interna de la red feed-forward
max_seq_length = 100 # Longitud máxima de la secuencia
dropout = 0.1 # Probabilidad de Dropout
# Inicializar el modelo
transformer = Transformer(src_vocab_size, tgt_vocab_size, d_model, num_heads, num_layers, d_ff, max_seq_length, dropout)
# Generar datos aleatorios
src_data = torch.randint(1, src_vocab_size, (64, max_seq_length)) # Secuencia de origen
tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length)) # Secuencia de destino
# Definir la función de pérdida y el optimizador
criterion = nn.CrossEntropyLoss(ignore_index=0) # Ignorar la pérdida de las partes de relleno (padding)
optimizer = optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)
# Bucle de entrenamiento
transformer.train()
for epoch in range(100):
optimizer.zero_grad() # Vaciar los gradientes para evitar la acumulación
# Al ingresar la secuencia de destino, eliminar la última palabra (para predecir la siguiente palabra)
output = transformer(src_data, tgt_data[:, :-1])
# Al calcular la pérdida, la secuencia de destino comienza desde la segunda palabra (es decir, predecir la siguiente palabra)
# Forma de output: (batch_size, seq_length-1, tgt_vocab_size)
# Forma del objetivo: (batch_size, seq_length-1)
loss = criterion(
output.contiguous().view(-1, tgt_vocab_size),
tgt_data[:, 1:].contiguous().view(-1)
)
loss.backward() # Retropropagación
optimizer.step() # Actualizar parámetros
print(f"Epoch: {epoch+1}, Loss: {loss.item()}")
Evaluación del modelo
Proceso de evaluación:Calcular la pérdida en los datos de validación para evaluar el rendimiento del modelo.
Ejemplo
# Generar datos de validación
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))
# Supongamos que la entrada es un lote de texto en inglés y su traducción al chino correspondiente (ya convertidos a índices)
# Datos de ejemplo:
# src_data: [[3, 14, 25, ..., 0, 0], ...] # Oraciones en inglés (0 es el símbolo de relleno)
# tgt_data: [[5, 20, 36, ..., 0, 0], ...] # Traducción al chino (0 es el símbolo de relleno)
# Nota: en aplicaciones reales, el texto debe preprocesarse con tokenización, codificación, relleno, etc.
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()}")