Import Datenklassen
Import Datensätze
Import Fackel
Import Fackel.nn als nn
Import tqdm
@Datenklassen.Datenklasse
Klasse BertConfig:
„“„Konfiguration für BERT-Modell.“„“
vocab_size: int = 30522
num_layers: int = 12
versteckte_Größe: int = 768
Anzahl_Köpfe: int = 12
dropout_prob: schweben = 0,1
pad_id: int = 0
max_seq_len: int = 512
Anzahl_Typen: int = 2
Klasse BertBlock(nn.Modul):
„“„Ein Transformatorblock in BERT.“„“
def __init__(selbst, versteckte_Größe: int, Anzahl_Köpfe: int, dropout_prob: schweben):
tremendous().__init__()
selbst.Aufmerksamkeit = nn.MultiheadAchtung(versteckte_Größe, Anzahl_Köpfe,
ausfallen=dropout_prob, batch_first=WAHR)
selbst.attn_norm = nn.LayerNorm(versteckte_Größe)
selbst.ff_norm = nn.LayerNorm(versteckte_Größe)
selbst.ausfallen = nn.Ausfallen(dropout_prob)
selbst.Feed_Forward = nn.Sequentiell(
nn.Linear(versteckte_Größe, 4 * versteckte_Größe),
nn.GELU(),
nn.Linear(4 * versteckte_Größe, versteckte_Größe),
)
def nach vorne(selbst, X: Fackel.Tensor, pad_mask: Fackel.Tensor) -> Fackel.Tensor:
#Selbstaufmerksamkeit mit Polstermaske und Postnorm
attn_output, _ = selbst.Aufmerksamkeit(X, X, X, key_padding_mask=pad_mask)
X = selbst.attn_norm(X + attn_output)
# Feed-Ahead mit GeLU-Aktivierung und Publish-Norm
ff_output = selbst.Feed_Forward(X)
X = selbst.ff_norm(X + selbst.ausfallen(ff_output))
zurückkehren X
Klasse BertPooler(nn.Modul):
„“„Pooler-Schicht für BERT zur Verarbeitung der (CLS-)Token-Ausgabe.“„“
def __init__(selbst, versteckte_Größe: int):
tremendous().__init__()
selbst.dicht = nn.Linear(versteckte_Größe, versteckte_Größe)
selbst.Aktivierung = nn.Tanh()
def nach vorne(selbst, X: Fackel.Tensor) -> Fackel.Tensor:
X = selbst.dicht(X)
X = selbst.Aktivierung(X)
zurückkehren X
Klasse BertModel(nn.Modul):
„“„Rückgrat des BERT-Modells.“„“
def __init__(selbst, config: BertConfig):
tremendous().__init__()
# Ebenen einbetten
selbst.word_embeddings = nn.Einbetten(config.vocab_size, config.versteckte_Größe,
padding_idx=config.pad_id)
selbst.type_embeddings = nn.Einbetten(config.Anzahl_Typen, config.versteckte_Größe)
selbst.position_embeddings = nn.Einbetten(config.max_seq_len, config.versteckte_Größe)
selbst.Einbettungen_norm = nn.LayerNorm(config.versteckte_Größe)
selbst.einbettungen_dropout = nn.Ausfallen(config.dropout_prob)
# Transformatorblöcke
selbst.Blöcke = nn.Modulliste((
BertBlock(config.versteckte_Größe, config.Anzahl_Köpfe, config.dropout_prob)
für _ In Reichweite(config.num_layers)
))
# (CLS) Pooler-Schicht
selbst.pooler = BertPooler(config.versteckte_Größe)
def nach vorne(selbst, input_ids: Fackel.Tensor, token_type_ids: Fackel.Tensor, pad_id: int = 0
) -> Tupel(Fackel.Tensor, Fackel.Tensor):
# Aufmerksamkeitsmaske zum Auffüllen von Token erstellen
pad_mask = input_ids == Unterlage_Ausweis
# Ganzzahl-Token in Einbettungsvektoren umwandeln
Batch_Größe, seq_len = input_ids.Kind
position_ids = Fackel.arrangieren(seq_len, Gerät=input_ids.Gerät).ausdrücken(0)
position_embeddings = selbst.position_embeddings(position_ids)
type_embeddings = selbst.type_embeddings(token_type_ids)
token_embeddings = selbst.word_embeddings(input_ids)
X = token_embeddings + type_embeddings + Place_Einbettungen
X = selbst.Einbettungen_norm(X)
X = selbst.einbettungen_dropout(X)
# Verarbeiten Sie die Sequenz mit Transformatorblöcken
für Block In selbst.Blöcke:
X = Block(X, pad_mask)
# Poolen Sie den verborgenen Standing des „(CLS)“-Tokens
pooled_output = selbst.pooler(X(:, 0, :))
zurückkehren X, pooled_output
Klasse BertPretrainingModel(nn.Modul):
def __init__(selbst, config: BertConfig):
tremendous().__init__()
selbst.Bert = BertModel(config)
selbst.mlm_head = nn.Sequentiell(
nn.Linear(config.versteckte_Größe, config.versteckte_Größe),
nn.GELU(),
nn.LayerNorm(config.versteckte_Größe),
nn.Linear(config.versteckte_Größe, config.vocab_size),
)
selbst.nsp_head = nn.Linear(config.versteckte_Größe, 2)
def nach vorne(selbst, input_ids: Fackel.Tensor, token_type_ids: Fackel.Tensor, pad_id: int = 0
) -> Tupel(Fackel.Tensor, Fackel.Tensor):
# Verarbeiten Sie die Sequenz mit dem BERT-Modell-Spine
X, pooled_output = selbst.Bert(input_ids, token_type_ids, pad_id)
# Sagen Sie die maskierten Token für die MLM-Aufgabe und die Klassifizierung für die NSP-Aufgabe voraus
mlm_logits = selbst.mlm_head(X)
nsp_logits = selbst.nsp_head(pooled_output)
zurückkehren mlm_logits, nsp_Protokolle
# Trainingsparameter
Epochen = 10
Lernrate = 1e–4
Batch_Größe = 32
# Datensatz laden und Dataloader einrichten
Datensatz = Datensätze.Datensatz.from_parquet(„wikitext-2_train_data.parquet“)
def collate_fn(Cost: Liste(dict)):
„“„Benutzerdefinierte Sortierfunktion zur Verarbeitung von Sequenzen variabler Länge im Datensatz.“„“
# immer mit maximaler Länge: tokens, segment_ids; immer Singleton: is_random_next
input_ids = Fackel.Tensor((Artikel(„Token“) für Artikel In Cost))
token_type_ids = Fackel.Tensor((Artikel(„segment_ids“) für Artikel In Cost)).Bauchmuskeln()
is_random_next = Fackel.Tensor((Artikel(„is_random_next“) für Artikel In Cost)).Zu(int)
# variable Länge: masked_positions, masked_labels
maskierte_pos = ((idx, Pos) für idx, Artikel In aufzählen(Cost) für Pos In Artikel(„maskierte_positionen“))
maskierte_labels = Fackel.Tensor((Etikett für Artikel In Cost für Etikett In Artikel(„masked_labels“)))
zurückkehren input_ids, token_type_ids, is_random_next, maskierte_pos, maskierte_labels
Datenlader = Fackel.Dienstprogramme.Daten.Datenlader(Datensatz, Batch_Größe=Batch_Größe, Shuffle=WAHR,
collate_fn=collate_fn, num_workers=8)
# Trainiere das Modell
Gerät = Fackel.Gerät(„Cuda“ Wenn Fackel.cuda.ist_verfügbar() anders „CPU“)
Modell = BertPretrainingModel(BertConfig()).Zu(Gerät)
Modell.Zug()
Optimierer = Fackel.optimum.AdamW(Modell.Parameter(), lr=Lernrate)
Planer = Fackel.optimum.lr_scheduler.StepLR(Optimierer, Schrittgröße=1, Gamma=0,1)
loss_fn = nn.CrossEntropyLoss()
für Epoche In Reichweite(Epochen):
pbar = tqdm.tqdm(Datenlader, absteigend=F„Epoche {epoch+1}/{epochs}“)
für Cost In pbar:
# Batch-Daten abrufen
input_ids, token_type_ids, is_random_next, maskierte_pos, maskierte_labels = Cost
input_ids = input_ids.Zu(Gerät)
token_type_ids = token_type_ids.Zu(Gerät)
is_random_next = is_random_next.Zu(Gerät)
maskierte_labels = maskierte_labels.Zu(Gerät)
# Ausgabe aus dem Modell extrahieren
mlm_logits, nsp_logits = Modell(input_ids, token_type_ids)
# MLM-Verlust: masked_positions ist eine Liste von Tupeln von (B, S), extrahieren Sie sie
# entsprechende Logits vom Tensor mlm_logits der Kind (B, S, V)
Batch_Indizes, token_positions = Reißverschluss(*maskierte_pos)
mlm_logits = mlm_logits(Batch_Indizes, token_positions)
mlm_loss = loss_fn(mlm_logits, maskierte_labels)
# Berechnen Sie den Verlust für die NSP-Aufgabe
nsp_loss = loss_fn(nsp_logits, is_random_next)
# rückwärts mit Totalschaden
total_loss = mlm_loss + nsp_loss
pbar.set_postfix(MLM=mlm_loss.Artikel(), NSP=nsp_loss.Artikel(), Gesamt=total_loss.Artikel())
Optimierer.null_grad()
total_loss.rückwärts()
Optimierer.Schritt()
Planer.Schritt()
pbar.aktualisieren(1)
pbar.schließen()
# Speichern Sie das Modell
Fackel.speichern(Modell.state_dict(), „bert_pretraining_model.pth“)
Fackel.speichern(Modell.Bert.state_dict(), „bert_model.pth“)