Treinamento de Redes Neurais com PyTorch

Objetivo

Dominar o ciclo completo de treinamento de redes neurais com PyTorch: construção de modelos robustos com nn.Module, técnicas avançadas de treinamento, callbacks, monitoramento e boas práticas de produção. Esta aula integra tudo que foi visto nas últimas três aulas em um framework de treinamento profissional.

1. Estrutura Profissional de Treinamento

Um loop de treinamento robusto vai muito além do básico. Em projetos reais você precisa de:

• Monitoramento de métricas por época
• Salvamento do melhor modelo automaticamente
• Early stopping com restauração do melhor estado
• Learning rate scheduling
• Gradient clipping para estabilidade
• Logging estruturado do progresso
• Suporte transparente a CPU e GPU

Vamos construir esse framework do zero.

2. Módulos e Datasets Personalizados

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,
                                TensorDataset, random_split)
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.datasets import make_classification, make_regression
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import (accuracy_score, f1_score,
                              roc_auc_score, r2_score,
                              mean_absolute_error)
from collections import defaultdict
import time
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}")

# ── Dataset Personalizado ────────────────────────────────
class TabelarDataset(Dataset):
    """
    Dataset genérico para dados tabulares.
    Herda de torch.utils.data.Dataset.
    Obrigatório implementar __len__ e __getitem__.
    """

    def __init__(self,
                  X: np.ndarray,
                  y: np.ndarray,
                  transform=None):
        self.X         = torch.from_numpy(
            X.astype(np.float32)
        )
        self.y         = torch.from_numpy(
            y.astype(np.float32)
        )
        self.transform = transform

    def __len__(self) -> int:
        return len(self.X)

    def __getitem__(self, idx: int):
        x = self.X[idx]
        y = self.y[idx]
        if self.transform:
            x = self.transform(x)
        return x, y

    @property
    def n_features(self) -> int:
        return self.X.shape[1]

# Dataset de exemplo
np.random.seed(42)
X_raw, y_raw = make_classification(
    n_samples=2000,
    n_features=20,
    n_informative=10,
    n_redundant=5,
    n_clusters_per_class=2,
    class_sep=0.8,
    random_state=42
)

# Divisão e normalização
X_tr_r, X_te_r, y_tr_r, y_te_r = train_test_split(
    X_raw, y_raw,
    test_size=0.2, random_state=42, stratify=y_raw
)
X_tr_r, X_va_r, y_tr_r, y_va_r = train_test_split(
    X_tr_r, y_tr_r,
    test_size=0.15, random_state=42, stratify=y_tr_r
)

sc = StandardScaler()
X_tr_r = sc.fit_transform(X_tr_r).astype(np.float32)
X_va_r = sc.transform(X_va_r).astype(np.float32)
X_te_r = sc.transform(X_te_r).astype(np.float32)

# Criar datasets
ds_tr = TabelarDataset(X_tr_r, y_tr_r)
ds_va = TabelarDataset(X_va_r, y_va_r)
ds_te = TabelarDataset(X_te_r, y_te_r)

# DataLoaders
dl_tr = DataLoader(ds_tr, batch_size=64,
                    shuffle=True,  drop_last=True)
dl_va = DataLoader(ds_va, batch_size=64,
                    shuffle=False, drop_last=False)
dl_te = DataLoader(ds_te, batch_size=64,
                    shuffle=False, drop_last=False)

print(f"Dataset: {X_raw.shape}")
print(f"Treino: {len(ds_tr)}, Val: {len(ds_va)}, "
      f"Teste: {len(ds_te)}")
print(f"Batches por época: {len(dl_tr)}")

3. Blocos de Construção de Modelos

# ── Bloco Residual (inspirado no ResNet) ─────────────────
class BlocoResidual(nn.Module):
    """
    Conexão residual: saída = F(x) + x
    Resolve o vanishing gradient em redes profundas.
    """

    def __init__(self, dim: int, dropout: float = 0.1):
        super().__init__()
        self.bloco = nn.Sequential(
            nn.Linear(dim, dim),
            nn.BatchNorm1d(dim),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(dim, dim),
            nn.BatchNorm1d(dim),
        )
        self.relu = nn.ReLU()

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.relu(self.bloco(x) + x)  # skip connection

# ── Modelo com Blocos Residuais ──────────────────────────
class RedeResidual(nn.Module):
    """
    Rede neural com blocos residuais para dados tabulares.
    Mais robusta que MLP simples para dados complexos.
    """

    def __init__(self,
                  n_entrada: int,
                  dim_oculta: int = 128,
                  n_blocos: int = 3,
                  dropout: float = 0.2,
                  n_saida: int = 1):
        super().__init__()

        # Projeção de entrada para dimensão oculta
        self.entrada = nn.Sequential(
            nn.Linear(n_entrada, dim_oculta),
            nn.BatchNorm1d(dim_oculta),
            nn.ReLU(),
        )

        # Blocos residuais
        self.blocos = nn.ModuleList([
            BlocoResidual(dim_oculta, dropout)
            for _ in range(n_blocos)
        ])

        # Cabeça de classificação
        self.saida = nn.Sequential(
            nn.Linear(dim_oculta, dim_oculta // 2),
            nn.ReLU(),
            nn.Dropout(dropout / 2),
            nn.Linear(dim_oculta // 2, n_saida)
        )

        # Inicialização de pesos
        self._inicializar_pesos()

    def _inicializar_pesos(self):
        """Inicialização de He para todas as camadas lineares."""
        for m in self.modules():
            if isinstance(m, nn.Linear):
                nn.init.kaiming_normal_(
                    m.weight, mode="fan_in",
                    nonlinearity="relu"
                )
                nn.init.zeros_(m.bias)
            elif isinstance(m, nn.BatchNorm1d):
                nn.init.ones_(m.weight)
                nn.init.zeros_(m.bias)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        x = self.entrada(x)
        for bloco in self.blocos:
            x = bloco(x)
        return self.saida(x)

    def contar_parametros(self) -> dict:
        total      = sum(p.numel() for p in self.parameters())
        treinavel  = sum(p.numel() for p in self.parameters()
                         if p.requires_grad)
        return {"total": total, "treinavel": treinavel}

# Inspecionar o modelo
modelo_res = RedeResidual(
    n_entrada=20,
    dim_oculta=128,
    n_blocos=3,
    dropout=0.2
)
info = modelo_res.contar_parametros()
print(f"\nRede Residual:")
print(f"  Parâmetros totais:     {info['total']:,}")
print(f"  Parâmetros treináveis: {info['treinavel']:,}")

# Verificar shape de saída
x_dummy = torch.randn(4, 20)
saida_dummy = modelo_res(x_dummy)
print(f"  Input:  {x_dummy.shape}")
print(f"  Output: {saida_dummy.shape}")

4. Framework de Treinamento Profissional

class Trainer:
    """
    Framework completo de treinamento PyTorch.
    Encapsula o loop de treino com todas as boas práticas.
    """

    def __init__(self,
                  modelo: nn.Module,
                  criterion,
                  optimizer,
                  scheduler=None,
                  device: torch.device = None,
                  gradient_clip: float = 1.0,
                  early_stopping_patience: int = 15,
                  checkpoint_path: str = "melhor_modelo.pt"):

        self.modelo     = modelo.to(
            device or torch.device("cpu")
        )
        self.criterion  = criterion
        self.optimizer  = optimizer
        self.scheduler  = scheduler
        self.device     = device or torch.device("cpu")
        self.grad_clip  = gradient_clip
        self.patience   = early_stopping_patience
        self.ckpt_path  = checkpoint_path

        self.historico  = defaultdict(list)
        self.melhor_val = float("inf")
        self.sem_melhora = 0
        self.epoca_melhor = 0

    def _epoca_treino(self,
                       dl: DataLoader) -> dict:
        """Executa uma época de treino."""
        self.modelo.train()
        losses, n_total, n_correto = [], 0, 0

        for X_b, y_b in dl:
            X_b = X_b.to(self.device)
            y_b = y_b.to(self.device)

            # ── Os 5 passos ──────────────────────────
            self.optimizer.zero_grad()
            logits = self.modelo(X_b).squeeze()
            loss   = self.criterion(logits, y_b)
            loss.backward()

            # Gradient clipping — evita exploding gradients
            if self.grad_clip > 0:
                nn.utils.clip_grad_norm_(
                    self.modelo.parameters(),
                    max_norm=self.grad_clip
                )

            self.optimizer.step()
            # ─────────────────────────────────────────

            losses.append(loss.item())

            # Acurácia do batch
            with torch.no_grad():
                preds    = (torch.sigmoid(logits) >= 0.5)
                n_correto += (preds == y_b.bool()).sum().item()
                n_total  += len(y_b)

        return {
            "loss": np.mean(losses),
            "acuracia": n_correto / n_total
        }

    @torch.no_grad()
    def _epoca_avaliacao(self,
                          dl: DataLoader) -> dict:
        """Executa uma época de avaliação."""
        self.modelo.eval()
        losses, n_total, n_correto = [], 0, 0
        todos_probs  = []
        todos_labels = []

        for X_b, y_b in dl:
            X_b    = X_b.to(self.device)
            y_b    = y_b.to(self.device)
            logits = self.modelo(X_b).squeeze()
            loss   = self.criterion(logits, y_b)

            losses.append(loss.item())
            probs = torch.sigmoid(logits)
            preds = (probs >= 0.5)

            n_correto += (preds == y_b.bool()).sum().item()
            n_total   += len(y_b)
            todos_probs.append(probs.cpu().numpy())
            todos_labels.append(y_b.cpu().numpy())

        todos_probs  = np.concatenate(todos_probs)
        todos_labels = np.concatenate(todos_labels)

        return {
            "loss":     np.mean(losses),
            "acuracia": n_correto / n_total,
            "auc":      roc_auc_score(todos_labels,
                                       todos_probs),
            "f1":       f1_score(todos_labels,
                                  (todos_probs >= 0.5),
                                  zero_division=0)
        }

    def treinar(self,
                dl_treino: DataLoader,
                dl_val: DataLoader,
                n_epocas: int = 100,
                verbose: int = 10) -> "Trainer":
        """
        Loop principal de treinamento.
        """
        print(f"Treinando por até {n_epocas} épocas "
              f"(patience={self.patience})...")
        t_inicio = time.time()

        for epoca in range(1, n_epocas + 1):
            # Treino e avaliação
            metr_tr = self._epoca_treino(dl_treino)
            metr_va = self._epoca_avaliacao(dl_val)

            # Registrar histórico
            for k, v in metr_tr.items():
                self.historico[f"treino_{k}"].append(v)
            for k, v in metr_va.items():
                self.historico[f"val_{k}"].append(v)

            # Learning rate scheduler
            if self.scheduler is not None:
                if isinstance(self.scheduler,
                               optim.lr_scheduler.ReduceLROnPlateau):
                    self.scheduler.step(metr_va["loss"])
                else:
                    self.scheduler.step()

            # Early stopping e checkpoint
            if metr_va["loss"] < self.melhor_val:
                self.melhor_val   = metr_va["loss"]
                self.sem_melhora  = 0
                self.epoca_melhor = epoca
                torch.save(self.modelo.state_dict(),
                            self.ckpt_path)
            else:
                self.sem_melhora += 1
                if self.sem_melhora >= self.patience:
                    print(f"\nEarly stopping na época {epoca}.")
                    print(f"Melhor: época {self.epoca_melhor}, "
                          f"val_loss={self.melhor_val:.4f}")
                    break

            # Log
            if verbose > 0 and epoca % verbose == 0:
                lr_atual = self.optimizer.param_groups[0]["lr"]
                print(f"Época {epoca:4d}/{n_epocas} | "
                      f"Loss: {metr_tr['loss']:.4f}/"
                      f"{metr_va['loss']:.4f} | "
                      f"Acc: {metr_tr['acuracia']:.4f}/"
                      f"{metr_va['acuracia']:.4f} | "
                      f"AUC: {metr_va['auc']:.4f} | "
                      f"LR: {lr_atual:.6f}")

        # Restaurar melhor modelo
        self.modelo.load_state_dict(
            torch.load(self.ckpt_path, weights_only=True)
        )
        t_total = time.time() - t_inicio
        print(f"\nTreinamento concluído em {t_total:.1f}s")
        print(f"Melhor época: {self.epoca_melhor}, "
              f"Val Loss: {self.melhor_val:.4f}")
        return self

    @torch.no_grad()
    def avaliar(self, dl: DataLoader) -> dict:
        """Avalia o modelo no DataLoader fornecido."""
        return self._epoca_avaliacao(dl)

    @torch.no_grad()
    def predict_proba(self,
                       X: np.ndarray) -> np.ndarray:
        """Retorna probabilidades para array NumPy."""
        self.modelo.eval()
        X_t = torch.from_numpy(
            X.astype(np.float32)
        ).to(self.device)
        logits = self.modelo(X_t).squeeze()
        return torch.sigmoid(logits).cpu().numpy()

    def plotar_historico(self,
                          metricas: list = None):
        """Plota curvas de aprendizado."""
        if metricas is None:
            metricas = ["loss", "acuracia", "auc"]

        n_metr = len(metricas)
        fig, axes = plt.subplots(
            1, n_metr, figsize=(5 * n_metr, 4)
        )
        if n_metr == 1:
            axes = [axes]

        for ax, m in zip(axes, metricas):
            ep = range(1, len(self.historico[f"treino_{m}"]) + 1)
            ax.plot(ep, self.historico[f"treino_{m}"],
                     label="Treino", color="steelblue",
                     linewidth=2)
            if f"val_{m}" in self.historico:
                ax.plot(ep, self.historico[f"val_{m}"],
                         label="Validação", color="coral",
                         linewidth=2, linestyle="--")
            ax.axvline(x=self.epoca_melhor, color="green",
                        linestyle=":", alpha=0.7,
                        label=f"Melhor (é.{self.epoca_melhor})")
            ax.set_title(m.capitalize())
            ax.set_xlabel("Época")
            ax.legend(fontsize=8)

        plt.suptitle("Curvas de Aprendizado", fontsize=13)
        plt.tight_layout()
        plt.show()

5. Treinando o Modelo

print("=" * 60)
print("TREINAMENTO COM FRAMEWORK PROFISSIONAL")
print("=" * 60)

# Instanciar modelo e mover para device
modelo = RedeResidual(
    n_entrada=ds_tr.n_features,
    dim_oculta=128,
    n_blocos=3,
    dropout=0.2
).to(device)

# Configurar treinamento
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.AdamW(
    modelo.parameters(),
    lr=1e-3,
    weight_decay=1e-4   # L2 regularização
)
scheduler = optim.lr_scheduler.CosineAnnealingLR(
    optimizer,
    T_max=100,
    eta_min=1e-5
)

# Instanciar Trainer
trainer = Trainer(
    modelo=modelo,
    criterion=criterion,
    optimizer=optimizer,
    scheduler=scheduler,
    device=device,
    gradient_clip=1.0,
    early_stopping_patience=20,
    checkpoint_path="rede_residual.pt"
)

# Treinar
trainer.treinar(
    dl_treino=dl_tr,
    dl_val=dl_va,
    n_epocas=150,
    verbose=25
)

# Avaliar no teste
metr_te = trainer.avaliar(dl_te)
print(f"\nResultados no Teste:")
print(f"  Loss:     {metr_te['loss']:.4f}")
print(f"  Acurácia: {metr_te['acuracia']:.4f}")
print(f"  AUC:      {metr_te['auc']:.4f}")
print(f"  F1-Score: {metr_te['f1']:.4f}")

# Curvas de aprendizado
trainer.plotar_historico(["loss", "acuracia", "auc"])

6. Learning Rate Schedulers

print("\n" + "=" * 60)
print("COMPARAÇÃO DE LEARNING RATE SCHEDULERS")
print("=" * 60)

# Visualizar o efeito de cada scheduler
fig, axes = plt.subplots(2, 3, figsize=(16, 8))
axes = axes.ravel()

def simular_lrs(nome_scheduler, scheduler_fn,
                n_epocas=100, lr_inicial=0.01):
    """Simula e retorna os LRs de um scheduler."""
    opt_fake = optim.SGD(
        [nn.Parameter(torch.zeros(1))],
        lr=lr_inicial
    )
    sch      = scheduler_fn(opt_fake)
    lrs      = []

    for _ in range(n_epocas):
        lrs.append(opt_fake.param_groups[0]["lr"])
        if isinstance(sch,
                       optim.lr_scheduler.ReduceLROnPlateau):
            sch.step(np.random.random())
        else:
            sch.step()

    return lrs

schedulers_demo = {
    "StepLR\n(reduz a cada N épocas)": lambda opt:
        optim.lr_scheduler.StepLR(opt, step_size=20,
                                   gamma=0.5),
    "ExponentialLR\n(decaimento exponencial)": lambda opt:
        optim.lr_scheduler.ExponentialLR(opt, gamma=0.95),
    "CosineAnnealingLR\n(ciclos coseno)": lambda opt:
        optim.lr_scheduler.CosineAnnealingLR(
            opt, T_max=50, eta_min=1e-5
        ),
    "ReduceLROnPlateau\n(reduz quando estagna)": lambda opt:
        optim.lr_scheduler.ReduceLROnPlateau(
            opt, patience=10, factor=0.5
        ),
    "OneCycleLR\n(warm-up + decaimento)": lambda opt:
        optim.lr_scheduler.OneCycleLR(
            opt, max_lr=0.1,
            total_steps=100,
            pct_start=0.3
        ),
    "CosineAnnealingWR\n(warm restarts)": lambda opt:
        optim.lr_scheduler.CosineAnnealingWarmRestarts(
            opt, T_0=25, T_mult=2
        ),
}

for ax, (nome, sch_fn) in zip(axes, schedulers_demo.items()):
    lrs = simular_lrs(nome, sch_fn)
    ax.plot(lrs, color="steelblue", linewidth=2)
    ax.set_title(nome, fontsize=9)
    ax.set_xlabel("Época")
    ax.set_ylabel("Learning Rate")
    ax.set_yscale("log")

plt.suptitle("Comparação de LR Schedulers", fontsize=13)
plt.tight_layout()
plt.show()

print("\nGuia de uso:")
print("  StepLR:            redução periódica e previsível")
print("  ExponentialLR:     decaimento suave e contínuo")
print("  CosineAnnealingLR: oscila suavemente — boa generalização")
print("  ReduceLROnPlateau: adapta ao progresso real do treino")
print("  OneCycleLR:        treino rápido com superconvergência")
print("  CosineAnnealingWR: re-explora espaço de parâmetros")

7. Gradient Clipping

# Demonstração do gradient clipping
print("\nGradient Clipping:")
print("=" * 50)

def demonstrar_gradient_clipping():
    """Mostra o efeito do gradient clipping."""

    np.random.seed(42)
    torch.manual_seed(42)

    # Dataset pequeno com dados mal escalados (causa gradientes grandes)
    X_clip = np.random.randn(200, 5).astype(np.float32) * 100
    y_clip = (X_clip[:, 0] > 0).astype(np.float32)

    modelo_gc = nn.Sequential(
        nn.Linear(5, 32), nn.ReLU(),
        nn.Linear(32, 1)
    )

    resultados_gc = {}

    for max_norm in [None, 10.0, 1.0, 0.1]:
        torch.manual_seed(42)
        modelo_gc_c = nn.Sequential(
            nn.Linear(5, 32), nn.ReLU(),
            nn.Linear(32, 1)
        )
        opt_gc = optim.SGD(modelo_gc_c.parameters(), lr=0.01)
        crit_gc = nn.BCEWithLogitsLoss()

        X_t_gc = torch.from_numpy(X_clip)
        y_t_gc = torch.from_numpy(y_clip)

        normas_grad = []
        losses_gc   = []

        for _ in range(50):
            opt_gc.zero_grad()
            logits_gc = modelo_gc_c(X_t_gc).squeeze()
            loss_gc   = crit_gc(logits_gc, y_t_gc)
            loss_gc.backward()

            # Calcular norma antes do clipping
            norma = 0.0
            for p in modelo_gc_c.parameters():
                if p.grad is not None:
                    norma += p.grad.data.norm(2).item() ** 2
            normas_grad.append(norma ** 0.5)

            if max_norm is not None:
                nn.utils.clip_grad_norm_(
                    modelo_gc_c.parameters(),
                    max_norm=max_norm
                )

            opt_gc.step()
            losses_gc.append(loss_gc.item())

        label = f"clip={max_norm}" if max_norm else "sem clip"
        resultados_gc[label] = {
            "normas": normas_grad,
            "losses": losses_gc
        }

    fig, axes = plt.subplots(1, 2, figsize=(14, 5))
    cores_gc = ["red", "coral", "steelblue", "mediumseagreen"]

    for (label, res), cor in zip(resultados_gc.items(),
                                   cores_gc):
        axes[0].plot(res["normas"], label=label,
                     color=cor, linewidth=2)
        axes[1].plot(res["losses"], label=label,
                     color=cor, linewidth=2)

    for ax, titulo in zip(axes,
                           ["Norma do Gradiente",
                            "Training Loss"]):
        ax.set_xlabel("Iteração")
        ax.set_ylabel(titulo)
        ax.set_title(titulo)
        ax.legend(fontsize=9)
        ax.set_ylim(0, ax.get_ylim()[1])

    plt.suptitle("Efeito do Gradient Clipping", fontsize=13)
    plt.tight_layout()
    plt.show()

demonstrar_gradient_clipping()

8. Rede Neural para Regressão

print("\n" + "=" * 60)
print("REDE NEURAL PARA REGRESSÃO")
print("=" * 60)

# Dataset de regressão
np.random.seed(42)
X_reg, y_reg = make_regression(
    n_samples=1500, n_features=15,
    n_informative=8, noise=25,
    random_state=42
)

X_tr_reg, X_te_reg, y_tr_reg, y_te_reg = train_test_split(
    X_reg, y_reg, test_size=0.2, random_state=42
)
X_tr_reg, X_va_reg, y_tr_reg, y_va_reg = train_test_split(
    X_tr_reg, y_tr_reg, test_size=0.15, random_state=42
)

sc_reg = StandardScaler()
X_tr_reg = sc_reg.fit_transform(X_tr_reg).astype(np.float32)
X_va_reg = sc_reg.transform(X_va_reg).astype(np.float32)
X_te_reg = sc_reg.transform(X_te_reg).astype(np.float32)

# Normalizar target também para regressão
y_mean    = y_tr_reg.mean()
y_std     = y_tr_reg.std()
y_tr_reg_n = ((y_tr_reg - y_mean) / y_std).astype(np.float32)
y_va_reg_n = ((y_va_reg - y_mean) / y_std).astype(np.float32)
y_te_reg_n = ((y_te_reg - y_mean) / y_std).astype(np.float32)

class RedeRegressao(nn.Module):
    """Rede para regressão: saída linear sem ativação final."""

    def __init__(self, n_entrada: int,
                  dims: list = None):
        super().__init__()
        if dims is None:
            dims = [128, 64, 32]

        camadas = []
        dim_ant = n_entrada

        for dim in dims:
            camadas.extend([
                nn.Linear(dim_ant, dim),
                nn.BatchNorm1d(dim),
                nn.ReLU(),
                nn.Dropout(0.1),
            ])
            dim_ant = dim

        # Camada de saída sem ativação
        camadas.append(nn.Linear(dim_ant, 1))
        self.rede = nn.Sequential(*camadas)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.rede(x)

# Trainer adaptado para regressão
class TrainerRegressao(Trainer):
    """Trainer especializado para regressão."""

    @torch.no_grad()
    def _epoca_avaliacao(self,
                          dl: DataLoader) -> dict:
        self.modelo.eval()
        losses   = []
        preds_l  = []
        labels_l = []

        for X_b, y_b in dl:
            X_b = X_b.to(self.device)
            y_b = y_b.to(self.device)

            saida = self.modelo(X_b).squeeze()
            loss  = self.criterion(saida, y_b)
            losses.append(loss.item())
            preds_l.append(saida.cpu().numpy())
            labels_l.append(y_b.cpu().numpy())

        preds_all  = np.concatenate(preds_l)
        labels_all = np.concatenate(labels_l)

        # Desnormalizar para métricas reais
        preds_dn   = preds_all  * y_std + y_mean
        labels_dn  = labels_all * y_std + y_mean

        return {
            "loss": np.mean(losses),
            "r2":   r2_score(labels_dn, preds_dn),
            "mae":  mean_absolute_error(labels_dn,
                                         preds_dn),
        }

    def _epoca_treino(self,
                       dl: DataLoader) -> dict:
        self.modelo.train()
        losses = []

        for X_b, y_b in dl:
            X_b = X_b.to(self.device)
            y_b = y_b.to(self.device)

            self.optimizer.zero_grad()
            saida = self.modelo(X_b).squeeze()
            loss  = self.criterion(saida, y_b)
            loss.backward()

            if self.grad_clip > 0:
                nn.utils.clip_grad_norm_(
                    self.modelo.parameters(),
                    max_norm=self.grad_clip
                )

            self.optimizer.step()
            losses.append(loss.item())

        return {"loss": np.mean(losses)}

# Criar datasets de regressão
ds_tr_reg = TabelarDataset(X_tr_reg, y_tr_reg_n)
ds_va_reg = TabelarDataset(X_va_reg, y_va_reg_n)
ds_te_reg = TabelarDataset(X_te_reg, y_te_reg_n)

dl_tr_reg = DataLoader(ds_tr_reg, batch_size=64,
                         shuffle=True, drop_last=True)
dl_va_reg = DataLoader(ds_va_reg, batch_size=64,
                         shuffle=False)
dl_te_reg = DataLoader(ds_te_reg, batch_size=64,
                         shuffle=False)

# Treinar modelo de regressão
modelo_reg = RedeRegressao(n_entrada=15).to(device)
opt_reg    = optim.AdamW(modelo_reg.parameters(),
                           lr=1e-3, weight_decay=1e-4)
sch_reg    = optim.lr_scheduler.ReduceLROnPlateau(
    opt_reg, patience=10, factor=0.5, verbose=False
)

trainer_reg = TrainerRegressao(
    modelo=modelo_reg,
    criterion=nn.MSELoss(),
    optimizer=opt_reg,
    scheduler=sch_reg,
    device=device,
    gradient_clip=1.0,
    early_stopping_patience=20,
    checkpoint_path="modelo_regressao.pt"
)

trainer_reg.treinar(
    dl_treino=dl_tr_reg,
    dl_val=dl_va_reg,
    n_epocas=200,
    verbose=50
)

metr_te_reg = trainer_reg.avaliar(dl_te_reg)
print(f"\nResultados de Regressão no Teste:")
print(f"  MSE Loss: {metr_te_reg['loss']:.4f}")
print(f"  R²:       {metr_te_reg['r2']:.4f}")
print(f"  MAE:      {metr_te_reg['mae']:.4f}")

trainer_reg.plotar_historico(["loss", "r2", "mae"])

9. Experimento Completo: Comparando Arquiteturas

print("\n" + "=" * 60)
print("EXPERIMENTO: COMPARANDO ARQUITETURAS")
print("=" * 60)

arquiteturas = {
    "MLP Simples": nn.Sequential(
        nn.Linear(20, 128), nn.ReLU(), nn.Dropout(0.2),
        nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.1),
        nn.Linear(64, 1)
    ),
    "MLP com BatchNorm": nn.Sequential(
        nn.Linear(20, 128), nn.BatchNorm1d(128),
        nn.ReLU(), nn.Dropout(0.2),
        nn.Linear(128, 64), nn.BatchNorm1d(64),
        nn.ReLU(), nn.Dropout(0.1),
        nn.Linear(64, 1)
    ),
    "Rede Residual": RedeResidual(
        n_entrada=20, dim_oculta=128,
        n_blocos=3, dropout=0.2
    ),
}

resultados_exp = {}
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
cores_exp = ["steelblue", "coral", "mediumseagreen"]

for (nome, arq), cor in zip(arquiteturas.items(),
                               cores_exp):
    torch.manual_seed(42)
    arq = arq.to(device)

    opt_exp = optim.AdamW(arq.parameters(),
                            lr=1e-3, weight_decay=1e-4)
    sch_exp = optim.lr_scheduler.CosineAnnealingLR(
        opt_exp, T_max=100, eta_min=1e-5
    )

    tr_exp = Trainer(
        modelo=arq, criterion=criterion,
        optimizer=opt_exp, scheduler=sch_exp,
        device=device, gradient_clip=1.0,
        early_stopping_patience=20,
        checkpoint_path=f"exp_{nome.replace(' ', '_')}.pt"
    )

    tr_exp.treinar(
        dl_tr, dl_va, n_epocas=100, verbose=0
    )
    metr = tr_exp.avaliar(dl_te)
    resultados_exp[nome] = metr

    ep = range(1, len(tr_exp.historico["val_loss"]) + 1)
    axes[0].plot(ep, tr_exp.historico["treino_loss"],
                  label=f"{nome} (tr)",
                  color=cor, alpha=0.5, linewidth=1.5)
    axes[1].plot(ep, tr_exp.historico["val_auc"],
                  label=nome, color=cor, linewidth=2)

for ax, titulo in zip(axes, ["Training Loss",
                               "Validation AUC"]):
    ax.set_xlabel("Época")
    ax.set_title(titulo)
    ax.legend(fontsize=9)

plt.suptitle("Comparação de Arquiteturas", fontsize=13)
plt.tight_layout()
plt.show()

print("\nResultados no Teste:")
print(f"{'Arquitetura':<22} {'Loss':>8} "
      f"{'Acc':>8} {'AUC':>8} {'F1':>8}")
print("-" * 58)
for nome, metr in resultados_exp.items():
    print(f"{nome:<22} {metr['loss']:>8.4f} "
          f"{metr['acuracia']:>8.4f} "
          f"{metr['auc']:>8.4f} "
          f"{metr['f1']:>8.4f}")

10. Salvamento e Deploy do Modelo

print("\n" + "=" * 60)
print("SALVAMENTO E DEPLOY DO MODELO")
print("=" * 60)

# ── Formas de salvar um modelo PyTorch ───────────────────

melhor_modelo = RedeResidual(
    n_entrada=20, dim_oculta=128,
    n_blocos=3, dropout=0.2
).to(device)

# 1. Salvar apenas os pesos (recomendado)
torch.save(melhor_modelo.state_dict(), "pesos.pt")

# 2. Salvar o modelo completo (mais simples, menos portável)
torch.save(melhor_modelo, "modelo_completo2.pt")

# 3. Checkpoint completo (melhor para retomar treino)
torch.save({
    "arquitetura": {
        "n_entrada":  20,
        "dim_oculta": 128,
        "n_blocos":   3,
        "dropout":    0.2,
    },
    "state_dict":   melhor_modelo.state_dict(),
    "optimizer":    optimizer.state_dict(),
    "epoca":        trainer.epoca_melhor,
    "val_loss":     trainer.melhor_val,
    "metricas_te":  metr_te,
    "scaler_mean":  sc.mean_.tolist(),
    "scaler_std":   sc.scale_.tolist(),
}, "checkpoint_completo.pt")

print("Arquivos salvos:")
import os
for f in ["pesos.pt", "modelo_completo2.pt",
           "checkpoint_completo.pt"]:
    if os.path.exists(f):
        size = os.path.getsize(f) / 1024
        print(f"  {f}: {size:.1f} KB")

# ── Carregando e usando em produção ──────────────────────
def carregar_modelo_producao(checkpoint_path: str,
                               device: torch.device):
    """Carrega modelo de um checkpoint completo."""
    ckpt  = torch.load(checkpoint_path,
                         map_location=device,
                         weights_only=False)
    arq   = ckpt["arquitetura"]
    model = RedeResidual(**arq).to(device)
    model.load_state_dict(ckpt["state_dict"])
    model.eval()
    return model, ckpt

modelo_prod, ckpt = carregar_modelo_producao(
    "checkpoint_completo.pt", device
)
print(f"\nModelo carregado (época {ckpt['epoca']}):")
print(f"  Val Loss: {ckpt['val_loss']:.4f}")
print(f"  Test AUC: {ckpt['metricas_te']['auc']:.4f}")

# ── Função de predição para produção ────────────────────
def predizer(X_novo: np.ndarray,
              scaler_mean: list,
              scaler_std: list,
              modelo: nn.Module,
              device: torch.device,
              limiar: float = 0.5) -> dict:
    """
    Pipeline completo de predição em produção.
    Inclui normalização e conversão de tipos.
    """
    # Normalizar com os parâmetros do treino
    mean  = np.array(scaler_mean, dtype=np.float32)
    std   = np.array(scaler_std,  dtype=np.float32)
    X_n   = ((X_novo - mean) / std).astype(np.float32)

    # Converter e mover para device
    X_t   = torch.from_numpy(X_n).to(device)

    # Inferência
    modelo.eval()
    with torch.no_grad():
        logits = modelo(X_t).squeeze()
        probs  = torch.sigmoid(logits).cpu().numpy()

    classes = (probs >= limiar).astype(int)
    return {
        "probabilidades": probs.tolist(),
        "classes":        classes.tolist()
    }

# Teste de produção
X_novos = np.random.randn(5, 20).astype(np.float32)
resultado = predizer(
    X_novos,
    ckpt["scaler_mean"],
    ckpt["scaler_std"],
    modelo_prod,
    device
)

print(f"\nPredição de novos exemplos:")
for i, (prob, cls) in enumerate(
    zip(resultado["probabilidades"],
        resultado["classes"])
):
    print(f"  Exemplo {i+1}: P={prob:.4f} → "
          f"Classe {cls}")

Resumo da Aula

• Dataset personalizado herda de torch.utils.data.Dataset e implementa __len__ e __getitem__
• Blocos residuais adicionam skip connections que resolvem vanishing gradient em redes profundas
• O Trainer encapsula todo o loop de treino: forward, backward, gradient clipping, early stopping e checkpointing
• model.train() e model.eval() alternam entre os modos que afetam Dropout e BatchNorm
• Gradient clipping (clip_grad_norm_) evita instabilidade por gradientes muito grandes
• LR schedulers adaptam o learning rate ao longo do treino: ReduceLROnPlateau reage ao progresso, CosineAnnealing oscila suavemente
• Para regressão, normalizar o target melhora a estabilidade do treinamento
• Checkpoints completos salvam tudo necessário para retomar o treino ou fazer deploy: pesos, otimizador, métricas e parâmetros de pré-processamento
• Em produção, inclua o pré-processamento na função de predição para garantir consistência
• O fluxo completo de produção é: checkpoint → normalização com parâmetros do treino → inferência → desnormalização se necessário

Exercícios

1. Explique o que é uma conexão residual (skip connection) e por que ela resolve o problema do vanishing gradient. Matematicamente, como a equação saída = F(x) + x afeta o fluxo do gradiente durante o backpropagation?

   Ver resposta

   ✓ Resposta:

   Uma conexão residual (skip connection) adiciona a entrada de um bloco diretamente à sua saída: saída = F(x) + x, onde F(x) é a transformação aprendida (camadas lineares + ativações). Em vez de aprender a transformação direta do dado, a rede aprende apenas o "resíduo" — a diferença entre a entrada e a saída desejada.

   O efeito no backpropagation é elegante. Ao calcular ∂L/∂x pela regra da cadeia:

   ∂L/∂x = ∂L/∂saída × ∂saída/∂x
          = ∂L/∂saída × (∂F(x)/∂x + 1)
   

   O termo +1 garante que o gradiente nunca seja menor que ∂L/∂saída — há sempre um caminho direto de gradiente não atenuado de volta às camadas iniciais. Mesmo que ∂F(x)/∂x seja muito pequeno (como acontece com sigmoid em redes profundas), o gradiente ainda flui via ∂L/∂saída × 1. Isso permite treinar redes com centenas de camadas que seriam impossíveis sem skip connections.

2. Por que é importante normalizar o target (variável dependente) em problemas de regressão com redes neurais? O que acontece se os valores do target forem muito grandes (ex: preços de imóveis em milhões)?

   Ver resposta

   ✓ Resposta:

   Normalizar o target em regressão é importante por três razões. Primeiro, estabilidade numérica: a função de perda MSE é quadrática no erro. Com targets na escala de milhões, erros pequenos em termos relativos produzem valores de MSE enormes, causando gradientes instáveis que dificultam a convergência.

   Segundo, compatibilidade com a escala das ativações: as ativações das camadas ocultas tipicamente ficam em torno de [-1, 1] ou [0, 1]. Se o target for 500.000, a última camada linear precisaria de pesos muito grandes para atingir essa escala, desbalanceando a rede.

   Terceiro, learning rate: um learning rate que funciona bem para targets normalizados (ex: 0.001) seria completamente inadequado para targets em escala bruta — precisaria ser ajustado manualmente para cada escala de dados diferente.

   Se os valores do target forem muito grandes sem normalização, a loss inicial será enorme, os gradientes serão grandes e instáveis, e o modelo pode divergir ou convergir para um ponto trivial (predizer sempre a média).

3. Explique gradient clipping. Por que gradientes muito grandes são um problema? Quais são as duas formas de clipping (por valor e por norma) e quando usar cada uma?

   Ver resposta

   ✓ Resposta:

   Gradientes muito grandes causam "exploding gradients": os pesos são atualizados por valores enormes em uma única iteração, saltando para regiões ruins do espaço de parâmetros. Isso pode fazer a loss aumentar drasticamente ou o modelo divergir completamente (loss = NaN).

   Clipping por valor: limita cada elemento do gradiente ao intervalo [-v, v]. Simples, mas pode distorcer a direção do gradiente.

   nn.utils.clip_grad_value_(model.parameters(), clip_value=1.0)
   

   Clipping por norma: escala o vetor de gradiente inteiro se sua norma L2 exceder um limiar. Preserva a direção do gradiente.

   nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
   

   Use clipping por norma (o padrão) quando quiser preservar a direção do gradiente. Use por valor para situações onde alguns parâmetros têm gradientes consistentemente maiores que outros. Na prática, max_norm=1.0 é um valor seguro para a maioria dos problemas.

4. Compare optim.Adam e optim.AdamW. Qual é a diferença matemática entre os dois? Por que AdamW é geralmente preferido em modelos modernos?

   Ver resposta

   ✓ Resposta:

   Adam implementa L2 regularização adicionando λ × w ao gradiente antes da divisão pelo segundo momento: g_efetivo = g + λw. O problema é que essa penalidade é escalada pelos momentos adaptativos — parâmetros com gradientes historicamente grandes recebem steps menores, e a regularização efetiva também fica menor para esses parâmetros. A regularização fica inconsistente entre parâmetros.

   AdamW separa a regularização da atualização adaptativa: primeiro aplica a atualização Adam com os gradientes originais, depois decai os pesos diretamente: w_novo = w - η(update_adam) - η × λ × w. O decaimento é aplicado com a taxa de aprendizado base, não com a adaptativa. Isso garante que todos os parâmetros são regularizados de forma consistente, independente de seus gradientes históricos.

   Em modelos modernos como Transformers, AdamW é preferido porque a regularização consistente produz modelos que generalizam melhor, especialmente em modelos grandes onde a regularização tem papel crítico.

5. Por que o checkpoint deve incluir os parâmetros do StandardScaler (média e desvio padrão) além dos pesos do modelo? O que aconteceria em produção se você normalizasse os novos dados com estatísticas calculadas nos próprios novos dados?

   Ver resposta

   ✓ Resposta:

   O StandardScaler aprende os parâmetros de normalização (média e desvio padrão) dos dados de treino. O modelo foi treinado com features normalizadas usando esses parâmetros específicos. As representações internas que o modelo aprendeu — os pesos de cada camada — assumem implicitamente que as entradas chegam nessa escala específica.

   Se em produção você normalizasse os novos dados com estatísticas calculadas nesses próprios novos dados, as features teriam distribuição diferente da que o modelo viu durante o treino. Por exemplo, se a feature "renda" no treino tinha média R$5.000 e no novo batch a média é R$8.000, normalizar com a média do novo batch distorce a relação entre as features. O modelo receberia entradas que parecem "similares" estatisticamente mas representam valores absolutamente diferentes.

   Isso causaria predições incorretas sem nenhum erro explícito — o bug mais difícil de detectar. Por isso o checkpoint deve incluir todos os parâmetros de pré-processamento: eles são parte do modelo tanto quanto os pesos da rede.

6. Explique o que é ReduceLROnPlateau e como ele funciona. Quais são os parâmetros patience e factor? Compare com CosineAnnealingLR e diga quando você escolheria cada um.

   Ver resposta

   ✓ Resposta:

   ReduceLROnPlateau monitora uma métrica (geralmente a validation loss) e reduz o learning rate por um fator quando a métrica para de melhorar por patience épocas consecutivas.

   patience=10 significa aguardar 10 épocas sem melhora antes de reduzir. factor=0.5 significa multiplicar o LR por 0.5 quando reduzir.

   Exemplo: LR inicial = 0.001. Após 10 épocas sem melhora → LR = 0.0005. Após mais 10 → LR = 0.00025, e assim por diante.

   CosineAnnealingLR reduz o LR seguindo uma curva cosseno de lr_max até lr_min ao longo de T_max épocas, independente do progresso do treino.

   Escolha ReduceLROnPlateau quando: o ritmo de aprendizado é imprevisível, o dataset é ruidoso, ou você não sabe quantas épocas serão necessárias. É mais robusto e adaptativo.

   Escolha CosineAnnealingLR quando: você tem um orçamento fixo de épocas, quer convergência suave e reproduzível, ou está fazendo fine-tuning de modelos pré-treinados onde oscilações suaves ajudam a explorar o espaço de parâmetros.

Referências

• PyTorch — Training Loop
• PyTorch — Saving and Loading Models
• PyTorch — LR Schedulers
• PyTorch — DataLoader
• ResNet original paper (He et al., 2016)
• AdamW paper (Loshchilov & Hutter, 2019)
• Gradient Clipping — PyTorch docs
• Real Python — PyTorch Tutorial
• PyTorch Lightning — alternativa de alto nível