Upload cnn_bigru/training/hypothesis_controller.py with huggingface_hub

cnn_bigru/training/hypothesis_controller.py

@@ -0,0 +1,276 @@
+"""hypothesis_controller.py — Mecanismo de hipóteses para autoaprendizado.
+
+Implementa o "acionamento de hipóteses quando ocorre punições no treinamento"
+descrito pelo usuário. Quando o verificador pune um passo (v < threshold),
+o sistema ativa camadas de hipótese que tentam N combinações alternativas
+de sinergias entre as redes, buscando a configuração de menor perda.
+
+Matemática:
+    Quando v_t < THRESHOLD_ERR:
+        hipótese_i = f_alternativa_i(input, params_perturbados_i)
+        loss_i = compute_loss(hipótese_i)
+        melhor = argmin(loss_i)
+        params <- params + lr * grad(melhor)
+
+As "camadas de hipótese" são camadas lineares paralelas (hipóteses)
+ativadas condicionalmente. A ativação é diferenciável via Gumbel-Softmax.
+"""
+from __future__ import annotations
+
+import logging
+import math
+import random
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class HypothesisConfig:
+    """Configuração do mecanismo de hipóteses."""
+    n_hypotheses: int = 4           # N tentativas de acerto
+    threshold_err: float = 0.5      # abaixo deste valor, ativa hipóteses
+    gumbel_temp: float = 1.0        # temperatura do Gumbel-Softmax
+    gumbel_hard: bool = False       # se True, usa argmax (não diferenciável)
+    perturbation_std: float = 0.05  # desvio padrão da perturbação
+
+
+class HypothesisLayer(nn.Module):
+    """Camada de hipótese: N transformações lineares paralelas.
+
+    Para cada amostra, calcula N hipóteses e seleciona a melhor via
+    Gumbel-Softmax (diferenciável) ou argmax (não diferenciável).
+
+    A "seleção" é feita baseada em um score externo (e.g. negativo da perda).
+    """
+
+    def __init__(
+        self,
+        in_dim: int,
+        out_dim: int,
+        n_hypotheses: int = 4,
+        activation: str = "relu",
+    ):
+        super().__init__()
+        self.n_hypotheses = n_hypotheses
+        self.in_dim = in_dim
+        self.out_dim = out_dim
+
+        # N transformações lineares paralelas
+        self.hypotheses = nn.ModuleList([
+            nn.Linear(in_dim, out_dim) for _ in range(n_hypotheses)
+        ])
+        # Inicialização diferente para cada hipótese (diversidade)
+        for i, h in enumerate(self.hypotheses):
+            nn.init.xavier_uniform_(h.weight, gain=0.5 + 0.2 * i)
+            nn.init.zeros_(h.bias)
+
+        # Scoreador: aprende a pontuar cada hipótese dado o input
+        self.scorer = nn.Linear(in_dim, n_hypotheses)
+
+        self.activation = activation
+
+    def _activate(self, x: torch.Tensor) -> torch.Tensor:
+        if self.activation == "relu":
+            return F.relu(x)
+        elif self.activation == "gelu":
+            return F.gelu(x)
+        elif self.activation == "tanh":
+            return torch.tanh(x)
+        return x
+
+    def forward(
+        self,
+        x: torch.Tensor,
+        gumbel_temp: float = 1.0,
+        gumbel_hard: bool = False,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        """
+        Args:
+            x: [B, in_dim]
+            gumbel_temp: temperatura do Gumbel-Softmax
+            gumbel_hard: se True, usa argmax
+
+        Returns:
+            output: [B, out_dim]
+            weights: [B, n_hypotheses]
+        """
+        B = x.size(0)
+
+        # Calcula todas as N hipóteses: [B, N, out_dim]
+        hyps = torch.stack([self._activate(h(x)) for h in self.hypotheses], dim=1)
+
+        # Scoreia cada hipótese
+        scores = self.scorer(x)  # [B, N]
+
+        # Gumbel-Softmax para seleção diferenciável
+        if gumbel_hard:
+            # Hard: argmax one-hot
+            weights = F.one_hot(scores.argmax(dim=-1), num_classes=self.n_hypotheses).float()
+        else:
+            weights = F.gumbel_softmax(scores, tau=gumbel_temp, hard=False)
+
+        # Combina: weighted sum
+        output = (hyps * weights.unsqueeze(-1)).sum(dim=1)  # [B, out_dim]
+
+        return output, weights
+
+
+class HypothesisController(nn.Module):
+    """Controlador de hipóteses: ativa camadas quando há punição.
+
+    Recebe:
+        - features do passo atual
+        - valor v do verificador (probabilidade de correção)
+        - loss atual
+
+    Se v < threshold, ativa as hipóteses e tenta N combinações.
+    Caso contrário, passa direto (identity).
+    """
+
+    def __init__(
+        self,
+        feat_dim: int,
+        out_dim: int,
+        config: HypothesisConfig,
+    ):
+        super().__init__()
+        self.cfg = config
+        self.feat_dim = feat_dim
+        self.out_dim = out_dim
+
+        # Camadas de hipótese paralelas
+        self.hyp_layer = HypothesisLayer(
+            in_dim=feat_dim,
+            out_dim=out_dim,
+            n_hypotheses=config.n_hypotheses,
+            activation="gelu",
+        )
+
+        # Porta de ativação (gating) baseada em v
+        # Se v alto, mantém identidade; se v baixo, ativa hipóteses
+        self.gate = nn.Linear(feat_dim + 1, 1)  # +1 para o valor v
+
+        # Projeto para combinar com input original
+        self.combine = nn.Linear(feat_dim + out_dim, out_dim)
+
+    def forward(
+        self,
+        x: torch.Tensor,
+        v: torch.Tensor,
+    ) -> Dict[str, torch.Tensor]:
+        """
+        Args:
+            x: [B, feat_dim]
+            v: [B, 1] probabilidade do verificador (0 = erro, 1 = correto)
+
+        Returns:
+            dict com:
+                output: [B, out_dim] features modificadas
+                gate: [B, 1] valor da porta (0 = identity, 1 = hipóteses)
+                weights: [B, N] pesos das hipóteses
+                activated: bool se as hipóteses foram ativadas (em qualquer amostra)
+        """
+        # Porta: gate = Sigmoid(linear(concat(x, v)))
+        # Como v baixo -> gate alto (ativa hipóteses), usamos (1 - v) na entrada
+        gate_input = torch.cat((x, 1.0 - v), dim=-1)  # [B, feat_dim + 1]
+        gate = torch.sigmoid(self.gate(gate_input))  # [B, 1]
+
+        # Hipóteses
+        hyp_out, weights = self.hyp_layer(
+            x, gumbel_temp=self.cfg.gumbel_temp, gumbel_hard=self.cfg.gumbel_hard
+        )
+
+        # Combina: out = combine(concat(x, hyp_out))
+        combined = self.combine(torch.cat((x, hyp_out), dim=-1))
+
+        # Aplica gate: se gate=0, mantém x (projetado); se gate=1, usa hipóteses
+        # Projeção de x para out_dim
+        x_proj = combined  # placeholder; usamos combined como base
+        output = gate * combined + (1 - gate) * x_proj
+
+        # Verifica se alguma amostra foi punida
+        activated = bool((v.squeeze(-1) < self.cfg.threshold_err).any().item())
+
+        return {
+            "output": output,
+            "gate": gate,
+            "weights": weights,
+            "activated": activated,
+        }
+
+
+class SynergySearcher:
+    """Busca sinérgica: tenta N combinações de hiperparâmetros e escolhe a melhor.
+
+    Dado um conjunto de "alavancas" (e.g. pesos das perdas, taxas de aprendizado),
+    explora N configurações e seleciona aquela que produz menor perda em um
+    mini-batch de validação.
+
+    Não é diferenciável — é uma busca heurística no espaço de configurações.
+    """
+
+    def __init__(
+        self,
+        n_attempts: int = 4,
+        seed: int = 42,
+    ):
+        self.n_attempts = n_attempts
+        self.rng = random.Random(seed)
+        self.history: List[Dict] = []
+
+    def sample_config(self, base: Dict) -> Dict:
+        """Amostra uma configuração perturbando a base."""
+        cfg = dict(base)
+        # Perturbação: multiplica pesos por fator aleatório próximo de 1
+        for key in ["alpha", "beta", "gamma_loss", "delta", "lambda_penal", "mu_exp_penal"]:
+            if key in cfg:
+                factor = 1.0 + self.rng.uniform(-0.2, 0.2)
+                cfg[key] = max(1e-6, cfg[key] * factor)
+        return cfg
+
+    def search(
+        self,
+        base_config: Dict,
+        eval_fn,
+    ) -> Dict:
+        """Executa N tentativas e retorna a melhor configuração.
+
+        Args:
+            base_config: configuração base (dict de hiperparâmetros)
+            eval_fn: função cfg -> loss (callable)
+
+        Returns:
+            melhor configuração encontrada
+        """
+        best_config = dict(base_config)
+        best_loss = float("inf")
+
+        for attempt in range(self.n_attempts):
+            try:
+                cfg = self.sample_config(base_config) if attempt > 0 else dict(base_config)
+                loss = float(eval_fn(cfg))
+                self.history.append({"attempt": attempt, "config": cfg, "loss": loss})
+                logger.info("SynergySearch attempt %d: loss=%.4f", attempt, loss)
+                if loss < best_loss:
+                    best_loss = loss
+                    best_config = cfg
+            except Exception as e:
+                logger.warning("SynergySearch attempt %d falhou: %s", attempt, e)
+                continue
+
+        logger.info("SynergySearch melhor loss=%.4f", best_loss)
+        return best_config
+
+
+__all__ = [
+    "HypothesisConfig",
+    "HypothesisLayer",
+    "HypothesisController",
+    "SynergySearcher",
+]