PyTorch LSTM / GRU
Las redes neuronales recurrentes (RNN) enfrentan el problema del desvanecimiento del gradiente al procesar datos secuenciales, lo que dificulta el aprendizaje de dependencias de largo alcance.
Red de memoria a largo y corto plazo (Long Short-Term Memory, LSTM)yUnidad recurrente con compuerta (Gated Recurrent Unit, GRU)Al resolver este problema mediante la introducción del mecanismo de compuerta, son modelos centrales para tareas secuenciales como series temporales, lenguaje natural, voz, etc.
1. Limitaciones de las RNN y el mecanismo de compuerta
La RNN estándar combina la entrada actual con el estado oculto del paso anterior en cada paso temporal para calcular el nuevo estado oculto:
\[h_t = \tanh(W_h \cdot h_{t-1} + W_x \cdot x_t + b)\]
Esta estructura tiene dos problemas principales:
Desvanecimiento del gradiente: Durante la retropropagación, el gradiente debe multiplicarse paso a paso por la matriz de pesos; cuando la secuencia es larga, el gradiente decae exponencialmente, lo que hace que los parámetros de los pasos temporales tempranos casi no se actualicen, y el modelo no puede aprender dependencias de largo alcance.
Explosión del gradiente: Cuando el valor propio máximo de la matriz de pesos es mayor que 1, el gradiente aumenta exponencialmente durante la retropropagación, lo que hace que el entrenamiento sea inestable (normalmente se mitiga con recorte de gradiente).
La idea central del mecanismo de compuerta es introducir "interruptores" aprendibles, permitiendo que la red decida autónomamente: en el paso temporal actual, qué información se debe recordar, qué se debe olvidar y qué nueva información se debe escribir en la memoria.
LSTM utiliza tres compuertas (compuerta de olvido, compuerta de entrada, compuerta de salida) más un estado de celda separado; GRU simplifica la estructura a dos compuertas (compuerta de reinicio, compuerta de actualización), con menos parámetros y entrenamiento más rápido.
2. Principios de LSTM
2.1 Estructura central y las tres compuertas
LSTM mantiene dos vectores de estado que se transmiten entre los pasos temporales:
- Estado de celda (Cell State)\(c_t\): portador de la memoria a largo plazo, en el que la información puede fluir casi sin pérdida
- Estado oculto (Hidden State)\(h_t\): memoria a corto plazo, que también es la salida del paso temporal actual
Las tres compuertas son transformaciones lineales con activación Sigmoid, cuyos valores de salida están entre 0 y 1, actuando como "válvulas":
遗忘门（Forget Gate）：决定从细胞状态中丢弃哪些信息 输入门（Input Gate）：决定将哪些新信息写入细胞状态 输出门（Output Gate）：决定基于细胞状态输出什么
2.2 Fórmulas de cálculo hacia adelante
\[
\begin{aligned}
\text{Entrada: } & x_t \text{ (entrada del paso temporal actual)}, h_{t-1} \text{ (estado oculto del paso anterior)}, c_{t-1} \text{ (estado de celda del paso anterior)} \\
\text{Compuerta de olvido: } & f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) \\
\text{Compuerta de entrada: } & i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) \\
\text{Valor candidato: } & \tilde{g}_t = \tanh(W_g \cdot [h_{t-1}, x_t] + b_g) \\
\text{Compuerta de salida: } & o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o) \\
\text{Actualizar estado de celda: } & c_t = f_t \odot c_{t-1} + i_t \odot \tilde{g}_t \\
\text{Actualizar estado oculto: } & h_t = o_t \odot \tanh(c_t)
\end{aligned}
\]
donde \(\odot\) denota la multiplicación elemento a elemento (producto de Hadamard), y \(\sigma\) denota la función Sigmoid.
Interpretación de la lógica de cálculo:
f_t ⊙ c_{t-1}: La compuerta de olvido determina cuánta memoria histórica conservar; cerca de 0 significa olvidar, cerca de 1 significa conservar.i_t ⊙ g_t: La compuerta de entrada decide cuánta información nueva escribir,g_tes el contenido nuevo candidatoo_t ⊙ tanh(c_t): La compuerta de salida decide qué extraer del estado de celda como salida del estado oculto.
3. Principios de GRU
3.1 Estructura central y las dos compuertas
GRU fusiona la compuerta de olvido y la compuerta de entrada de LSTM en unacompuerta de actualización, y elimina el estado de celda independiente, conservando solo el estado oculto, lo que hace la estructura más simple.
重置门（Reset Gate）：决定忽略多少历史状态来计算候选隐藏状态 更新门（Update Gate）：决定保留多少历史状态，写入多少新状态
3.2 Fórmulas de cálculo hacia adelante
\[
\begin{aligned}
\text{Entrada: } & x_t \text{ (entrada del paso temporal actual)}, h_{t-1} \text{ (estado oculto del paso anterior)} \\
\text{Compuerta de reinicio: } & r_t = \sigma(W_r \cdot [h_{t-1}, x_t] + b_r) \\
\text{Compuerta de actualización: } & z_t = \sigma(W_z \cdot [h_{t-1}, x_t] + b_z) \\
\text{Valor candidato: } & \tilde{h}_t = \tanh(W_h \cdot [r_t \odot h_{t-1}, x_t] + b_h) \\
\text{Actualizar estado oculto: } & h_t = (1 - z_t) \odot h_{t-1} + z_t \odot \tilde{h}_t
\end{aligned}
\]
Interpretación de la lógica de cálculo:
- Compuerta de reinicio
r_tCuando está cerca de 0, el estado candidatoh̃_tcasi no depende del historial, lo que equivale a empezar de nuevo - Compuerta de actualización
z_tCuando está cerca de 1, el nuevo estado adopta más el valor candidato; cuando está cerca de 0, conserva más el estado histórico - GRU no tiene un estado de celda independiente, y la cantidad de parámetros es aproximadamente el 75% de la de LSTM.
4. LSTM en PyTorch
Esta sección detalla los parámetros de nn.LSTM, las formas de entrada/salida y los métodos de inicialización del estado oculto.
4.1 Explicación detallada de los parámetros de nn.LSTM
Ejemplo
import torch
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimensión del vector de entrada en cada paso temporal
hidden_size=128, # Dimensión del estado oculto (y del estado de celda)
num_layers=2, # Número de capas apiladas, por defecto 1
bias=True, # Si se usa sesgo (bias), por defecto True
batch_first=False, # Si batch está en la primera dimensión del shape de entrada/salida, por defecto False
dropout=0.0, # Probabilidad de dropout entre capas (solo tiene efecto cuando num_layers > 1)
bidirectional=False, # Si se usa LSTM bidireccional, por defecto False
proj_size=0, # Dimensión de la capa de proyección (LSTM con proyección); por defecto 0 indica que no se usa
)
# Ver la cantidad de parámetros
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Cantidad de parámetros LSTM: {total_params:,}")
# Con input_size=64, hidden_size=128, num_layers=2, es aproximadamente 197,632
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Dimensión del vector de entrada en cada paso temporal
hidden_size=128, # Dimensión del estado oculto (y del estado de celda)
num_layers=2, # Número de capas apiladas, por defecto 1
bias=True, # Si se usa sesgo (bias), por defecto True
batch_first=False, # Si batch está en la primera dimensión del shape de entrada/salida, por defecto False
dropout=0.0, # Probabilidad de dropout entre capas (solo tiene efecto cuando num_layers > 1)
bidirectional=False, # Si se usa LSTM bidireccional, por defecto False
proj_size=0, # Dimensión de la capa de proyección (LSTM con proyección); por defecto 0 indica que no se usa
)
# Ver la cantidad de parámetros
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Cantidad de parámetros LSTM: {total_params:,}")
# Con input_size=64, hidden_size=128, num_layers=2, es aproximadamente 197,632
Fórmula de estimación de la cantidad de parámetros (capa única unidireccional):
\[
\begin{aligned}
\text{Número de parámetros por capa} & = 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}
\]
donde 4 corresponde a cuatro grupos de matrices de pesos: compuerta de olvido, compuerta de entrada, valor candidato y compuerta de salida.
4.2 Formas de entrada y salida
Esta es la parte más propensa a errores al usar LSTM, se necesita especial atención.batch_firstLa influencia de los parámetros.
Ejemplo
import torch
import torch.nn as nn
# ── batch_first=False (predeterminado) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Forma de entrada: (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"forma de output: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Salida del estado oculto en cada paso de tiempo
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Estado oculto del último paso de tiempo
print(f"forma de c_n: {c_n.shape}")
# torch.Size([1, 4, 64]) → igual que h_n, estado de celda del último paso de tiempo
# ── batch_first=True (recomendado, más intuitivo) ────────────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Forma de entrada: (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"forma de output: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Nota: la forma de h_n no se ve afectada por batch_first
# ── Forma de salida de LSTM multicapa bidireccional ──────────────────────
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"forma de output: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, porque es concatenación bidireccional
print(f"forma 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 (predeterminado) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Forma de entrada: (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"forma de output: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Salida del estado oculto en cada paso de tiempo
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Estado oculto del último paso de tiempo
print(f"forma de c_n: {c_n.shape}")
# torch.Size([1, 4, 64]) → igual que h_n, estado de celda del último paso de tiempo
# ── batch_first=True (recomendado, más intuitivo) ────────────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Forma de entrada: (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"forma de output: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Nota: la forma de h_n no se ve afectada por batch_first
# ── Forma de salida de LSTM multicapa bidireccional ──────────────────────
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"forma de output: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, porque es concatenación bidireccional
print(f"forma de h_n: {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
Resumen de las formas de salida:
| 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 (bidireccional) o 1 (unidireccional)
4.3 Inicialización del estado oculto
Ejemplo
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étodo 1: no pasar estado inicial, PyTorch usa automáticamente inicialización con ceros
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Método 2: inicializar manualmente con ceros (equivalente al método 1, pero muestra claramente la intención)
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étodo 3: modo con estado (stateful) — transferir el estado entre batches
# Adecuado para segmentos de secuencias muy largas (modelos de lenguaje, generación de texto largo, etc.)
# Es necesario llamar a detach() para cortar el grafo computacional del batch anterior y evitar fugas de memoria
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() # Corta el gradiente, solo conserva los valores
c = c.detach()
# Método 4: usar inicialización de Xavier o distribución normal (converge más rápido en algunos escenarios)
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) # Inicialización ortogonal, ayuda a estabilizar el entrenamiento
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étodo 1: no pasar estado inicial, PyTorch usa automáticamente inicialización con ceros
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Método 2: inicializar manualmente con ceros (equivalente al método 1, pero muestra claramente la intención)
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étodo 3: modo con estado (stateful) — transferir el estado entre batches
# Adecuado para segmentos de secuencias muy largas (modelos de lenguaje, generación de texto largo, etc.)
# Es necesario llamar a detach() para cortar el grafo computacional del batch anterior y evitar fugas de memoria
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() # Corta el gradiente, solo conserva los valores
c = c.detach()
# Método 4: usar inicialización de Xavier o distribución normal (converge más rápido en algunos escenarios)
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) # Inicialización ortogonal, ayuda a estabilizar el entrenamiento
return h, c
5. GRU en PyTorch
La interfaz de GRU es casi idéntica a la de LSTM; la principal diferencia es que no tiene estado de celda.c。
5.1 Explicación detallada de los parámetros de nn.GRU
Ejemplo
import torch.nn as nn
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Se recomienda establecerlo en True
dropout=0.3, # dropout entre capas
bidirectional=False,
)
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Se recomienda establecerlo en True
dropout=0.3, # dropout entre capas
bidirectional=False,
)
5.2 Ejemplo básico de uso
Ejemplo
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 solo devuelve output y h_n, no tiene c_n
output, h_n = gru(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Tomar la salida del último paso de tiempo (para tareas como clasificación)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# o de forma equivalente:
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 solo devuelve output y h_n, no tiene c_n
output, h_n = gru(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"forma de h_n: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Tomar la salida del último paso de tiempo (para tareas como clasificación)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# o de forma equivalente:
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
6. Variantes de LSTM y GRU
Esta sección presenta estructuras bidireccionales, apilamiento multicapa y estructuras multicapa con Dropout.
6.1 LSTM / GRU bidireccional
Los modelos bidireccionales procesan información de la secuencia tanto en dirección directa como inversa; la salida de cada paso de tiempo contiene contexto pasado y futuro, adecuados para tareas que requieren contexto global, como clasificación de texto, reconocimiento de entidades nombradas, etc.
Ejemplo
import torch
import torch.nn as nn
# LSTM bidireccional
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Habilitar bidireccional
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 10, 128])
# 64 dimensiones hacia adelante + 64 dimensiones hacia atrás = 128 dimensiones
print(f"forma de h_n: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Separar el último estado oculto del forward y del backward
# Orden de h_n: [forward layer0, backward layer0, forward layer1, backward layer1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) última capa forward
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) última capa backward
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# El uso de GRU bidireccional es exactamente el mismo
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 bidireccional
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Habilitar bidireccional
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 10, 128])
# 64 dimensiones hacia adelante + 64 dimensiones hacia atrás = 128 dimensiones
print(f"forma de h_n: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Separar el último estado oculto del forward y del backward
# Orden de h_n: [forward layer0, backward layer0, forward layer1, backward layer1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) última capa forward
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) última capa backward
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# El uso de GRU bidireccional es exactamente el mismo
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
6.2 Apilamiento de múltiples capas
Ejemplo
import torch
import torch.nn as nn
# LSTM apilado de 3 capas
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Apilar 3 capas
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 20, 128]) solo contiene la salida de la capa superior
print(f"forma de h_n: {h_n.shape}")
# torch.Size([3, 8, 128]) incluye el estado oculto final de cada capa
# Obtener el estado oculto final de cada capa
h_layer1 = h_n[0] # (batch_size, hidden_size) primera capa
h_layer2 = h_n[1] # (batch_size, hidden_size) segunda capa
h_layer3 = h_n[2] # (batch_size, hidden_size) tercera capa (la más superior)
import torch.nn as nn
# LSTM apilado de 3 capas
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Apilar 3 capas
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"forma de output: {output.shape}")
# torch.Size([8, 20, 128]) solo contiene la salida de la capa superior
print(f"forma de h_n: {h_n.shape}")
# torch.Size([3, 8, 128]) incluye el estado oculto final de cada capa
# Obtener el estado oculto final de cada capa
h_layer1 = h_n[0] # (batch_size, hidden_size) primera capa
h_layer2 = h_n[1] # (batch_size, hidden_size) segunda capa
h_layer3 = h_n[2] # (batch_size, hidden_size) tercera capa (la más superior)
6.3 Estructura multicapa con Dropout
nn.LSTMincorporadodropoutEl parámetro solo actúaentre capasy no actúa sobre la salida de la última capa. Si se necesita añadir Dropout también después de la última capa, debe añadirse manualmente:
Ejemplo
import torch
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
LSTM multicapa + Dropout entre capas + Dropout de salida
"""
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,
# cuando num_layers=1, dropout no tiene efecto; establécelo en 0 para evitar advertencias
)
self.dropout = nn.Dropout(dropout) # Dropout después de la última capa
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Tomar el estado oculto del último paso de tiempo
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 multicapa + Dropout entre capas + Dropout de salida
"""
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,
# cuando num_layers=1, dropout no tiene efecto; establécelo en 0 para evitar advertencias
)
self.dropout = nn.Dropout(dropout) # Dropout después de la última capa
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Tomar el estado oculto del último paso de tiempo
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. Procesamiento de secuencias de longitud variable
En tareas reales, las longitudes de las secuencias dentro de un mismo batch suelen ser diferentes. PyTorch proporcionapack_padded_sequenceypad_packed_sequencepara manejar este problema y evitar que LSTM realice cálculos ineficaces en las posiciones de relleno (padding).
7.1 pack_padded_sequence
Ejemplo
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Simular secuencias de longitud variable en un batch (ordenadas de más larga a más corta)
seq1 = torch.randn(5, 32) # Longitud de secuencia 5
seq2 = torch.randn(3, 32) # Longitud de secuencia 3
seq3 = torch.randn(2, 32) # Longitud de secuencia 2
# pad_sequence rellena con ceros automáticamente, alineando a la secuencia más larga
# cuando batch_first=True, la forma de salida es (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"forma después del relleno con ceros: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: comprime el relleno, indica a LSTM la longitud real
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Las secuencias deben estar ordenadas por longitud en orden descendente
# enforce_sorted=False # permite cualquier orden (ordenamiento automático interno), se recomienda establecerlo en 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
# Simular secuencias de longitud variable en un batch (ordenadas de más larga a más corta)
seq1 = torch.randn(5, 32) # Longitud de secuencia 5
seq2 = torch.randn(3, 32) # Longitud de secuencia 3
seq3 = torch.randn(2, 32) # Longitud de secuencia 2
# pad_sequence rellena con ceros automáticamente, alineando a la secuencia más larga
# cuando batch_first=True, la forma de salida es (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"forma después del relleno con ceros: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: comprime el relleno, indica a LSTM la longitud real
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Las secuencias deben estar ordenadas por longitud en orden descendente
# enforce_sorted=False # permite cualquier orden (ordenamiento automático interno), se recomienda establecerlo en False
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
7.2 pad_packed_sequence
Ejemplo
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Pasar el PackedSequence a la LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: restaurar al tensor con relleno de ceros
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"forma de output tras la restauración: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# La salida en las posiciones de relleno es 0
print(f"longitud real de cada secuencia: {output_lengths}")
# tensor([5, 3, 2])
# Pasar el PackedSequence a la LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: restaurar al tensor con relleno de ceros
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"forma de output tras la restauración: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# La salida en las posiciones de relleno es 0
print(f"longitud real de cada secuencia: {output_lengths}")
# tensor([5, 3, 2])
7.3 Flujo de procesamiento completo para secuencias de longitud variable
Ejemplo
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) — índices de palabras
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Empaquetar
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Propagación hacia adelante de LSTM (omitiendo posiciones de relleno)
packed_output, (h_n, c_n) = self.lstm(packed)
# Método 1: Usar h_n de la última capa como representación de secuencia
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Método 2: después de desempaquetar, tomar la salida de la última posición real de cada secuencia (equivalente al método 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)
# Simular un batch
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Longitud real de cada secuencia
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) — índices de palabras
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Empaquetar
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Propagación hacia adelante de LSTM (omitiendo posiciones de relleno)
packed_output, (h_n, c_n) = self.lstm(packed)
# Método 1: Usar h_n de la última capa como representación de secuencia
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Método 2: después de desempaquetar, tomar la salida de la última posición real de cada secuencia (equivalente al método 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)
# Simular un batch
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Longitud real de cada secuencia
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
8. Práctica completa: Clasificación de sentimiento en texto
Tomando como ejemplo la clasificación binaria de sentimiento de reseñas de películas IMDB, se muestra el flujo completo de procesamiento de texto con LSTM bidireccional:
Ejemplo
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")
# ── Definición del modelo ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Modelo de clasificación de texto con LSTM bidireccional
Estructura: 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
)
# Al usar vectores de palabras preentrenados:
# 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 bidireccional: concatenar los últimos estados ocultos hacia adelante y hacia atrás
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)
# Tomar los estados ocultos hacia adelante y hacia atrás de la última capa y concatenarlos
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)
# ── Dataset personalizado ─────────────────────────────────
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 personalizado: rellenar con ceros secuencias de longitud variable y registrar la longitud real"""
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
# ── Funciones de entrenamiento y evaluación ────────────────────────────────
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()
# Recorte de gradiente: evitar la explosión del gradiente en 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
# ── Inicialización y entrenamiento ──────────────────────────────────
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"Epoch {epoch:2d}/{EPOCHS} | "
f"Train Loss: {train_loss:.4f}, Acc: {train_acc:.4f} | "
f"Val Loss: {val_loss:.4f}, Acc: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Guardar el mejor modelo, 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")
# ── Definición del modelo ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Modelo de clasificación de texto con LSTM bidireccional
Estructura: 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
)
# Al usar vectores de palabras preentrenados:
# 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 bidireccional: concatenar los últimos estados ocultos hacia adelante y hacia atrás
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)
# Tomar los estados ocultos hacia adelante y hacia atrás de la última capa y concatenarlos
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)
# ── Dataset personalizado ─────────────────────────────────
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 personalizado: rellenar con ceros secuencias de longitud variable y registrar la longitud real"""
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
# ── Funciones de entrenamiento y evaluación ────────────────────────────────
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()
# Recorte de gradiente: evitar la explosión del gradiente en 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
# ── Inicialización y entrenamiento ──────────────────────────────────
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"Epoch {epoch:2d}/{EPOCHS} | "
f"Train Loss: {train_loss:.4f}, Acc: {train_acc:.4f} | "
f"Val Loss: {val_loss:.4f}, Acc: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Guardar el mejor modelo, Val Acc: {best_acc:.4f}")
9. Práctica completa: Predicción de series temporales
Tomando como ejemplo la predicción de series temporales de múltiples pasos, se usa LSTM para predecir los valores de los próximos N pasos:
Ejemplo
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")
# ── Dataset de ventana deslizante ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Segmentación de series temporales con ventana deslizante
Entrada: pasos pasados input_len
Objetivo: pasos futuros output_len
"""
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,)
# ── Definición del modelo ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Modelo de predicción de series temporales de múltiples pasos
Estructura: 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)
# Tomar la salida del último paso de tiempo
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Preparación de datos (con onda sinusoidal como ejemplo)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Usar los 60 pasos pasados
OUTPUT_LEN = 10 # Predecir los próximos 10 pasos
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)
# ── Entrenamiento ──────────────────────────────────────────
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"Epoch {epoch:2d}/30 | Train MSE: {train_loss:.6f} | Val MSE: {val_loss:.6f}")
# ── Predicción recursiva de múltiples pasos (otra estrategia)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Predicción recursiva: predecir un paso cada vez, agregar el valor predicho a la secuencia y luego predecir el siguiente paso
Adecuado para modelos de predicción de un solo paso con 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")
# ── Dataset de ventana deslizante ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Segmentación de series temporales con ventana deslizante
Entrada: pasos pasados input_len
Objetivo: pasos futuros output_len
"""
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,)
# ── Definición del modelo ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Modelo de predicción de series temporales de múltiples pasos
Estructura: 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)
# Tomar la salida del último paso de tiempo
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Preparación de datos (con onda sinusoidal como ejemplo)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Usar los 60 pasos pasados
OUTPUT_LEN = 10 # Predecir los próximos 10 pasos
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)
# ── Entrenamiento ──────────────────────────────────────────
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"Epoch {epoch:2d}/30 | Train MSE: {train_loss:.6f} | Val MSE: {val_loss:.6f}")
# ── Predicción recursiva de múltiples pasos (otra estrategia)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Predicción recursiva: predecir un paso cada vez, agregar el valor predicho a la secuencia y luego predecir el siguiente paso
Adecuado para modelos de predicción de un solo paso con 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. Comparación entre LSTM y GRU y selección de modelo
Esta sección compara las diferencias estructurales entre LSTM y GRU, y ofrece recomendaciones de selección.
Comparación de estructuras
| Elemento de comparación | LSTM | GRU |
|---|---|---|
| Número de compuertas | 3 (olvido, entrada, salida) | 2 (reinicio, actualización) |
| Vector de estado | Estado de celda + estado oculto | Solo estado oculto |
| Cantidad de parámetros (mismo hidden_size) | Referencia | Aproximadamente 75% |
| Velocidad de entrenamiento | Más lenta | Más rápida |
| Rendimiento en secuencias largas | Generalmente mejor | Cercano |
| Rendimiento en secuencias cortas | Similar | Similar |
| Complejidad de implementación | Alta | Baja |
Recomendaciones de selección
数据量小、训练资源有限
-> 优先选 GRU，参数少，不容易过拟合，训练快
序列较长（> 100 步）、长距离依赖重要
-> 优先选 LSTM，细胞状态更擅长保留远期信息
需要快速实验和基线对比
-> 先用 GRU，效果差再换 LSTM
任务对准确率要求极高，有充足数据
-> 两者都试，结合交叉验证选择
2020 年后的新项目
-> 考虑 Transformer 架构（BERT、GPT），在大数据量下通常优于 LSTM/GRU
LSTM/GRU 仍在边缘设备、低延迟推理、数据量较小的场景中有优势
Consulta rápida de preguntas frecuentes
| Problema | Causa | Solución |
|---|---|---|
| La pérdida de entrenamiento no disminuye | Tasa de aprendizaje demasiado alta o explosión del gradiente | Reducir la tasa de aprendizaje; agregar recorte de gradienteclip_grad_norm_ |
| La pérdida del conjunto de validación es mucho mayor que la del conjunto de entrenamiento | Sobreajuste | Aumentar Dropout; reducir el número de capas o hidden_size |
| Predicción deficiente al final de la secuencia | Desvanecimiento del gradiente | Aumentar el número de capas; usar estructura bidireccional; considerar el mecanismo de atención |
| Uso excesivo de memoria de video | Secuencia demasiado larga o batch demasiado grande | Reducir seq_len o batch_size; usar puntos de control de gradiente |
| Pérdida anormal en entrenamiento con estado | El estado transferido entre batches no se ha separado (detach) | Llamar después de cada batchh.detach_() |
| Error de dimensión de concatenación en LSTM bidireccional | Comprensión incorrecta del índice de h_n | usarh_n[-2](hacia adelante) yh_n[-1](hacia atrás) |
| Error de PackedSequence | Las longitudes de secuencia no están ordenadas en orden descendente | Configuraciónenforce_sorted=False |
| Confusión con batch_first | Olvidar configurar de manera uniforme | Se recomienda usar en todo el procesobatch_first=True |