Nous allons entraîner un étiqueteur de séquences à prédire les groupes nominaux représentant des entités visuelles dans une phrase. Les données sont extraites du corpus Flicker30k entities et le problème est formalisé comme la prédiction d'une étiquette par mot indiquant si une entité commence sur le mot (B), si elle continue (I) ou si on est à l'exterieur d'une entité. Le type de modèle que nous allons concevoir pourrait être appliqué à un taggeur en parties de discours.
Nous allons entraîner un RNN à faire ces prédictions et les embeddings de mots seront initialisés avec des embeddings préentrainés avec fasttext. Nous en profiterons pour créer un RNN moins dépendant de la taille des séquences et capable de traiter en test des séquences plus longues que celles qu'il a vu en entraînement.
Les données sont sous la forme d'un mot par ligne suivi de son étiquette. Les phrases sont séparées par des lignes vides.
%%bash
if [ ! -f f30kE-captions-bio.txt ]; then
wget -q https://raw.githubusercontent.com/benob/dl4nlp-tutorials-data/master/f30kE-captions-bio.txt
fi
head -20 f30kE-captions-bio.txt
La première chose à faire est charger les données. Un premier tableau contiendra les mots de chaque phrase, et un second les étiquettes associées.
texts = []
labels = []
with open('f30kE-captions-bio.txt') as fp:
text = []
label = []
for line in fp:
tokens = line.strip().split()
if len(tokens) == 0:
texts.append(text)
labels.append(label)
text = []
label = []
else:
text.append(tokens[0])
label.append(tokens[1])
print(len(texts))
print(texts[12])
print(labels[12])
Il faut ensuite convertir les étiquettes et les mots en entiers.
Nous allons devoir créer des séquences de taille fixe, donc il faut réserver une étiquette de padding, ici <eos>.
Toutefois, pour l'instant nous nous contentons de convertir les étiquettes sans appliquer le padding.
Un defaultdict est utilisé pour créer le vocabulaire des étiquettes. À chaque fois qu'une nouvelle étiquette est rencontrée, il l'associe à une nouvelle valeur.
import collections
label_vocab = collections.defaultdict(lambda: len(label_vocab))
label_vocab['<eos>'] = 0
int_labels = []
for label in labels:
int_labels.append([label_vocab[token] for token in label])
print(int_labels[12])
print(label_vocab)
Il en va de même pour les textes à qui nous allons dédier un vocabulaire différent de celui des étiquettes. Les embeddings préentraînés seront chargés à partir de ce vocabulaire. Il faut veiller à mettre les mots en minuscules car le vocabulaire des embeddings fasttext est en minuscules.
vocab = collections.defaultdict(lambda: len(vocab))
vocab['<eos>'] = 0
int_texts = []
for text in texts:
int_texts.append([vocab[token.lower()] for token in text])
print(int_texts[12])
len(vocab)
Nous pouvons maintenant créer des vocabulaires inversés pour pouvoir revenir vers les étiquettes et mots. Celà et utile pour vérifier le contenu d'un tenseur, et surtout pour convertir les décisions du système en étiquettes textuelles.
rev_label_vocab = {y: x for x, y in label_vocab.items()}
rev_vocab = {y: x for x, y in vocab.items()}
Importons les modules habituels de pytorch. Les constantes suivantes sont définies :
max_len est la longueur maximale d'une phrase en apprentissagebatch_size est la taille des batchesembed_size est la taille des embeddings préentraînés (nous utiliserons les embeddings téléchargeables sur le site de fasttext qui sont de taille 300.hidden_size est la taille de l'état caché du RNNimport torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
max_len = 16
batch_size = 64
embed_size = 300
hidden_size = 128
Il faut maintenant créer des tenseurs pytorch avec les données phrases et les étiquettes associées. X et Y sont des tenseurs d'entiers de taille (le nombre d'exemples du corpus, la longueur maximale d'une phrase). Ils sont initialisés à zéro car c'est la valeur de padding.
Pour chaque phrase et séquence d'étiquettes associées, nous calculons la longueur effective à intégrer en fonction de la longueur maximale (les phrases et séquences d'étiquettes trop longues sont coupées). Puis nous pouvons les intégrer dans les tenseurs en début de ligne, les fins de lignes étant remplies de zéros (<eos>).
X = torch.zeros(len(int_texts), max_len).long()
Y = torch.zeros(len(int_labels), max_len).long()
for i, (text, label) in enumerate(zip(int_texts, int_labels)):
length = min(max_len, len(text))
X[i,:length] = torch.LongTensor(text[:length])
Y[i,:length] = torch.LongTensor(label[:length])
print(X[12])
print(Y[12])
Pour vérifier les performances du système, nous le séparons en un ensemble d'entraînement et un ensemble de développement. Comme il y a 5500 exemple, nous utiliserons les 5000 premiers pour l'entraînement et les 500 suivants pour la validation.
X_train = X[:5000]
Y_train = Y[:5000]
X_valid = X[5000:]
Y_valid = Y[5000:]
Torch contient des classes permettant facilement de charger des batches d'exemples déjà mélangés pour l'entraînement. Nous allons en tirer parti.
from torch.utils.data import TensorDataset, DataLoader
train_set = TensorDataset(X_train, Y_train)
valid_set = TensorDataset(X_valid, Y_valid)
train_loader = DataLoader(train_set, batch_size=batch_size, shuffle=True)
valid_loader = DataLoader(valid_set, batch_size=batch_size)
La prochaine étape consiste en le chargement des embeddings préentraînés. Les embeddings fasttext peuvent être téléchargés depuis la page suivante : https://github.com/facebookresearch/fastText/blob/master/pretrained-vectors.md
Toutefois, ces embeddings couvrant plusieurs millions de mots, le fichier fait plusieurs GiB et une version filtrée par rapport au vocabulaire du corpus est disponible. Notez qu'en production, il faudrait utiliser tous les embeddings pour avoir une bonne couverture du vocabulaire que l'on peut rencontrer.
![ -f wiki.en.filtered.vec ] || wget -q https://raw.githubusercontent.com/benob/dl4nlp-tutorials-data/master/wiki.en.filtered.vec.gz -O - | gunzip > wiki.en.filtered.vec
Les embeddings ont le format suivant :
Nous créons donc un tenseur à zéro, puis plaçons l'embedding de chaque mot du vocabulaire rencontré à la bonne ligne dans le tenseur. Les embeddings des mots non couverts dans ce fichier resteront à zéro.
pretrained_weights = torch.zeros(len(vocab), 300)
with open('wiki.en.filtered.vec') as fp:
for line in fp:
tokens = line.strip().split()
if tokens[0] in vocab:
pretrained_weights[vocab[tokens[0]]] = torch.FloatTensor([float(x) for x in tokens[1:]])
pretrained_weights[12][:10]
Le modèle est une class qui étend nn.Module. Son constructeur appelle le constructeur de la class mère, puis déclare les différentes couches : une couche d'embeddings, une couche de GRU et une couche de décision. Cette couche de décision sera appliquée à chaque position de la phrase pour prédire l'étiquette associée.
Le premier détail important que nous allons traiter est le problème du padding. Pour pouvoir prendre en entrée des batches de phrases de la même taille, nous avons complété ces dernières avec des 0. Le mot d'indice zéro a un embedding et le système peut donc l'utiliser pour apprendre des régularités des données. Si on utilise un RNN bidirectionnel, l'état caché après chaque padding peut aussi contribuer à la représentation créée. Ces deux problème font que le modèle aura un comportement différent si on change la taille des séquences traitées.
Une solution est de forcer l'état caché à rester à zéro en présence de padding. Pour celà il faut spécifier le padding_idx de la couche d'emebdding pour la représentation associée au padding soit toujours le vecteur nul. Ensuite, dans le RNN, l'état caché est calculé comme une transformation affine à partir de l'embedding en entrée et de l'état caché précédent. Comme l'état caché initial est à zéro, et que l'embedding du padding est à zéro, si on désactive le bias dans les calculs (en donnant l'option bias=False au GRU), cela va forcer l'état caché à rester à zéro tout au long du padding. Le modèle peut ainsi traiter des séquences de taille arbitraire, même quand il est bidirectionnel.
Le deuxième détail est l'initialisation des embeddings. La méthode diffère selon les versions de pytorch, mais la façon présentée ici marche dans la plupart des cas. Nous allons directement modifier le champ weight de la couche d'embeddings et remplacer ce paramètre par nos embeddings préentraînés. requires_grad=False permet de geler la couche d'embeddings et donc de ne pas les modifier lors de l'apprentissage. Cela permettra en prédiction d'utiliser des embeddings fasttext pour les mots que nous n'avons pas observés dans le corpus d'apprentissage, en espérant qu'un mot d'embedding proche s'y trouvait. Si on omet cette option, les embeddings sont fine-tunés pour maximiser les performances du système.
class RNN(nn.Module):
def __init__(self):
super().__init__()
self.embed = nn.Embedding(len(vocab), embed_size, padding_idx=vocab['<eos>'])
self.embed.weight = nn.Parameter(pretrained_weights, requires_grad=False)
self.rnn = nn.GRU(embed_size, hidden_size, bias=False, num_layers=1, bidirectional=False, batch_first=True)
self.dropout = nn.Dropout(0.3)
self.decision = nn.Linear(hidden_size * 1 * 1, len(label_vocab))
def forward(self, x):
embed = self.embed(x)
output, hidden = self.rnn(embed)
return self.decision(self.dropout(output))
rnn_model = RNN()
rnn_model
with torch.no_grad():
print(rnn_model(X[:2]).size())
Il est possible de tester le modèle avec un batch créé à partir des exemples du corpus. Nous pouvons remarquer que la taille du tenseur produit est (batch_size, sequence_length, num_labels), il y a bien une prédiction par position dans les séquences.
La fonction qui calcule les performances d'un modèle doit accomoder la nouvelle forme des données.
En particulier, le critère d'entropie croisée n'accepte que des matrices à deux dimensions, donc il faut redimentionner les scores produits pour qu'ils ait la taille (batch_size * sequence_length, num_labels) et les références pour quelles aient la taille (batch_size * sequence_length).
Ensuite, il faut modifier le max qui calcule les prédictions pour qu'il agisse sur la dernière dimension du tenseur y_scores.
Et finalement, pour calculer le score d'une séquence, nous devons ignorer le padding. Pour celà une matrice mask est crée qui contient 1 pour les éléments non nuls de la matrice contenant les étiquettes et 0 sinon.
On peut calculer le nombre de corrects ainsi que le numérateur de la fonction de performance en appliquant le masque.
Le loss, pour être comparable au loss de l'entraînement, n'est pas modifié.
def perf(model, loader):
criterion = nn.CrossEntropyLoss()
model.eval()
total_loss = correct = num_loss = num_perf = 0
for x, y in loader:
with torch.no_grad():
y_scores = model(x)
loss = criterion(y_scores.view(y.size(0) * y.size(1), -1), y.view(y.size(0) * y.size(1)))
y_pred = torch.max(y_scores, 2)[1]
mask = (y != 0)
correct += torch.sum((y_pred.data == y) * mask)
total_loss += loss.item()
num_loss += len(y)
num_perf += torch.sum(mask).item()
return total_loss / num_loss, correct.item() / num_perf
perf(rnn_model, valid_loader)
La fonction d'apprentissage est modifiée en deux endroits. Tout d'abord, pytorch refuse d'entraîner un modèle contenant des paramètres sans gradient. Nous devons donc filtrer la liste des paramètres passés à l'optimiseur pour qu'elle ne contienne pas les embeddings "gelés".
Ensuite, nous appliquons le même redimenssionnement des scores et des équeittes pour accomoder le critère d'apprentissage.
def fit(model, epochs):
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(filter(lambda param: param.requires_grad, model.parameters()))
for epoch in range(epochs):
model.train()
total_loss = num = 0
for x, y in train_loader:
optimizer.zero_grad()
y_scores = model(x)
loss = criterion(y_scores.view(y.size(0) * y.size(1), -1), y.view(y.size(0) * y.size(1)))
loss.backward()
optimizer.step()
total_loss += loss.item()
num += len(y)
print(epoch, total_loss / num, *perf(model, valid_loader))
rnn_model = RNN()
fit(rnn_model, 10)
On souhaite mainenant ajouter à ce modèle une couche de décision de type Conditional Random Field (CRF). Un CRF est un modèle probabiliste d'un étiquetage $y$ d'une séquence $x$ définit par : \begin{eqnarray} P(y|x) &=& \frac{1}{Z(x)} \prod_i \exp\left(\psi_i(y, x)\right) \\ Z(x) &=& \sum_{y'} \prod_i \exp\left(\psi_i(y', x)\right) \end{eqnarray}
où $\psi_i$ donne un score au voisinage du slot $i$ en fonction d'une clique de $y$ et de $x$ (potentiellement dans son intégralité), et $Z$ est un facteur de normalisation pour obtenir une probabilité.
Pour une séquence linéaire, une façon populaire de construire le voisinage est de décomposer $\psi_i$ en deux fonctions : $f$ donne un score à la transition $(y_i, y_{i-1})$, $g$ donne un score à l'émission $(y_i, x)$.
\begin{eqnarray} \psi_i &=& f(y_i,y_{i-1}) + g(y_i, x) \end{eqnarray}Dans notre modèle, $f$ sera un jeu de paramètres avec un paramètre pour chaque couple d'étiquettes, et $g$ utilisera les scores fournis par un RNN tel que celui utilisé précédemment.
On entraîne un CRF en maximisant la log-vraisemblance des données : \begin{eqnarray} LL &=& \sum_{(x, y)} \log P(y|x) \\ &=& \sum_{(x, y)} - \log Z(x) + \sum_i \psi_i \end{eqnarray}
Dans le cadre de pytorch, cela revient à minimiser $-LL$. Nous utiliserons pour cela le module pytorch-crf (documentation à https://pytorch-crf.readthedocs.io). Notez que nos tenseurs ont pour première dimension le batch, ce qu'il faut spécifier.
!pip install pytorch_crf
from torchcrf import CRF
crf_model = CRF(len(label_vocab), batch_first=True)
On peut tester le modèle. Ce dernier accepte deux fonctions :
forward() calcule la log likelihood $LL$decode() retourne la séquence d'étiquettes de plus haute probabilité avec l'algorithme de ViterbiCes deux fonctions peuvent prendre un masque en paramètre pour indiquer le padding (ici l'étiquette 0). Elles prennent en entrée un tenseur représentant les emissions calculées à partir des entrées. Nous les produirons avec le modèle RNN inchangé (notez que ce dernier n'a pas a être entraîné préalablement, l'apprentissage se fait de bout en bout).
with torch.no_grad():
mask = (Y[:2] != 0)
print('log likelihood', crf_model(rnn_model(X[:2]), Y[:2], mask=mask))
print('viterbi output', crf_model.decode(rnn_model(X[:2]), mask=mask))
Le calcul des performances sur le test est similaire à celui pour le RNN, à la différence qu'il faut utiliser decode() pour obtenir la séquence d'étiquettes prédite et calculer le nombre d'étiquettes correctes manuellement. Techniquement, le loss renvoyé est la log-vraisemblance négative moyenne par phrase.
def perf_crf(emission_model, crf_model, loader):
def num_correct(y_pred, y):
correct = 0
for y_pred_current, y_current in zip(y_pred, y):
for i in range(len(y_pred_current)):
if y_pred_current[i] == y_current[i]:
correct += 1
return correct
emission_model.eval()
crf_model.eval()
loss = num_loss = correct = num_perf = 0
for x, y in loader:
with torch.no_grad():
mask = (y != 0)
emissions = emission_model(x)
y_pred = crf_model.decode(emissions, mask)
correct += num_correct(y_pred, y)
num_perf += torch.sum(mask).item()
loss -= crf_model(emissions, y, mask).item()
num_loss += len(y)
return loss / num_loss, correct / num_perf
perf_crf(rnn_model, crf_model, valid_loader)
L'apprentissage du modèle contenant le CRF doit aussi être modifié pour tenir compte du fait que le loss est spécifique à ce module. Notez que l'on doit appliquer l'optimiseur sur les paramètres des deux modèles, et que crf_model.forward(), appelée implicitement par crf_model(), renvoie la log-vraisemblence positive alors que l'on souhaite minimiser la log-vraisemblance negative.
Une meilleure façon d'implémenter le CRF serait de construire une classe incluant les deux modèles (émissions et CRF) de manière à écrire les fonctions perf() et fit() de manière plus naturelle.
def fit_crf(emission_model, crf_model, epochs):
all_parameters = list(emission_model.parameters()) + list(crf_model.parameters())
optimizer = optim.Adam(filter(lambda param: param.requires_grad, all_parameters))
for epoch in range(epochs):
emission_model.train()
crf_model.train()
total_loss = num = 0
for x, y in train_loader:
optimizer.zero_grad()
mask = (y != 0)
emissions = emission_model(x)
loss = -crf_model(emissions, y, mask)
loss.backward()
optimizer.step()
total_loss += loss.item()
num += len(y)
print(epoch, total_loss / num, *perf_crf(emission_model, crf_model, valid_loader))
rnn_model = RNN()
crf_model = CRF(len(label_vocab), batch_first=True)
fit_crf(rnn_model, crf_model, 10)
À la fin de l'entraînement, il est possible de visualiser les transitions apprises par le modèle. On voit que les transitions de plus haut score sont $B \to I$, $I\to I$ et $O\to B$. De la même manière, $O\to I$ est la plus pénalisée.
print(label_vocab)
print(crf_model.transitions)
# create a list with (label1, label2, transition-score)
values = [(rev_label_vocab[i], rev_label_vocab[j], crf_model.transitions[i, j].item()) for i in range(4) for j in range(4)]
# sort and show highest values
values.sort(key=lambda x: -x[2])
for i, j, value in values[:3]:
print(i + '->' + j, value)
Une fois le modèle entraîné, nous pouvons écrire une fonction qui génère les prédictions pour une phrase. Cette fonction devra :
Notez que cette fonction peut prendre en entrée des phrases plus longues que max_len et que le temps de calcul n'est pas gaspillé par le padding.
predict(sentence) qui prend en entrée une phrase sous la forme d'une chaîne de caractères et renvoie une liste de couples (étiquette, mot).
Créer un système de reconnaissance d'entités nommées à partir des données de CoNLL-2003 (https://github.com/davidsbatista/NER-datasets/tree/master/CONLL2003). Vous utiliserez uniquement les formes (1e colone) et créerez une fonction prenant une chaîne de caractères en entrée et renvoyant la séquence d'entités nommées trouvées. Vous pouvez vous aider de Spacy pour la tokenization (https://spacy.io/api/tokenizer).