Travaux pratiques - RNN pour la génération de texte

L’objectif de cette séance de travaux pratiques est d’illustrer la mise en application des réseaux de neurones récurrents sur des données séquentielles. En particulier, nous allons nous intéresser aux modèles auto-régressifs pour la génération de texte.

Génération de poésie

Une première application des réseaux de neurones récurents est la génération de texte. Pour démarrer, nous allons extraire les textes d’un recueil de poésies, « Les fleurs du mal » (1857) de l’écrivain Charles Baudelaire (1821-1867). Cet ensemble de textes va constituer notre corpus d’entraînement.

!wget -nc "http://cedric.cnam.fr/~thomen/cours/US330X/fleurs_mal.txt"

Dans notre application, nous nous intéressons à la génération de texte au travers de la prédiction du mot suivant. En considérant un texte comme une suite de mots \((x_1, x_2, ..., x_m)\), nous allons entraîner un réseau de neurones récurrent de sorte à prédire le bon mot \(x_n\) à partir des mots \((x_1, x_2, ..., x_{n-1})\) qui le précèdent dans une phrase.

Création du jeu de données d’entraînement

Le code ci-dessous va nous servir à générer les données et les étiquettes correspondantes. On va commencer par parser le ficher d’entrée pour récupérer le texte et effectuer quelques pré-traitements simples:

# Lire le fichier texte et ajouter toutes les lignes dans une liste
with open("fleurs_mal.txt", 'r' , encoding = 'utf8') as f:
    lines = f.readlines()

for idx, line in enumerate(lines):
    if "Charles Baudelaire avait un ami" in line:
        first_line = idx
    if "End of the Project Gutenberg EBook of Les Fleurs du Mal, by Charles Baudelaire" in line:
        last_line = idx

lines = lines[first_line:last_line]
lines = [l.lower().strip().replace('_', '') for l in lines if len(l) > 1]
text = " ".join(lines)
characters = sorted(set(text))
n_characters = len(characters)

Question

Que contient la variable characters ? Que représente n_characters ? La documentation des Set en Python peut vous aider.

Dans la suite de ce TP, nous allons considérer le texte comme une suite de caractères. Nous n’allons donc pas raisonner au niveau du mot mais au niveau du symbole. Chaque caractère du texte d’entrée sera représenté en entrée du réseau de neurones par un encodage one-hot sur le dictionnaire de symboles. Autrement dit, pour un dictionnaire simplifié (” “, a, b, c, d), la lettre a serait représentée par le vecteur \((0, 1, 0, 0, 0, 0)\) tandis que l’espace “ “ serait représenté par le vecteur \((1, 0, 0, 0, 0)\).

Nous allons désormais entraîner un réseau de neurones récurrent. Celui-ci va recevoir en entrée une séquence de SEQLEN caractères. Son objectif sera de prédire en sortie le caractère suivant dans le corpus. Par exemple, pour la phrase :

Le vélo est rouge.

le modèle devra prédire l à partir de la séquence Le , puis o à partir de la séquence Le vél, et ainsi de suite. Il s’agit donc d’un problème de classification à n_characters classes différentes (une classe par symbole).

L’étiquette de classe est obtenue automatiquement à partir du corpus. Comme il n’y a eu aucune annotation manuelle du jeu de données, cet objectif de prédiction du caractère suivant représente un problème d’apprentissage dit auto-supervisé (ou self-supervised). La supervision est construite artificiellement à partir des données elles mêmes.

Les données d’entraînement consistent donc en l’ensemble des séquences de caractères du corpus dont la taille est inférieure à SEQLEN. L’étiquette de la classe cible correspondante est celle de l’indice du prochain caractère à prédire, c’est-à-dire le caractère suivant dans le corpus.

# SEQLEN représente la taille de la séquence de lettres à passer en entrée
SEQLEN = 10
step = 1
input_characters, labels = [], []
# On parcourt le corpus de texte avec une fenêtre glissante
for i in range(0, len(text) - SEQLEN, step):
    input_characters.append(text[i:i + SEQLEN])
    labels.append(text[i + SEQLEN])

print(f"Il y a {len(input_characters)} séquences de {SEQLEN} caractères dans le corpus d'entraînement.")

Question

Afficher une séquence de SEQLEN caractères et l’étiquette de classe correspondante, c’est-à-dire le caractère suivant.

Nous pouvons maintenant vectoriser les données d’entraînement en utilisant le dictionnaire et un encodage one-hot pour chaque caractère :

# Encodage caractère -> indice du dictionaire
char2index = dict((c, i) for i, c in enumerate(characters))
# Encodage de l'indice vers le caractère (utilisé pour décoder les prédictions du modèle)
index2char = dict((i, c) for i, c in enumerate(characters)) # mapping index -> char in dictionary

Chaque séquence d’entraînement est donc représentée par une matrice de taille \(SEQLEN \times m\), correspondant à une longueur de SEQLEN caractères, chaque caractère étant encodé par un vecteur binaire correspondant à un encodage one-hot. \(m\) représente la taille du dictionnaire, c’est-à-dire le nombre de symboles uniques dans le corpus.

  • L’ensemble des données d’entraînement X seront donc constituées par un tenseur de taille \(N \times SEQLEN \times m\)\(N\) est le nombre de séquences de SEQLEN caractères dans le corpus.

  • L’ensemble des labels d’entraînement y seront représentées par un tenseur de \(N \times m\), où la sortie pour chaque exemple correspond à l’indice dans le dictionnaire du caractère suivant la séquence

Question

Compléter le code ci-dessous afin de créer les tenseurs X et y contenant les données d’entraînement (séquences de caractères dans X et étiquettes de classe dans y. Vous pourrez notamment utiliser à bon escient le dictionnaire char2index qui permet de transformer un caractère en son indice entier.

import numpy as np
X = np.zeros((len(input_characters), SEQLEN, n_characters), dtype=bool)
y = np.zeros((len(input_characters), n_characters), dtype=bool)

for idx_seq, sequence in enumerate(input_characters):
    # À compléter
    ...

Comme à l’accoutumée, nous allons séparer le jeu de données en deux : un ensemble d’apprentissage et un ensemble de validation. Le jeu de validation nous permettra notamment d’évaluer les performances du modèle et d’éviter le sur-apprentissage.

from sklearn.model_selection import train_test_split

# 80% des données en apprentissage, 20% en validation
X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.8, random_state=0)

Apprentissage d’un modèle auto-supervisé pour la génération de texte

Maintenant que les données ont été formatées, nous pouvons commencer à définir le modèle que nous allons utiliser. Nous allons l’implémenter sous la forme d’un modèle Keras séquentiel (Sequential).

from tensorflow.keras.layers import SimpleRNN
from tensorflow.keras.models import Sequential

model = Sequential()

Pour l’instant, ce modèle est vide. Nous allons lui ajouter une couche récurrente avec un modèle de type SimpleRNN (la cellule récurrente la plus simple) :

h_size = 128
model.add(SimpleRNN(h_size, return_sequences=False, input_shape=(SEQLEN, n_characters), unroll=True))

Question

À l’aide de la documentation de SimpleRNN dans Keras, expliquer à quoi correspondent les paramètres h_size = 128 et return_sequences=False.

Note

L’argument optionnel unroll=True permet simplement d’accélérer les calculs en « déroulant » le réseau récurrent plutôt que d’utiliser une boucle for en interne.

Pour terminer notre modèle, nous ajoutons enfin une couche entièrement connectée suivie d’une fonction softmax qui effectuera la classification du caractère suivant la séquence.

from tensorflow.keras.layers import Dense, Activation

model.add(Dense(n_characters))
model.add(Activation("softmax"))

Empiriquement, il a été constaté que l’optimisation des réseaux récurrents est plus rapide et la convergence plus robuste lorsque l’on utilise des méthodes de descente de gradient à pas adaptatif, telles que RMSprop [TH12].

from tensorflow.keras.optimizers import RMSprop

learning_rate = 0.001
optim = RMSprop(learning_rate=learning_rate)

Nous pouvons donc compiler le modèle et utiliser la méthode summary() de Keras pour visualiser le nombre de paramètres du réseaux

model.compile(loss="categorical_crossentropy", optimizer=optim, metrics=['accuracy'])
model.summary()

L’entraînement s’effectue de la manière habituelle à l’aide de la méthode fit():

batch_size = 128
num_epochs = 50
model.fit(X_train, y_train, batch_size=batch_size, epochs=num_epochs)

Nous pouvons utiliser la méthode .evaluate() pour calculer automatiquement les métriques spécifiées à la compilation du modèle sur un jeu de données. Calculons par exemple les scores de taux de bonne classification (accuracy) sur le jeu d’apprentissage et sur le jeu de test :

scores_train = model.evaluate(X_train, y_train, verbose=1)
scores_test = model.evaluate(X_test, y_test, verbose=1)
print(f"Performances (apprentissage, {model.metrics_names[1]}) = {scores_train[1]*100:.2f}")
print(f"Performances (validation, {model.metrics_names[1]}) = {scores_test[1]*100:.2f}")

Note: Il est possible de sauvegarder les paramètres du modèle appris à l’aide de la méthode .save(). Cela permet notamment de réutiliser le modèle plus tard, sans avoir à l’entraîner de nouveau.

model_name = f"SimpleRNN_{h_size}_{num_epochs}epochs"
model.save(model_name)

Analyse de l’apprentissage

Question

Quels taux de classification obtient-on en apprentissage ? En validation ? Commenter les performances obtenues.

Question

En quoi le problème est-il différents des problèmes de classification abordés jusqu’ici ? Par exemple, faire une recherche de la séquence d’entrée « la mort de », et analyser les labels cibles présents dans le corpus d’apprentissage.

Génération de texte à partir du modèle appris

Nous pouvons désormais nous servir du modèle entraîné pour générer du texte qui va « imiter » le style du corpus de poésie initial (Les Fleurs du mal). Si nécessaire, commençons par charger les paramètres du réseau récurrent précédemment entraîné à l’aide de la fonction loadModel :

from tensorflow.keras.models import load_model
model = load_model(model_name)
model.compile(loss='categorical_crossentropy', optimizer='RMSprop', metrics=['accuracy'])
model.summary()

Comme le modèle a été entraîné pour prédire le caractère suivant à partir d’une séquence de SEQLEN caractères précédent, nous devons l’initialiser avec une chaîne de caractères de départ. Le modèle pourra ensuite générer du texte en prédisant les caractères suivants un par un.

Reprenons un texte initial issu de notre corpus d’entraînement :

idx = 10
# index2char permet de repasser de l'encodage one-hot au caractère du dictionnaire
initial_characters = [index2char[np.argmax(c)] for c in X_train[idx]]
initial_text = "".join(initial_characters)
print(f"La séquence n°{idx} est : '{initial_text}'")

Nous pouvons maintenant extraire la représentation en encodage one-hot de ce texte, que nous passerons ensuite dans le réseau entraîné pour obtenir la prédiction du caractère suivant.

test_sequence = np.zeros((1, SEQLEN, n_characters), dtype=bool)
test_sequence[0] = X_train[idx]
prediction = model.predict(test_sequence)
print(prediction)

Question

À l’aide du dictionnaire index2char, déterminer quel est le prochain caractère prédit par le modèle.

Au lieu de prédire systématiquement le symbole dont la probabilité est maximale, nous pouvons ajouter du non-déterministe (et donc de l’aléatoire) dans la génération de texte en échantillonnant selon la distribution de probabilités en sortie du softmax. Autrement dit, plus un symbole aura une forte activation après le softmax, plus il aura de chances d’être tiré.

Pour contrôler à quel point cet échantillonnage sera non-déterministe, nous allons introduire un nouveau paramètre permettant de contrôler la forme de la distribution. En notant \(T\) ce nouveau paramètre, nous allons altérer les probabilités de sortie en les remplaçant par la formule suivante :

\[z_{i}^N = \frac{z_{i}^{\frac{1}{T}}}{\sum\limits_{j=1}^C z_{j}^{\frac{1}{T}} }\]

\(T\) est un paramètre appelé température. Si \(T=1\), alors il s’agit du softmax habituel.

La figure ci-dessous montre l’impact sur la distribution de cette renormalisation :

Nous pourrons par la suite utiliser la fonction ci-dessous qui tire aléatoirement un symbole en échantillonnant selon la distribution, éventuellement modifiée par la température.

def sample(probabilities, temperature=1.0):
    probabilities = np.asarray(probabilities).astype('float64')
    # Modifie la distribution selon la valeur de la température
    probabilities = pow(probabilities, 1.0/temperature)
    probabilities /= np.sum(probabilities)
    # Tire des variables aléatoires selon la distribution multinomiale transformée
    random_values = np.random.multinomial(1, probabilities, 1)
    # Renvoie le symbole échantillonné
    return np.argmax(random_values)

Question

Comment est-ce que la température modifie la distribution lorsqu’elle augmente (\(T \rightarrow +\infty\)) ou diminue (\(T \rightarrow 0\)) ? Comment cela va-t-il influer sur l’échantillonnage du caractère suivant ?

Pour terminer, nous pouvons mettre en place la génération de texte à partir d’une séquence de SEQLEN caractères initiaux. Pour ce faire, nous allons créer une boucle qui :

  1. extraie les SEQLEN derniers caractères du texte généré

  2. calcule les probabilités après softmax du réseau (méthode .predict())

  3. échantillonne un caractère dans ces probabilités (fonction sample())

  4. ajoute ce caractère au texte généré

Question

Compléter le code de génération de texte ci-dessous.

# Longueur du texte à générer (en caractères)
text_length = 200
# Température
temperature  = 0.5


generated_text = initial_text
network_input = test_sequence

for i in range(text_length):
    last_characters = generated_text[-SEQLEN:]
    # Compléter le code
    ...

Question

Évaluer l’impact du paramètre de température dans la génération, ainsi que le nombre d’époques dans l’apprentissage. Commenter les points forts et points faibles du générateur.

Pour aller plus loin

Comment doit-on modifier le jeu de données ou modèle de RNN pour améliorer les performances du générateur ? Essayez notamment d’augmenter SEQLEN. Quel problème cela résout-il ? Un modèle avec plus de paramètres donne-t-il de meilleurs résultats ?