Objetivo
Entender como redes recorrentes processam sequências, por que LSTMs resolvem o problema de dependências de longo prazo, implementar RNNs e LSTMs com PyTorch, e aplicar essas arquiteturas em classificação de texto e previsão de séries temporais.
1. Por que Dados Sequenciais São Diferentes?
MLPs e CNNs assumem que os exemplos são independentes entre si. Mas muitos problemas do mundo real têm estrutura sequencial onde a ordem importa:
- Texto: "O banco está na margem do rio" vs "Fui ao banco sacar dinheiro"
- Séries temporais: preço de ações, temperatura, vendas mensais
- Áudio: fala, música
- Código fonte: a estrutura sintática depende do que veio antes
- Genomas: sequências de DNA
Para processar sequências, precisamos de uma arquitetura que: - Mantenha memória do que foi processado anteriormente - Lide com sequências de tamanho variável - Capture dependências entre elementos distantes
2. Redes Recorrentes (RNN)
A ideia central de uma RNN é um estado oculto que é passado de um passo para o próximo, funcionando como uma "memória de curto prazo".
hₜ = tanh(Wₓ × xₜ + Wₕ × hₜ₋₁ + b)
yₜ = Wᵧ × hₜ + bᵧ
Onde: - xₜ é a entrada no passo t - hₜ é o estado oculto no passo t - hₜ₋₁ é o estado oculto do passo anterior - yₜ é a saída no passo t
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import accuracy_score, classification_report
from sklearn.preprocessing import MinMaxScaler
import warnings
warnings.filterwarnings("ignore")
torch.manual_seed(42)
np.random.seed(42)
sns.set_theme(style="whitegrid")
device = torch.device("cuda" if torch.cuda.is_available()
else "cpu")
print(f"Dispositivo: {device}")
# ── Implementação manual de RNN ──────────────────────────
class RNNManual(nn.Module):
"""
RNN vanilla implementada manualmente.
Mostra explicitamente como o estado oculto é propagado.
"""
def __init__(self, n_entrada: int,
n_oculto: int,
n_saida: int):
super().__init__()
self.n_oculto = n_oculto
# Pesos
self.Wx = nn.Linear(n_entrada, n_oculto, bias=False)
self.Wh = nn.Linear(n_oculto, n_oculto, bias=True)
self.Wy = nn.Linear(n_oculto, n_saida, bias=True)
def forward(self, x: torch.Tensor,
h0: torch.Tensor = None) -> tuple:
"""
x: (batch, seq_len, n_entrada)
h0: (batch, n_oculto) — estado inicial (opcional)
"""
batch_size, seq_len, _ = x.shape
# Estado oculto inicial (zeros se não fornecido)
if h0 is None:
h = torch.zeros(batch_size, self.n_oculto,
device=x.device)
else:
h = h0
saidas = []
# Processar cada passo da sequência
for t in range(seq_len):
xt = x[:, t, :] # (batch, n_entrada)
h = torch.tanh(
self.Wx(xt) + self.Wh(h)
) # (batch, n_oculto)
y = self.Wy(h) # (batch, n_saida)
saidas.append(y.unsqueeze(1))
saidas = torch.cat(saidas, dim=1) # (batch, seq, n_saida)
return saidas, h
# Teste da RNN manual
rnn_manual = RNNManual(n_entrada=4,
n_oculto=8,
n_saida=2)
x_seq = torch.randn(3, 10, 4) # batch=3, seq=10, features=4
saidas, h_final = rnn_manual(x_seq)
print(f"RNN Manual:")
print(f" Entrada: {x_seq.shape}")
print(f" Saídas: {saidas.shape}")
print(f" Estado final: {h_final.shape}")
# Comparar com nn.RNN do PyTorch
rnn_pytorch = nn.RNN(
input_size=4,
hidden_size=8,
num_layers=1,
batch_first=True, # (batch, seq, features)
nonlinearity="tanh"
)
saidas_pt, h_pt = rnn_pytorch(x_seq)
print(f"\nnn.RNN (PyTorch):")
print(f" Saídas: {saidas_pt.shape}")
print(f" Estado final: {h_pt.shape}")
3. O Problema do Vanishing Gradient em RNNs
def demonstrar_vanishing_rnn():
"""
Mostra como gradientes desaparecem em RNNs longas.
"""
seq_lengths = [5, 10, 20, 50, 100]
normas_grad = []
for seq_len in seq_lengths:
rnn = nn.RNN(input_size=1, hidden_size=16,
batch_first=True)
x = torch.randn(1, seq_len, 1, requires_grad=True)
out, h = rnn(x)
loss = h.sum()
loss.backward()
# Norma do gradiente em relação à entrada
norma = x.grad.norm().item()
normas_grad.append(norma)
print(f" seq_len={seq_len:3d}: norma_grad={norma:.6f}")
fig, ax = plt.subplots(figsize=(10, 5))
ax.semilogy(seq_lengths, normas_grad,
marker="o", color="coral",
linewidth=2.5, markersize=8)
ax.set_xlabel("Comprimento da Sequência")
ax.set_ylabel("Norma do Gradiente (log)")
ax.set_title("Vanishing Gradient em RNNs:\n"
"Gradiente encolhe exponencialmente "
"com o comprimento da sequência")
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
print("Vanishing Gradient em RNNs:")
demonstrar_vanishing_rnn()
print("\nConsequência: RNNs vanilla esquecem dependências longas.")
print("Solução: LSTM e GRU com mecanismos de portas (gates).")
4. LSTM — Long Short-Term Memory
A LSTM resolve o vanishing gradient com dois estados: o estado oculto h (memória de curto prazo) e o estado de célula c (memória de longo prazo), controlados por portas (gates).
Porta de esquecimento: f = σ(Wf × [h,x] + bf)
Porta de entrada: i = σ(Wi × [h,x] + bi)
Candidato de célula: g = tanh(Wg × [h,x] + bg)
Porta de saída: o = σ(Wo × [h,x] + bo)
Atualização da célula: c_t = f ⊙ c_{t-1} + i ⊙ g
Estado oculto: h_t = o ⊙ tanh(c_t)
class LSTMManual(nn.Module):
"""
LSTM implementada manualmente — mostra cada porta explicitamente.
"""
def __init__(self, n_entrada: int, n_oculto: int):
super().__init__()
self.n_oculto = n_oculto
n = n_entrada + n_oculto # dimensão concatenada
# Quatro portas em uma matriz (eficiente)
# Ordem: forget, input, gate, output
self.W = nn.Linear(n, 4 * n_oculto, bias=True)
def forward(self, x: torch.Tensor,
estado: tuple = None) -> tuple:
"""
x: (batch, seq_len, n_entrada)
estado: (h0, c0) — estados iniciais
"""
batch, seq_len, _ = x.shape
if estado is None:
h = torch.zeros(batch, self.n_oculto,
device=x.device)
c = torch.zeros(batch, self.n_oculto,
device=x.device)
else:
h, c = estado
saidas = []
for t in range(seq_len):
xt = x[:, t, :]
# Concatenar entrada e estado oculto
hx = torch.cat([h, xt], dim=1)
# Calcular todas as portas de uma vez
portas = self.W(hx)
# Separar as portas
f = torch.sigmoid(portas[:, 0:self.n_oculto])
i = torch.sigmoid(portas[:, self.n_oculto:2*self.n_oculto])
g = torch.tanh(portas[:, 2*self.n_oculto:3*self.n_oculto])
o = torch.sigmoid(portas[:, 3*self.n_oculto:])
# Atualizar estado de célula (memória longa)
c = f * c + i * g
# Calcular estado oculto (memória curta)
h = o * torch.tanh(c)
saidas.append(h.unsqueeze(1))
saidas = torch.cat(saidas, dim=1)
return saidas, (h, c)
# Comparar LSTM manual com PyTorch
lstm_manual = LSTMManual(n_entrada=4, n_oculto=8)
lstm_pytorch = nn.LSTM(input_size=4, hidden_size=8,
batch_first=True)
x_seq = torch.randn(3, 10, 4)
saidas_manual, (h_m, c_m) = lstm_manual(x_seq)
saidas_pt, (h_p, c_p) = lstm_pytorch(x_seq)
print("LSTM Manual vs PyTorch:")
print(f" Manual — saídas: {saidas_manual.shape}, "
f"h: {h_m.shape}, c: {c_m.shape}")
print(f" PyTorch — saídas: {saidas_pt.shape}, "
f"h: {h_p.shape}, c: {c_p.shape}")
# Visualizar as portas
def visualizar_portas_lstm(lstm: LSTMManual,
sequencia: torch.Tensor):
"""Visualiza os valores das portas ao longo da sequência."""
lstm.eval()
batch, seq_len, _ = sequencia.shape
h = torch.zeros(batch, lstm.n_oculto)
c = torch.zeros(batch, lstm.n_oculto)
f_vals, i_vals, g_vals, o_vals = [], [], [], []
c_vals = []
with torch.no_grad():
for t in range(seq_len):
xt = sequencia[:, t, :]
hx = torch.cat([h, xt], dim=1)
portas = lstm.W(hx)
f = torch.sigmoid(portas[:, :lstm.n_oculto])
i = torch.sigmoid(portas[:, lstm.n_oculto:2*lstm.n_oculto])
g = torch.tanh(portas[:, 2*lstm.n_oculto:3*lstm.n_oculto])
o = torch.sigmoid(portas[:, 3*lstm.n_oculto:])
c = f * c + i * g
h = o * torch.tanh(c)
f_vals.append(f[0].mean().item())
i_vals.append(i[0].mean().item())
g_vals.append(g[0].mean().item())
o_vals.append(o[0].mean().item())
c_vals.append(c[0].mean().item())
fig, axes = plt.subplots(2, 1, figsize=(12, 6))
t = range(seq_len)
axes[0].plot(t, f_vals, label="Forget gate",
color="coral", linewidth=2)
axes[0].plot(t, i_vals, label="Input gate",
color="steelblue", linewidth=2)
axes[0].plot(t, o_vals, label="Output gate",
color="mediumseagreen", linewidth=2)
axes[0].set_ylabel("Valor da Porta (0-1)")
axes[0].set_title("Ativações das Portas LSTM")
axes[0].legend()
axes[0].set_ylim(0, 1)
axes[1].plot(t, c_vals, label="Estado de Célula",
color="mediumpurple", linewidth=2)
axes[1].set_xlabel("Passo da Sequência")
axes[1].set_ylabel("Valor Médio")
axes[1].set_title("Evolução do Estado de Célula")
axes[1].legend()
plt.tight_layout()
plt.show()
# Sequência sintética com padrão
x_demo_lstm = torch.sin(
torch.linspace(0, 4*np.pi, 30)
).reshape(1, 30, 1).repeat(1, 1, 4)
visualizar_portas_lstm(lstm_manual, x_demo_lstm)
5. GRU — Gated Recurrent Unit
GRU é uma simplificação da LSTM com duas portas (vs quatro), sem o estado de célula separado, mas com performance similar na maioria dos casos.
# Comparação LSTM vs GRU
print("Comparação LSTM vs GRU:")
print(f"{'Aspecto':<25} {'LSTM':>20} {'GRU':>20}")
print("-" * 67)
comparacoes = [
("Estados internos", "h + c (2 estados)", "h (1 estado)"),
("Portas", "4 (f, i, g, o)", "2 (r, z)"),
("Parâmetros (h=128)", "~264K", "~198K"),
("Memória longa", "Excelente", "Boa"),
("Velocidade", "Moderada", "Mais rápida"),
("Seq. muito longas", "Melhor", "Ligeiramente pior"),
]
for aspecto, lstm_v, gru_v in comparacoes:
print(f"{aspecto:<25} {lstm_v:>20} {gru_v:>20}")
# Implementação com PyTorch
gru = nn.GRU(input_size=4, hidden_size=8, batch_first=True)
x_seq = torch.randn(3, 10, 4)
saidas_gru, h_gru = gru(x_seq)
print(f"\nGRU PyTorch:")
print(f" Saídas: {saidas_gru.shape}")
print(f" Estado final: {h_gru.shape}")
6. Classificação de Texto com LSTM
# Dataset sintético de análise de sentimentos
class SentimentDataset(Dataset):
"""
Dataset sintético de classificação de sentimentos.
Simula reviews positivos e negativos.
"""
def __init__(self, n: int = 2000,
max_len: int = 30,
vocab_size: int = 500):
np.random.seed(42)
self.max_len = max_len
self.vocab_size = vocab_size
# Palavras positivas (índices 1-100)
# Palavras negativas (índices 101-200)
# Palavras neutras (índices 201-vocab_size)
self.sequencias = []
self.labels = []
for _ in range(n):
label = np.random.randint(0, 2)
if label == 1: # positivo
# Mais palavras positivas
n_pos = np.random.randint(5, 15)
n_neg = np.random.randint(0, 3)
else: # negativo
n_pos = np.random.randint(0, 3)
n_neg = np.random.randint(5, 15)
n_neutro = np.random.randint(5, 15)
seq_len = n_pos + n_neg + n_neutro
palavras_pos = np.random.randint(1, 101, n_pos)
palavras_neg = np.random.randint(101, 201, n_neg)
palavras_neut = np.random.randint(201,
vocab_size,
n_neutro)
seq = np.concatenate([palavras_pos,
palavras_neg,
palavras_neut])
np.random.shuffle(seq)
# Padding ou truncamento
if len(seq) >= max_len:
seq = seq[:max_len]
else:
seq = np.pad(seq,
(0, max_len - len(seq)),
constant_values=0)
self.sequencias.append(seq)
self.labels.append(label)
self.sequencias = np.array(self.sequencias,
dtype=np.int64)
self.labels = np.array(self.labels,
dtype=np.float32)
def __len__(self) -> int:
return len(self.labels)
def __getitem__(self, idx: int):
return (torch.from_numpy(self.sequencias[idx]),
torch.tensor(self.labels[idx]))
# Criar dataset e dataloaders
ds_sent = SentimentDataset(n=3000, max_len=30,
vocab_size=500)
n_tr_s = int(0.7 * len(ds_sent))
n_va_s = int(0.15 * len(ds_sent))
n_te_s = len(ds_sent) - n_tr_s - n_va_s
from torch.utils.data import random_split
ds_tr_s, ds_va_s, ds_te_s = random_split(
ds_sent, [n_tr_s, n_va_s, n_te_s],
generator=torch.Generator().manual_seed(42)
)
dl_tr_s = DataLoader(ds_tr_s, batch_size=64, shuffle=True)
dl_va_s = DataLoader(ds_va_s, batch_size=64)
dl_te_s = DataLoader(ds_te_s, batch_size=64)
print(f"Dataset de Sentimentos:")
print(f" Treino: {len(ds_tr_s)}, "
f"Val: {len(ds_va_s)}, "
f"Teste: {len(ds_te_s)}")
# Modelo LSTM para classificação de sentimentos
class LSTMSentimentos(nn.Module):
"""
LSTM bidirecional para classificação de sentimentos.
Embedding + BiLSTM + Attention + Classificador.
"""
def __init__(self,
vocab_size: int,
embed_dim: int = 64,
hidden_dim: int = 128,
n_layers: int = 2,
dropout: float = 0.3,
n_classes: int = 1):
super().__init__()
# Embedding: mapeia índices para vetores densos
self.embedding = nn.Embedding(
num_embeddings=vocab_size,
embedding_dim=embed_dim,
padding_idx=0 # índice 0 = padding (zeros)
)
# LSTM bidirecional (processa sequência em ambas direções)
self.lstm = nn.LSTM(
input_size=embed_dim,
hidden_size=hidden_dim,
num_layers=n_layers,
batch_first=True,
dropout=dropout if n_layers > 1 else 0,
bidirectional=True # 2× mais estados ocultos
)
self.dropout = nn.Dropout(dropout)
# Classificador
# Bidirecional → hidden_dim × 2
self.classificador = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim, n_classes)
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""
x: (batch, seq_len) — índices das palavras
"""
# Embedding: (batch, seq) → (batch, seq, embed_dim)
embedded = self.dropout(self.embedding(x))
# LSTM: (batch, seq, embed) → (batch, seq, hidden*2)
output, (h, c) = self.lstm(embedded)
# Usar os estados finais de ambas direções
# h shape: (n_layers*2, batch, hidden)
# Pegar última camada, ambas direções
h_forward = h[-2, :, :] # última camada, forward
h_backward = h[-1, :, :] # última camada, backward
h_combined = torch.cat([h_forward, h_backward],
dim=1) # (batch, hidden*2)
h_combined = self.dropout(h_combined)
return self.classificador(h_combined)
def contar_parametros(self) -> int:
return sum(p.numel() for p in self.parameters()
if p.requires_grad)
# Instanciar e treinar
modelo_sent = LSTMSentimentos(
vocab_size=500,
embed_dim=64,
hidden_dim=128,
n_layers=2,
dropout=0.3
).to(device)
print(f"\nModelo LSTM Sentimentos:")
print(f" Parâmetros: {modelo_sent.contar_parametros():,}")
criterion_sent = nn.BCEWithLogitsLoss()
optimizer_sent = optim.AdamW(
modelo_sent.parameters(),
lr=1e-3, weight_decay=1e-4
)
scheduler_sent = optim.lr_scheduler.ReduceLROnPlateau(
optimizer_sent, patience=5, factor=0.5
)
n_epocas_sent = 30
hist_sent = {"treino_loss": [], "treino_acc": [],
"val_loss": [], "val_acc": []}
melhor_val_sent = float("inf")
print(f"\nTreinando LSTM por {n_epocas_sent} épocas...")
for epoca in range(n_epocas_sent):
# Treino
modelo_sent.train()
losses_tr, n_corr_tr, n_tot_tr = [], 0, 0
for X_b, y_b in dl_tr_s:
X_b = X_b.to(device)
y_b = y_b.to(device)
optimizer_sent.zero_grad()
logits = modelo_sent(X_b).squeeze()
loss = criterion_sent(logits, y_b)
loss.backward()
nn.utils.clip_grad_norm_(
modelo_sent.parameters(), 1.0
)
optimizer_sent.step()
losses_tr.append(loss.item())
preds = (torch.sigmoid(logits) >= 0.5)
n_corr_tr += (preds == y_b.bool()).sum().item()
n_tot_tr += len(y_b)
# Validação
modelo_sent.eval()
losses_va, n_corr_va, n_tot_va = [], 0, 0
with torch.no_grad():
for X_b, y_b in dl_va_s:
X_b = X_b.to(device)
y_b = y_b.to(device)
logits = modelo_sent(X_b).squeeze()
loss = criterion_sent(logits, y_b)
losses_va.append(loss.item())
preds = (torch.sigmoid(logits) >= 0.5)
n_corr_va += (preds == y_b.bool()).sum().item()
n_tot_va += len(y_b)
loss_tr = np.mean(losses_tr)
acc_tr = n_corr_tr / n_tot_tr
loss_va = np.mean(losses_va)
acc_va = n_corr_va / n_tot_va
hist_sent["treino_loss"].append(loss_tr)
hist_sent["treino_acc"].append(acc_tr)
hist_sent["val_loss"].append(loss_va)
hist_sent["val_acc"].append(acc_va)
scheduler_sent.step(loss_va)
if loss_va < melhor_val_sent:
melhor_val_sent = loss_va
torch.save(modelo_sent.state_dict(),
"lstm_sent_melhor.pt")
if (epoca + 1) % 10 == 0:
print(f" Época {epoca+1:2d}/{n_epocas_sent} | "
f"Loss: {loss_tr:.4f}/{loss_va:.4f} | "
f"Acc: {acc_tr:.4f}/{acc_va:.4f}")
# Avaliar
modelo_sent.load_state_dict(
torch.load("lstm_sent_melhor.pt", weights_only=True)
)
modelo_sent.eval()
preds_s, labels_s = [], []
with torch.no_grad():
for X_b, y_b in dl_te_s:
X_b = X_b.to(device)
logits = modelo_sent(X_b).squeeze()
p = (torch.sigmoid(logits) >= 0.5).cpu().numpy()
preds_s.extend(p.astype(int))
labels_s.extend(y_b.numpy().astype(int))
acc_sent = accuracy_score(labels_s, preds_s)
print(f"\nAcurácia no teste: {acc_sent:.4f}")
print(classification_report(
labels_s, preds_s,
target_names=["Negativo", "Positivo"]
))
7. Previsão de Séries Temporais
# ── Dataset de Série Temporal ────────────────────────────
def gerar_serie_temporal(n: int = 2000,
freq: float = 0.1,
ruido: float = 0.1) -> np.ndarray:
"""Gera série temporal com múltiplos componentes."""
t = np.arange(n)
# Tendência + sazonalidade + ruído
serie = (0.3 * np.sin(2 * np.pi * freq * t)
+ 0.2 * np.sin(2 * np.pi * freq * 3 * t)
+ 0.1 * np.cos(2 * np.pi * freq * 5 * t)
+ 0.05 * t / n # tendência leve
+ np.random.normal(0, ruido, n))
return serie.astype(np.float32)
class SerieTemporalDataset(Dataset):
"""
Dataset de série temporal para previsão.
Dado uma janela de tamanho `janela`, prever o próximo valor.
"""
def __init__(self, serie: np.ndarray,
janela: int = 30,
horizonte: int = 1):
self.janela = janela
self.horizonte = horizonte
self.X = []
self.y = []
for i in range(len(serie) - janela - horizonte + 1):
self.X.append(serie[i:i + janela])
self.y.append(serie[i + janela:
i + janela + horizonte])
self.X = np.array(self.X).reshape(-1, janela, 1)
self.y = np.array(self.y).reshape(-1, horizonte)
def __len__(self):
return len(self.X)
def __getitem__(self, idx):
return (torch.from_numpy(self.X[idx]),
torch.from_numpy(self.y[idx]))
# Gerar e preparar dados
np.random.seed(42)
serie_bruta = gerar_serie_temporal(n=3000)
# Normalizar
scaler_ts = MinMaxScaler()
serie_norm = scaler_ts.fit_transform(
serie_bruta.reshape(-1, 1)
).ravel().astype(np.float32)
# Dividir em treino/teste (temporal — não aleatório!)
n_treino = int(0.7 * len(serie_norm))
n_val = int(0.15 * len(serie_norm))
n_teste = len(serie_norm) - n_treino - n_val
serie_tr = serie_norm[:n_treino]
serie_va = serie_norm[n_treino:n_treino + n_val]
serie_te = serie_norm[n_treino + n_val:]
JANELA = 30
ds_ts_tr = SerieTemporalDataset(serie_tr, JANELA)
ds_ts_va = SerieTemporalDataset(serie_va, JANELA)
ds_ts_te = SerieTemporalDataset(serie_te, JANELA)
dl_ts_tr = DataLoader(ds_ts_tr, batch_size=64, shuffle=True)
dl_ts_va = DataLoader(ds_ts_va, batch_size=64)
dl_ts_te = DataLoader(ds_ts_te, batch_size=64)
print(f"Série Temporal:")
print(f" Total: {len(serie_norm)} pontos")
print(f" Treino: {len(ds_ts_tr)}, "
f"Val: {len(ds_ts_va)}, "
f"Teste: {len(ds_ts_te)} janelas")
# Modelo LSTM para previsão de série temporal
class LSTMPrevisao(nn.Module):
"""LSTM para previsão de série temporal."""
def __init__(self,
n_features: int = 1,
hidden_dim: int = 64,
n_layers: int = 2,
horizonte: int = 1,
dropout: float = 0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=n_features,
hidden_size=hidden_dim,
num_layers=n_layers,
batch_first=True,
dropout=dropout if n_layers > 1 else 0,
)
self.saida = nn.Sequential(
nn.Linear(hidden_dim, 32),
nn.ReLU(),
nn.Linear(32, horizonte)
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""x: (batch, janela, n_features)"""
output, (h, c) = self.lstm(x)
# Usar apenas o último estado oculto
return self.saida(output[:, -1, :])
# Treinar modelo de série temporal
modelo_ts = LSTMPrevisao(
n_features=1, hidden_dim=64,
n_layers=2, horizonte=1, dropout=0.2
).to(device)
opt_ts = optim.Adam(modelo_ts.parameters(), lr=1e-3)
sch_ts = optim.lr_scheduler.ReduceLROnPlateau(
opt_ts, patience=5, factor=0.5
)
n_epocas_ts = 50
hist_ts = {"treino": [], "val": []}
melhor_ts = float("inf")
print(f"\nTreinando LSTM para série temporal...")
for epoca in range(n_epocas_ts):
modelo_ts.train()
losses_tr_ts = []
for X_b, y_b in dl_ts_tr:
X_b = X_b.to(device)
y_b = y_b.to(device)
opt_ts.zero_grad()
pred = modelo_ts(X_b)
loss = F.mse_loss(pred, y_b)
loss.backward()
nn.utils.clip_grad_norm_(modelo_ts.parameters(), 1.0)
opt_ts.step()
losses_tr_ts.append(loss.item())
modelo_ts.eval()
losses_va_ts = []
with torch.no_grad():
for X_b, y_b in dl_ts_va:
X_b = X_b.to(device)
y_b = y_b.to(device)
pred = modelo_ts(X_b)
loss = F.mse_loss(pred, y_b)
losses_va_ts.append(loss.item())
loss_tr_ts = np.mean(losses_tr_ts)
loss_va_ts = np.mean(losses_va_ts)
hist_ts["treino"].append(loss_tr_ts)
hist_ts["val"].append(loss_va_ts)
sch_ts.step(loss_va_ts)
if loss_va_ts < melhor_ts:
melhor_ts = loss_va_ts
torch.save(modelo_ts.state_dict(), "lstm_ts_melhor.pt")
if (epoca + 1) % 10 == 0:
print(f" Época {epoca+1:2d}/{n_epocas_ts} | "
f"MSE Treino: {loss_tr_ts:.6f} | "
f"MSE Val: {loss_va_ts:.6f}")
# Avaliar e visualizar predições
modelo_ts.load_state_dict(
torch.load("lstm_ts_melhor.pt", weights_only=True)
)
modelo_ts.eval()
preds_ts = []
labels_ts = []
with torch.no_grad():
for X_b, y_b in dl_ts_te:
X_b = X_b.to(device)
pred = modelo_ts(X_b).cpu().numpy()
preds_ts.extend(pred.ravel())
labels_ts.extend(y_b.numpy().ravel())
preds_ts = np.array(preds_ts)
labels_ts = np.array(labels_ts)
# Desnormalizar
preds_orig = scaler_ts.inverse_transform(
preds_ts.reshape(-1, 1)
).ravel()
labels_orig = scaler_ts.inverse_transform(
labels_ts.reshape(-1, 1)
).ravel()
rmse_ts = np.sqrt(np.mean((preds_orig - labels_orig)**2))
mae_ts = np.mean(np.abs(preds_orig - labels_orig))
print(f"\nResultados Série Temporal (escala original):")
print(f" RMSE: {rmse_ts:.6f}")
print(f" MAE: {mae_ts:.6f}")
# Visualizar
fig, axes = plt.subplots(2, 1, figsize=(14, 8))
# Curvas de aprendizado
ep_ts = range(1, n_epocas_ts + 1)
axes[0].semilogy(ep_ts, hist_ts["treino"],
label="Treino", color="steelblue",
linewidth=2)
axes[0].semilogy(ep_ts, hist_ts["val"],
label="Validação", color="coral",
linewidth=2, linestyle="--")
axes[0].set_xlabel("Época")
axes[0].set_ylabel("MSE Loss (log)")
axes[0].set_title("Convergência LSTM — Série Temporal")
axes[0].legend()
# Predições vs Real
n_plot = min(200, len(preds_orig))
axes[1].plot(labels_orig[:n_plot], label="Real",
color="steelblue", linewidth=1.5, alpha=0.8)
axes[1].plot(preds_orig[:n_plot], label="Predito",
color="coral", linewidth=1.5,
linestyle="--", alpha=0.8)
axes[1].set_xlabel("Passo de Tempo")
axes[1].set_ylabel("Valor")
axes[1].set_title("Predição vs Real (conjunto de teste)")
axes[1].legend()
plt.suptitle("LSTM para Série Temporal", fontsize=13)
plt.tight_layout()
plt.show()
8. Comparação RNN vs LSTM vs GRU
print("\n" + "=" * 60)
print("COMPARAÇÃO: RNN vs LSTM vs GRU")
print("=" * 60)
# Modelos para comparação
modelos_seq = {
"RNN": nn.RNN(
input_size=1, hidden_size=64,
num_layers=2, batch_first=True,
dropout=0.2
),
"LSTM": nn.LSTM(
input_size=1, hidden_size=64,
num_layers=2, batch_first=True,
dropout=0.2
),
"GRU": nn.GRU(
input_size=1, hidden_size=64,
num_layers=2, batch_first=True,
dropout=0.2
),
}
resultados_seq = {}
for nome, rnn_base in modelos_seq.items():
# Wrapper para saída consistente
class ModeloSeq(nn.Module):
def __init__(self, rnn):
super().__init__()
self.rnn = rnn
self.fc = nn.Sequential(
nn.Linear(64, 32),
nn.ReLU(),
nn.Linear(32, 1)
)
def forward(self, x):
if isinstance(self.rnn, nn.LSTM):
out, (h, _) = self.rnn(x)
else:
out, h = self.rnn(x)
return self.fc(out[:, -1, :])
modelo_c = ModeloSeq(rnn_base).to(device)
n_params = sum(p.numel() for p in modelo_c.parameters())
opt_c = optim.Adam(modelo_c.parameters(), lr=1e-3)
melhor_c = float("inf")
hist_c = []
for epoca in range(30):
modelo_c.train()
for X_b, y_b in dl_ts_tr:
X_b = X_b.to(device)
y_b = y_b.to(device)
opt_c.zero_grad()
pred = modelo_c(X_b)
loss = F.mse_loss(pred, y_b)
loss.backward()
nn.utils.clip_grad_norm_(
modelo_c.parameters(), 1.0
)
opt_c.step()
modelo_c.eval()
losses_va_c = []
with torch.no_grad():
for X_b, y_b in dl_ts_va:
X_b = X_b.to(device)
y_b = y_b.to(device)
loss = F.mse_loss(modelo_c(X_b), y_b)
losses_va_c.append(loss.item())
loss_va_c = np.mean(losses_va_c)
hist_c.append(loss_va_c)
if loss_va_c < melhor_c:
melhor_c = loss_va_c
# Teste
modelo_c.eval()
losses_te_c = []
with torch.no_grad():
for X_b, y_b in dl_ts_te:
X_b = X_b.to(device)
y_b = y_b.to(device)
loss = F.mse_loss(modelo_c(X_b), y_b)
losses_te_c.append(loss.item())
resultados_seq[nome] = {
"teste_mse": np.mean(losses_te_c),
"n_params": n_params,
"hist": hist_c
}
print(f"\n{nome}:")
print(f" Parâmetros: {n_params:,}")
print(f" Melhor Val: {melhor_c:.6f}")
print(f" Teste MSE: {np.mean(losses_te_c):.6f}")
# Gráfico comparativo
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
cores_seq = ["coral", "steelblue", "mediumseagreen"]
for (nome, res), cor in zip(resultados_seq.items(),
cores_seq):
axes[0].semilogy(res["hist"], label=nome,
color=cor, linewidth=2)
axes[0].set_xlabel("Época")
axes[0].set_ylabel("Validation MSE (log)")
axes[0].set_title("Convergência: RNN vs LSTM vs GRU")
axes[0].legend()
nomes_m = list(resultados_seq.keys())
mses_te = [resultados_seq[n]["teste_mse"] for n in nomes_m]
params_m = [resultados_seq[n]["n_params"] for n in nomes_m]
x_pos = np.arange(len(nomes_m))
bars = axes[1].bar(x_pos, mses_te,
color=cores_seq, alpha=0.8)
axes[1].set_xticks(x_pos)
axes[1].set_xticklabels(nomes_m)
axes[1].set_ylabel("Teste MSE")
axes[1].set_title("Desempenho no Teste")
for bar, p in zip(bars, params_m):
axes[1].text(
bar.get_x() + bar.get_width()/2,
bar.get_height() * 1.02,
f"{p:,}\nparams",
ha="center", fontsize=8
)
plt.suptitle("RNN vs LSTM vs GRU — Série Temporal",
fontsize=13)
plt.tight_layout()
plt.show()
9. Arquiteturas Avançadas
# Seq2Seq com Encoder-Decoder
class EncoderLSTM(nn.Module):
"""Codifica a sequência de entrada em um contexto."""
def __init__(self, n_features: int,
hidden_dim: int,
n_layers: int):
super().__init__()
self.lstm = nn.LSTM(
n_features, hidden_dim, n_layers,
batch_first=True, dropout=0.2
)
def forward(self, x):
_, (h, c) = self.lstm(x)
return h, c # contexto
class DecoderLSTM(nn.Module):
"""Decodifica o contexto em previsões futuras."""
def __init__(self, hidden_dim: int,
n_layers: int,
horizonte: int):
super().__init__()
self.lstm = nn.LSTM(
1, hidden_dim, n_layers,
batch_first=True, dropout=0.2
)
self.fc = nn.Linear(hidden_dim, 1)
self.horizonte = horizonte
def forward(self, contexto_h, contexto_c,
batch_size: int,
device: torch.device):
# Entrada inicial: zeros (start token)
decoder_input = torch.zeros(
batch_size, 1, 1, device=device
)
h, c = contexto_h, contexto_c
saidas = []
for _ in range(self.horizonte):
out, (h, c) = self.lstm(decoder_input, (h, c))
pred = self.fc(out)
saidas.append(pred)
decoder_input = pred # autoregressive
return torch.cat(saidas, dim=1)
class Seq2Seq(nn.Module):
"""Arquitetura Encoder-Decoder para previsão multi-step."""
def __init__(self, n_features: int = 1,
hidden_dim: int = 64,
n_layers: int = 2,
horizonte: int = 5):
super().__init__()
self.encoder = EncoderLSTM(n_features, hidden_dim,
n_layers)
self.decoder = DecoderLSTM(hidden_dim, n_layers,
horizonte)
self.horizonte = horizonte
def forward(self, x: torch.Tensor) -> torch.Tensor:
h, c = self.encoder(x)
saidas = self.decoder(h, c,
batch_size=x.shape[0],
device=x.device)
return saidas.squeeze(-1)
# Testar Seq2Seq
seq2seq = Seq2Seq(horizonte=5).to(device)
x_s2s = torch.randn(4, 30, 1).to(device)
y_s2s = seq2seq(x_s2s)
print(f"Seq2Seq:")
print(f" Entrada: {x_s2s.shape}")
print(f" Saída: {y_s2s.shape} (4 exemplos, 5 passos)")
n_s2s = sum(p.numel() for p in seq2seq.parameters())
print(f" Parâmetros: {n_s2s:,}")
10. Exemplo Completo: Pipeline Sequencial
print("\n" + "=" * 60)
print("PIPELINE COMPLETO — LSTM BIDIRECIONAL")
print("=" * 60)
# Modelo final otimizado para série temporal
class LSTMBiDirecional(nn.Module):
"""
LSTM bidirecional com atenção para série temporal.
Processa sequência em ambas direções.
"""
def __init__(self, n_features: int = 1,
hidden_dim: int = 64,
n_layers: int = 2,
dropout: float = 0.2):
super().__init__()
self.lstm = nn.LSTM(
n_features, hidden_dim, n_layers,
batch_first=True, dropout=dropout,
bidirectional=True
)
# *2 por ser bidirecional
self.atencao = nn.Linear(hidden_dim * 2, 1)
self.saida = nn.Sequential(
nn.Linear(hidden_dim * 2, 32),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(32, 1)
)
def forward(self, x: torch.Tensor) -> torch.Tensor:
output, _ = self.lstm(x)
# Atenção simples: peso para cada passo
pesos = torch.softmax(
self.atencao(output), dim=1
)
contexto = (pesos * output).sum(dim=1)
return self.saida(contexto)
modelo_bi = LSTMBiDirecional().to(device)
n_bi = sum(p.numel() for p in modelo_bi.parameters())
print(f"\nLSTM Bidirecional com Atenção: {n_bi:,} parâmetros")
opt_bi = optim.AdamW(modelo_bi.parameters(),
lr=1e-3, weight_decay=1e-4)
sch_bi = optim.lr_scheduler.OneCycleLR(
opt_bi, max_lr=5e-3,
steps_per_epoch=len(dl_ts_tr),
epochs=40, pct_start=0.2
)
hist_bi = {"treino": [], "val": []}
melhor_bi = float("inf")
for epoca in range(40):
modelo_bi.train()
for X_b, y_b in dl_ts_tr:
X_b = X_b.to(device)
y_b = y_b.to(device)
opt_bi.zero_grad()
pred = modelo_bi(X_b).squeeze()
loss = F.mse_loss(pred, y_b.squeeze())
loss.backward()
nn.utils.clip_grad_norm_(modelo_bi.parameters(), 1.0)
opt_bi.step()
sch_bi.step()
modelo_bi.eval()
losses_va_bi = []
with torch.no_grad():
for X_b, y_b in dl_ts_va:
X_b = X_b.to(device)
y_b = y_b.to(device)
pred = modelo_bi(X_b).squeeze()
loss = F.mse_loss(pred, y_b.squeeze())
losses_va_bi.append(loss.item())
loss_va_bi = np.mean(losses_va_bi)
hist_bi["val"].append(loss_va_bi)
if loss_va_bi < melhor_bi:
melhor_bi = loss_va_bi
torch.save(modelo_bi.state_dict(), "lstm_bi_melhor.pt")
if (epoca + 1) % 10 == 0:
print(f" Época {epoca+1:2d}/40 | "
f"Val MSE: {loss_va_bi:.6f}")
modelo_bi.load_state_dict(
torch.load("lstm_bi_melhor.pt", weights_only=True)
)
modelo_bi.eval()
preds_bi, labels_bi = [], []
with torch.no_grad():
for X_b, y_b in dl_ts_te:
X_b = X_b.to(device)
pred = modelo_bi(X_b).squeeze().cpu().numpy()
preds_bi.extend(np.atleast_1d(pred))
labels_bi.extend(y_b.numpy().ravel())
preds_bi = np.array(preds_bi)
labels_bi = np.array(labels_bi)
rmse_bi = np.sqrt(np.mean(
(scaler_ts.inverse_transform(preds_bi.reshape(-1, 1)).ravel()
- scaler_ts.inverse_transform(labels_bi.reshape(-1, 1)).ravel()) ** 2
))
print(f"\nLSTM Bidirecional — RMSE no teste: {rmse_bi:.6f}")
print(f"LSTM Simples — RMSE no teste: {rmse_ts:.6f}")
# Visualização final
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# Comparação de predições
n_plot = 100
preds_bi_orig = scaler_ts.inverse_transform(
preds_bi[:n_plot].reshape(-1, 1)
).ravel()
preds_orig_cmp = scaler_ts.inverse_transform(
preds_ts[:n_plot].reshape(-1, 1)
).ravel()
labels_orig_cmp = scaler_ts.inverse_transform(
labels_bi[:n_plot].reshape(-1, 1)
).ravel()
axes[0].plot(labels_orig_cmp, label="Real",
color="gray", linewidth=2, alpha=0.7)
axes[0].plot(preds_orig_cmp, label="LSTM Simples",
color="coral", linewidth=1.5, linestyle="--")
axes[0].plot(preds_bi_orig, label="LSTM BiDirecional",
color="steelblue", linewidth=1.5,
linestyle=":")
axes[0].set_title("Predições no Conjunto de Teste")
axes[0].set_xlabel("Passo de Tempo")
axes[0].legend()
# Tabela resumo
modelos_resumo = {
"LSTM Simples": rmse_ts,
"LSTM BiDirecional": rmse_bi,
}
cores_r = ["coral", "steelblue"]
barras_r = axes[1].bar(
modelos_resumo.keys(),
modelos_resumo.values(),
color=cores_r, alpha=0.85
)
for b, v in zip(barras_r, modelos_resumo.values()):
axes[1].text(b.get_x() + b.get_width()/2,
v * 1.02, f"{v:.5f}",
ha="center", fontsize=10)
axes[1].set_title("RMSE por Modelo (menor = melhor)")
axes[1].set_ylabel("RMSE")
plt.suptitle("Pipeline Sequencial Completo", fontsize=13)
plt.tight_layout()
plt.show()
Resumo da Aula
- RNNs processam sequências mantendo um estado oculto propagado de passo em passo
- O problema do vanishing gradient em RNNs impede aprender dependências longas
- LSTMs resolvem isso com dois estados (h e c) e quatro portas: forget, input, gate e output
- A porta de esquecimento controla o que esquecer; a de entrada controla o que memorizar
- GRUs são simplificações das LSTMs com duas portas, performance similar e menos parâmetros
- LSTMs bidirecionais processam a sequência em ambas as direções para contexto completo
- Para série temporal, dividir os dados em ordem cronológica (não aleatória) é fundamental
- O estado
output[:, -1, :]contém o resumo de toda a sequência para classificação - Seq2Seq (encoder-decoder) permite previsão de múltiplos passos futuros
- Atenção pondera quais passos da sequência são mais relevantes para a predição
Exercícios
-
Explique matematicamente como o estado oculto de uma RNN é atualizado a cada passo. Por que usar tanh como ativação em vez de ReLU? O que aconteceria com o estado oculto se não houvesse função de ativação?
✓ Resposta:A atualização do estado oculto a cada passo t é:
hₜ = tanh(Wₓxₜ + Wₕhₜ₋₁ + b).A função tanh é usada por dois motivos. Primeiro, ela produz valores no intervalo (-1, 1), o que mantém o estado oculto em uma escala controlada ao longo de muitos passos — sem isso, o estado poderia crescer indefinidamente. Segundo, ela é centrada em zero, o que acelera a convergência em comparação com a sigmoid.
Se não houvesse função de ativação:
hₜ = Wₓxₜ + Wₕhₜ₋₁ + b. Isso seria uma transformação puramente linear. Após n passos, hₜ seria uma transformação linear de todas as entradas anteriores. O modelo perderia a capacidade de aprender dependências não lineares — essencialmente seria uma regressão linear sobre a sequência, sem memória seletiva.ReLU em RNNs causa instabilidade porque o estado oculto pode crescer indefinidamente (valores maiores que 1 não são limitados pela ativação), levando a exploding gradients. Existem variantes como IRNN que usam ReLU com inicialização identidade para mitigar isso, mas não são o padrão.
-
Descreva as quatro portas da LSTM (forget, input, gate, output) e o papel de cada uma. Como o estado de célula
cé diferente do estado ocultohem termos de o que cada um representa?✓ Resposta:Porta de esquecimento
f = σ(Wf[h,x] + bf): decide que proporção do estado de célula anterior manter. Valores próximos de 1 significam "lembrar completamente", próximos de 0 significam "esquecer completamente". Permite que a rede aprenda a descartar informação irrelevante.Porta de entrada
i = σ(Wi[h,x] + bi): decide quais novas informações serão escritas no estado de célula. É multiplicada pelo candidato g para filtrar quais valores novos são relevantes.Candidato de célula
g = tanh(Wg[h,x] + bg): os novos valores candidatos a serem adicionados ao estado de célula, sem filtro ainda.Porta de saída
o = σ(Wo[h,x] + bo): decide quais partes do estado de célula serão expostas como estado oculto para a próxima camada.O estado de célula
cé a "memória de longo prazo": flui com alterações graduais (adição e esquecimento), podendo carregar informação por centenas de passos sem atenuação. O estado ocultohé a "memória de curto prazo" ou saída: é o estado de célula filtrado pela porta de saída, representando o que o modelo quer comunicar para a próxima camada.carmazena;hcomunica. -
Por que ao trabalhar com séries temporais você NÃO deve embaralhar os dados antes de dividir em treino/teste? Qual problema isso causaria e como você deve fazer a divisão corretamente?
✓ Resposta:Séries temporais têm dependência temporal: o valor de amanhã é influenciado pelos valores passados. Se você embaralhar e dividir aleatoriamente, o conjunto de teste pode incluir datas anteriores às datas de treino. O modelo "veria o futuro" durante o treino — data leakage temporal.
Exemplo concreto: se sua série vai de janeiro a dezembro e você embaralha, o modelo de treino pode ter aprendido com dados de novembro e dezembro, e então ser avaliado em outubro. Na prática, outubro é anterior a novembro — o modelo usou dados "futuros" para aprender a prever o passado. Isso produz métricas irrealisticamente boas que não se repetem na produção.
A divisão correta é estritamente cronológica: - Treino: período mais antigo (ex: jan-ago) - Validação: período intermediário (ex: set-out) - Teste: período mais recente (ex: nov-dez)
Isso simula fielmente o cenário real de produção: o modelo aprende com o passado e é avaliado em dados futuros que nunca viu.
-
O que é uma LSTM bidirecional? Quais são suas vantagens e desvantagens em comparação com uma LSTM unidirecional? Para qual tipo de tarefa uma LSTM bidirecional NÃO seria adequada?
✓ Resposta:Uma LSTM bidirecional executa dois LSTMs em paralelo: um processa a sequência da esquerda para a direita (forward), o outro da direita para a esquerda (backward). Em cada passo t, os estados ocultos das duas direções são concatenados, dando ao modelo acesso ao contexto tanto anterior quanto posterior ao passo atual.
Vantagens: contexto completo da sequência — ao classificar uma palavra no meio de uma frase, o modelo considera tanto o que vem antes quanto o que vem depois. Geralmente superior para tarefas como classificação de texto e NER.
Desvantagens: dobra o número de parâmetros e o tempo de processamento. Requer a sequência inteira antes de processar qualquer passo — não pode ser usada em tempo real.
LSTM bidirecional NÃO seria adequada para: previsão de série temporal (você não pode usar valores futuros para prever o presente), geração de texto (você precisa gerar token por token, sem acesso ao futuro), e qualquer tarefa online onde os dados chegam em tempo real e precisam de resposta imediata.
-
Explique o conceito de word embeddings em modelos de NLP. Por que usar embeddings aprendíveis é melhor que representação one-hot para palavras? O que o parâmetro
padding_idx=0faz nonn.Embedding?✓ Resposta:Word embeddings mapeiam palavras (índices inteiros) para vetores densos de dimensão fixa (ex: 64 ou 300 dimensões) que o modelo aprende durante o treinamento. Palavras semanticamente similares tendem a ter vetores próximos no espaço de embeddings.
One-hot é problemático por dois motivos. Primeiro, dimensionalidade: para um vocabulário de 50.000 palavras, cada palavra seria um vetor de 50.000 dimensões com 49.999 zeros — extremamente esparso e ineficiente. Segundo, sem semântica: "gato" e "felino" são tão distantes quanto "gato" e "avião" na representação one-hot. Embeddings aprendem que palavras similares têm representações similares.
padding_idx=0: quando uma sequência é menor que o comprimento máximo, ela é preenchida com zeros (padding). O parâmetropadding_idx=0instrui a camada de Embedding a sempre retornar um vetor de zeros para o índice 0 e a não atualizar esses vetores durante o backpropagation. Isso garante que o padding não influencie os embeddings das palavras reais. -
Compare LSTMs com Transformers para processamento de sequências. Quais são as limitações das LSTMs que os Transformers resolvem? Por que LSTMs ainda são usadas em alguns cenários?
✓ Resposta:Limitações das LSTMs que Transformers resolvem:
Processamento sequencial: LSTMs processam tokens um por vez, impedindo paralelização durante o treino. Transformers processam toda a sequência simultaneamente usando atenção — muito mais rápidos em hardware moderno.
Dependências longas: embora LSTMs sejam melhores que RNNs simples, ainda têm dificuldade com dependências muito longas (centenas de tokens). Transformers têm acesso direto a qualquer posição via atenção — a distância entre dois tokens não aumenta o custo de conectá-los.
Contexto global: cada passo da LSTM só "vê" diretamente o passo anterior. No Transformer, o mecanismo de atenção calcula relações entre todos os pares de tokens simultaneamente.
LSTMs ainda são usadas quando: a sequência é gerada token-por-token em tempo real (streaming), o dataset é pequeno e Transformers sofreriam overfitting, a latência de inferência é crítica e modelos simples são suficientes, ou recursos computacionais são limitados (dispositivos embarcados, IoT).