PyTorch LSTM / GRU
Rekurrente neuronale Netze (RNNs) stehen bei der Verarbeitung sequenzieller Daten vor dem Problem des verschwindenden Gradienten, was das Erlernen langfristiger Abhängigkeiten erschwert.
Long Short-Term Memory (LSTM)undGated Recurrent Unit (GRU)Durch die Einführung von Gating-Mechanismen lösen sie dieses Problem und sind Kernmodelle für die Verarbeitung von Sequenzaufgaben wie Zeitreihen, natürlicher Sprache und gesprochener Sprache.
1. Grenzen von RNN und Gating-Mechanismen
Ein Standard-RNN kombiniert bei jedem Zeitschritt die aktuelle Eingabe mit dem vorherigen verborgenen Zustand, um einen neuen verborgenen Zustand zu berechnen:
\[h_t = \tanh(W_h \cdot h_{t-1} + W_x \cdot x_t + b)\]
Diese Struktur hat zwei Kernprobleme:
Verschwindender Gradient: Während der Backpropagation wird der Gradient Schritt für Schritt mit der Gewichtsmatrix multipliziert. Wenn die Sequenz lang ist, klingt der Gradient exponentiell ab, wodurch die Parameter an frühen Zeitschritten kaum aktualisiert werden und das Modell keine langreichweitigen Abhängigkeiten lernen kann.
Explodierender Gradient: Wenn der größte Eigenwert der Gewichtsmatrix größer als 1 ist, wächst der Gradient während der Backpropagation exponentiell, wodurch das Training instabil wird (üblicherweise durch Gradienten-Clipping abgemildert).
Die Kernidee von Gating-Mechanismen besteht darin, lernbare „Schalter“ einzuführen, die das Netzwerk autonom entscheiden lassen: Am aktuellen Zeitschritt, welche Information erinnert, welche vergessen und welche neue Information in das Gedächtnis geschrieben werden soll.
LSTM verwendet drei Tore (Vergessenstor, Eingangstor, Ausgangstor) plus einen separaten Zellzustand; GRU vereinfacht die Struktur auf zwei Tore (Reset-Tor, Update-Tor), mit weniger Parametern und schnellerem Training.
2. LSTM-Prinzip
2.1 Kernstruktur und drei Tore
LSTM verwaltet zwei Zustandsvektoren, die zwischen Zeitschritten weitergegeben werden:
- Zellzustand\(c_t\): der Träger des Langzeitgedächtnisses, in dem Informationen nahezu verlustfrei fließen können
- Verborgener Zustand\(h_t\): Kurzzeitgedächtnis, auch die Ausgabe des aktuellen Zeitschritts
Alle drei Tore sind lineare Transformationen, die durch Sigmoid aktiviert werden, mit Ausgabewerten zwischen 0 und 1, die als „Ventile“ wirken:
遗忘门（Forget Gate）：决定从细胞状态中丢弃哪些信息 输入门（Input Gate）：决定将哪些新信息写入细胞状态 输出门（Output Gate）：决定基于细胞状态输出什么
2.2 Formeln zur Vorwärtsberechnung
\[
\begin{aligned}
\text{Eingabe:} & x_t \text{ (Eingabe des aktuellen Zeitschritts)}, h_{t-1} \text{ (vorheriger verborgener Zustand)}, c_{t-1} \text{ (vorheriger Zellzustand)} \\
\text{Vergessens-Gate:} & f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) \\
\text{Eingabe-Gate:} & i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) \\
\text{Kandidatenwert:} & \tilde{g}_t = \tanh(W_g \cdot [h_{t-1}, x_t] + b_g) \\
\text{Ausgabe-Gate:} & o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o) \\
\text{Zellzustand aktualisieren:} & c_t = f_t \odot c_{t-1} + i_t \odot \tilde{g}_t \\
\text{Verborgenen Zustand aktualisieren:} & h_t = o_t \odot \tanh(c_t)
\end{aligned}
\]
wobei \(\odot\) die elementweise Multiplikation (Hadamard-Produkt) bezeichnet und \(\sigma\) die Sigmoid-Funktion bezeichnet.
f_t ⊙ c_{t-1}Interpretation der Berechnungslogik:i_t ⊙ g_t: Das Vergessens-Gate entscheidet, wie viel historisches Gedächtnis beibehalten wird; nahe 0 vergisst es, nahe 1 behält es.g_t: Das Eingabe-Gate entscheidet, wie viel neue Information eingeschrieben wird,o_t ⊙ tanh(c_t)ist der Kandidat für den neuen Inhalt
3. GRU-Prinzip
3.1 Kernstruktur und zwei Gates
3.1 Kernstruktur und zwei GatesGRU vereint das Vergessens-Gate und das Eingabe-Gate von LSTM zumUpdate-Gate
重置门（Reset Gate）：决定忽略多少历史状态来计算候选隐藏状态 更新门（Update Gate）：决定保留多少历史状态，写入多少新状态
3.2 Formeln zur Vorwärtsberechnung
\[
\begin{aligned}
\text{Eingabe:} & x_t \text{ (Eingabe des aktuellen Zeitschritts)}, h_{t-1} \text{ (vorheriger verborgener Zustand)} \\
\text{Reset-Gate:} & r_t = \sigma(W_r \cdot [h_{t-1}, x_t] + b_r) \\
\text{Update-Gate:} & z_t = \sigma(W_z \cdot [h_{t-1}, x_t] + b_z) \\
\text{Kandidatenwert:} & \tilde{h}_t = \tanh(W_h \cdot [r_t \odot h_{t-1}, x_t] + b_h) \\
\text{Verborgenen Zustand aktualisieren:} & h_t = (1 - z_t) \odot h_{t-1} + z_t \odot \tilde{h}_t
\end{aligned}
\]
Interpretation der Berechnungslogik:
- Reset-Gate
r_tWenn nahe bei 0, der Kandidatenzustandh̃_tkaum von der Historie abhängt, gleichbedeutend mit einem Neuanfang - Update-Gate
z_tWenn nahe bei 1, übernimmt der neue Zustand mehr vom Kandidatenwert; wenn nahe bei 0, behält er mehr vom historischen Zustand - GRU hat keinen separaten Zellzustand, und seine Parameteranzahl beträgt etwa 75% der von LSTM
4. LSTM in PyTorch
Dieser Abschnitt beschreibt detailliert die Parameter, Eingabe-/Ausgabeformen und Initialisierungsmethoden des verborgenen Zustands von nn.LSTM.
4.1 Detaillierte Erklärung der Parameter von nn.LSTM
Beispiel
import torch
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimension des Eingabevektors bei jedem Zeitschritt
hidden_size=128, # Dimension des verborgenen Zustands (und Zellzustands)
num_layers=2, # Anzahl der gestapelten Schichten, Standard ist 1
bias=True, # Ob Bias-Terme verwendet werden, Standard ist True
batch_first=False, # Ob Batch in der ersten Dimension der Eingabe-/Ausgabeform liegt, Standard: False
dropout=0.0, # Dropout-Wahrscheinlichkeit zwischen Schichten (nur wirksam, wenn num_layers > 1)
bidirectional=False, # Ob bidirektionales LSTM verwendet werden soll, Standard: False
proj_size=0, # Dimension der Projektionsschicht (LSTM mit Projektion), Standard: 0 bedeutet nicht verwendet
)
# Parameteranzahl anzeigen
total_params = sum(p.numel() for p in lstm.parameters())
print(f"LSTM-Parameteranzahl: {total_params:,}")
# Wenn input_size=64, hidden_size=128, num_layers=2, beträgt es ungefähr 197,632
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimension des Eingabevektors bei jedem Zeitschritt
hidden_size=128, # Dimension des verborgenen Zustands (und Zellzustands)
num_layers=2, # Anzahl der gestapelten Schichten, Standard ist 1
bias=True, # Ob Bias-Terme verwendet werden, Standard ist True
batch_first=False, # Ob Batch in der ersten Dimension der Eingabe-/Ausgabeform liegt, Standard: False
dropout=0.0, # Dropout-Wahrscheinlichkeit zwischen Schichten (nur wirksam, wenn num_layers > 1)
bidirectional=False, # Ob bidirektionales LSTM verwendet werden soll, Standard: False
proj_size=0, # Dimension der Projektionsschicht (LSTM mit Projektion), Standard: 0 bedeutet nicht verwendet
)
# Parameteranzahl anzeigen
total_params = sum(p.numel() for p in lstm.parameters())
print(f"LSTM-Parameteranzahl: {total_params:,}")
# Wenn input_size=64, hidden_size=128, num_layers=2, beträgt es ungefähr 197,632
Formel zur Schätzung der Parameteranzahl (einschichtig, unidirektional):
\[
\begin{aligned}
\text{Parameter pro Schicht} & = 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}
\]
wobei 4 vier Gruppen von Gewichtsmatrizen entspricht: Vergessensgatter, Eingangsgatter, Kandidatenwert und Ausgangsgatter.
4.2 Formen von Eingabe und Ausgabe
Dies ist der fehleranfälligste Teil bei der Verwendung von LSTM, daher ist besondere Aufmerksamkeit erforderlich.batch_firstDer Einfluss der Parameter.
Beispiel
import torch
import torch.nn as nn
# ── batch_first=False (Standard) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Eingabeform: (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"Ausgabeform: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Hidden-State-Ausgabe bei jedem Zeitschritt
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Versteckter Zustand am letzten Zeitschritt
print(f"c_n shape: {c_n.shape}")
# torch.Size([1, 4, 64]) → wie h_n, Zellzustand am letzten Zeitschritt
# ── batch_first=True (empfohlen, intuitiver) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Eingabeform: (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"Ausgabeform: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Hinweis: Die Form von h_n wird durch batch_first nicht beeinflusst
# ── Ausgabeform eines mehrschichtigen bidirektionalen LSTM ──────────────────────
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"Ausgabeform: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, wegen bidirektionaler Verkettung
print(f"h_n-Form: {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
import torch.nn as nn
# ── batch_first=False (Standard) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Eingabeform: (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"Ausgabeform: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Hidden-State-Ausgabe bei jedem Zeitschritt
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Versteckter Zustand am letzten Zeitschritt
print(f"c_n shape: {c_n.shape}")
# torch.Size([1, 4, 64]) → wie h_n, Zellzustand am letzten Zeitschritt
# ── batch_first=True (empfohlen, intuitiver) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Eingabeform: (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"Ausgabeform: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Hinweis: Die Form von h_n wird durch batch_first nicht beeinflusst
# ── Ausgabeform eines mehrschichtigen bidirektionalen LSTM ──────────────────────
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"Ausgabeform: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, wegen bidirektionaler Verkettung
print(f"h_n-Form: {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
Zusammenfassung der Ausgabeformen:
| Variablen | 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 (bidirektional) oder 1 (unidirektional)
4.3 Initialisierung des verborgenen Zustands
Beispiel
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)
# Methode 1: Keinen Anfangszustand übergeben; PyTorch verwendet automatisch die Nullinitialisierung
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Methode 2: Manuell mit Nullen initialisieren (entspricht Methode 1, zeigt aber die Absicht explizit)
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))
# Methode 3: Zustandsbehafteter Modus — Zustand über Batches hinweg übergeben
# Geeignet zum Aufteilen ultra-langer Sequenzen (Sprachmodelle, Langtextgenerierung usw.)
# detach() muss aufgerufen werden, um den Berechnungsgraphen vom vorherigen Batch zu trennen, um GPU-Speicherlecks zu vermeiden
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() # Gradienten abschneiden, nur die Werte behalten
c = c.detach()
# Methode 4: Xavier- oder Normalverteilungsinitialisierung verwenden (konvergiert in manchen Szenarien schneller)
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) # Orthogonale Initialisierung hilft, das Training zu stabilisieren
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)
# Methode 1: Keinen Anfangszustand übergeben; PyTorch verwendet automatisch die Nullinitialisierung
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Methode 2: Manuell mit Nullen initialisieren (entspricht Methode 1, zeigt aber die Absicht explizit)
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))
# Methode 3: Zustandsbehafteter Modus — Zustand über Batches hinweg übergeben
# Geeignet zum Aufteilen ultra-langer Sequenzen (Sprachmodelle, Langtextgenerierung usw.)
# detach() muss aufgerufen werden, um den Berechnungsgraphen vom vorherigen Batch zu trennen, um GPU-Speicherlecks zu vermeiden
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() # Gradienten abschneiden, nur die Werte behalten
c = c.detach()
# Methode 4: Xavier- oder Normalverteilungsinitialisierung verwenden (konvergiert in manchen Szenarien schneller)
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) # Orthogonale Initialisierung hilft, das Training zu stabilisieren
return h, c
5. GRU in PyTorch
Die GRU-Schnittstelle ist nahezu identisch mit LSTM; der Hauptunterschied besteht darin, dass es keinen Zellzustand gibt.c。
5.1 Detaillierte Erläuterung der Parameter von nn.GRU
Beispiel
import torch.nn as nn
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Empfohlen, auf True zu setzen
dropout=0.3, # Dropout zwischen den Schichten
bidirectional=False,
)
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Empfohlen, auf True zu setzen
dropout=0.3, # Dropout zwischen den Schichten
bidirectional=False,
)
5.2 Einfaches Anwendungsbeispiel
Beispiel
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 gibt nur output und h_n zurück, kein c_n
output, h_n = gru(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Die Ausgabe am letzten Zeitschritt entnehmen (für Aufgaben wie Klassifikation)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Oder äquivalent:
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 gibt nur output und h_n zurück, kein c_n
output, h_n = gru(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"h_n-Form: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Die Ausgabe am letzten Zeitschritt entnehmen (für Aufgaben wie Klassifikation)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Oder äquivalent:
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
6. Varianten von LSTM und GRU
Dieser Abschnitt führt bidirektionale, mehrschichtig gestapelte und mehrschichtige Strukturen mit Dropout ein.
6.1 Bidirektionale LSTM / GRU
Bidirektionale Modelle verarbeiten Informationen gleichzeitig aus der Vorwärts- und der Rückwärtsrichtung der Sequenz. Die Ausgabe an jedem Zeitschritt enthält vergangenen und zukünftigen Kontext, wodurch sie für Aufgaben geeignet sind, die globalen Kontext erfordern, wie Textklassifikation und Named-Entity-Erkennung.
Beispiel
import torch
import torch.nn as nn
# Bidirektionales LSTM
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Bidirektionalität aktivieren
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 10, 128])
# Vorwärts 64-dimensional + Rückwärts 64-dimensional = 128-dimensional
print(f"h_n-Form: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Die finalen Hidden States der Vorwärts- und Rückwärtsrichtung trennen
# h_n-Reihenfolge: [Vorwärts-Layer0, Rückwärts-Layer0, Vorwärts-Layer1, Rückwärts-Layer1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) letzter Vorwärts-Layer
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) letzter Rückwärts-Layer
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# Die Verwendung eines bidirektionalen GRU ist genau dieselbe
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
# Bidirektionales LSTM
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Bidirektionalität aktivieren
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 10, 128])
# Vorwärts 64-dimensional + Rückwärts 64-dimensional = 128-dimensional
print(f"h_n-Form: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Die finalen Hidden States der Vorwärts- und Rückwärtsrichtung trennen
# h_n-Reihenfolge: [Vorwärts-Layer0, Rückwärts-Layer0, Vorwärts-Layer1, Rückwärts-Layer1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) letzter Vorwärts-Layer
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) letzter Rückwärts-Layer
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# Die Verwendung eines bidirektionalen GRU ist genau dieselbe
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
6.2 Mehrschichtige Stapelung
Beispiel
import torch
import torch.nn as nn
# 3-schichtiges gestapeltes LSTM
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # 3 Schichten stapeln
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 20, 128]) enthält nur die Ausgabe der obersten Schicht
print(f"h_n-Form: {h_n.shape}")
# torch.Size([3, 8, 128]) enthält den finalen verborgenen Zustand jeder Schicht
# Hole den finalen verborgenen Zustand jeder Schicht
h_layer1 = h_n[0] # (batch_size, hidden_size) erste Schicht
h_layer2 = h_n[1] # (batch_size, hidden_size) zweite Schicht
h_layer3 = h_n[2] # (batch_size, hidden_size) dritte Schicht (oberste Schicht)
import torch.nn as nn
# 3-schichtiges gestapeltes LSTM
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # 3 Schichten stapeln
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"Ausgabeform: {output.shape}")
# torch.Size([8, 20, 128]) enthält nur die Ausgabe der obersten Schicht
print(f"h_n-Form: {h_n.shape}")
# torch.Size([3, 8, 128]) enthält den finalen verborgenen Zustand jeder Schicht
# Hole den finalen verborgenen Zustand jeder Schicht
h_layer1 = h_n[0] # (batch_size, hidden_size) erste Schicht
h_layer2 = h_n[1] # (batch_size, hidden_size) zweite Schicht
h_layer3 = h_n[2] # (batch_size, hidden_size) dritte Schicht (oberste Schicht)
6.3 Mehrschichtige Struktur mit Dropout
nn.LSTMDie eingebautedropoutDer Parameter gilt nur für den Bereichzwischen den Schichten, nicht für die Ausgabe der letzten Schicht. Wenn Sie Dropout auch nach der letzten Schicht hinzufügen möchten, müssen Sie es manuell hinzufügen:
Beispiel
import torch
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
Mehrschichtiges LSTM + Dropout zwischen den Schichten + Dropout in der Ausgabe
"""
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,
# Wenn num_layers=1, ist dropout ungültig; setzen Sie es auf 0, um Warnungen zu vermeiden
)
self.dropout = nn.Dropout(dropout) # Dropout nach der letzten Schicht
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Nimm den Hidden State am letzten Zeitschritt
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):
"""
Mehrschichtiges LSTM + Dropout zwischen den Schichten + Dropout in der Ausgabe
"""
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,
# Wenn num_layers=1, ist dropout ungültig; setzen Sie es auf 0, um Warnungen zu vermeiden
)
self.dropout = nn.Dropout(dropout) # Dropout nach der letzten Schicht
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Nimm den Hidden State am letzten Zeitschritt
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. Umgang mit Sequenzen variabler Länge
In realen Aufgaben haben Sequenzen innerhalb desselben Batches normalerweise unterschiedliche Längen. PyTorch bietetpack_padded_sequenceundpad_packed_sequenceum dieses Problem zu behandeln, wodurch vermieden wird, dass LSTM ungültige Berechnungen an Padding-Positionen durchführt.
7.1 pack_padded_sequence
Beispiel
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Simuliere Sequenzen variabler Länge in einem Batch (bereits vom längsten zum kürzesten sortiert)
seq1 = torch.randn(5, 32) # Sequenzlänge 5
seq2 = torch.randn(3, 32) # Sequenzlänge 3
seq3 = torch.randn(2, 32) # Sequenzlänge 2
# pad_sequence füllt automatisch mit Nullen auf, um an die längste Sequenz anzugleichen
# Wenn batch_first=True, ist die Ausgabeform (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"Form nach Padding: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: komprimiert Padding, teilt dem LSTM die tatsächlichen Längen mit
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Sequenzen müssen in absteigender Reihenfolge ihrer Länge sortiert sein
# enforce_sorted=False # Ermöglicht jede Reihenfolge (wird intern automatisch sortiert), es wird empfohlen, False zu setzen
)
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
# Simuliere Sequenzen variabler Länge in einem Batch (bereits vom längsten zum kürzesten sortiert)
seq1 = torch.randn(5, 32) # Sequenzlänge 5
seq2 = torch.randn(3, 32) # Sequenzlänge 3
seq3 = torch.randn(2, 32) # Sequenzlänge 2
# pad_sequence füllt automatisch mit Nullen auf, um an die längste Sequenz anzugleichen
# Wenn batch_first=True, ist die Ausgabeform (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"Form nach Padding: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: komprimiert Padding, teilt dem LSTM die tatsächlichen Längen mit
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Sequenzen müssen in absteigender Reihenfolge ihrer Länge sortiert sein
# enforce_sorted=False # Ermöglicht jede Reihenfolge (wird intern automatisch sortiert), es wird empfohlen, False zu setzen
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
7.2 pad_packed_sequence
Beispiel
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# PackedSequence an die LSTM übergeben
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: zu einem gepolsterten Tensor wiederherstellen
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Wiederhergestellte Ausgabeform: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# Die Ausgabe an Padding-Positionen ist 0
print(f"Tatsächliche Längen jeder Sequenz: {output_lengths}")
# tensor([5, 3, 2])
# PackedSequence an die LSTM übergeben
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: zu einem gepolsterten Tensor wiederherstellen
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Wiederhergestellte Ausgabeform: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# Die Ausgabe an Padding-Positionen ist 0
print(f"Tatsächliche Längen jeder Sequenz: {output_lengths}")
# tensor([5, 3, 2])
7.3 Vollständige Verarbeitungspipeline für Sequenzen variabler Länge
Beispiel
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) — Wortindizes
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Packen
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# LSTM-Vorwärtsdurchlauf (Padding-Positionen überspringen)
packed_output, (h_n, c_n) = self.lstm(packed)
# Methode 1: h_n der letzten Schicht als Sequenzrepräsentation verwenden
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Methode 2: Nach dem Entpacken die Ausgabe an der tatsächlich letzten Position jeder Sequenz nehmen (äquivalent zu Methode 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)
# Simuliere einen Batch
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Tatsächliche Länge jeder Sequenz
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) — Wortindizes
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Packen
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# LSTM-Vorwärtsdurchlauf (Padding-Positionen überspringen)
packed_output, (h_n, c_n) = self.lstm(packed)
# Methode 1: h_n der letzten Schicht als Sequenzrepräsentation verwenden
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Methode 2: Nach dem Entpacken die Ausgabe an der tatsächlich letzten Position jeder Sequenz nehmen (äquivalent zu Methode 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)
# Simuliere einen Batch
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Tatsächliche Länge jeder Sequenz
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
8. Vollständiges Praxisbeispiel: Text-Sentiment-Klassifikation
Am Beispiel der binären Sentiment-Klassifikation von IMDB-Filmkritiken wird die vollständige Pipeline zur Verarbeitung von Text mit bidirektionalen LSTMs demonstriert:
Beispiel
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")
# ── Modell-Definition ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Bidirektionales LSTM-Textklassifikationsmodell
Architektur: 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
)
# Bei Verwendung vortrainierter Wortvektoren:
# 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)
# Bidirektionales LSTM: Vorwärts- und Rückwärts-Hidden-States der letzten Schicht verketten
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)
# Die Vorwärts- und Rückwärts-Hidden-States der letzten Schicht nehmen und sie verketten
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)
# ── Benutzerdefinierter Datensatz ────────────────────────────────
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):
"""Benutzerdefiniertes Collate: Sequenzen variabler Länge mit Nullen auffüllen, tatsächliche Längen erfassen"""
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
# ── Trainings- und Evaluierungsfunktionen ────────────────────────────────
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()
# Gradientenclipping: Gradientenexplosion in RNN verhindern
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
# ── Initialisierung und Training ──────────────────────────────────
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"Epoche {epoch:2d}/{EPOCHS} | "
f"Trainingsverlust: {train_loss:.4f}, Genauigkeit: {train_acc:.4f} | "
f"Validierungsverlust: {val_loss:.4f}, Genauigkeit: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Speichere bestes Modell, Val Acc: {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")
# ── Modell-Definition ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Bidirektionales LSTM-Textklassifikationsmodell
Architektur: 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
)
# Bei Verwendung vortrainierter Wortvektoren:
# 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)
# Bidirektionales LSTM: Vorwärts- und Rückwärts-Hidden-States der letzten Schicht verketten
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)
# Die Vorwärts- und Rückwärts-Hidden-States der letzten Schicht nehmen und sie verketten
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)
# ── Benutzerdefinierter Datensatz ────────────────────────────────
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):
"""Benutzerdefiniertes Collate: Sequenzen variabler Länge mit Nullen auffüllen, tatsächliche Längen erfassen"""
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
# ── Trainings- und Evaluierungsfunktionen ────────────────────────────────
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()
# Gradientenclipping: Gradientenexplosion in RNN verhindern
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
# ── Initialisierung und Training ──────────────────────────────────
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"Epoche {epoch:2d}/{EPOCHS} | "
f"Trainingsverlust: {train_loss:.4f}, Genauigkeit: {train_acc:.4f} | "
f"Validierungsverlust: {val_loss:.4f}, Genauigkeit: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Speichere bestes Modell, Val Acc: {best_acc:.4f}")
9. Vollständige praktische Anwendung: Zeitreihenvorhersage
Am Beispiel der mehrstufigen Zeitreihenvorhersage: Verwenden Sie LSTM, um Werte für die nächsten N Schritte vorherzusagen:
Beispiel
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")
# ── Gleitfenster-Datensatz ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Gleitfenster-Segmentierung von Zeitreihen
Eingabe: vergangene input_len Schritte
Ziel: zukünftige output_len Schritte
"""
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,)
# ── Modell-Definition ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Mehrschritt-Zeitreihenvorhersagemodell
Architektur: 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)
# Nimm die Ausgabe des letzten Zeitschritts
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Datenvorbereitung (mit Sinuswelle als Beispiel)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Verwende die vergangenen 60 Schritte
OUTPUT_LEN = 10 # Sage die nächsten 10 Schritte voraus
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)
# ── Training ──────────────────────────────────────────
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, c):
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"Epoche {epoch:2d}/30 | Trainings-MSE: {train_loss:.6f} | Validierungs-MSE: {val_loss:.6f}")
# ── Mehrschrittige rekursive Vorhersage (eine andere Strategie)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Rekursive Vorhersage: Jeweils einen Schritt vorhersagen, den vorhergesagten Wert an die Sequenz anhängen, dann den nächsten Schritt vorhersagen.
Geeignet für Einzelschritt-Vorhersagemodelle mit output_len=1
"""
model.eval()
sequence = list(init_sequence)
predictions = []
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")
# ── Gleitfenster-Datensatz ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Gleitfenster-Segmentierung von Zeitreihen
Eingabe: vergangene input_len Schritte
Ziel: zukünftige output_len Schritte
"""
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,)
# ── Modell-Definition ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Mehrschritt-Zeitreihenvorhersagemodell
Architektur: 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)
# Nimm die Ausgabe des letzten Zeitschritts
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Datenvorbereitung (mit Sinuswelle als Beispiel)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Verwende die vergangenen 60 Schritte
OUTPUT_LEN = 10 # Sage die nächsten 10 Schritte voraus
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)
# ── Training ──────────────────────────────────────────
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, c):
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"Epoche {epoch:2d}/30 | Trainings-MSE: {train_loss:.6f} | Validierungs-MSE: {val_loss:.6f}")
# ── Mehrschrittige rekursive Vorhersage (eine andere Strategie)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Rekursive Vorhersage: Jeweils einen Schritt vorhersagen, den vorhergesagten Wert an die Sequenz anhängen, dann den nächsten Schritt vorhersagen.
Geeignet für Einzelschritt-Vorhersagemodelle mit output_len=1
"""
model.eval()
sequence = list(init_sequence)
predictions = []
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. Vergleich und Auswahl von LSTM und GRU
Dieser Abschnitt vergleicht die strukturellen Unterschiede zwischen LSTM und GRU und gibt Auswahlempfehlungen.
Struktureller Vergleich
| Vergleichspunkt | LSTM | GRU |
|---|---|---|
| Anzahl der Gates | 3 (Vergessen, Eingabe, Ausgabe) | 2 (Zurücksetzen, Aktualisieren) |
| Zustandsvektoren | Zellzustand + verborgener Zustand | Nur verborgener Zustand |
| Parameteranzahl (bei gleichem hidden_size) | Baseline | Etwa 75% |
| Trainingsgeschwindigkeit | Langsamer | Schneller |
| Leistung bei langen Sequenzen | Normalerweise besser | Comparable |
| Leistung bei kurzen Sequenzen | Ähnlich | Ähnlich |
| Implementierungskomplexität | Höher | Niedriger |
Empfehlungen zur Auswahl
数据量小、训练资源有限
-> 优先选 GRU，参数少，不容易过拟合，训练快
序列较长（> 100 步）、长距离依赖重要
-> 优先选 LSTM，细胞状态更擅长保留远期信息
需要快速实验和基线对比
-> 先用 GRU，效果差再换 LSTM
任务对准确率要求极高，有充足数据
-> 两者都试，结合交叉验证选择
2020 年后的新项目
-> 考虑 Transformer 架构（BERT、GPT），在大数据量下通常优于 LSTM/GRU
LSTM/GRU 仍在边缘设备、低延迟推理、数据量较小的场景中有优势
Kurzreferenz für häufige Fragen
| Problem | Ursache | Lösung |
|---|---|---|
| Trainingsverlust nimmt nicht ab | Lernrate zu hoch oder Gradientenexplosion | Lernrate reduzieren; Gradient Clipping hinzufügenclip_grad_norm_ |
| Validierungs-Loss viel höher als Trainings-Loss | Überanpassung (Overfitting) | Dropout erhöhen; Anzahl der Schichten oder hidden_size reduzieren |
| Schlechte Vorhersage am Sequenzende | Verschwindender Gradient | Anzahl der Schichten erhöhen; bidirektionale Struktur verwenden; Attention-Mechanismus in Betracht ziehen |
| Übermäßige GPU-Speichernutzung | Sequenz zu lang oder Batch zu groß | seq_len oder batch_size reduzieren; Gradient Checkpointing verwenden |
| Abnormaler Loss beim Stateful-Training | Zwischen Batches übergebene States nicht detached | Nach jedem Batch aufrufenh.detach_() |
| Dimensionsfehler bei der Konkatenation von bidirektionalem LSTM | Missverständnis beim h_n-Index | Verwenden Sieh_n[-2](vorwärts) undh_n[-1](rückwärts) |
| PackedSequence-Fehler | Sequenzlängen nicht in absteigender Reihenfolge sortiert | Setenforce_sorted=False |
| batch_first-Verwirrung | Vergessen, konsistent zu setzen | Empfohlen, durchgängig zu verwendenbatch_first=True |