PyTorch LSTM / GRU
Les réseaux de neurones récurrents (RNNs) sont confrontés au problème de la disparition du gradient lors du traitement de données séquentielles, ce qui rend difficile l'apprentissage des dépendances à long terme.
Long Short-Term Memory (LSTM)etGated Recurrent Unit (GRU)En introduisant des mécanismes de portes, ils résolvent ce problème et sont des modèles essentiels pour le traitement de tâches séquentielles telles que les séries temporelles, le langage naturel et la parole.
1. Limitations des RNN et mécanismes de portes
Un RNN standard combine l'entrée actuelle avec l'état caché précédent à chaque pas de temps pour calculer un nouvel état caché :
\[h_t = \tanh(W_h \cdot h_{t-1} + W_x \cdot x_t + b)\]
Cette structure présente deux problèmes fondamentaux :
Disparition du gradient: lors de la rétropropagation, le gradient est multiplié pas à pas par la matrice de poids. Lorsque la séquence est longue, le gradient décroît exponentiellement, ce qui fait que les paramètres des premiers pas de temps sont à peine mis à jour et que le modèle ne peut pas apprendre les dépendances à longue portée.
Explosion du gradient: lorsque la plus grande valeur propre de la matrice de poids est supérieure à 1, le gradient augmente exponentiellement pendant la rétropropagation, ce qui rend l'entraînement instable (généralement atténué par l'écrêtage du gradient).
L'idée centrale des mécanismes de portes est d'introduire des « interrupteurs » apprenables qui permettent au réseau de décider de manière autonome : au pas de temps courant, quelles informations doivent être mémorisées, lesquelles doivent être oubliées et quelles nouvelles informations doivent être écrites en mémoire.
LSTM utilise trois portes (porte d'oubli, porte d'entrée, porte de sortie) plus un état de cellule séparé ; GRU simplifie la structure en deux portes (porte de réinitialisation, porte de mise à jour), avec moins de paramètres et un entraînement plus rapide.
2. Principe de LSTM
2.1 Structure centrale et trois portes
LSTM maintient deux vecteurs d'état transmis entre les pas de temps :
- État de cellule\(c_t\) : le porteur de la mémoire à long terme, dans lequel les informations peuvent circuler presque sans perte
- État caché\(h_t\) : mémoire à court terme, également la sortie du pas de temps courant
Les trois portes sont des transformations linéaires activées par Sigmoid, avec des valeurs de sortie comprises entre 0 et 1, agissant comme des « vannes » :
遗忘门（Forget Gate）：决定从细胞状态中丢弃哪些信息 输入门（Input Gate）：决定将哪些新信息写入细胞状态 输出门（Output Gate）：决定基于细胞状态输出什么
2.2 Formules de calcul en propagation avant
\[
\begin{aligned}
\text{Entrée :} & x_t \text{ (entrée au pas de temps courant)}, h_{t-1} \text{ (état caché précédent)}, c_{t-1} \text{ (état de cellule précédent)} \\
\text{Porte d'oubli :} & f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) \\
\text{Porte d'entrée :} & i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) \\
\text{Valeur candidate :} & \tilde{g}_t = \tanh(W_g \cdot [h_{t-1}, x_t] + b_g) \\
\text{Porte de sortie :} & o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o) \\
\text{Mise à jour de l'état de cellule :} & c_t = f_t \odot c_{t-1} + i_t \odot \tilde{g}_t \\
\text{Mise à jour de l'état caché :} & h_t = o_t \odot \tanh(c_t)
\end{aligned}
\]
où \(\odot\) désigne la multiplication élément par élément (produit de Hadamard), et \(\sigma\) désigne la fonction Sigmoïde.
f_t ⊙ c_{t-1}Interprétation de la logique de calcul :i_t ⊙ g_t: La porte d'oubli décide quelle quantité de mémoire historique conserver ; lorsqu'elle est proche de 0, elle oublie, lorsqu'elle est proche de 1, elle conserveg_t: La porte d'entrée décide quelle quantité de nouvelles informations écrire,o_t ⊙ tanh(c_t)est le nouveau contenu candidat
3. Principe du GRU
3.1 Structure centrale et deux portes
3.1 Structure centrale et deux portesGRU fusionne la porte d'oubli et la porte d'entrée du LSTM en laporte de mise à jour
重置门（Reset Gate）：决定忽略多少历史状态来计算候选隐藏状态 更新门（Update Gate）：决定保留多少历史状态，写入多少新状态
3.2 Formules de calcul en propagation avant
\[
\begin{aligned}
\text{Entrée :} & x_t \text{ (entrée au pas de temps courant)}, h_{t-1} \text{ (état caché précédent)} \\
\text{Porte de réinitialisation :} & r_t = \sigma(W_r \cdot [h_{t-1}, x_t] + b_r) \\
\text{Porte de mise à jour :} & z_t = \sigma(W_z \cdot [h_{t-1}, x_t] + b_z) \\
\text{Valeur candidate :} & \tilde{h}_t = \tanh(W_h \cdot [r_t \odot h_{t-1}, x_t] + b_h) \\
\text{Mise à jour de l'état caché :} & h_t = (1 - z_t) \odot h_{t-1} + z_t \odot \tilde{h}_t
\end{aligned}
\]
Interprétation de la logique de calcul :
- Porte de réinitialisation
r_tLorsqu'il est proche de 0, l'état candidath̃_tne dépend presque pas de l'historique, équivalent à repartir de zéro - Porte de mise à jour
z_tLorsqu'il est proche de 1, le nouvel état adopte davantage la valeur candidate ; lorsqu'il est proche de 0, il conserve davantage l'état historique - GRU n'a pas d'état de cellule séparé, et son nombre de paramètres représente environ 75% de celui de LSTM
4. LSTM dans PyTorch
Cette section détaille les paramètres, les formes d'entrée/sortie et les méthodes d'initialisation de l'état caché de nn.LSTM.
4.1 Explication détaillée des paramètres de nn.LSTM
Exemple
import torch
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimension du vecteur d'entrée à chaque pas de temps
hidden_size=128, # Dimension de l'état caché (et de l'état de cellule)
num_layers=2, # Nombre de couches empilées, par défaut 1
bias=True, # Indique s'il faut utiliser des termes de biais, par défaut True
batch_first=False, # Indique si le batch est dans la première dimension de la forme d'entrée/sortie, False par défaut
dropout=0.0, # Probabilité de dropout entre les couches (n'a d'effet que lorsque num_layers > 1)
bidirectional=False, # Indique s'il faut utiliser un LSTM bidirectionnel, False par défaut
proj_size=0, # Dimension de la couche de projection (LSTM avec projection), 0 par défaut signifie non utilisée
)
# Afficher le nombre de paramètres
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Nombre de paramètres du LSTM : {total_params:,}")
# Lorsque input_size=64, hidden_size=128, num_layers=2, cela fait environ 197,632
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimension du vecteur d'entrée à chaque pas de temps
hidden_size=128, # Dimension de l'état caché (et de l'état de cellule)
num_layers=2, # Nombre de couches empilées, par défaut 1
bias=True, # Indique s'il faut utiliser des termes de biais, par défaut True
batch_first=False, # Indique si le batch est dans la première dimension de la forme d'entrée/sortie, False par défaut
dropout=0.0, # Probabilité de dropout entre les couches (n'a d'effet que lorsque num_layers > 1)
bidirectional=False, # Indique s'il faut utiliser un LSTM bidirectionnel, False par défaut
proj_size=0, # Dimension de la couche de projection (LSTM avec projection), 0 par défaut signifie non utilisée
)
# Afficher le nombre de paramètres
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Nombre de paramètres du LSTM : {total_params:,}")
# Lorsque input_size=64, hidden_size=128, num_layers=2, cela fait environ 197,632
Formule d'estimation du nombre de paramètres (monocouche unidirectionnelle) :
\[
\begin{aligned}
\text{Paramètres par couche} & = 4 \times (hidden\_size \times input\_size + hidden\_size \times hidden\_size + hidden\_size) \\
& = 4 \times hidden\_size \times (input\_size + hidden\_size + 1)
\end{aligned}
\]
où 4 correspond à quatre groupes de matrices de poids : porte d'oubli, porte d'entrée, valeur candidate et porte de sortie.
4.2 Formes de l'entrée et de la sortie
C'est la partie la plus source d'erreurs lors de l'utilisation de LSTM, une attention particulière est donc requise.batch_firstL'impact des paramètres.
Exemple
import torch
import torch.nn as nn
# ── batch_first=False (par défaut) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Forme d'entrée : (seq_len, batch_size, input_size)
seq_len, batch_size, input_size = 10, 4, 32
x = torch.randn(seq_len, batch_size, input_size)
output, (h_n, c_n) = lstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# État caché de sortie à chaque pas de temps
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# État caché au dernier pas de temps
print(f"c_n shape: {c_n.shape}")
# torch.Size([1, 4, 64]) → identique à h_n, état de cellule au dernier pas de temps
# ── batch_first=True (recommandé, plus intuitif) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Forme de l'entrée : (batch_size, seq_len, input_size)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bf(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Remarque : la forme de h_n n'est pas affectée par batch_first
# ── Forme de sortie d'un LSTM bidirectionnel multicouche ──────────────────────
lstm_bd = nn.LSTM(input_size=32, hidden_size=64,
num_layers=3, bidirectional=True, batch_first=True)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bd(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, car concaténation bidirectionnelle
print(f"forme de h_n : {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
import torch.nn as nn
# ── batch_first=False (par défaut) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Forme d'entrée : (seq_len, batch_size, input_size)
seq_len, batch_size, input_size = 10, 4, 32
x = torch.randn(seq_len, batch_size, input_size)
output, (h_n, c_n) = lstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# État caché de sortie à chaque pas de temps
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# État caché au dernier pas de temps
print(f"c_n shape: {c_n.shape}")
# torch.Size([1, 4, 64]) → identique à h_n, état de cellule au dernier pas de temps
# ── batch_first=True (recommandé, plus intuitif) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Forme de l'entrée : (batch_size, seq_len, input_size)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bf(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Remarque : la forme de h_n n'est pas affectée par batch_first
# ── Forme de sortie d'un LSTM bidirectionnel multicouche ──────────────────────
lstm_bd = nn.LSTM(input_size=32, hidden_size=64,
num_layers=3, bidirectional=True, batch_first=True)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bd(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, car concaténation bidirectionnelle
print(f"forme de h_n : {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
Résumé des formes de sortie :
| Variables | batch_first=False | batch_first=True |
|---|---|---|
output |
(seq_len, N, H * D) |
(N, seq_len, H * D) |
h_n |
(L * D, N, H) |
(L * D, N, H) |
c_n |
(L * D, N, H) |
(L * D, N, H) |
\(N\) = batch_size, \(H\) = hidden_size, \(L\) = num_layers, \(D\) = 2 (bidirectionnel) ou 1 (unidirectionnel)
4.3 Initialisation de l'état caché
Exemple
import torch
import torch.nn as nn
lstm = nn.LSTM(input_size=32, hidden_size=64, num_layers=2, batch_first=True)
batch_size = 8
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
lstm = lstm.to(device)
# Méthode 1 : ne pas passer d'état initial ; PyTorch utilise automatiquement une initialisation à zéro
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Méthode 2 : initialiser manuellement à zéro (équivalent à la méthode 1, mais montre explicitement l'intention)
num_layers, num_directions = 2, 1
h_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
c_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
output, (h_n, c_n) = lstm(x, (h_0, c_0))
# Méthode 3 : mode avec état — transmettre l'état entre les lots
# Convient pour découper des séquences ultra-longues (modèles de langue, génération de texte long, etc.)
# Il faut appeler detach() pour couper le graphe de calcul du lot précédent afin d'éviter une fuite de mémoire GPU
h, c = h_0, c_0
for batch_x in data_loader:
batch_x = batch_x.to(device)
output, (h, c) = lstm(batch_x, (h, c))
h = h.detach() # Couper les gradients, ne garder que les valeurs
c = c.detach()
# Méthode 4 : utiliser l'initialisation Xavier ou une distribution normale (converge plus rapidement dans certains scénarios)
def init_hidden(lstm_module, batch_size, device):
num_layers = lstm_module.num_layers
hidden_size = lstm_module.hidden_size
directions = 2 if lstm_module.bidirectional else 1
h = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
c = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
nn.init.orthogonal_(h) # L'initialisation orthogonale aide à stabiliser l'entraînement
return h, c
import torch.nn as nn
lstm = nn.LSTM(input_size=32, hidden_size=64, num_layers=2, batch_first=True)
batch_size = 8
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
lstm = lstm.to(device)
# Méthode 1 : ne pas passer d'état initial ; PyTorch utilise automatiquement une initialisation à zéro
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Méthode 2 : initialiser manuellement à zéro (équivalent à la méthode 1, mais montre explicitement l'intention)
num_layers, num_directions = 2, 1
h_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
c_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
output, (h_n, c_n) = lstm(x, (h_0, c_0))
# Méthode 3 : mode avec état — transmettre l'état entre les lots
# Convient pour découper des séquences ultra-longues (modèles de langue, génération de texte long, etc.)
# Il faut appeler detach() pour couper le graphe de calcul du lot précédent afin d'éviter une fuite de mémoire GPU
h, c = h_0, c_0
for batch_x in data_loader:
batch_x = batch_x.to(device)
output, (h, c) = lstm(batch_x, (h, c))
h = h.detach() # Couper les gradients, ne garder que les valeurs
c = c.detach()
# Méthode 4 : utiliser l'initialisation Xavier ou une distribution normale (converge plus rapidement dans certains scénarios)
def init_hidden(lstm_module, batch_size, device):
num_layers = lstm_module.num_layers
hidden_size = lstm_module.hidden_size
directions = 2 if lstm_module.bidirectional else 1
h = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
c = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
nn.init.orthogonal_(h) # L'initialisation orthogonale aide à stabiliser l'entraînement
return h, c
5. GRU dans PyTorch
L'interface GRU est presque identique à celle de LSTM ; la principale différence est qu'il n'y a pas d'état de cellule.c。
5.1 Explication détaillée des paramètres de nn.GRU
Exemple
import torch.nn as nn
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Recommandé de définir sur True
dropout=0.3, # Dropout entre les couches
bidirectional=False,
)
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Recommandé de définir sur True
dropout=0.3, # Dropout entre les couches
bidirectional=False,
)
5.2 Exemple d'utilisation de base
Exemple
import torch
import torch.nn as nn
gru = nn.GRU(input_size=32, hidden_size=64, batch_first=True)
batch_size, seq_len = 8, 10
x = torch.randn(batch_size, seq_len, 32)
# GRU ne renvoie que output et h_n, pas c_n
output, h_n = gru(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Prendre la sortie au dernier pas de temps (pour des tâches telles que la classification)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Ou de manière équivalente :
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
import torch.nn as nn
gru = nn.GRU(input_size=32, hidden_size=64, batch_first=True)
batch_size, seq_len = 8, 10
x = torch.randn(batch_size, seq_len, 32)
# GRU ne renvoie que output et h_n, pas c_n
output, h_n = gru(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forme de h_n : {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Prendre la sortie au dernier pas de temps (pour des tâches telles que la classification)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Ou de manière équivalente :
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
6. Variantes de LSTM et GRU
Cette section présente des structures bidirectionnelles, empilées multicouches et multicouches avec Dropout.
6.1 LSTM / GRU bidirectionnels
Les modèles bidirectionnels traitent simultanément les informations provenant des directions avant et arrière de la séquence. La sortie à chaque pas de temps contient le contexte passé et futur, ce qui les rend adaptés aux tâches nécessitant un contexte global, telles que la classification de texte et la reconnaissance d'entités nommées.
Exemple
import torch
import torch.nn as nn
# LSTM bidirectionnel
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Activer le mode bidirectionnel
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 10, 128])
# 64 dimensions en avant + 64 dimensions en arrière = 128 dimensions
print(f"forme de h_n : {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Séparer les états cachés finaux des directions avant et arrière
# ordre de h_n : [couche avant 0, couche arrière 0, couche avant 1, couche arrière 1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) dernière couche avant
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) dernière couche arrière
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# L'utilisation d'un GRU bidirectionnel est exactement la même
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
import torch.nn as nn
# LSTM bidirectionnel
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Activer le mode bidirectionnel
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 10, 128])
# 64 dimensions en avant + 64 dimensions en arrière = 128 dimensions
print(f"forme de h_n : {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Séparer les états cachés finaux des directions avant et arrière
# ordre de h_n : [couche avant 0, couche arrière 0, couche avant 1, couche arrière 1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) dernière couche avant
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) dernière couche arrière
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# L'utilisation d'un GRU bidirectionnel est exactement la même
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
6.2 Empilement multicouche
Exemple
import torch
import torch.nn as nn
# LSTM empilé à 3 couches
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Empiler 3 couches
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 20, 128]) ne contient que la sortie de la couche supérieure
print(f"forme de h_n : {h_n.shape}")
# torch.Size([3, 8, 128]) contient l'état caché final de chaque couche
# Obtenir l'état caché final de chaque couche
h_layer1 = h_n[0] # (batch_size, hidden_size) première couche
h_layer2 = h_n[1] # (batch_size, hidden_size) deuxième couche
h_layer3 = h_n[2] # (batch_size, hidden_size) troisième couche (couche supérieure)
import torch.nn as nn
# LSTM empilé à 3 couches
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Empiler 3 couches
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"forme de sortie : {output.shape}")
# torch.Size([8, 20, 128]) ne contient que la sortie de la couche supérieure
print(f"forme de h_n : {h_n.shape}")
# torch.Size([3, 8, 128]) contient l'état caché final de chaque couche
# Obtenir l'état caché final de chaque couche
h_layer1 = h_n[0] # (batch_size, hidden_size) première couche
h_layer2 = h_n[1] # (batch_size, hidden_size) deuxième couche
h_layer3 = h_n[2] # (batch_size, hidden_size) troisième couche (couche supérieure)
6.3 Structure multicouche avec Dropout
nn.LSTMLa fonction intégréedropoutle paramètre ne s'applique qu'àentre les couches, pas à la sortie de la dernière couche. Si vous devez également ajouter Dropout après la dernière couche, vous devez l'ajouter manuellement :
Exemple
import torch
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
LSTM multicouche + Dropout inter-couche + Dropout de sortie
"""
def __init__(self, input_size, hidden_size, num_layers,
num_classes, dropout=0.3):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
# Lorsque num_layers=1, dropout est invalide ; mettez-le à 0 pour éviter les avertissements
)
self.dropout = nn.Dropout(dropout) # Dropout après la dernière couche
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Prendre l'état caché au dernier pas de temps
last_output = output[:, -1, :] # (batch_size, hidden_size)
last_output = self.dropout(last_output)
return self.fc(last_output)
model = StackedLSTM(input_size=64, hidden_size=128,
num_layers=3, num_classes=5, dropout=0.3)
x = torch.randn(16, 20, 64)
print(model(x).shape) # torch.Size([16, 5])
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
LSTM multicouche + Dropout inter-couche + Dropout de sortie
"""
def __init__(self, input_size, hidden_size, num_layers,
num_classes, dropout=0.3):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
# Lorsque num_layers=1, dropout est invalide ; mettez-le à 0 pour éviter les avertissements
)
self.dropout = nn.Dropout(dropout) # Dropout après la dernière couche
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Prendre l'état caché au dernier pas de temps
last_output = output[:, -1, :] # (batch_size, hidden_size)
last_output = self.dropout(last_output)
return self.fc(last_output)
model = StackedLSTM(input_size=64, hidden_size=128,
num_layers=3, num_classes=5, dropout=0.3)
x = torch.randn(16, 20, 64)
print(model(x).shape) # torch.Size([16, 5])
7. Gestion des séquences de longueur variable
Dans les tâches réelles, les séquences d'un même lot ont généralement des longueurs différentes. PyTorch fournitpack_padded_sequenceetpad_packed_sequencepour gérer ce problème, en évitant que le LSTM effectue des calculs non valides sur les positions de padding.
7.1 pack_padded_sequence
Exemple
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Simuler des séquences de longueur variable dans un lot (déjà triées du plus long au plus court)
seq1 = torch.randn(5, 32) # Longueur de séquence 5
seq2 = torch.randn(3, 32) # Longueur de séquence 3
seq3 = torch.randn(2, 32) # Longueur de séquence 2
# pad_sequence ajoute automatiquement des zéros pour aligner sur la séquence la plus longue
# Lorsque batch_first=True, la forme de sortie est (batch_size, max_seq_len, input_size)
padded = pad_sequence([seq1, seq2, seq3], batch_first=True, padding_value=0.0)
lengths = torch.tensor([5, 3, 2])
print(f"Forme après padding : {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence : compresse le padding, indique à LSTM les longueurs réelles
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Les séquences doivent être triées par ordre décroissant de longueur
# enforce_sorted=False # Autorise n'importe quel ordre (trié automatiquement en interne), il est recommandé de le définir à False
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Simuler des séquences de longueur variable dans un lot (déjà triées du plus long au plus court)
seq1 = torch.randn(5, 32) # Longueur de séquence 5
seq2 = torch.randn(3, 32) # Longueur de séquence 3
seq3 = torch.randn(2, 32) # Longueur de séquence 2
# pad_sequence ajoute automatiquement des zéros pour aligner sur la séquence la plus longue
# Lorsque batch_first=True, la forme de sortie est (batch_size, max_seq_len, input_size)
padded = pad_sequence([seq1, seq2, seq3], batch_first=True, padding_value=0.0)
lengths = torch.tensor([5, 3, 2])
print(f"Forme après padding : {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence : compresse le padding, indique à LSTM les longueurs réelles
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Les séquences doivent être triées par ordre décroissant de longueur
# enforce_sorted=False # Autorise n'importe quel ordre (trié automatiquement en interne), il est recommandé de le définir à False
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
7.2 pad_packed_sequence
Exemple
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Transmettre la PackedSequence au LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence : restaurer en un tenseur avec padding
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Forme de sortie restaurée : {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# La sortie aux positions de padding est 0
print(f"Longueurs réelles de chaque séquence: {output_lengths}")
# tensor([5, 3, 2])
# Transmettre la PackedSequence au LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence : restaurer en un tenseur avec padding
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Forme de sortie restaurée : {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# La sortie aux positions de padding est 0
print(f"Longueurs réelles de chaque séquence: {output_lengths}")
# tensor([5, 3, 2])
7.3 Pipeline complet de traitement de séquences à longueur variable
Exemple
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence
class LSTMClassifier(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_size, num_classes, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=padding_idx)
self.lstm = nn.LSTM(embed_dim, hidden_size, batch_first=True)
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len) — indices de mots
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Empaqueter
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Passe avant du LSTM (ignorer les positions de remplissage)
packed_output, (h_n, c_n) = self.lstm(packed)
# Méthode 1 : utiliser le h_n de la dernière couche comme représentation de la séquence
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Méthode 2 : après dépaquetage, prendre la sortie à la véritable dernière position de chaque séquence (équivalent à la méthode 1)
# output, _ = pad_packed_sequence(packed_output, batch_first=True)
# last_hidden = output[range(len(lengths)), lengths - 1, :]
return self.fc(last_hidden)
model = LSTMClassifier(vocab_size=10000, embed_dim=128,
hidden_size=256, num_classes=5)
# Simuler un lot
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Longueur réelle de chaque séquence
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence
class LSTMClassifier(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_size, num_classes, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=padding_idx)
self.lstm = nn.LSTM(embed_dim, hidden_size, batch_first=True)
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len) — indices de mots
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Empaqueter
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Passe avant du LSTM (ignorer les positions de remplissage)
packed_output, (h_n, c_n) = self.lstm(packed)
# Méthode 1 : utiliser le h_n de la dernière couche comme représentation de la séquence
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Méthode 2 : après dépaquetage, prendre la sortie à la véritable dernière position de chaque séquence (équivalent à la méthode 1)
# output, _ = pad_packed_sequence(packed_output, batch_first=True)
# last_hidden = output[range(len(lengths)), lengths - 1, :]
return self.fc(last_hidden)
model = LSTMClassifier(vocab_size=10000, embed_dim=128,
hidden_size=256, num_classes=5)
# Simuler un lot
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Longueur réelle de chaque séquence
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
8. Pratique complète : Classification de sentiments textuels
En prenant comme exemple la classification binaire de sentiments d'avis de films IMDB, démontrer le pipeline complet de traitement de texte par LSTM bidirectionnel :
Exemple
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pack_padded_sequence, pad_sequence
from torch.optim.lr_scheduler import ReduceLROnPlateau
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Définition du modèle ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Modèle de classification de texte par LSTM bidirectionnel
Architecture : Embedding -> BiLSTM -> Dropout -> FC
"""
def __init__(self, vocab_size, embed_dim, hidden_size,
num_layers, num_classes, dropout=0.5, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(
vocab_size, embed_dim, padding_idx=padding_idx
)
# Lors de l'utilisation de vecteurs de mots pré-entraînés :
# self.embedding = nn.Embedding.from_pretrained(pretrained_vectors)
self.lstm = nn.LSTM(
input_size=embed_dim,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
bidirectional=True,
)
self.dropout = nn.Dropout(dropout)
# LSTM bidirectionnel : concaténer les derniers états cachés avant et arrière
self.fc = nn.Linear(hidden_size * 2, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len)
embedded = self.dropout(self.embedding(x))
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
packed_output, (h_n, c_n) = self.lstm(packed)
# h_n: (num_layers * 2, batch_size, hidden_size)
# Prendre les états cachés avant et arrière de la dernière couche et les concaténer
h_forward = h_n[-2, :, :] # (batch_size, hidden_size)
h_backward = h_n[-1, :, :] # (batch_size, hidden_size)
h_combined = torch.cat([h_forward, h_backward], dim=-1)
# (batch_size, hidden_size * 2)
out = self.dropout(h_combined)
return self.fc(out)
# ── Jeu de données personnalisé ────────────────────────────────
class TextDataset(Dataset):
def __init__(self, texts, labels, vocab, max_len=200):
self.data = texts
self.labels = labels
self.vocab = vocab
self.max_len = max_len
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
tokens = self.data[idx][:self.max_len]
ids = [self.vocab.get(t, 1) for t in tokens] # 1 = <UNK>
return torch.tensor(ids, dtype=torch.long), torch.tensor(self.labels[idx])
def collate_fn(batch):
"""Collate personnalisé : remplir avec des zéros les séquences de longueur variable, enregistrer les longueurs réelles"""
sequences, labels = zip(*batch)
lengths = torch.tensor([len(s) for s in sequences])
padded = pad_sequence(sequences, batch_first=True, padding_value=0)
labels = torch.stack(labels)
return padded, lengths, labels
# ── Fonctions d'entraînement et d'évaluation ────────────────────────────────
def train_epoch(model, loader, optimizer, criterion):
model.train()
total_loss, correct = 0.0, 0
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
loss.backward()
# Écrêtage du gradient: éviter l'explosion du gradient dans les RNN
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
def eval_epoch(model, loader, criterion):
model.eval()
total_loss, correct = 0.0, 0
with torch.no_grad():
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
# ── Initialisation et entraînement ──────────────────────────────────
VOCAB_SIZE = 50000
EMBED_DIM = 128
HIDDEN_SIZE = 256
NUM_LAYERS = 2
NUM_CLASSES = 2
DROPOUT = 0.5
EPOCHS = 15
model = BiLSTMClassifier(VOCAB_SIZE, EMBED_DIM, HIDDEN_SIZE,
NUM_LAYERS, NUM_CLASSES, DROPOUT).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
scheduler = ReduceLROnPlateau(optimizer, mode="max",
patience=3, factor=0.5)
best_acc = 0.0
for epoch in range(1, EPOCHS + 1):
train_loss, train_acc = train_epoch(model, train_loader, optimizer, criterion)
val_loss, val_acc = eval_epoch(model, val_loader, criterion)
scheduler.step(val_acc)
print(f"Époque {epoch:2d}/{EPOCHS} | "
f"Perte d'entraînement : {train_loss:.4f}, Précision : {train_acc:.4f} | "
f"Perte de validation : {val_loss:.4f}, Précision : {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Enregistrement du meilleur modèle, Préc. Val : {best_acc:.4f}")
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pack_padded_sequence, pad_sequence
from torch.optim.lr_scheduler import ReduceLROnPlateau
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Définition du modèle ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Modèle de classification de texte par LSTM bidirectionnel
Architecture : Embedding -> BiLSTM -> Dropout -> FC
"""
def __init__(self, vocab_size, embed_dim, hidden_size,
num_layers, num_classes, dropout=0.5, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(
vocab_size, embed_dim, padding_idx=padding_idx
)
# Lors de l'utilisation de vecteurs de mots pré-entraînés :
# self.embedding = nn.Embedding.from_pretrained(pretrained_vectors)
self.lstm = nn.LSTM(
input_size=embed_dim,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
bidirectional=True,
)
self.dropout = nn.Dropout(dropout)
# LSTM bidirectionnel : concaténer les derniers états cachés avant et arrière
self.fc = nn.Linear(hidden_size * 2, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len)
embedded = self.dropout(self.embedding(x))
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
packed_output, (h_n, c_n) = self.lstm(packed)
# h_n: (num_layers * 2, batch_size, hidden_size)
# Prendre les états cachés avant et arrière de la dernière couche et les concaténer
h_forward = h_n[-2, :, :] # (batch_size, hidden_size)
h_backward = h_n[-1, :, :] # (batch_size, hidden_size)
h_combined = torch.cat([h_forward, h_backward], dim=-1)
# (batch_size, hidden_size * 2)
out = self.dropout(h_combined)
return self.fc(out)
# ── Jeu de données personnalisé ────────────────────────────────
class TextDataset(Dataset):
def __init__(self, texts, labels, vocab, max_len=200):
self.data = texts
self.labels = labels
self.vocab = vocab
self.max_len = max_len
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
tokens = self.data[idx][:self.max_len]
ids = [self.vocab.get(t, 1) for t in tokens] # 1 = <UNK>
return torch.tensor(ids, dtype=torch.long), torch.tensor(self.labels[idx])
def collate_fn(batch):
"""Collate personnalisé : remplir avec des zéros les séquences de longueur variable, enregistrer les longueurs réelles"""
sequences, labels = zip(*batch)
lengths = torch.tensor([len(s) for s in sequences])
padded = pad_sequence(sequences, batch_first=True, padding_value=0)
labels = torch.stack(labels)
return padded, lengths, labels
# ── Fonctions d'entraînement et d'évaluation ────────────────────────────────
def train_epoch(model, loader, optimizer, criterion):
model.train()
total_loss, correct = 0.0, 0
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
loss.backward()
# Écrêtage du gradient: éviter l'explosion du gradient dans les RNN
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
def eval_epoch(model, loader, criterion):
model.eval()
total_loss, correct = 0.0, 0
with torch.no_grad():
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
# ── Initialisation et entraînement ──────────────────────────────────
VOCAB_SIZE = 50000
EMBED_DIM = 128
HIDDEN_SIZE = 256
NUM_LAYERS = 2
NUM_CLASSES = 2
DROPOUT = 0.5
EPOCHS = 15
model = BiLSTMClassifier(VOCAB_SIZE, EMBED_DIM, HIDDEN_SIZE,
NUM_LAYERS, NUM_CLASSES, DROPOUT).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
scheduler = ReduceLROnPlateau(optimizer, mode="max",
patience=3, factor=0.5)
best_acc = 0.0
for epoch in range(1, EPOCHS + 1):
train_loss, train_acc = train_epoch(model, train_loader, optimizer, criterion)
val_loss, val_acc = eval_epoch(model, val_loader, criterion)
scheduler.step(val_acc)
print(f"Époque {epoch:2d}/{EPOCHS} | "
f"Perte d'entraînement : {train_loss:.4f}, Précision : {train_acc:.4f} | "
f"Perte de validation : {val_loss:.4f}, Précision : {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Enregistrement du meilleur modèle, Préc. Val : {best_acc:.4f}")
9. Cas pratique complet : Prédiction de séries temporelles
En prenant l'exemple de la prédiction de séries temporelles à plusieurs pas, utilisez LSTM pour prédire les valeurs des N prochaines étapes :
Exemple
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import Dataset, DataLoader
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Jeu de données à fenêtre glissante ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Segmentation par fenêtre glissante des séries temporelles
Entrée : input_len étapes passées
Cible : output_len étapes futures
"""
def __init__(self, series, input_len, output_len):
self.data = torch.tensor(series, dtype=torch.float32)
self.input_len = input_len
self.output_len = output_len
def __len__(self):
return len(self.data) - self.input_len - self.output_len + 1
def __getitem__(self, idx):
x = self.data[idx : idx + self.input_len]
y = self.data[idx + self.input_len : idx + self.input_len + self.output_len]
return x.unsqueeze(-1), y # x: (input_len, 1), y: (output_len,)
# ── Définition du modèle ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Modèle de prédiction de séries temporelles multi-étapes
Architecture : LSTM -> Dropout -> FC
"""
def __init__(self, input_size, hidden_size, num_layers,
output_len, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_len)
def forward(self, x):
# x: (batch_size, input_len, input_size)
output, (h_n, c_n) = self.lstm(x)
# Prendre la sortie du dernier pas de temps
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Préparation des données (en utilisant une onde sinusoïdale comme exemple)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Utiliser les 60 pas passés
OUTPUT_LEN = 10 # Prédire les 10 prochains pas
split = int(len(series) * 0.8)
train_data = series[:split]
val_data = series[split:]
train_dataset = TimeSeriesDataset(train_data, INPUT_LEN, OUTPUT_LEN)
val_dataset = TimeSeriesDataset(val_data, INPUT_LEN, OUTPUT_LEN)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False)
# ── Entraînement ──────────────────────────────────────────
model = LSTMForecaster(input_size=1, hidden_size=128,
num_layers=2, output_len=OUTPUT_LEN).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
def train_epoch_ts(model, loader, optimizer, criterion):
model.train()
total_loss = 0.0
for x, y in loader:
x, y = x.to(device), y.to(device)
optimizer.zero_grad()
pred = model(x)
loss = criterion(pred, y)
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * x.size(0)
return total_loss / len(loader.dataset)
def eval_epoch_ts(model, loader, criterion):
model.eval()
total_loss = 0.0
with torch.no_grad():
for x, y in loader:
x, y = x.to(device), y.to(device)
pred = model(x)
total_loss += criterion(pred, y).item() * x.size(0)
return total_loss / len(loader.dataset)
for epoch in range(1, 31):
train_loss = train_epoch_ts(model, train_loader, optimizer, criterion)
val_loss = eval_epoch_ts(model, val_loader, criterion)
print(f"Époque {epoch:2d}/30 | MSE d'entraînement: {train_loss:.6f} | MSE de validation: {val_loss:.6f}")
# ── Prédiction récursive multi-étapes (autre stratégie)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Prédiction récursive : prédire une étape à la fois, ajouter la valeur prédite à la séquence, puis prédire l'étape suivante
Adapté aux modèles de prédiction à une seule étape avec output_len=1
"""
model.eval()
sequence = list(init_sequence)
prédictions = []
with torch.inference_mode():
for _ in range(steps):
x = torch.tensor(sequence[-INPUT_LEN:], dtype=torch.float32)
x = x.unsqueeze(0).unsqueeze(-1).to(device) # (1, input_len, 1)
pred = model(x).item()
predictions.append(pred)
sequence.append(pred)
return predictions
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import Dataset, DataLoader
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Jeu de données à fenêtre glissante ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Segmentation par fenêtre glissante des séries temporelles
Entrée : input_len étapes passées
Cible : output_len étapes futures
"""
def __init__(self, series, input_len, output_len):
self.data = torch.tensor(series, dtype=torch.float32)
self.input_len = input_len
self.output_len = output_len
def __len__(self):
return len(self.data) - self.input_len - self.output_len + 1
def __getitem__(self, idx):
x = self.data[idx : idx + self.input_len]
y = self.data[idx + self.input_len : idx + self.input_len + self.output_len]
return x.unsqueeze(-1), y # x: (input_len, 1), y: (output_len,)
# ── Définition du modèle ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Modèle de prédiction de séries temporelles multi-étapes
Architecture : LSTM -> Dropout -> FC
"""
def __init__(self, input_size, hidden_size, num_layers,
output_len, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_len)
def forward(self, x):
# x: (batch_size, input_len, input_size)
output, (h_n, c_n) = self.lstm(x)
# Prendre la sortie du dernier pas de temps
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Préparation des données (en utilisant une onde sinusoïdale comme exemple)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Utiliser les 60 pas passés
OUTPUT_LEN = 10 # Prédire les 10 prochains pas
split = int(len(series) * 0.8)
train_data = series[:split]
val_data = series[split:]
train_dataset = TimeSeriesDataset(train_data, INPUT_LEN, OUTPUT_LEN)
val_dataset = TimeSeriesDataset(val_data, INPUT_LEN, OUTPUT_LEN)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False)
# ── Entraînement ──────────────────────────────────────────
model = LSTMForecaster(input_size=1, hidden_size=128,
num_layers=2, output_len=OUTPUT_LEN).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
def train_epoch_ts(model, loader, optimizer, criterion):
model.train()
total_loss = 0.0
for x, y in loader:
x, y = x.to(device), y.to(device)
optimizer.zero_grad()
pred = model(x)
loss = criterion(pred, y)
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * x.size(0)
return total_loss / len(loader.dataset)
def eval_epoch_ts(model, loader, criterion):
model.eval()
total_loss = 0.0
with torch.no_grad():
for x, y in loader:
x, y = x.to(device), y.to(device)
pred = model(x)
total_loss += criterion(pred, y).item() * x.size(0)
return total_loss / len(loader.dataset)
for epoch in range(1, 31):
train_loss = train_epoch_ts(model, train_loader, optimizer, criterion)
val_loss = eval_epoch_ts(model, val_loader, criterion)
print(f"Époque {epoch:2d}/30 | MSE d'entraînement: {train_loss:.6f} | MSE de validation: {val_loss:.6f}")
# ── Prédiction récursive multi-étapes (autre stratégie)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Prédiction récursive : prédire une étape à la fois, ajouter la valeur prédite à la séquence, puis prédire l'étape suivante
Adapté aux modèles de prédiction à une seule étape avec output_len=1
"""
model.eval()
sequence = list(init_sequence)
prédictions = []
with torch.inference_mode():
for _ in range(steps):
x = torch.tensor(sequence[-INPUT_LEN:], dtype=torch.float32)
x = x.unsqueeze(0).unsqueeze(-1).to(device) # (1, input_len, 1)
pred = model(x).item()
predictions.append(pred)
sequence.append(pred)
return predictions
10. Comparaison et sélection de LSTM et GRU
Cette section compare les différences structurelles entre LSTM et GRU et fournit des recommandations de sélection.
Comparaison structurelle
| Élément de comparaison | LSTM | GRU |
|---|---|---|
| Nombre de portes | 3 (Oubli, Entrée, Sortie) | 2 (Réinitialisation, Mise à jour) |
| Vecteurs d'état | État de cellule + État caché | État caché uniquement |
| Nombre de paramètres (même hidden_size) | Référence | Environ 75% |
| Vitesse d'entraînement | Plus lent | Plus rapide |
| Performance sur séquences longues | Généralement meilleur | Comparable |
| Performance sur séquences courtes | Similaire | Similaire |
| Complexité d'implémentation | Plus élevé | Inférieur |
Recommandations de sélection
数据量小、训练资源有限
-> 优先选 GRU，参数少，不容易过拟合，训练快
序列较长（> 100 步）、长距离依赖重要
-> 优先选 LSTM，细胞状态更擅长保留远期信息
需要快速实验和基线对比
-> 先用 GRU，效果差再换 LSTM
任务对准确率要求极高，有充足数据
-> 两者都试，结合交叉验证选择
2020 年后的新项目
-> 考虑 Transformer 架构（BERT、GPT），在大数据量下通常优于 LSTM/GRU
LSTM/GRU 仍在边缘设备、低延迟推理、数据量较小的场景中有优势
Référence rapide pour les questions courantes
| Problème | Cause | Solution |
|---|---|---|
| La perte d'entraînement ne diminue pas | Taux d'apprentissage trop élevé ou explosion du gradient | Réduire le taux d'apprentissage ; ajouter un écrêtage du gradientclip_grad_norm_ |
| Perte de validation bien plus élevée que la perte d'entraînement | Surapprentissage | Augmenter le Dropout ; réduire le nombre de couches ou hidden_size |
| Prédiction médiocre en fin de séquence | Disparition du gradient | Augmenter le nombre de couches ; utiliser une structure bidirectionnelle ; envisager un mécanisme d'attention |
| Utilisation excessive de la mémoire GPU | Séquence trop longue ou batch trop grand | Réduire seq_len ou batch_size ; utiliser le gradient checkpointing |
| Perte anormale dans l'entraînement avec état | États transmis entre les lots sans être détachés | Appeler après chaque loth.detach_() |
| Erreur de dimension de concaténation de LSTM bidirectionnel | Malentendu sur l'indice h_n | useh_n[-2](forward) eth_n[-1](backward) |
| Erreur PackedSequence | Longueurs de séquence non triées par ordre décroissant | Setenforce_sorted=False |
| Confusion sur batch_first | Oubli de définir de manière cohérente | Recommandé à utiliser partoutbatch_first=True |