Implementação Manual de RNN e LSTM com NumPy: Propagação Direta, Retropropagação e Validação de Gradientes

Fundamentos da Implementação de Redes Recorrentes

A utilização de bibliotecas de alto nível frequentemente oculta a mecânica interna das redes neurais recorrentes. Compreender a alocação de tensores, o compartilhamento de parâmetros ao longo do tempo e o fluxo de gradientes é essencial para diagnosticar problemas de convergência e instabilidade numérica. Este guia detalha a construção completa de arquiteturas RNN e LSTM utilizando exclusivamente NumPy, abordando a propagação direta, a retropropagação através do tempo (BPTT) e técnicas de verificação numérica de gradientes.

Arquitetura RNN: Fluxo Direto e Inicialização de Parâmetros

A dinâmica de uma rede recorrente padrão baseia-se na atualização sequencial de um estado oculto. Para uma sequência de entrada $x_1, ..., x_T$, a transição em cada passo $t$ segue:

  • $h_t = \tanh(W_{hh} h_{t-1} + W_{xh} x_t + b_h)$
  • $y_t = W_{hy} h_t + b_y$

Os pesos $W_{hh}$, $W_{xh}$ e $W_{hy}$ são compartilhados em todas as iterações temporais. Diferente de redes feedforward, a profundidade aqui é definida pelo comprimento da sequência. Para viabilizar o cálculo de gradientes, é impérativo armaznear os estados intermediários.

import numpy as np

class RedeRecorrenteSimples:
    def __init__(self, dim_entrada, dim_oculta, dim_saida, taxa_aprendizado=0.01):
        self.dim_entrada = dim_entrada
        self.dim_oculta = dim_oculta
        self.taxa = taxa_aprendizado

        # Inicialização com distribuição uniforme ajustada
        escala = np.sqrt(2.0 / (dim_entrada + dim_oculta))
        self.pesos_entrada = np.random.uniform(-escala, escala, (dim_oculta, dim_entrada))
        self.pesos_recorrentes = np.random.uniform(-escala, escala, (dim_oculta, dim_oculta))
        self.pesos_saida = np.random.uniform(-escala, escala, (dim_saida, dim_oculta))

        self.vieses_oculto = np.zeros((dim_oculta, 1))
        self.vieses_saida = np.zeros((dim_saida, 1))

    def propagar(self, sequencia_entrada):
        passos = len(sequencia_entrada)
        estado_atual = np.zeros((self.dim_oculta, 1))

        # Armazenamento explícito para BPTT
        self.cache_entradas = {}
        self.cache_estados = {-1: np.copy(estado_atual)}
        self.cache_saidas = {}

        for t in range(passos):
            self.cache_entradas[t] = sequencia_entrada[t]
            estado_atual = np.tanh(
                self.pesos_entrada @ sequencia_entrada[t] +
                self.pesos_recorrentes @ estado_atual +
                self.vieses_oculto
            )
            self.cache_estados[t] = estado_atual
            self.cache_saidas[t] = self.pesos_saida @ estado_atual + self.vieses_saida

        return self.cache_saidas

A escolha de float64 é recomendada durante a fase de desenvolvimento e validação de gradientes para minimizar erros de precisão. O uso de dicionários para o cache facilita o acesso ao estado inicial $h_{-1}$ sem necessidade de tratamento especial de índices negativos.

Retropropagação Temporal (BPTT) e Acúmulo de Gradientes

O algoritmo BPTT desdobra a rede ao longo do eixo temporal, tratando-a como um grafo computacional profundo. Como os parâmetros são compartilhados, os gradientes calculados em cada passo devem ser somados.

    def retropropagar(self, gradientes_saida):
        passos = len(gradientes_saida)
        grad_pesos_entrada = np.zeros_like(self.pesos_entrada)
        grad_pesos_recorrentes = np.zeros_like(self.pesos_recorrentes)
        grad_pesos_saida = np.zeros_like(self.pesos_saida)
        grad_vieses_oculto = np.zeros_like(self.vieses_oculto)
        grad_vieses_saida = np.zeros_like(self.vieses_saida)

        grad_estado_futuro = np.zeros((self.dim_oculta, 1))

        for t in reversed(range(passos)):
            dy = gradientes_saida[t]

            # Gradientes da camada de saída
            grad_pesos_saida += dy @ self.cache_estados[t].T
            grad_vieses_saida += dy

            # Gradiente total do estado oculto (saída atual + próximo passo temporal)
            dh = self.pesos_saida.T @ dy + grad_estado_futuro

            # Derivada do tanh: 1 - h^2
            dh_nao_linear = (1 - self.cache_estados[t] ** 2) * dh

            grad_vieses_oculto += dh_nao_linear
            grad_pesos_entrada += dh_nao_linear @ self.cache_entradas[t].T
            grad_pesos_recorrentes += dh_nao_linear @ self.cache_estados[t - 1].T

            # Propagação para o passo anterior
            grad_estado_futuro = self.pesos_recorrentes.T @ dh_nao_linear

        # Clipagem para estabilidade
        for g in [grad_pesos_entrada, grad_pesos_recorrentes, grad_pesos_saida,
                  grad_vieses_oculto, grad_vieses_saida]:
            np.clip(g, -5.0, 5.0, out=g)

        return grad_pesos_entrada, grad_pesos_recorrentes, grad_pesos_saida, grad_vieses_oculto, grad_vieses_saida

Verificação Numérica de Gradientes

A validação da implementação analítica exige comparação com aproximações numéricas via diferença central. Perturbações infinitesimais $\epsilon$ são aplicadas a cada parâmetro para estimar a derivada.

def verificar_gradientes_numericamente(modelo, entradas, saidas_alvo, nome_parametro, epsilon=1e-5):
    parametro = getattr(modelo, nome_parametro)
    gradiente_aproximado = np.zeros_like(parametro)

    iterador = np.nditer(parametro, flags=['multi_index'], op_flags=['readwrite'])
    while not iterador.finished:
        idx = iterador.multi_index
        valor_original = parametro[idx]

        parametro[idx] = valor_original + epsilon
        pred_mais = modelo.propagar(entradas)
        perda_mais = 0.5 * sum(np.sum((saidas_alvo[k] - pred_mais[k])**2) for k in range(len(entradas)))

        parametro[idx] = valor_original - epsilon
        pred_menos = modelo.propagar(entradas)
        perda_menos = 0.5 * sum(np.sum((saidas_alvo[k] - pred_menos[k])**2) for k in range(len(entradas)))

        parametro[idx] = valor_original
        gradiente_aproximado[idx] = (perda_mais - perda_menos) / (2 * epsilon)
        iterador.iternext()

    return gradiente_aproximado

A discrepância relativa entre o gradiente analítico e o numérico deve permanecer abaixo de $10^{-6}$. Valores superiores indicam erros na cadeia de derivadas ou no acúmulo de gradientes.

Instabilidade em Redes Recorrentes: Vanishing e Exploding Gradients

A propagação de erros através de múltiplos passos temporais envolve multiplicações matriciais sucessivas por $W_{hh}$ e pela derivada da função de ativação. Se o raio espectral de $W_{hh}$ combinado com a derivada do $\tanh$ for menor que 1, o gradiente decai exponencialmente (vanishing). Se for maior que 1, ele cresce descontroladamente (exploding). A clipagem de norma global mitiga a explosão, mas não resolve o desaparecimento, que exige modificações arquiteturais.

Arquitetura LSTM: Mecanismos de Controle de Fluxo

A Long Short-Term Memory introduz um estado de célula ($c_t$) que opera como uma via linear de preservação de informação, contornando a compressão não linear repetida. O fluxo é regulado por três portas sigmoidais e um candidato $\tanh$:

  • Porta de esquecimento ($f_t$): filtra informações do estado anterior.
  • Porta de entrada ($i_t$): controla a atualização com novos dados.
  • Candidato ($\tilde{c}_t$): proposta de nova memória.
  • Porta de saída ($o_t$): expõe seletivamente o estado processado.

As equações de atualização são: $c_t = f_t \odot c_{t-1} + i_t \odot \tilde{c}_t$ e $h_t = o_t \odot \tanh(c_t)$.

Implementação NumPy da LSTM

A concatenação de $h_{t-1}$ e $x_t$ simplifica as operações matriciais, unificando o cálculo das quatro portas.

class MemoriaLongoCurtoPrazo:
    def __init__(self, dim_entrada, dim_oculta, dim_saida, taxa=0.01):
        self.dim_entrada = dim_entrada
        self.dim_oculta = dim_oculta
        self.taxa = taxa
        dim_concat = dim_oculta + dim_entrada

        escala = np.sqrt(2.0 / dim_concat)
        self.W_esq = np.random.uniform(-escala, escala, (dim_oculta, dim_concat))
        self.W_ent = np.random.uniform(-escala, escala, (dim_oculta, dim_concat))
        self.W_sai = np.random.uniform(-escala, escala, (dim_oculta, dim_concat))
        self.W_cand = np.random.uniform(-escala, escala, (dim_oculta, dim_concat))

        self.b_esq = np.zeros((dim_oculta, 1))
        self.b_ent = np.zeros((dim_oculta, 1))
        self.b_sai = np.zeros((dim_oculta, 1))
        self.b_cand = np.zeros((dim_oculta, 1))

        self.W_proj = np.random.uniform(-escala, escala, (dim_saida, dim_oculta))
        self.b_proj = np.zeros((dim_saida, 1))

    def _sigmoide(self, val):
        return 1.0 / (1.0 + np.exp(-np.clip(val, -500, 500)))

    def propagar(self, seq_entrada):
        T = len(seq_entrada)
        h = np.zeros((self.dim_oculta, 1))
        c = np.zeros((self.dim_oculta, 1))

        self.cache_x, self.cache_h, self.cache_c = {}, {}, {}
        self.cache_f, self.cache_i, self.cache_o, self.cache_c_til = {}, {}, {}, {}
        self.cache_h[-1], self.cache_c[-1] = np.copy(h), np.copy(c)
        self.cache_y = {}

        for t in range(T):
            x_t = seq_entrada[t]
            vetor_concat = np.vstack((h, x_t))
            self.cache_x[t] = x_t

            self.cache_f[t] = self._sigmoide(self.W_esq @ vetor_concat + self.b_esq)
            self.cache_i[t] = self._sigmoide(self.W_ent @ vetor_concat + self.b_ent)
            self.cache_o[t] = self._sigmoide(self.W_sai @ vetor_concat + self.b_sai)
            self.cache_c_til[t] = np.tanh(self.W_cand @ vetor_concat + self.b_cand)

            c = self.cache_f[t] * c + self.cache_i[t] * self.cache_c_til[t]
            h = self.cache_o[t] * np.tanh(c)

            self.cache_c[t] = c
            self.cache_h[t] = h
            self.cache_y[t] = self.W_proj @ h + self.b_proj

        return self.cache_y

Retropropagação na LSTM

O cálculo reverso exige atenção ao fluxo duplo de gradientes: um através do estado oculto $h_t$ e outro através do estado da célula $c_t$. A derivada deve ser decomposta porta a porta.

    def retropropagar(self, grads_saida):
        T = len(grads_saida)
        # Inicialização de acumuladores de gradiente
        dW_esq = np.zeros_like(self.W_esq)
        dW_ent = np.zeros_like(self.W_ent)
        dW_sai = np.zeros_like(self.W_sai)
        dW_cand = np.zeros_like(self.W_cand)
        db_esq = np.zeros_like(self.b_esq)
        db_ent = np.zeros_like(self.b_ent)
        db_sai = np.zeros_like(self.b_sai)
        db_cand = np.zeros_like(self.b_cand)
        dW_proj = np.zeros_like(self.W_proj)
        db_proj = np.zeros_like(self.b_proj)

        dh_prox = np.zeros((self.dim_oculta, 1))
        dc_prox = np.zeros((self.dim_oculta, 1))

        for t in reversed(range(T)):
            dy = grads_saida[t]
            dW_proj += dy @ self.cache_h[t].T
            db_proj += dy

            dh = self.W_proj.T @ dy + dh_prox

            # Decomposição em relação às portas e estado da célula
            tanh_c = np.tanh(self.cache_c[t])
            do_porta = dh * tanh_c
            dc = dh * self.cache_o[t] * (1 - tanh_c**2) + dc_prox

            df_porta = dc * self.cache_c[t-1]
            di_porta = dc * self.cache_c_til[t]
            dc_til = dc * self.cache_i[t] * (1 - self.cache_c_til[t]**2)

            # Aplicação das derivadas das ativações (sigmoide e tanh)
            df_raw = df_porta * self.cache_f[t] * (1 - self.cache_f[t])
            di_raw = di_porta * self.cache_i[t] * (1 - self.cache_i[t])
            do_raw = do_porta * self.cache_o[t] * (1 - self.cache_o[t])

            vec_concat = np.vstack((self.cache_h[t-1], self.cache_x[t]))

            dW_esq += df_raw @ vec_concat.T
            dW_ent += di_raw @ vec_concat.T
            dW_sai += do_raw @ vec_concat.T
            dW_cand += dc_til @ vec_concat.T
            db_esq += df_raw
            db_ent += di_raw
            db_sai += do_raw
            db_cand += dc_til

            # Retropropagação para o passo t-1
            d_concat = (self.W_esq.T @ df_raw + self.W_ent.T @ di_raw +
                        self.W_sai.T @ do_raw + self.W_cand.T @ dc_til)
            dh_prox = d_concat[:self.dim_oculta]
            dc_prox = self.cache_f[t] * dc

        # Clipagem e retorno
        for g in [dW_esq, dW_ent, dW_sai, dW_cand, dW_proj,
                  db_esq, db_ent, db_sai, db_cand, db_proj]:
            np.clip(g, -5.0, 5.0, out=g)

        return dW_esq, dW_ent, dW_sai, dW_cand, dW_proj, db_esq, db_ent, db_sai, db_cand, db_proj

Pontos críticos na implementação reversa incluem a soma correta de dc_prox ao gradiente da célula atual e a multiplicação por f_t ao repassar o gradiente para c_{t-1}. Erros nessas linhas quebram a capacidade de memória de longo prazo.

Validação Cruzada e Comportamento Empírico

Para garantir a fidelidade da implementação, recomenda-se a inicialização idêntica de pesos e a comparação passo a passo com frameworks consolidados. Diferenças de precisão entre float32 e float64 devem ser consideradas, aceitando-se tolerâncias na ordem de $10^{-4}$ para saídas e gradientes.

Em tarefas de dependência temporal longa, como soma cumulativa de sequências binárias ou previsão de sinais não estacionários, a RNN padrão apresenta degradação rápida do desempenho após 15-20 passos devido ao vanishing gradient. A LSTM mantém a coerência das previsões graças ao caminho aditivo do estado $c_t$, que permite o fluxo de gradientes sem multiplicação repetida por derivadas de ativação. A inspeção dos valores médios das portas durante o treinamento revela padrões de retenção e descarte de informação alinhados com a estrutura dos dados, validando o mecanismo de controle aprendido.

Tags: NumPy rnn LSTM bptt gradient-checking

Publicado em 10-13 03:01