Delete models/cyclic_reasoning.py with huggingface_hub

models/cyclic_reasoning.py

@@ -1,411 +0,0 @@
-"""cyclic_reasoning.py — Raciocínio Cíclico (Cyclic Reasoning) para CNN-BiGRU.
-
-Implementa um módulo de raciocínio iterativo que refina a representação
-do modelo através de múltiplos ciclos, cada um alimentado pela saída do
-ciclo anterior combinada com uma hipótese nova. O processo para quando
-a convergência é atingida (norma da diferença < epsilon) ou quando o
-número máximo de ciclos é alcançado.
-
-==============================================================================
-ANÁLISE MATEMÁTICA E LÓGICA
-==============================================================================
-
-Seja h_0 ∈ R^d a representação inicial produzida pelo modelo (por exemplo,
-a saída fused da CooperativeCNNBiGRU). Em cada ciclo c = 1, 2, ..., C_max,
-computamos:
-
-    g_c   = HypothesisController(h_{c-1}, penalty_c)   ∈ R^d   (hipótese)
-    r_c   = RefinementLayer(concat(h_{c-1}, g_c))      ∈ R^d   (refinamento)
-    h_c   = LayerNorm(h_{c-1} + alpha_c * r_c)         ∈ R^d   (novo estado)
-
-onde:
-    - alpha_c ∈ [0, 1] é um coeficiente de confiança aprendido:
-        alpha_c = sigmoid(w_alpha^T h_{c-1} + b_alpha)
-    - penalty_c é a penalidade do verificador no ciclo c (se > threshold,
-      uma nova hipótese é ativada — ponte com o HypothesisController)
-    - O critério de parada é:
-        ||h_c - h_{c-1}||_2 < epsilon
-      ou c == C_max.
-
-GARANTIAS MATEMÁTICAS:
-    1. CONTRAÇÃO: como LayerNorm normaliza para norma unitária (≈1) e
-       alpha_c ≤ 1, então ||h_c|| ≤ ||h_{c-1}|| + ||r_c|| (limitado em R^d).
-       A sequência {h_c} é de Cauchy se o refinement for Lipschitz com L < 1.
-    2. CONVERGÊNCIA: empírica e empiricamente observada em ≤ 4 ciclos para
-       tarefas NLP comuns.
-    3. DIFERENCIABILIDADE: todas as operações são diferenciáveis, permitindo
-       backpropagation através dos ciclos (BPTT — Backprop Through Time).
-       Para reduzir o custo computacional, suportamos truncamento (truncated
-       BPTT) preservando apenas os últimos K ciclos no grafo.
-
-INTEGRAÇÃO COM OUTROS MÓDULOS:
-    - HypothesisController (cnn_bigru.training.hypothesis_controller): gera
-      g_c quando penalty_c > threshold.
-    - EWC (cnn_bigru.utils.ewc): os parâmetros do refinement são protegidos
-      pelo EWC entre tarefas.
-    - AntiHallucinationLayer: aplicada em h_c para validar coerência lógica
-      (tau ≥ tau_min antes de aceitar o ciclo).
-
-Autor: CNN-BiGRU Project
-"""
-from __future__ import annotations
-
-import logging
-import math
-from dataclasses import dataclass, field
-from typing import Callable, Dict, List, Optional, Tuple
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-
-logger = logging.getLogger(__name__)
-
-
-# ============================================================================
-# Configuração
-# ============================================================================
-
-@dataclass
-class CyclicReasoningConfig:
-    """Configuração do módulo de raciocínio cíclico."""
-    embed_dim: int = 128                  # dimensão da representação h
-    max_cycles: int = 4                   # número máximo de ciclos C_max
-    convergence_eps: float = 1e-3         # epsilon para parada antecipada
-    use_truncated_bptt: bool = True       # se True, detach após K ciclos
-    bptt_truncate: int = 2                # K ciclos preservados no grafo
-    use_layer_norm: bool = True           # LayerNorm após cada ciclo
-    use_residual: bool = True             # conexão residual h_{c-1} + alpha * r_c
-    hypothesis_input_dim: Optional[int] = None  # dim do vetor de penalidade (default: embed_dim)
-    refinement_hidden_mult: int = 4       # multiplicador da camada oculta
-    dropout: float = 0.1                  # dropout no refinement
-    penalty_threshold: float = 0.5        # ativa nova hipótese se penalty > threshold
-    use_anti_hallucination_gate: bool = True  # porta lógica fuzzy no h_c
-    device: str = "cpu"
-
-
-# ============================================================================
-# Refinement Layer (refina a representação a cada ciclo)
-# ============================================================================
-
-class RefinementLayer(nn.Module):
-    """Camada de refinamento que combina estado + hipótese.
-
-    R(conv(h, g)) = Linear(GELU(Linear(concat(h, g))))
-    """
-
-    def __init__(
-        self,
-        embed_dim: int,
-        hypothesis_dim: Optional[int] = None,
-        hidden_mult: int = 4,
-        dropout: float = 0.1,
-    ):
-        super().__init__()
-        h_dim = hypothesis_dim or embed_dim
-        # Camada de fusão: concat(h, g) -> hidden
-        self.fuse = nn.Linear(embed_dim + h_dim, embed_dim * hidden_mult)
-        self.dropout = nn.Dropout(dropout)
-        # Camada de projeção: hidden -> embed_dim
-        self.proj = nn.Linear(embed_dim * hidden_mult, embed_dim)
-        # Inicialização ortogonal para estabilidade (dados.txt linha ~880)
-        for layer in (self.fuse, self.proj):
-            for name, p in layer.named_parameters():
-                if "weight" in name and p.dim() >= 2:
-                    try:
-                        nn.init.orthogonal_(p)
-                    except Exception:
-                        nn.init.xavier_uniform_(p)
-                elif "bias" in name:
-                    nn.init.zeros_(p)
-
-    def forward(self, h: torch.Tensor, g: Optional[torch.Tensor] = None) -> torch.Tensor:
-        """
-        Args:
-            h: [B, D] estado atual
-            g: [B, D_h] hipótese (opcional — se None, usa zeros)
-        Returns:
-            r: [B, D] refinamento
-        """
-        if g is None:
-            g = torch.zeros_like(h)
-        elif g.size(-1) != h.size(-1):
-            # projetar g para dim h
-            if not hasattr(self, "_g_proj"):
-                self._g_proj = nn.Linear(g.size(-1), h.size(-1), bias=False).to(g.device)
-                nn.init.orthogonal_(self._g_proj.weight)
-            g = self._g_proj(g)
-        x = torch.cat([h, g], dim=-1)
-        x = self.dropout(F.gelu(self.fuse(x)))
-        return self.proj(x)
-
-
-# ============================================================================
-# Confidence Gate (alpha_c)
-# ============================================================================
-
-class ConfidenceGate(nn.Module):
-    """Computa alpha_c = sigmoid(w^T h + b) ∈ [0, 1].
-
-    Determina o quanto da hipótese refinada deve ser aceita no novo estado.
-    """
-
-    def __init__(self, embed_dim: int):
-        super().__init__()
-        self.gate = nn.Linear(embed_dim, 1)
-        nn.init.zeros_(self.gate.bias)
-
-    def forward(self, h: torch.Tensor) -> torch.Tensor:
-        # Retorna [B, 1] para broadcast
-        return torch.sigmoid(self.gate(h))
-
-
-# ============================================================================
-# Anti-Hallucination Gate (lógica fuzzy de Łukasiewicz simplificada)
-# ============================================================================
-
-class FuzzyLogicGate(nn.Module):
-    """Porta lógica fuzzy baseada em Łukasiewicz (conforme dados.txt linhas 510-540).
-
-    Computa um valor de verdade suave tau ∈ [0, 1] que mede a coerência
-    lógica do estado h_c. Se tau < tau_min, o ciclo é rejeitado (h permanece).
-    """
-
-    def __init__(self, embed_dim: int, tau_min: float = 0.3):
-        super().__init__()
-        # Projeção para um escalar (representando o "valor de verdade")
-        self.proj = nn.Linear(embed_dim, 1)
-        self.tau_min = tau_min
-        nn.init.zeros_(self.proj.bias)
-
-    def forward(self, h: torch.Tensor) -> torch.Tensor:
-        """Retorna tau ∈ [0, 1] — [B, 1]."""
-        # sigmoid(w^T h + b) — suaviza para [0, 1]
-        return torch.sigmoid(self.proj(h))
-
-    def accept(self, tau: torch.Tensor) -> torch.Tensor:
-        """Máscara booleana [B, 1]: True se tau >= tau_min."""
-        return (tau >= self.tau_min).float()
-
-
-# ============================================================================
-# Cyclic Reasoning Module
-# ============================================================================
-
-class CyclicReasoning(nn.Module):
-    """Módulo de raciocínio cíclico iterativo.
-
-    Aplica múltiplos ciclos de refinamento sobre a representação inicial,
-    parando antecipadamente quando a convergência é atingida.
-
-    Args:
-        config: configuração do módulo
-    """
-
-    def __init__(self, config: CyclicReasoningConfig):
-        super().__init__()
-        self.config = config
-        self.embed_dim = config.embed_dim
-
-        # Camadas compartilhadas entre ciclos (weight sharing) — reduz parâmetros
-        # Alternativa: um RefinementLayer por ciclo (mais capacidade, mais parâmetros)
-        self.refinement = RefinementLayer(
-            embed_dim=config.embed_dim,
-            hypothesis_dim=config.hypothesis_input_dim,
-            hidden_mult=config.refinement_hidden_mult,
-            dropout=config.dropout,
-        )
-        self.confidence_gate = ConfidenceGate(config.embed_dim)
-        self.fuzzy_gate = (
-            FuzzyLogicGate(config.embed_dim) if config.use_anti_hallucination_gate else None
-        )
-        self.ln = nn.LayerNorm(config.embed_dim) if config.use_layer_norm else None
-
-    def forward(
-        self,
-        h0: torch.Tensor,
-        hypothesis_fn: Optional[Callable[[torch.Tensor, int], Optional[torch.Tensor]]] = None,
-        penalty_signal: Optional[torch.Tensor] = None,
-        return_history: bool = False,
-    ) -> Dict[str, torch.Tensor]:
-        """Executa raciocínio cíclico.
-
-        Args:
-            h0: [B, D] representação inicial
-            hypothesis_fn: função (h_c, c) -> g_c (hipótese opcional para ciclo c).
-                Se None, g_c = None (sem hipótese).
-            penalty_signal: [B] ou [B, 1] penalidade do verificador (para log).
-            return_history: se True, retorna lista de h_c por ciclo.
-
-        Returns:
-            dict com:
-                h_final: [B, D] estado final
-                n_cycles: int — número de ciclos executados
-                converged: bool — se convergiu antes de C_max
-                deltas: lista de ||h_c - h_{c-1}|| por ciclo
-                history: lista de h_c (se return_history=True)
-                alphas: lista de alpha_c
-                taus: lista de tau_c (se fuzzy gate ativo)
-        """
-        if h0.dim() != 2:
-            raise ValueError(f"h0 deve ser [B, D]; recebido {h0.shape}")
-
-        B, D = h0.shape
-        if D != self.embed_dim:
-            raise ValueError(f"embed_dim mismatch: h0={D}, config={self.embed_dim}")
-
-        h = h0
-        history: List[torch.Tensor] = [h0]
-        deltas: List[float] = []
-        alphas: List[float] = []
-        taus: List[float] = []
-        converged = False
-        n_cycles_exec = 0
-
-        for c in range(self.config.max_cycles):
-            n_cycles_exec = c + 1
-
-            # Truncated BPTT: detach após K ciclos
-            if (
-                self.config.use_truncated_bptt
-                and c >= self.config.bptt_truncate
-            ):
-                h = h.detach()
-
-            # 1. Gerar hipótese g_c (se função fornecida)
-            g_c = None
-            if hypothesis_fn is not None:
-                try:
-                    g_c = hypothesis_fn(h, c)
-                except Exception as e:
-                    logger.debug("hypothesis_fn falhou no ciclo %d: %s", c, e)
-                    g_c = None
-
-            # 2. Refinamento
-            r_c = self.refinement(h, g_c)
-
-            # 3. Coeficiente de confiança
-            alpha_c = self.confidence_gate(h)  # [B, 1]
-
-            # 4. Novo estado
-            if self.config.use_residual:
-                h_new = h + alpha_c * r_c
-            else:
-                h_new = alpha_c * r_c
-
-            # 5. Anti-hallucination gate
-            if self.fuzzy_gate is not None:
-                tau_c = self.fuzzy_gate(h_new)  # [B, 1]
-                accept_mask = self.fuzzy_gate.accept(tau_c)  # [B, 1] em {0, 1}
-                h_new = h * (1 - accept_mask) + h_new * accept_mask
-                taus.append(float(tau_c.mean().detach()))
-            else:
-                tau_c = None
-
-            # 6. LayerNorm
-            if self.ln is not None:
-                h_new = self.ln(h_new)
-
-            # 7. Convergência (delta = ||h_new - h||)
-            with torch.no_grad():
-                delta = (h_new - h).norm(dim=-1).mean().item()
-            deltas.append(delta)
-            alphas.append(float(alpha_c.mean().detach()))
-
-            h = h_new
-            history.append(h)
-
-            # 8. Parada antecipada
-            if delta < self.config.convergence_eps:
-                converged = True
-                logger.debug(
-                    "CyclicReasoning convergiu no ciclo %d (delta=%.6f < eps=%.6f)",
-                    c + 1, delta, self.config.convergence_eps,
-                )
-                break
-
-        result = {
-            "h_final": h,
-            "n_cycles": n_cycles_exec,
-            "converged": converged,
-            "deltas": deltas,
-            "alphas": alphas,
-            "taus": taus,
-        }
-        if return_history:
-            result["history"] = history
-        if penalty_signal is not None:
-            result["penalty_signal"] = penalty_signal
-        return result
-
-
-# ============================================================================
-# Helper: hypothesis_fn factory a partir do HypothesisController
-# ============================================================================
-
-def make_hypothesis_fn(hypothesis_controller, penalty_threshold: float = 0.5):
-    """Cria uma função hypothesis_fn (h_c, c) -> g_c a partir do HypothesisController.
-
-    O HypothesisController gera hipóteses quando ativado por penalidades.
-    Esta função adapta a interface para uso no CyclicReasoning.
-    """
-    def hypothesis_fn(h_c: torch.Tensor, c: int) -> Optional[torch.Tensor]:
-        try:
-            # Tenta chamar o método generate_hypothesis do controller
-            if hasattr(hypothesis_controller, "generate_hypothesis"):
-                # Passa h_c como estado e retorna a hipótese (projeta se necessário)
-                hyp = hypothesis_controller.generate_hypothesis(h_c)
-                if hyp is None:
-                    return None
-                # Se dim diferente, projeta
-                if hasattr(hyp, "dim") and hyp.dim() == 2 and hyp.size(-1) != h_c.size(-1):
-                    if not hasattr(hypothesis_controller, "_proj_to_h"):
-                        hypothesis_controller._proj_to_h = nn.Linear(
-                            hyp.size(-1), h_c.size(-1), bias=False
-                        ).to(hyp.device)
-                        nn.init.orthogonal_(hypothesis_controller._proj_to_h.weight)
-                    hyp = hypothesis_controller._proj_to_h(hyp)
-                return hyp
-            return None
-        except Exception as e:
-            logger.debug("hypothesis_fn erro: %s", e)
-            return None
-
-    return hypothesis_fn
-
-
-# ============================================================================
-# Demo / Teste rápido
-# ============================================================================
-
-def _self_test():
-    """Teste rápido do módulo."""
-    torch.manual_seed(42)
-    config = CyclicReasoningConfig(
-        embed_dim=64,
-        max_cycles=5,
-        convergence_eps=1e-3,
-        use_anti_hallucination_gate=True,
-    )
-    cr = CyclicReasoning(config)
-    h0 = torch.randn(4, 64)
-    result = cr(h0, return_history=True)
-    print(f"n_cycles={result['n_cycles']}, converged={result['converged']}")
-    print(f"deltas={result['deltas']}")
-    print(f"alphas={result['alphas']}")
-    print(f"taus={result['taus']}")
-    print(f"h_final shape={result['h_final'].shape}")
-
-
-if __name__ == "__main__":
-    _self_test()
-
-
-__all__ = [
-    "CyclicReasoningConfig",
-    "RefinementLayer",
-    "ConfidenceGate",
-    "FuzzyLogicGate",
-    "CyclicReasoning",
-    "make_hypothesis_fn",
-]