Tout le monde a entendu parler des moteurs Google Translation et DeepL qui vous permettent de traduire un texte d’une langue à une autre sans aucune intervention humaine et vous vous demandez probablement comment ils le font.
Les deux systèmes qui ont connu le succès au fil des ans sont :
Traduction automatique statistique (TAS)
Ces systèmes n’utilisent aucune règle linguistique pour effectuer la traduction. Ils traduisent plutôt à l’aide de modèles statistiques construits automatiquement à partir des corpora. Le logiciel de traduction automatique analyse une base de données différente pour chaque langue, ce qui permet de produire des traductions relativement fluides mais pas toujours très logiques.
Traduction automatique neuronale (TAN)
Ces systèmes utilisent des réseaux neuronaux pour prédire la distribution de probabilité des mots dans la séquence. Ils permettent d’obtenir des traductions de meilleure qualité qui sont à la pointe de la technologie. La majorité des modèles sont basés sur le paradigme Seq2Seq où un encodeur résume les informations contenues dans la phrase source dans un vecteur caché transmis au décodeur pour générer des prédictions.
Dans cet article, nous parlerons du modèle Seq2Seq en utilisant les RNNs et particulièrement l’architecture GRU.
Alors comment fonctionne une architecture Seq2Seq ?
Chaque mot de la phrase d’entrée est encodé dans des états cachés par l’intermédiaire d’une cellule neuronale récurrente dans notre cas, puis cette information est exploitée pour générer un vecteur “contexte” caché unique utilisé comme entrée pour le décodeur qui reçoit également la phrase cible. Le but du décodeur est de générer un vecteur caché nous permettant à travers une couche softmax de prédire la séquence correcte des mots.
La partie ci-dessous est destinée aux lecteurs avancés qui sont à l’aise avec Python et particulièrement PyTorch.
Dans ce tutoriel, nous nous concentrons sur la traduction allemand-anglais. Le langage utilisé est Python et la bibliothèque d’apprentissage profond est Pytorch. Il s’inspire du tutoriel Pytorch Seq2Seq sur GitHub.
Préparation des données
Tout d’abord, nous importons les modules nécessaires.
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torchtext.datasets import TranslationDataset, Multi30k
from torchtext.data import Field, BucketIterator
import spacy
import random
import math
import time
Tout le traitement de texte se fera à l’aide de spaCy et Torchtext. Les objets Field permettent de stocker toutes les opérations de prétraitement et de post-traitement prévues et sont utilisés pour construire les objets du jeu de données utilisés par les Iterators.
Nous chargeons les modèles spaCy allemand et anglais.
spacy_de = spacy.load('de')
spacy_en = spacy.load('en')
Nous créons les tokenizers et les objets Field.
def tokenize_de(text):
"""
Tokenizes German text from a string into a list of strings
"""
return [tok.text for tok in spacy_de.tokenizer(text)]
def tokenize_en(text):
"""
Tokenizes English text from a string into a list of strings
"""
return [tok.text for tok in spacy_en.tokenizer(text)]
SRC = Field(tokenize = tokenize_de,
init_token = '<sos>',
eos_token = '<eos>',
lower = True)
TRG = Field(tokenize = tokenize_en,
init_token = '<sos>',
eos_token = '<eos>',
lower = True)
Nous chargeons l’ensemble de données Multi30k et utilisons l’ensemble de données d’apprentissage pour construire le vocabulaire.
train_data, valid_data, test_data = Multi30k.splits(exts = ('.de', '.en'),
fields = (SRC, TRG))
SRC.build_vocab(train_data, min_freq = 2)
TRG.build_vocab(train_data, min_freq = 2)
Ensuite, nous définissons le dispositif (CPU/GPU) et créons les itérateurs qui seront utilisés dans la boucle d’apprentissage.
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
BATCH_SIZE = 128
train_iterator, valid_iterator, test_iterator = BucketIterator.splits(
(train_data, valid_data, test_data),
batch_size = BATCH_SIZE,
device = device)
Le modèle Seq2Seq
Dans notre architecture, l’encodeur est un GRU bidirectionnel (Gated Recurrent Unit) tandis que le décodeur est un simple GRU unidirectionnel. L’attention est incorporée dans le modèle pour réduire la perte d’information : la phrase source encodée est partagée avec le décodeur.
L’encodeur
Ici nous initialisons les couches de l’encodeur et les différents paramètres. Nous devons spécifier la dimension d’entrée, la dimension d’incorporation (“Embedding”) pour transformer les mots en vecteurs, la dimension cachée de l’encodeur, la dimension cachée du décodeur car l’entrée du décodeur est la sortie de l’encodeur et éventuellement le taux de dropout.
class Encoder(nn.Module):
def __init__(self, input_dim, emb_dim, enc_hid_dim, dec_hid_dim, dropout):
super().__init__()
self.input_dim = input_dim
self.emb_dim = emb_dim
self.enc_hid_dim = enc_hid_dim
self.dec_hid_dim = dec_hid_dim
self.dropout = dropout
self.embedding = nn.Embedding(input_dim, emb_dim)
self.rnn = nn.GRU(emb_dim, enc_hid_dim, bidirectional = True)
self.fc = nn.Linear(enc_hid_dim * 2, dec_hid_dim)
self.dropout = nn.Dropout(dropout)
Nous implémentons maintenant la fonction forward qui sera appelée lors de l’utilisation de l’encodeur.
def forward(self, src):
embedded = self.dropout(self.embedding(src))
outputs, hidden = self.rnn(embedded)
#initial decoder hidden is final hidden state of the forwards and backwards
# encoder RNNs fed through a linear layer
hidden = torch.tanh(self.fc(torch.cat((hidden[-2,:,:], hidden[-1,:,:]), dim = 1)))
return outputs, hidden
La couche suivante est la couche d’attention. Ceci prendra en compte l’état caché précédent du décodeur, et tous les états cachés avant et arrière empilés de l’encodeur. Cette couche produira un vecteur d’attention qui est la longueur de la phrase source, chaque élément est compris entre 0 et 1 et le vecteur entier somme à 1.
Nous devons initialiser la couche d’attention qui est un réseau entièrement connecté recevant les sorties de l’encodeur concaténées à l’état caché précédent du décodeur et sortant un vecteur de la bonne dimension.
class Attention(nn.Module):
def __init__(self, enc_hid_dim, dec_hid_dim):
super().__init__()
self.enc_hid_dim = enc_hid_dim
self.dec_hid_dim = dec_hid_dim
self.attn = nn.Linear((enc_hid_dim * 2) + dec_hid_dim, dec_hid_dim)
self.v = nn.Parameter(torch.rand(dec_hid_dim))
Maintenant, nous implémentons la fonction forward pour produire un vecteur qui représente les mots de la phrase source auxquels nous devons prêter le plus d’attention afin de prédire correctement le mot suivant à décoder.
def forward(self, hidden, encoder_outputs):
#hidden = [batch size, dec hid dim]
#encoder_outputs = [src sent len, batch size, enc hid dim * 2]
batch_size = encoder_outputs.shape[1]
src_len = encoder_outputs.shape[0]
#repeat encoder hidden state src_len times
hidden = hidden.unsqueeze(1).repeat(1, src_len, 1)
encoder_outputs = encoder_outputs.permute(1, 0, 2)
#hidden = [batch size, src sent len, dec hid dim]
#encoder_outputs = [batch size, src sent len, enc hid dim * 2]
energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim = 2)))
#energy = [batch size, src sent len, dec hid dim]
energy = energy.permute(0, 2, 1)
#energy = [batch size, dec hid dim, src sent len]
#v = [dec hid dim]
v = self.v.repeat(batch_size, 1).unsqueeze(1)
#v = [batch size, 1, dec hid dim]
attention = torch.bmm(v, energy).squeeze(1)
#attention= [batch size, src len]
return F.softmax(attention, dim=1)
Le décodeur reçoit à chaque étape un mot (à T = 0, c’est le token), le vecteur attention et l’état caché précédent.
Nous initialisons les paramètres multiples, en particulier la couche GRU et la dernière couche produisant un vecteur de la forme output_dim. Chaque valeur de ce vecteur est associée à un mot du vocabulaire.
class Decoder(nn.Module):
def __init__(self, output_dim, emb_dim, enc_hid_dim, dec_hid_dim, dropout, attention):
super().__init__()
self.emb_dim = emb_dim
self.enc_hid_dim = enc_hid_dim
self.dec_hid_dim = dec_hid_dim
self.output_dim = output_dim
self.dropout = dropout
self.attention = attention
self.embedding = nn.Embedding(output_dim, emb_dim)
self.rnn = nn.GRU((enc_hid_dim * 2) + emb_dim, dec_hid_dim)
self.out = nn.Linear((enc_hid_dim * 2) + dec_hid_dim + emb_dim, output_dim)
self.dropout = nn.Dropout(dropout)
Maintenant la fonction forward.
def forward(self, input, hidden, encoder_outputs):
#input = [batch size]
#hidden = [batch size, dec hid dim]
#encoder_outputs = [src sent len, batch size, enc hid dim * 2]
input = input.unsqueeze(0)
embedded = self.dropout(self.embedding(input))
a = self.attention(hidden, encoder_outputs)
a = a.unsqueeze(1)
encoder_outputs = encoder_outputs.permute(1, 0, 2)
weighted = torch.bmm(a, encoder_outputs)
weighted = weighted.permute(1, 0, 2)
rnn_input = torch.cat((embedded, weighted), dim = 2)
output, hidden = self.rnn(rnn_input, hidden.unsqueeze(0))
embedded = embedded.squeeze(0)
output = output.squeeze(0)
weighted = weighted.squeeze(0)
output = self.out(torch.cat((output, weighted, embedded), dim = 1))
#output = [bsz, output dim]
return output, hidden.squeeze(0)
Maintenant, nous devons faire fonctionner toutes ces briques ensemble.
class Seq2Seq(nn.Module):
def __init__(self, encoder, decoder, device):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.device = device
def forward(self, src, trg, teacher_forcing_ratio = 0.5):
#src = [src sent len, batch size]
#trg = [trg sent len, batch size]
#teacher_forcing_ratio is probability to use teacher forcing
#e.g. if teacher_forcing_ratio is 0.75 we use teacher forcing 75% of the time
batch_size = src.shape[1]
max_len = trg.shape[0]
trg_vocab_size = self.decoder.output_dim
#tensor to store decoder outputs
outputs = torch.zeros(max_len, batch_size, trg_vocab_size).to(self.device)
#encoder_outputs is all hidden states of the input sequence, back and forwards
#hidden is the final forward and backward hidden states, passed through a linear layer
encoder_outputs, hidden = self.encoder(src)
#first input to the decoder is the <sos> tokens
output = trg[0,:]
for t in range(1, max_len):
output, hidden = self.decoder(output, hidden, encoder_outputs)
outputs[t] = output
teacher_force = random.random() < teacher_forcing_ratio
top1 = output.max(1)[1]
output = (trg[t] if teacher_force else top1)
return outputs
Vous remarquerez que nous avons introduit la notion de “teacher forcing” ci-dessus. Lorsqu’il est utilisé, le mot prédit précédent n’est plus une entrée dans le décodeur, il est remplacé par le mot réel.
Nous initialisons nos paramètres, l’encodeur, le décodeur et le modèle seq2seq.
INPUT_DIM = len(SRC.vocab)
OUTPUT_DIM = len(TRG.vocab)
ENC_EMB_DIM = 256
DEC_EMB_DIM = 256
ENC_HID_DIM = 512
DEC_HID_DIM = 512
ENC_DROPOUT = 0.5
DEC_DROPOUT = 0.5
attn = Attention(ENC_HID_DIM, DEC_HID_DIM)
enc = Encoder(INPUT_DIM, ENC_EMB_DIM, ENC_HID_DIM, DEC_HID_DIM, ENC_DROPOUT)
dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, ENC_HID_DIM, DEC_HID_DIM, DEC_DROPOUT, attn)
model = Seq2Seq(enc, dec, device).to(device)
optimizer = optim.Adam(model.parameters())
PAD_IDX = TRG.vocab.stoi['<pad>']
criterion = nn.CrossEntropyLoss(ignore_index = PAD_IDX)
Nous implémentations notre boucle d’apprentissage.
def train(model, iterator, optimizer, criterion, clip):
model.train()
epoch_loss = 0
for i, batch in enumerate(iterator):
src = batch.src
trg = batch.trg
optimizer.zero_grad()
output = model(src, trg)
#trg = [trg sent len, batch size]
#output = [trg sent len, batch size, output dim]
output = output[1:].view(-1, output.shape[-1])
trg = trg[1:].view(-1)
#trg = [(trg sent len - 1) * batch size]
#output = [(trg sent len - 1) * batch size, output dim]
loss = criterion(output, trg)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), clip)
optimizer.step()
epoch_loss += loss.item()
return epoch_loss / len(iterator)
Ensuite, nous spécifions le nombre d’épochs et nous lançons l’apprentissage.
N_EPOCHS = 10
CLIP = 1
best_valid_loss = float('inf')
for epoch in range(N_EPOCHS):
start_time = time.time()
train_loss = train(model, train_iterator, optimizer, criterion, CLIP)
end_time = time.time()
epoch_mins, epoch_secs = epoch_time(start_time, end_time)
print(f'Epoch: {epoch+1:02} | Time: {epoch_mins}m {epoch_secs}s')
print(f'\tTrain Loss: {train_loss:.3f} | Train PPL: {math.exp(train_loss):7.3f}')
J’espère que ce tutoriel a été utile. N’hésitez pas à me faire part de vos remarques.
