L'analyse de sentiment consiste à prédire à partir d'un texte la polarité du sentiment associé. Le système prend en entrée une séquence de mots et génère en sortie une unique prédiction.
Il faut commencer par télécharger les données pour ce problème d'analyse de sentiment.
L'archive contient sanders-twitter-sentiment.csv qui est un petit corpus de tweets annotés
avec une cible (apple, microsoft, google...) et une étiquette de sentiment (positif, negatif, neutre, non pertient).
%%bash
[ -f sanders-twitter-sentiment.csv ] || wget -q https://pageperso.lis-lab.fr/benoit.favre/files/sanders-twitter-sentiment.csv
head -3 sanders-twitter-sentiment.csv
Les colonnes qui nous intéressent dans le csv sont les colonnes 4, 5 et 6 contenant le tweet, l'étiquette et la cible.
On peut charger rapidement les données avec la classe python qui lit des csv. Pour faciliter les traitements, nous allons ajouter la cible (row[5]) en début de tweet (row[3]) entre chevrons. L'étiquette est dans row[4].
Pour vérifier que tout s'est bien passé on peut afficher le nombre d'exemples chargés ainsi qu'un exemple arbitrire.
texts = []
labels = []
import csv
with open('sanders-twitter-sentiment.csv', 'r', encoding='utf8') as csvfile:
reader = csv.reader(csvfile, delimiter=',', quotechar='"')
for row in reader:
texts.append("<" + row[5] + "> " + row[3])
labels.append(row[4])
print(len(texts))
print(texts[12])
print(labels[12])
La premier chose à faire est de créer un dictionnaire pour faire correspondre les étiquettes à des entiers. Les étiquettes ainsi converties seront stockées dans la liste python int_labels.
label_vocab = {label: i for i, label in enumerate(set(labels))}
print(label_vocab)
int_labels = [label_vocab[label] for label in labels]
print(int_labels[:10])
Nous pouvons opérer de la même manière pour convertir les mots en entiers. Notez que notre système ne fait aucun prétraitement sur le texte du tweet, il se contente de le découper selon les espaces. Un système d'analyse de sentiment plus évolué ferait une tokenisation plus fine, mettrait les mots en minuscues, et irait même jusqu'à les lemmatiser. Notez qu'on utilise un defaultdict pour stoquer le vocabulaire. Il attribue un entier à chaque nouveau mot qu'il rencontre. Le mot d'indice 0 est reservé pour un symbole <eos> que nous utilserons plus tard.
import collections
vocab = collections.defaultdict(lambda: len(vocab))
vocab['<eos>'] = 0
int_texts = []
for text in texts:
int_texts.append([vocab[token] for token in text.split()])
print(int_texts[12])
print(int_labels[12])
On peut vérifier que les mots ont bien été convertis en faisant la conversion dans l'autre sens. Il est interessant de regarder la taille du vocabulaire et la taille maximale d'un tweet dans ce corpus.
rev_vocab = {y: x for x, y in vocab.items()}
print([rev_vocab[word_id] for word_id in int_texts[12]])
print(len(vocab))
print(max([len(text) for text in int_texts]))
Nous sommes prêts à convertir les données en tenseurs pytorch. Il faut importer les modules de pytorch, et définir quelques constantes qui s'assurent que le problème est de taille raisonnable pour tourner sur CPU. La constante max_len est importante car les tweets qui ont une taille superieure à cette dernière seront coupés.
import 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 = 128
hidden_size = 128
device = torch.device('cpu') # can be changed to cuda if available
Afin de rendre les calculs rapides, il est souhaitable de mettre toutes les données dans des tenseurs. Comme les textes sont de taille variable, nous allons les mettre dans des matrices de taille (nombre de tweets, taille maximum d'un tweet). Les tweets trop grands par rapport à la taille maximale fixée seront coupés, et ceux qui sont plus courts seronts garnis de symboles de padding <eos> dont la valeur est 0.
Techniquement, il n'est pas nécessaire de mettre toutes les données dans un seul tenseur. En particulier cela devient inefficace quand les textes ont des longueurs très différentes. La seule contrainte est que pour un batch donné, les séquences aient la même taille. De nombreuses bibliothèques (comme torchtext) génèrent des batches à la volée qui font la bonne taille.
On notera que textes[12] a une longeur de 15 symboles et donc X[12] se retrouve paddé avec un symbole <eos>.
X = torch.zeros(len(int_texts), max_len).long()
for i, text in enumerate(int_texts):
length = min(max_len, len(text))
X[i,:length] = torch.LongTensor(text[:length])
Y = torch.LongTensor(int_labels)
X = X.to(device)
Y = Y.to(device)
print(X.size(), Y.size())
print(X[12], Y[12])
Pour pouvoir estimer les performances du modèle, nous allons diviser les données en un jeu d'entraînement et un jeu de validation.
X_train = X[:5000]
Y_train = Y[:5000]
X_valid = X[5000:]
Y_valid = Y[5000:]
pytorch fournit un générateur de batches qui s'occupe de mélanger les données. Utilisons le pour nos deux sources de données.
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)
Nous allons d'abord définir une fonction d'évaluation d'un modèle qui calcule le loss moyen sur un ensemble de test, ainsi que le taux d'exemples corrects. Cette fonction utilise l'entropie croisée comme fonction de loss.
Il faut tout d'abord mettre le modèle en mode evaluation pour que les traitements propres à l'apprentissage, comme le dropout, soient désactivés. Il ne faudra pas oublier de le remettre en mode entrainement lors de l'entraînement.
Puis pour chaque batch produit par le loader, on peut calculer les scores produits par le modèle pour chaque étiquette, en déduire le loss, et calculer les prédictions en prenant l'indice de la classe de score max.
Nous pourrons tester cette fonction lorsque nous aurons créé un modèle.
def perf(model, loader):
criterion = nn.CrossEntropyLoss()
model.eval()
total_loss = correct = num = 0
for x, y in loader:
with torch.no_grad():
y_scores = model(x)
loss = criterion(y_scores, y)
y_pred = torch.max(y_scores, 1)[1]
correct += torch.sum(y_pred.data == y)
total_loss += loss.item()
num += len(y)
return total_loss / num, correct.item() / num
La fonction d'apprentissage n'est pas très différente. Elle contient en plus un optimiseur (Adam est utilisé ici car il a des performances raisonnables lorsqu'on a pas encore exploré les hyper-paramètres). Puis pour chaque époque, on n'oublie pas de remettre le modèle en mode entraînement, et cette fois on parcourt les données d'entraînement batch par batch pendant plusieurs époques.
L'entraînement nécessite de remettre à zero les accumulateurs de gradient dans le modèle, de calculer les scores prédits pour le batch, d'en déduire la fonction de coût, puis de faire la back-propgation. L'optimiseur peut alors appliquer le gradient aux paramètres du modèle.
À la fin de chaque époque, on appelle la fonction perf définie précédemment pour se faire une idée des performances sur le jeu de validation.
def fit(model, epochs):
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(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, y)
loss.backward()
optimizer.step()
total_loss += loss.item()
num += len(y)
print(epoch, total_loss / num, *perf(model, valid_loader))
Le premier modèle que nous allons créer est un réseaux de neurones recurrent. Il prend en entrée une séquence d'entiers représentant les mots d'un tweet, puis les projette dans un espace d'embedding. Ces embeddings sont initialement aléatoires et seront appris avec le reste du modèle. Ensuite, il passe par une couche recurrente de type Gated Recurrent Units qui est un peu plus rapide que des LSTM pour des performances similaires. Enfin, l'état caché à la fin de la séquence est utilisé en entrée d'une couche de décision. Cette couche projette l'état caché du RNN vers un espace de dimension le nombre d'étiquettes.
Sur GPU, couche recurrente bénéficie d'une accélération grâce à la librairie CuDNN qui nécessite de passer la séquence complète d'entrées plutôt que de manuellement la traiter symbole par symbole (ceci permet de parallèliser plus d'éléments, mais on perd l'acceleration lorsque l'on veut customiser le comportement de cette couche). L'entrée attendue est de taille (batch_size, sequence_size, embed_size) si l'on a activé l'option batch_first.
La couche d'embeddings prend en entrée une matrice de taille (batch_size, sequence_size) et renvoie un tenseur de taille (batch_size, sequence_size, embed_size). La taille de l'état caché produit par les couches RNN de pytorch est (num_layers * num_directions, batch, hidden_size). Donc si on augmente le nombre de couches ou qu'on rend le RNN bidirectionnel, il faudra changer la taille de la couche de sortie en hidden_size * num_layers * num_directions. La couche de décision attend en entrée une matrice de taille (batch_size, in_size), donc il faut transposer les deux premières dimensions de hidden et la redimensionner (il faut rendre le tenseur contigu pour que pytorch veuille bien en changer la taille).
class RNN(nn.Module):
def __init__(self):
super().__init__()
self.embed = nn.Embedding(len(vocab), embed_size)
self.rnn = nn.GRU(embed_size, hidden_size, 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)
drop = self.dropout(hidden)
return self.decision(drop.transpose(0, 1).contiguous().view(x.size(0), -1))
rnn_model = RNN()
rnn_model.to(device)
On peut tester le modèle en lui passant comme batch les 3 premiers exemples du corpus encapsulés dans une Variable. Le résultat est une matrice de taille (batch_size, num_labels).
rnn_model(X[:3])
Nous pouvons alors entraîner le modèle quelques époques avec la fonction fit. Il faut vérifier que le loss diminue sur les données d'apprentissage, surveiller le loss sur les données de développement (s'il ne diminue pas, c'est que le modèle ne généralise pas), et les performances calculées par rapport à la métrique que nous intéresse vraiment. Dans la pratique, on sauvegarderait le modèle à chaque fois que les performances sur les données de validation augmentent de manière à ne pas tomber victime de sur-apprentissage. On testerait aussi de nombreux hyper-paramètrages afin de sélectionner le meilleur modèle.
fit(rnn_model, 10)
Le second modèle que nous allons créer est un réseau convolutionnel. La convolution permet d'extraire des n-grammes d'embeddings de mots, puis on la suit d'un max-pooling sur la séquence pour que la position des n-grammes soit invariante. La couche Conv1d génère la convolution sur la séquence d'embeddings. Elle attend en entrée un tenseur de taille (batch_size, embedding_size, sequence_length), ce qui va demander de transposer les deux dernières dimensions du tenseur produit par la couche d'embedding. La couche de convolution applique fait un produit entre sa matrice de paramètres et les n-grammes extraits de manière à pouvoir apprendre à sélectionner plusieurs n-grammes. Ses sorties sont de taille (batch_size, num_filters, sequence_length). kernel_size permet de régler la taille des n-grammes extraits. La couche de convolution étant linéaire, nous appliquons une fonction non linéaire de type ReLU (rectified linear unit).
La deuxième étape est d'appliquer le max pooling. Il existe une couche de max pooling mais elle est plus adaptée au traitement des images, nous allons donc utiliser la fonction max_pool1d qui prend en entrée le tenseur produit par la couche de convolution et la fenêtre sur laquelle appliquer le max (ici, toute la longueur de la séquence). Cette couche renvoie un tenseur de taille (batch_size, num_filters, num_max_windows) avec num_max_windows=1 pour nous. On pourrait imaginer un CNN multi-couches qui fasse progressivement le max sur des sous-fenêtres pour obtenir des représentations plus dépendentes de la position des mots. Il suffit ensuite de redimensionner le tenseur produit par la couche de pooling pour le passer à la couche de décision.
class CNN(nn.Module):
def __init__(self):
super().__init__()
self.embed = nn.Embedding(len(vocab), embed_size)
self.conv = nn.Conv1d(embed_size, hidden_size, kernel_size=2)
self.dropout = nn.Dropout(.3)
self.decision = nn.Linear(hidden_size, len(label_vocab))
def forward(self, x):
embed = self.embed(x)
conv = F.relu(self.conv(embed.transpose(1,2)))
pool = F.max_pool1d(conv, conv.size(2))
drop = self.dropout(pool)
return self.decision(drop.view(x.size(0), -1))
cnn_model = CNN()
cnn_model.to(device)
On peut tester que le modèle renvoie bien une matrice de taille (batch_size, num_labels).
cnn_model(X[:3])
Et l'entraîner avec la fonction fit.
fit(cnn_model, 10)
Créer un modèle RNN+CNN qui donne en entrée du CNN la sortie du RNN au lieu des embeddings. La taille de la sortie du RNN est (batch, seq_len, hidden_size * num_directions).
Appliquez la même méthodologie que dans ce tutoriel pour le corpus 20newsgroups (http://qwone.com/~jason/20Newsgroups/).
torch.cat([x1, x2, ...], dim) permet de concatèner pluseurs tenseurs sur une dimension donnée.