PyTorch Transformer-Modell erstellen

Transformer ist eines der leistungsstärksten Modelle im modernen maschinellen Lernen.

Das Transformer-Modell ist eine Deep-Learning-Architektur, die auf dem Self-Attention-Mechanismus basiert. Es hat den Bereich der Verarbeitung natürlicher Sprache (NLP) revolutioniert und ist zur Grundlage moderner Deep-Learning-Modelle wie BERT, GPT usw. geworden.

Transformer ist eine Kernarchitektur im modernen NLP. Mit seiner leistungsstarken Modellierungsfähigkeit für weitreichende Abhängigkeiten und effizienten Vorteilen bei der parallelen Berechnung übertrifft es traditionelle Long Short-Term Memory (LSTM)-Netzwerke bei Aufgaben wie Sprachübersetzung und Textzusammenfassung.

Wenn du Transformer noch nicht verstehst, kannst du nachlesen:Einführung in das Transformer-Modell。

Erstellen eines Transformer-Modells mit PyTorch

Die Schritte zum Erstellen eines Transformer-Modells sind wie folgt:

1. Importiere die notwendigen Bibliotheken und Module

Importiere die PyTorch-Kernbibliothek, das Modul für neuronale Netzwerke, das Optimierer-Modul, Datenverarbeitungswerkzeuge sowie Mathematik- und Objektkopiermodule, um Unterstützung für die Definition der Modellarchitektur, die Verwaltung von Daten und den Trainingsprozess zu bieten.

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

Beschreibung:

torch: PyTorchs Kernbibliothek, verwendet für Tensoroperationen und automatische Differenzierung.
torch.nn: PyTorchs Modul für neuronale Netzwerke, das verschiedene Schichten und Verlustfunktionen enthält.
torch.optim: Optimierungsalgorithmus-Modul, wie Adam, SGD usw.
math: Mathematikfunktionsbibliothek, verwendet zur Berechnung von Quadratwurzeln usw.
copy: Wird zum tiefen Kopieren von Objekten verwendet.

Grundlegende Bausteine definieren: Multi-Head-Attention, positionsweises Feed-Forward-Netzwerk, Positionskodierung

Multi-Head-AttentionDurch die Berechnung der Beziehung zwischen jedem Positionspaar in der Sequenz über mehrere "Attention-Heads" kann sie verschiedene Merkmale und Muster der Eingabesequenz erfassen.

Die Klasse MultiHeadAttention kapselt den in Transformer-Modellen üblicherweise verwendeten Multi-Head-Attention-Mechanismus. Sie ist dafür verantwortlich, die Eingabe in mehrere Attention-Heads aufzuteilen, auf jeden Head Attention anzuwenden und dann die Ergebnisse zu kombinieren, sodass das Modell verschiedene Beziehungen in den Eingabedaten auf unterschiedlichen Skalen erfassen und seine Ausdruckskraft verbessern kann.

Beispiel

class MultiHeadAttention(nn.Module):
def __init__(self, d_model, num_heads):
super(MultiHeadAttention, self).__init__()
assert d_model % num_heads == 0, "d_model muss durch num_heads teilbar sein"

self.d_model = d_model # Modell-Dimension (z. B. 512)
self.num_heads = num_heads # Anzahl der Attention-Heads (z. B. 8)
self.d_k = d_model // num_heads # Dimension jedes Heads (z. B. 64)

# Lineare Transformationsschichten definieren (ohne Bias)
self.W_q = nn.Linear(d_model, d_model) # Query-Transformation
self.W_k = nn.Linear(d_model, d_model) # Key-Transformation
self.W_v = nn.Linear(d_model, d_model) # Value-Transformation
self.W_o = nn.Linear(d_model, d_model) # Output-Transformation

def scaled_dot_product_attention(self, Q, K, V, mask=Keine):
"""
Skalierte Skalarprodukt-Attention berechnen
Eingabeform:
Q: (batch_size, num_heads, seq_length, d_k)
K, V: wie Q
Ausgabeform: (batch_size, num_heads, seq_length, d_k)
"""
# Attention-Scores berechnen (Skalarprodukt von Q und K)
attn_scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)

# Maske anwenden (z. B. Padding-Maske oder Maske für zukünftige Informationen)
if mask is not None:
attn_scores = attn_scores.masked_fill(mask == 0, -1e9)

# Attention-Gewichte berechnen (Softmax-Normalisierung)
attn_probs = torch.softmax(attn_scores, dim=-1)

# Gewichtete Summe der Value-Vektoren
output = torch.matmul(attn_probs, V)
return output

def split_heads(self, x):
"""
Teile den Eingabe-Tensor in mehrere Heads auf
Eingabeform: (batch_size, seq_length, d_model)
Ausgabeform: (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):
"""
Führe die Ausgaben mehrerer Heads wieder zur ursprünglichen Form zusammen
Eingabeform: (batch_size, num_heads, seq_length, d_k)
Ausgabeform: (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=Keine):
"""
Vorwärtspropagierung
Eingabeform: Q/K/V: (batch_size, seq_length, d_model)
Ausgabeform: (batch_size, seq_length, d_model)
"""
# Lineare Transformation und Aufteilung in mehrere Köpfe
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))

# Attention berechnen
attn_output = self.scaled_dot_product_attention(Q, K, V, mask)

# Köpfe zusammenführen und Ausgabetransformation anwenden
output = self.W_o(self.combine_heads(attn_output))
return output

Beschreibung:

Mehrköpfiger Aufmerksamkeitsmechanismus: Teile die Eingabe in mehrere Köpfe auf, jeder Kopf berechnet die Aufmerksamkeit unabhängig, und führe schließlich die Ergebnisse zusammen.
Skalierte Skalarprodukt-Aufmerksamkeit: Berechne das Skalarprodukt von Abfragen und Schlüsseln, skaliere es, verwende Softmax, um Aufmerksamkeitsgewichte zu berechnen, und bilde schließlich eine gewichtete Summe der Werte.
Maske: Wird verwendet, um ungültige Positionen zu maskieren (z. B. aufgefüllte Teile).

Positionsweises Feed-Forward-Netzwerk

Beispiel

class PositionWiseFeedForward(nn.Module):
def __init__(self, d_model, d_ff):
super(PositionWiseFeedForward, self).__init__()
self.fc1 = nn.Linear(d_model, d_ff) # Erste vollständig verbundene Schicht
self.fc2 = nn.Linear(d_ff, d_model) # Zweite vollständig verbundene Schicht
self.relu = nn.ReLU() # Aktivierungsfunktion

def forward(self, x):
# Berechnung des Feed-Forward-Netzwerks
return self.fc2(self.relu(self.fc1(x)))

Feed-Forward-Netzwerk:Es besteht aus zwei vollständig verbundenen Schichten und einer ReLU-Aktivierungsfunktion, die verwendet wird, um die Ausgabe des Aufmerksamkeitsmechanismus weiterzuverarbeiten.

Positionale Codierung

Positionskodierung wird verwendet, um Positionsinformationen jedes Tokens in der Eingabesequenz einzufügen.

Verwenden Sie Sinus- und Kosinusfunktionen unterschiedlicher Frequenzen, um Positionskodierungen zu erzeugen.

Beispiel

class PositionalEncoding(nn.Module):
def __init__(self, d_model, max_seq_length):
super(PositionalEncoding, self).__init__()
pe = torch.zeros(max_seq_length, d_model) # Initialisiere die Positionskodierungsmatrix
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) # Sinusfunktion für gerade Positionen verwenden
pe[:, 1::2] = torch.cos(position * div_term) # Kosinusfunktion für ungerade Positionen verwenden
self.register_buffer('pe', pe.unsqueeze(0)) # Als Puffer registrieren

def forward(self, x):
# Positionskodierung zur Eingabe hinzufügen
return x + self.pe[:, :x.size(1)]

Encoder-Block erstellen (Encoder-Schicht)

Encoder-Schicht:Enthält einen Self-Attention-Mechanismus und ein Feed-Forward-Netzwerk, wobei auf jede Unterschicht Residualverbindungen und Layer-Normalisierung folgen.

Beispiel

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-Attention-Mechanismus
self.feed_forward = PositionWiseFeedForward(d_model, d_ff) # Feed-Forward-Netzwerk
self.norm1 = nn.LayerNorm(d_model) # Layer-Normalisierung
self.norm2 = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout) # Dropout

def forward(self, x, mask):
# Self-Attention-Mechanismus
attn_output = self.self_attn(x, x, x, mask)
x = self.norm1(x + self.dropout(attn_output)) # Residualverbindung und Layer-Normalisierung

# Feed-Forward-Netzwerk
ff_output = self.feed_forward(x)
x = self.norm2(x + self.dropout(ff_output)) # Restverbindung und Layer-Normalisierung
return x

Decoder-Modul erstellen

Decoder-Schicht:Enthält einen Self-Attention-Mechanismus, einen Cross-Attention-Mechanismus und ein Feed-Forward-Netzwerk, wobei auf jede Unterschicht Restverbindungen und Layer-Normalisierung folgen.

Beispiel

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-Attention-Mechanismus
self.cross_attn = MultiHeadAttention(d_model, num_heads) # Cross-Attention-Mechanismus
self.feed_forward = PositionWiseFeedForward(d_model, d_ff) # Feed-Forward-Netzwerk
self.norm1 = nn.LayerNorm(d_model) # Layer-Normalisierung
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):
# Self-Attention-Mechanismus
attn_output = self.self_attn(x, x, x, tgt_mask)
x = self.norm1(x + self.dropout(attn_output)) # Residualverbindung und Layer-Normalisierung

# Cross-Attention-Mechanismus
attn_output = self.cross_attn(x, enc_output, enc_output, src_mask)
x = self.norm2(x + self.dropout(attn_output)) # Residualverbindung und Layer-Normalisierung

# Feed-Forward-Netzwerk
ff_output = self.feed_forward(x)
x = self.norm3(x + self.dropout(ff_output)) # Residualverbindung und Layer-Normalisierung
return x

Erstelle das vollständige Transformer-Modell

Beispiel

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) # Encoder-Worteinbettung
self.decoder_embedding = nn.Embedding(tgt_vocab_size, d_model) # Decoder-Worteinbettung
self.positional_encoding = PositionalEncoding(d_model, max_seq_length) # Positionskodierung

# Encoder- und Decoder-Schichten
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) # Abschließende vollständig verbundene Schicht
self.dropout = nn.Dropout(dropout) # Dropout

def generate_mask(self, src, tgt):
# Quellmaske: Padding-Tokens maskieren (Padding-Index ist 0)
# Form: (batch_size, 1, 1, seq_length)
src_mask = (src != 0).unsqueeze(1).unsqueeze(2)

# Zielmaske: Padding-Tokens und zukünftige Informationen maskieren
# Form: (batch_size, 1, seq_length, 1)
tgt_mask = (tgt != 0).unsqueeze(1).unsqueeze(3)
seq_length = tgt.size(1)
# Erzeuge eine obere Dreiecksmatrix-Maske, um zu verhindern, dass während der Dekodierung zukünftige Informationen gesehen werden
nopeak_mask = (1 - torch.triu(torch.ones(1, seq_length, seq_length), diagonal=1)).bool()
tgt_mask = tgt_mask & nopeak_mask # Kombiniere die Padding-Maske und die Maske für zukünftige Informationen
return src_mask, tgt_mask

def forward(self, src, tgt):
# Masken generieren
src_mask, tgt_mask = self.generate_mask(src, tgt)

# Encoder-Teil
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)

# Decoder-Teil
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)

# Endausgabe
output = self.fc(dec_output)
return output

Beschreibung:

Transformer-ModellEnthält Encoder- und Decoder-Teile, die jeweils aus mehreren gestapelten Schichten zusammengesetzt sind.
MaskengenerierungWird verwendet, um ungültige Positionen und zukünftige Informationen zu maskieren.
VorwärtspropagationDurchläuft nacheinander den Encoder und Decoder und gibt schließlich über eine vollständig verbundene Schicht aus.

Beschreibung der Modellinitialisierungsparameter:

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概率
    ):

Das PyTorch-Transformer-Modell trainieren

Zufallsdaten zum Trainieren des Modells verwenden, den Verlust berechnen und Parameter aktualisieren.

Beispiel

# Hyperparameter
src_vocab_size = 5000 # Größe des Quellvokabulars
tgt_vocab_size = 5000 # Größe des Zielvokabulars
d_model = 512 # Modell-Dimension
num_heads = 8 # Anzahl der Attention-Heads
num_layers = 6 # Anzahl der Encoder- und Decoder-Schichten
d_ff = 2048 # Innere Dimension des Feed-Forward-Netzwerks
max_seq_length = 100 # Maximale Sequenzlänge
dropout = 0.1 # Dropout-Wahrscheinlichkeit

# Modell initialisieren
transformer = Transformer(src_vocab_size, tgt_vocab_size, d_model, num_heads, num_layers, d_ff, max_seq_length, dropout)

# Zufällige Daten generieren
src_data = torch.randint(1, src_vocab_size, (64, max_seq_length)) # Quellsequenz
tgt_data = torch.randint(1, tgt_vocab_size, (64, max_seq_length)) # Zielsequenz

# Verlustfunktion und Optimierer definieren
criterion = nn.CrossEntropyLoss(ignore_index=0) # Ignoriere den Verlust des Padding-Teils
optimizer = optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)

# Trainingsschleife
transformer.train()
for epoch in range(100):
optimizer.zero_grad() # Gradienten zurücksetzen, um Akkumulation zu verhindern

# Entferne das letzte Wort bei der Eingabe der Zielsequenz (wird verwendet, um das nächste Wort vorherzusagen)
output = transformer(src_data, tgt_data[:, :-1])

# Bei der Berechnung des Verlusts beginnt die Zielsequenz beim zweiten Wort (d. h. das nächste Wort vorhersagen)
# Ausgabeform: (batch_size, seq_length-1, tgt_vocab_size)
# Zielform: (batch_size, seq_length-1)
loss = criterion(
output.contiguous().view(-1, tgt_vocab_size),
tgt_data[:, 1:].contiguous().view(-1)
)

loss.backward() # Backpropagation
optimizer.step() # Parameter aktualisieren
print(f"Epoche: {epoch+1}, Verlust: {loss.item()}")

Modellbewertung

Evaluierungsprozess:Berechnen Sie den Verlust auf den Validierungsdaten, um die Modellleistung zu bewerten.

Beispiel

transformer.eval()
# Validierungsdaten generieren
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))
# Angenommen, die Eingabe ist ein Batch englischer Sätze und ihrer entsprechenden chinesischen Übersetzungen (in Indizes umgewandelt)
# Beispieldaten:
# src_data: [[3, 14, 25, ..., 0, 0], ...] # Englische Sätze (0 ist das Padding-Token)
# tgt_data: [[5, 20, 36, ..., 0, 0], ...] # Chinesische Übersetzungen (0 ist das Padding-Token)
# Hinweis: In praktischen Anwendungen muss Text vorverarbeitet werden, z. B. Tokenisierung, Kodierung und Padding
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().Ansicht(-1))
print(f"Validierungsverlust: {val_loss.item()}")

Weitere Erweiterungen