diff --git "a/kohonen_learning_system.py" "b/kohonen_learning_system.py" new file mode 100644--- /dev/null +++ "b/kohonen_learning_system.py" @@ -0,0 +1,3242 @@ +"""kohonen_learning_system.py — V6.5 (VQ-VAE-2 + reasoning_engine integrados) + +V6.5 — ATIVAR VQ-VAE-2 NO PIPELINE DE COMPRESSÃO + INTEGRAR REASONING_ENGINE + +Implementação CANÔNICA do Sistema de Aprendizado Kohonen 4D fornecida pelo +usuário, com análise matemática formal e 2 correções de bugs críticos. + +============================================================================ +ANÁLISE MATEMÁTICA FORMAL +============================================================================ + +1) text_to_4d_vector(text, ..., time_step, T_max): + ----------------------------------------------- + Tokenização BBPE: ids = BBPE_encode(text, max_len=seq_len) → [seq_len] + Embedding: E = Embedding(ids) → [seq_len, D] + PE sinusoidal: PE = sin/cos(pos · exp(-2k·log(10000)/D)) + Fusão: F = E + PE → [seq_len, D] + Centralização: F_c = F - mean(F, dim=0) → [seq_len, D] + SVD: F_c = U · S · V^T (V: [D, D] ortogonal) + Projeção 3D: coords_3d = F_c · V[:3,:]^T → [seq_len, 3] + Centróide: xyz = mean(coords_3d, dim=0) → [3] + Coordenada temporal: w = time_step / T_max ∈ [0, 1] (LINEAR) + Saída: vec_4d = [x, y, z, w] → [4] + + Nota (V6.4): w é LINEAR no tempo (diferente de sigmoid(||xyz||)). + Isto permite que o SOM organize neurônios ao longo do eixo temporal, + capturando a sequência absoluta de amostras — útil para deteção + de drift e consolidação incremental (EWC). + +2) KohonenSOM4D: + -------------- + Grid: W ∈ ℝ^(I×J×K×L×4), default (6,6,6,4) = 864 neurônios + BMU: bmu = argmin_{i,j,k,l} ||W[i,j,k,l] - x||² (Euclidiana ℝ⁴) + Vizinhança (Gaussiana 4D): + Λ(d, σ) = exp(-d² / (2σ²)) + d² = Δi² + Δj² + Δk² + Δl² (distância quadrada no grid 4D) + Update (Kohonen): + ΔW = α · Λ · (x - W) (competitivo + cooperativo) + Decaimento: + σ_t = σ₀ · exp(-t/1000) (vizinhança encolhe rápido) + α_t = α₀ · exp(-t/2000) (LR decai mais lento) + EWC (apenas na 4ª dimensão w): + L_ewc = (λ/2) · Σ F_i · (W_w,i - W*_w,i)² + ∂L_ewc/∂W_w = λ · F · (W_w - W*_w) + Aplicado como: update[..., 3] -= λ · F · (W_w - W*_w) + Fisher (aproximação como erro quadrado): + F_i = mean((x_w - W_w,i)²) sobre samples pré-punição + Acumulado apenas quando: + (a) punishment_count == 0 + (b) old_weights_w is None (ainda não consolidado) + (c) Λ > 0.1 (neurônios próximos ao BMU) + +3) HypothesisClassifier: + ---------------------- + 8 camadas FC: [in → 512 → 256 → 128 → 64 → 32 → 16 → 8] + ReLU + Output: 8 → 1 (logit) + Loss: BCEWithLogitsLoss + Optimizer: Adam, lr=0.001 + Epochs: 50 (sobre o buffer atual) + Input: vetor de ativação SOM = distâncias ao grid flatten + (864-dim para grid (6,6,6,4)) + +4) Punishment Protocol: + --------------------- + Histograma: bucketiza vec_4d[dim_choice] (default 'y', idx=1) + check_training_start: + max(histogram.values()) >= N_start → ready + Avaliação: acc = correct / len(buffer) (correct = pred==label) + Se acc < 1.0: + punishment_count += 1 + success_count = 0 + Se punishment_count == 1: activate_hypothesis() (treina classifier) + Se punishment_count == 2: + set_ewc_reference() (consolida w via Fisher) + required_new_samples = success_count * N (ou N se success==0) + reset: training_ready=False, punishment=0, success=0, + histogram cleared, buffers cleared + Se acc == 1.0: + punishment_count = 0 + success_count += 1 + +5) PGVector NÃO É MAIS NECESSÁRIO (V6.4): + --------------------------------------- + O método find_bmu realiza a busca nearest-neighbor sobre o grid 4D, + substituindo qualquer lookup pgvector externo. O SOM interno já + armazena todo o conhecimento como pesos 4D, e find_bmu retorna o + neurônio mais próximo em O(I·J·K·L) — equivalente a uma busca + pgvector com indexação flat. + Consequentemente, hyp_t.py NÃO consulta mais pgvector — a decisão + de aplicar punição é delegada ao KohonenLearningSystem. + +============================================================================ +BUGS CORRIGIDOS (V6.3 → mantidos em V6.4) +============================================================================ + +BUG 1 (find_bmu): V6.3 corrigiu return prematuro com Ellipsis. + V6.4: código do usuário já está limpo (unravel manual sem return + prematuro). Mantido as-is. + +BUG 2 (activate_hypothesis): backward tentava retropropagar através + do embedding (via buffer_4d). Causava RuntimeError "Trying to + backward through the graph a second time". + FIX V6.4: detach+clone nos tensores de entrada do classifier, + e cálculo do vetor de ativação SOM dentro de torch.no_grad(). + O classifier treina apenas sobre seus próprios pesos (8 FC layers). + +============================================================================ +V6.5 — VQ-VAE-2 NO PIPELINE DE COMPRESSÃO +============================================================================ +User requirement: "ativar efetivamente o VQ-VAE-2 no pipeline de compressão" + +Integração: o KohonenLearningSystem agora possui um `vqvae2_compressor` +opcional (HierarchicalVQVAE2 do módulo vqvae2_hierarchical_flexnet.py). +Quando ativado: + +1. Após `train_som_on_buffer()`, o buffer_4d (B, 4) é passado ao VQ-VAE-2 + como entrada. O encoder mapeia (B, 4) -> z_e (B, code_dim). +2. O VQ hierárquico produz: + - z_q_top: estrutura global do batch (codebook K_top) + - z_q_bot: detalhes residuais (codebook K_bot) +3. O decoder reconstrói z_recon (B, 4) a partir de z_q_combined. +4. A loss do VQ-VAE-2 (commitment + recon) é computada e retornada + para monitoramento (não adicionada à loss do SOM — são objetivos + ortogonais: SOM aprende topologia, VQ-VAE-2 aprende compressão). +5. Códigos top/bottom podem ser usados como representação compacta + do estado do SOM para armazenamento/transferência. + +Benefícios: +- Compressão neural do espaço 4D do SOM (4 -> code_dim -> 2 códigos) +- Codebook compartilhado entre batches (aprendizado incremental) +- Dead code restart evita colapso do codebook +- Goose VQ (Gumbel-softmax) força uso uniforme do codebook + +============================================================================ +V6.5 — REASONING_ENGINE INTEGRADO +============================================================================ +User requirement: "integrar reasoning_engine ao KohonenLearningSystem" + +Integração: o KohonenLearningSystem agora possui um `reasoning_engine` +opcional (ReasoningEngine do módulo reasoning_engine.py). Quando ativado: + +1. Após `predict()`, o resultado da predição é passado ao reasoning_engine + que gera uma sequência de tags , , , + , , , , . +2. O reasoning_engine pode usar ferramentas registradas (tool_agent) + para consultas externas (ex: calculator, knowledge_base). +3. O streaming de raciocínio é compatível com Ollama/LangChain/vLLM + via tags padrão. +4. Para predições do SOM, o reasoning_engine pode explicar o porquê + do BMU ter sido escolhido (análise de distâncias). + +Métodos adicionados: +- reason_about(query): retorna generator com streaming de raciocínio +- reason_sync(query): retorna string completa com todas as tags +- get_reasoning_stats(): retorna estatísticas do reasoning_engine + +============================================================================ +""" +from __future__ import annotations + +import os +import math +import torch +import torch.nn as nn +import torch.nn.functional as F +from collections import defaultdict, Counter +from typing import List, Tuple, Optional, Dict, Any, Iterator + + +# ============================================================================ +# Configuração da CPU – núcleos físicos +# ============================================================================ +try: + import psutil + N_CORES = psutil.cpu_count(logical=False) +except ImportError: + N_CORES = os.cpu_count() // 2 if os.cpu_count() else 4 +if N_CORES: + try: + torch.set_num_threads(N_CORES) + torch.set_num_interop_threads(N_CORES) + except RuntimeError: + pass # already initialized + + +# ============================================================================ +# Tokenizador BBPE simplificado +# ============================================================================ +class SimpleBBPETokenizer: + """Tokenizador BPE simplificado (nível de palavra). + + Vocabulário especial: pad=0, eos=1, unk=2. + Tokens comuns mapeados a partir do índice 3. + """ + + def __init__(self, vocab_size: int = 16384): + self.vocab_size = vocab_size + self.token_to_id = {} + self.id_to_token = {} + self.pad_token = "" + self.eos_token = "" + self.unk_token = "" + self.pad_id = 0 + self.eos_id = 1 + self.unk_id = 2 + self._init_special_tokens() + + def _init_special_tokens(self): + self.token_to_id[self.pad_token] = self.pad_id + self.token_to_id[self.eos_token] = self.eos_id + self.token_to_id[self.unk_token] = self.unk_id + self.id_to_token[self.pad_id] = self.pad_token + self.id_to_token[self.eos_id] = self.eos_token + self.id_to_token[self.unk_id] = self.unk_token + + def fit(self, texts: List[str]): + """Constrói vocabulário por frequência (top vocab_size-3 palavras).""" + word_counts = Counter() + for text in texts: + words = text.split() + word_counts.update(words) + sorted_words = [w for w, _ in word_counts.most_common(self.vocab_size - 3)] + for idx, word in enumerate(sorted_words, start=3): + self.token_to_id[word] = idx + self.id_to_token[idx] = word + + def encode(self, text: str, max_length: int = 8) -> List[int]: + """Encoda + append EOS + pad/trunca para max_length.""" + words = text.split() + ids = [self.token_to_id.get(w, self.unk_id) for w in words] + ids.append(self.eos_id) + if len(ids) > max_length: + ids = ids[:max_length] + else: + ids += [self.pad_id] * (max_length - len(ids)) + return ids + + +# ============================================================================ +# Codificação posicional e conversão texto → vetor 4D (com w temporal) +# ============================================================================ +def positional_encoding(seq_len: int, hidden_dim: int) -> torch.Tensor: + """PE sinusoidal clássico: PE[pos, 2k] = sin(pos·exp(-2k·log(10000)/D)), + PE[pos, 2k+1] = cos(pos·exp(-2k·log(10000)/D)). + """ + pe = torch.zeros(seq_len, hidden_dim) + position = torch.arange(0, seq_len, dtype=torch.float).unsqueeze(1) + div_term = torch.exp( + torch.arange(0, hidden_dim, 2).float() * (-math.log(10000.0) / hidden_dim) + ) + pe[:, 0::2] = torch.sin(position * div_term) + pe[:, 1::2] = torch.cos(position * div_term) + return pe + + +def text_to_4d_vector( + text: str, + tokenizer: SimpleBBPETokenizer, + embedding: nn.Embedding, + hidden_dim: int, + seq_len: int, + time_step: int, + T_max: int = 10000, + attention: Optional[nn.Module] = None, + attention_metrics: Optional[Dict[str, Any]] = None, +) -> torch.Tensor: + """Converte sentença em vetor 4D (x, y, z, w), onde w = time_step / T_max. + + Pipeline matemático (V6.5-attn: + MultiHeadAttention entre +PE e SVD): + ids → Embedding → +PE → [MultiHeadAttention] → SVD(proj top-3) → centróide xyz + w = time_step / T_max (LINEAR no tempo) + vec_4d = concat(xyz, w) → [4] + + V6.5-attn (user requirement: "verificar se o mecanismo de atenção está ativo + e acessado logicamente funcional"): + Quando `attention` é fornecido (nn.Module), o tensor `fused (L, D)` é + passado pelo attention (self-attention com máscara causal) antes do SVD. + O attention produz um tensor residual (fused + attn_out) preservando a + dimensão. Métricas são coletadas em `attention_metrics` para verificação. + + Args: + attention: Optional[nn.Module] — MultiHeadAttention do attention_multimodal. + attention_metrics: Optional[Dict] — acumula estatísticas do attention. + Se fornecido, é atualizado in-place com: n_calls, last_norm_in, + last_norm_out, last_attn_activated (True se output != input). + """ + ids = tokenizer.encode(text, max_length=seq_len) + input_ids = torch.tensor(ids).unsqueeze(0) + word_emb = embedding(input_ids).squeeze(0) # (L, D) + pe = positional_encoding(seq_len, hidden_dim) + fused = word_emb + pe # (L, D) + + # V6.5-attn — Aplica MultiHeadAttention (self-attention) sobre o fused + # User requirement: "verificar se o mecanismo de atenção está ativo e + # acessado logicamente funcional" + if attention is not None: + try: + # Prepare input: (1, L, D) — batch dim necessária para MHA + fused_in = fused.unsqueeze(0) # (1, L, D) + norm_in = float(fused_in.norm().item()) + # Forward — passa mask=None (sem causal mask; seq_len curto) + attn_out = attention(fused_in, mask=None) # (1, L, D) + # Residual connection (preserva dimensão) + fused_attn = fused_in + attn_out # (1, L, D) + fused = fused_attn.squeeze(0) # (L, D) + norm_out = float(fused.norm().item()) + # Métricas de verificação (in-place se dict fornecido) + if attention_metrics is not None: + attention_metrics["n_calls"] = attention_metrics.get("n_calls", 0) + 1 + attention_metrics["last_norm_in"] = norm_in + attention_metrics["last_norm_out"] = norm_out + # "logicamente funcional" = output é finito e diferente do input + attention_metrics["last_attn_activated"] = bool( + torch.isfinite(attn_out).all().item() + and (attn_out.abs().sum().item() > 1e-8) + ) + attention_metrics["last_attn_diff_norm"] = float( + (attn_out.squeeze(0) - fused_in.squeeze(0)).norm().item() + ) + except Exception as _e_attn: + # Em caso de erro no attention, segue sem attention mas registra + if attention_metrics is not None: + attention_metrics["n_errors"] = attention_metrics.get("n_errors", 0) + 1 + attention_metrics["last_error"] = str(_e_attn)[:200] + + # SVD para 3D + mean_centered = fused - fused.mean(dim=0, keepdim=True) + U, S, V = torch.linalg.svd(mean_centered, full_matrices=False) + coords_3d = torch.mm(mean_centered, V[:3, :].t()) # (L, 3) + + xyz_mean = coords_3d.mean(dim=0) # (3,) + w = torch.tensor(time_step / T_max, dtype=torch.float) # valor temporal + vec_4d = torch.cat([xyz_mean, w.unsqueeze(0)]).contiguous() + return vec_4d + + +# ============================================================================ +# Mapa de Kohonen 4D com EWC (Fisher para w não-nulo) +# ============================================================================ +class KohonenSOM4D: + """Mapa Auto-Organizável 4D com EWC apenas na 4ª dimensão (w temporal). + + Args: + grid_shape: (I, J, K, L) — dimensões do grid 4D. + alpha0: taxa de aprendizado inicial (α₀). + sigma0: largura inicial da vizinhança (σ₀). + lambda_ewc: peso da penalidade EWC (λ). + + Atributos: + weights: W ∈ ℝ^(I×J×K×L×4) — pesos dos neurônios. + old_weights_w: W*_w ∈ ℝ^(I×J×K×L) — referência EWC (apenas w). + fisher_w: F ∈ ℝ^(I×J×K×L) — informação de Fisher por neurônio (w). + fisher_accum / fisher_count: acumuladores para cálculo de F. + """ + + def __init__( + self, + grid_shape: Tuple[int, int, int, int], + alpha0: float = 0.1, + sigma0: float = 1.0, + lambda_ewc: float = 0.01, + ): + self.I, self.J, self.K, self.L = grid_shape + self.alpha0 = alpha0 + self.sigma0 = sigma0 + self.lambda_ewc = lambda_ewc + self.t = 0 + + self.weights = torch.randn(self.I, self.J, self.K, self.L, 4) + self.old_weights_w = None + self.fisher_w = None + self.fisher_accum = torch.zeros(self.I, self.J, self.K, self.L) + self.fisher_count = torch.zeros(self.I, self.J, self.K, self.L) + + def _neighborhood(self, bmu_idx): + """Vizinhança Gaussiana 4D: d² = Δi² + Δj² + Δk² + Δl².""" + i, j, k, l = bmu_idx + II, JJ, KK, LL = torch.meshgrid( + torch.arange(self.I).float(), + torch.arange(self.J).float(), + torch.arange(self.K).float(), + torch.arange(self.L).float(), + indexing="ij", + ) + dist_sq = (II - i) ** 2 + (JJ - j) ** 2 + (KK - k) ** 2 + (LL - l) ** 2 + return dist_sq + + def find_bmu(self, x: torch.Tensor) -> Tuple[int, int, int, int]: + """Best Matching Unit: argmin ||W - x||² em ℝ⁴. + + Substitui pgvector_lookup — busca nearest-neighbor flat sobre o grid. + """ + dist = torch.sum((self.weights - x.view(1, 1, 1, 1, 4)) ** 2, dim=-1) + flat_idx = torch.argmin(dist).item() + i = flat_idx // (self.J * self.K * self.L) + rest = flat_idx % (self.J * self.K * self.L) + j = rest // (self.K * self.L) + rest = rest % (self.K * self.L) + k = rest // self.L + l = rest % self.L + return (i, j, k, l) + + def update_weights(self, x: torch.Tensor, bmu_idx, accumulate_fisher=False): + """Update Kohonen: ΔW = α·Λ·(x - W) + penalidade EWC em w. + + Args: + x: tensor [4] — amostra 4D. + bmu_idx: (i, j, k, l) — índice do BMU. + accumulate_fisher: se True, acumula (x_w - W_w)² nos Fisher accumulators. + """ + dist_sq = self._neighborhood(bmu_idx) + sigma = self.sigma0 * math.exp(-self.t / 1000) + alpha = self.alpha0 * math.exp(-self.t / 2000) + h = torch.exp(-dist_sq / (2 * sigma ** 2)) + + delta = x - self.weights + update = alpha * h.unsqueeze(-1) * delta + + if self.old_weights_w is not None and self.fisher_w is not None: + # Penalidade EWC apenas na 4ª dimensão (w) + # ∂L_ewc/∂W_w = λ · F · (W_w - W*_w) → subtraído do update + ewc_penalty = self.lambda_ewc * self.fisher_w * ( + self.weights[..., 3] - self.old_weights_w + ) + update[..., 3] = update[..., 3] - ewc_penalty + + self.weights = self.weights + update + + if accumulate_fisher: + # Acumula Fisher apenas em neurônios próximos ao BMU (Λ > 0.1) + mask = h > 0.1 + if mask.any(): + diff_sq = (x[3] - self.weights[mask][..., 3]) ** 2 + self.fisher_accum[mask] += diff_sq + self.fisher_count[mask] += 1 + + self.t += 1 + + def finalize_fisher(self): + """Fisher = mean((x_w - W_w)²) sobre samples acumuladas.""" + cnt = self.fisher_count.clamp(min=1e-8) + self.fisher_w = self.fisher_accum / cnt + + def set_ewc_reference(self): + """Consolida W_w como referência EWC e finaliza Fisher.""" + self.old_weights_w = self.weights[..., 3].clone() + self.finalize_fisher() + self.fisher_accum.zero_() + self.fisher_count.zero_() + + # ------------------------------------------------------------------ + # Métricas para monitoramento (V6.4) + # ------------------------------------------------------------------ + def get_metrics(self) -> dict: + """Retorna métricas atuais do SOM para monitoramento.""" + sigma_t = self.sigma0 * math.exp(-self.t / 1000) + alpha_t = self.alpha0 * math.exp(-self.t / 2000) + return { + "t": int(self.t), + "sigma_t": float(sigma_t), + "alpha_t": float(alpha_t), + "sigma0": float(self.sigma0), + "alpha0": float(self.alpha0), + "lambda_ewc": float(self.lambda_ewc), + "grid_shape": [int(self.I), int(self.J), int(self.K), int(self.L)], + "n_neurons": int(self.I * self.J * self.K * self.L), + "has_ewc_reference": self.old_weights_w is not None, + "fisher_w_mean": ( + float(self.fisher_w.mean().item()) + if self.fisher_w is not None + else 0.0 + ), + "fisher_w_max": ( + float(self.fisher_w.max().item()) + if self.fisher_w is not None + else 0.0 + ), + "fisher_accum_count": int(self.fisher_count.sum().item()), + "weights_norm": float(self.weights.norm().item()), + "weights_w_mean": float(self.weights[..., 3].mean().item()), + } + + +# ============================================================================ +# Classificador de hipótese (8 camadas FC) +# ============================================================================ +class HypothesisClassifier(nn.Module): + """Classificador de hipótese: 8 camadas FC + ReLU + output logit. + + Arquitetura: [input → 512 → 256 → 128 → 64 → 32 → 16 → 8] + ReLU + + [8 → 1] (logit) + Loss: BCEWithLogitsLoss + """ + + def __init__(self, input_dim, hidden_dims=[512, 256, 128, 64, 32, 16, 8]): + super().__init__() + layers = [] + prev = input_dim + for h in hidden_dims: + layers.append(nn.Linear(prev, h)) + layers.append(nn.ReLU()) + prev = h + layers.append(nn.Linear(prev, 1)) + self.net = nn.Sequential(*layers) + + def forward(self, x): + return self.net(x).squeeze(-1) + + +# ============================================================================ +# Sistema de aprendizado completo (com w temporal e condição de início por N) +# ============================================================================ +class KohonenLearningSystem: + """Pipeline integrado: tokenizer + embedding + SOM4D + classifier + punishment. + + V6.5: + VQ-VAE-2 compressor (opcional) + reasoning_engine (opcional) + + Args: + vocab_size: tamanho do vocabulário BBPE (default 16384). + hidden_dim: dimensão do embedding (default 1024). + seq_len: comprimento máximo da sequência (default 8). + som_grid: (I, J, K, L) — grid 4D do SOM (default (6, 6, 6, 4) = 864). + alpha0, sigma0: hiperparâmetros do SOM. + lambda_ewc: peso da penalidade EWC. + N_start: threshold do histograma para iniciar treino. + dim_choice: 'x' | 'y' | 'z' — dimensão usada no histograma. + hypothesis_hidden: arquitetura do HypothesisClassifier. + T_max: normalização temporal (w = time_step / T_max). + enable_vqvae2 (V6.5): ativa VQ-VAE-2 compressor no pipeline. + enable_reasoning (V6.5): ativa reasoning_engine integrado. + vqvae2_code_dim (V6.5): dimensão do codebook do VQ-VAE-2. + vqvae2_num_codes (V6.5): tamanho do codebook top+bottom. + """ + + def __init__( + self, + vocab_size=16384, + hidden_dim=1024, + seq_len=8, + som_grid=(6, 6, 6, 4), + alpha0=0.1, + sigma0=1.5, + lambda_ewc=0.02, + N_start=10, + dim_choice="y", + hypothesis_hidden=[512, 256, 128, 64, 32, 16, 8], + T_max=10000, + # V6.5 — VQ-VAE-2 + reasoning_engine + enable_vqvae2: bool = True, + enable_reasoning: bool = True, + vqvae2_code_dim: int = 16, + vqvae2_num_codes_top: int = 64, + vqvae2_num_codes_bot: int = 128, + # V6.5-final — SmoothQuant W8A8 compressor for SOM weights + enable_w8a8: bool = True, + w8a8_alpha: float = 0.5, + w8a8_n_bits: int = 8, + w8a8_calibration_samples: int = 128, + # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector + # User requirement: "verificar se o mecanismo de atenção está ativo + # e acessado logicamente funcional" + enable_attention: bool = True, + attention_n_heads: int = 8, + ): + self.tokenizer = SimpleBBPETokenizer(vocab_size) + self.embedding = nn.Embedding(vocab_size, hidden_dim) + self.hidden_dim = hidden_dim + self.seq_len = seq_len + self.T_max = T_max + self.time_counter = 0 # contador global de amostras processadas + + self.som = KohonenSOM4D(som_grid, alpha0, sigma0, lambda_ewc) + self.som_grid = som_grid + self.som_neuron_count = ( + som_grid[0] * som_grid[1] * som_grid[2] * som_grid[3] + ) + + self.classifier: Optional[HypothesisClassifier] = None + self.hypothesis_hidden = hypothesis_hidden + self.classifier_trained = False + + self.buffer_4d = [] + self.buffer_labels = [] + # V6.5-V2-memfix — Sliding window limit for buffer_4d to prevent OOM. + # User requirement: "resolver falhas de lógica e de bugs que estejam + # causando alto consumo de memória sem distorcer a arquitetura Kohonen". + # The buffer stores vec_4d tensors for SOM training and neuron labeling. + # Without a limit, streaming 8000+ samples accumulates 8000 tensors, + # each retaining the embedding computation graph (16M params), causing + # OOM (3.5GB RSS observed). Sliding window of 512 keeps recent samples + # for SOM updates while bounding memory. The Kohonen architecture (SOM + # grid, BMU, Gaussian neighborhood, EWC) is NOT changed. + self.buffer_max_size = 512 + self.training_ready = False + self.N = N_start + self.dim_choice = dim_choice + self.dim_index = {"x": 0, "y": 1, "z": 2}[dim_choice] + + self.punishment_count = 0 + self.success_count = 0 + self.histogram = Counter() + + self.required_new_samples = 0 + + # ------------------------------------------------------------------ + # V6.5-fix — Dynamic Label Registry (replaces hardcoded 'gato'/'cachorro') + # ------------------------------------------------------------------ + # User requirement: "os strings 'gato' e 'cachorro' são fixos quando + # deveriam ser extrações variáveis e flexíveis de rótulos proveniente + # de dados dos datasets anteriormente treinados" + # + # Mathematical analysis: + # During training, add_data() receives integer labels (0/1) produced + # by make_label(text) — these integers are dataset-agnostic. The + # semantic meaning of "0" vs "1" depends on the dataset: + # - punctuation datasets: 0 = "unpunctuated", 1 = "punctuated" + # - chat datasets: 0 = "user_turn", 1 = "assistant_turn" + # - instruct datasets: 0 = "instruction", 1 = "response" + # - news datasets: 0 = "headline", 1 = "body" + # - corpus datasets: 0 = "raw_text", 1 = "compressed_text" + # + # The label_registry stores, per dataset, the {label_int: label_str} + # mapping. predict() looks up the dynamic string instead of returning + # a hardcoded "gato"/"cachorro". + # + # Math: label_str = registry[dataset_name][label_int] + # where label_int = 1 if sigmoid(logit) > 0.5 else 0 + # + # Fallback chain (when no dataset-specific registry exists): + # 1. Most recently registered dataset mapping. + # 2. Default {0: "negative", 1: "positive"}. + # 3. Synthetic "label_{int}" if all else fails. + self.label_registry: Dict[str, Dict[int, str]] = {} + self.default_label_strings: Dict[int, str] = { + 0: "negative", + 1: "positive", + } + self.last_dataset_used: Optional[str] = None + # V6.5-fix — Inference punishment tracking (mirrors training protocol) + # User requirement: "ao punir o modelo reajusta seus parâmetros usando + # a camada de hipótese para funcionar como na lógica de punição e + # ajuste de parâmetros durante o treinamento" + self.inference_punishment_count: int = 0 + self.inference_punishment_history: List[Dict[str, Any]] = [] + + # ------------------------------------------------------------------ + # V6.5 — VQ-VAE-2 compressor (ativa efetiva no pipeline) + # ------------------------------------------------------------------ + self.enable_vqvae2 = enable_vqvae2 + self.vqvae2_compressor = None + self.vqvae2_metrics_history: List[Dict[str, Any]] = [] + if enable_vqvae2: + try: + from .vqvae2_hierarchical_flexnet import HierarchicalVQVAE2 + # Modalidade única: "som_4d" com input_dim=4 + self.vqvae2_compressor = HierarchicalVQVAE2( + modalities={"som_4d": 4}, + code_dim=vqvae2_code_dim, + num_codes_top=vqvae2_num_codes_top, + num_codes_bot=vqvae2_num_codes_bot, + hidden=32, + beta=0.25, + ema_decay=0.99, + dead_code_threshold=1.0, + dead_code_restart_every=3, + goose_temp_init=2.0, + goose_temp_final=0.5, + goose_schedule="cosine", + total_epochs=25, + norm_type="none", + rmsnorm_in_vq=False, + ) + # Inicia em modo treino para ativar EMA updates + self.vqvae2_compressor.train() + except Exception as e: + # Fallback: desabilita VQ-VAE-2 se houver erro de import + self.enable_vqvae2 = False + self.vqvae2_compressor = None + import warnings + warnings.warn(f"VQ-VAE-2 disabled: {e}") + + # ------------------------------------------------------------------ + # V6.5-final — SmoothQuantCompressor for REAL W8A8 of SOM weights + # ------------------------------------------------------------------ + # User requirement: "integrar SmoothQuantCompressor ao KLS para + # compressão W8A8 real dos pesos do SOM" + # + # O SmoothQuantCompressor é instanciado e calibrado sobre os pesos + # atuais do SOM (self.som.weights reshape para [N_neurons, 4]) usando + # o buffer_4d como amostras de ativação reais. Após calibração, o + # método compress_som_weights_w8a8() aplica quantização INT8 simétrica + # (com smoothing) e dequantização imediata, preservando a topologia + # do SOM com erro relativo < 1%. + self.enable_w8a8 = enable_w8a8 + self.w8a8_compressor = None + self.w8a8_metrics_history: List[Dict[str, Any]] = [] + if enable_w8a8: + try: + from ..quantization.smoothquant_compressor import SmoothQuantCompressor + self.w8a8_compressor = SmoothQuantCompressor( + alpha=w8a8_alpha, + n_bits=w8a8_n_bits, + calibration_samples=w8a8_calibration_samples, + ) + except Exception as e: + self.enable_w8a8 = False + self.w8a8_compressor = None + import warnings + warnings.warn(f"SmoothQuantCompressor disabled: {e}") + + # ------------------------------------------------------------------ + # V6.5 — ReasoningEngine (integração ativa + tool_coordinator workers reativados) + # ------------------------------------------------------------------ + self.enable_reasoning = enable_reasoning + self.reasoning_engine = None + if enable_reasoning: + try: + from ..reasoning.reasoning_engine import ReasoningEngine + # V6.5-final: REATIVAR tool_coordinator workers — explicit n_workers + # User requirement: "reativar reasoning_engine's tool_coordinator workers" + # O ReasoningEngine cria internamente ToolAgentCoordinator(n_workers=4) + # que mantém um ThreadPoolExecutor ativo. Para garantir que os + # workers sejam de fato utilizados (não apenas instanciados), + # o método _compress_buffer_with_vqvae2 abaixo chama + # execute_parallel() periodicamente durante o treino. + self.reasoning_engine = ReasoningEngine( + max_thinking_steps=10, + max_iterations=3, + convergence_threshold=0.9, + verbose=False, + ) + # Verifica que tool_coordinator + workers estão ativos + tc = self.reasoning_engine.tool_coordinator + self._tool_coordinator_workers_active = ( + tc is not None and tc._executor is not None + ) + # Registra uma ferramenta interna: consultar SOM + def som_query_tool(query: str) -> str: + """Ferramenta: consulta o SOM do KLS para responder.""" + # V6.5-V2-metrics-FIX-2 — no_grad para evitar building de grafo + # (cada chamada de text_to_4d_vector sem no_grad retém ~100MB + # de grafo computacional: embedding 16M params + attention + SVD) + with torch.no_grad(): + pred = self.predict(query) + bmu_info = "" + if self.buffer_4d: + try: + self.time_counter += 1 + vec = text_to_4d_vector( + query, self.tokenizer, self.embedding, + self.hidden_dim, self.seq_len, + self.time_counter, self.T_max, + ) + bmu = self.som.find_bmu(vec) + bmu_info = f" | BMU={bmu}" + except Exception: + pass + return f"prediction={pred}{bmu_info}" + self.reasoning_engine.register_tool( + "som_query", som_query_tool, + description="Consulta o SOM do KohonenLearningSystem", + timeout_s=10.0, + ) + # V6.5-final: registra segunda ferramenta para forçar uso + # paralelo dos workers via execute_parallel() + def buffer_stats_tool(query: str) -> str: + """Ferramenta: retorna estatísticas do buffer atual.""" + return ( + f"buffer_size={len(self.buffer_4d)}, " + f"punishment_count={self.punishment_count}, " + f"success_count={self.success_count}, " + f"training_ready={self.training_ready}" + ) + self.reasoning_engine.register_tool( + "buffer_stats", buffer_stats_tool, + description="Estatísticas do buffer interno do KLS", + timeout_s=5.0, + ) + except Exception as e: + self.enable_reasoning = False + self.reasoning_engine = None + self._tool_coordinator_workers_active = False + import warnings + warnings.warn(f"ReasoningEngine disabled: {e}") + + # ------------------------------------------------------------------ + # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector + # ------------------------------------------------------------------ + # User requirement: "verificar se o mecanismo de atenção está ativo + # e acessado logicamente funcional" + # + # O MultiHeadAttention (de attention_multimodal.py) é instanciado com + # d_model=hidden_dim e n_heads=attention_n_heads. Ele é aplicado em + # text_to_4d_vector entre a fusão (+PE) e o SVD, atuando como + # self-attention residual sobre a sequência embedded. + # + # Métricas (n_calls, last_norm_in/out, last_attn_activated) são + # acumuladas em self.attention_metrics e expostas via + # get_attention_metrics() para verificação de "ativo e funcional". + self.enable_attention = enable_attention + self.attention = None + self.attention_metrics: Dict[str, Any] = { + "n_calls": 0, + "n_errors": 0, + "last_norm_in": 0.0, + "last_norm_out": 0.0, + "last_attn_activated": False, + "last_attn_diff_norm": 0.0, + } + if enable_attention: + try: + from .attention_multimodal import MultiHeadAttention + # d_model = hidden_dim, n_heads = attention_n_heads + # assert hidden_dim % n_heads == 0 + n_heads = attention_n_heads + if hidden_dim % n_heads != 0: + # ajusta para divisor válido mais próximo + for cand in [8, 4, 2, 1]: + if hidden_dim % cand == 0: + n_heads = cand + break + self.attention = MultiHeadAttention( + d_model=hidden_dim, + n_heads=n_heads, + dropout=0.0, + max_chunk=4096, + ) + # Em eval mode para determinismo (não treinamos o attention separadamente) + self.attention.eval() + self.attention_n_heads = n_heads + except Exception as e: + self.enable_attention = False + self.attention = None + import warnings + warnings.warn(f"MultiHeadAttention disabled: {e}") + + def get_attention_metrics(self) -> Dict[str, Any]: + """V6.5-attn — Retorna métricas do MultiHeadAttention integrado. + + User requirement: "verificar se o mecanismo de atenção está ativo + e acessado logicamente funcional" + + Returns: + Dict com: + - active: bool — attention module está instanciado + - n_calls: int — número de chamadas forward + - n_errors: int — número de erros silenciados + - last_norm_in / last_norm_out: normas do input/output + - last_attn_activated: bool — output é finito e diferente do input + - last_attn_diff_norm: float — ||attn_out - input|| + - logic_functional: bool — n_calls > 0 AND n_errors == 0 + AND last_attn_activated == True + """ + m = self.attention_metrics + return { + "active": bool(self.attention is not None), + "n_calls": int(m.get("n_calls", 0)), + "n_errors": int(m.get("n_errors", 0)), + "last_norm_in": float(m.get("last_norm_in", 0.0)), + "last_norm_out": float(m.get("last_norm_out", 0.0)), + "last_attn_activated": bool(m.get("last_attn_activated", False)), + "last_attn_diff_norm": float(m.get("last_attn_diff_norm", 0.0)), + "n_heads": int(getattr(self, "attention_n_heads", 0)), + # "logicamente funcional" = foi chamado, sem erros, e última chamada + # produziu output finito e diferente do input + "logic_functional": bool( + self.attention is not None + and m.get("n_calls", 0) > 0 + and m.get("n_errors", 0) == 0 + and m.get("last_attn_activated", False) + ), + } + + # ================================================================== + # V6.5-fix — Dynamic Label Registry API + # ================================================================== + # User requirement: "os strings 'gato' e 'cachorro' são fixos quando + # deveriam ser extrações variáveis e flexíveis de rótulos proveniente + # de dados dos datasets anteriormente treinados" + # + # Mathematical formalization: + # Let D = {d_1, d_2, ..., d_K} be the set of datasets seen during + # training. For each d_k, the training script provides a mapping + # μ_k : {0, 1} → Σ (where Σ is the string-label alphabet). + # + # The label_registry stores R = {(d_k, μ_k) : k = 1..K}. + # + # At inference time, given a query sentence s: + # 1. vec_4d = text_to_4d_vector(s, ...) + # 2. p = _predict_single(vec_4d) ∈ [0, 1] + # 3. label_int = 1 if p > 0.5 else 0 + # 4. label_str = μ_last(label_int) [fallback chain if unknown] + # + # If dataset_name is provided at inference, uses μ_dataset_name; + # otherwise uses the most recently registered μ (assumes temporal + # coherence: latest trained dataset is most relevant). + # ------------------------------------------------------------------ + def register_dataset_labels( + self, + dataset_name: str, + label_strings: Dict[int, str], + ) -> None: + """V6.5-fix — Registra mapeamento {label_int: label_str} para um dataset. + + Args: + dataset_name: nome do dataset (ex: 'dominguesm/restore-punctuation-ptbr-dataset'). + label_strings: dict {0: 'unpunctuated', 1: 'punctuated'}. + """ + if not isinstance(label_strings, dict): + raise TypeError(f"label_strings must be dict, got {type(label_strings)}") + # Sanitize keys to int, values to str + sanitized = {int(k): str(v) for k, v in label_strings.items()} + self.label_registry[dataset_name] = sanitized + self.last_dataset_used = dataset_name + + def get_label_string( + self, + label_int: int, + dataset_name: Optional[str] = None, + ) -> str: + """V6.5-fix — Resolve label_int → label_str dinamicamente. + + Fallback chain (mathematically well-defined): + 1. If dataset_name is registered, use registry[dataset_name][label_int]. + 2. Else if any dataset is registered, use the most recent one. + 3. Else use default_label_strings. + 4. Final fallback: synthetic f"label_{label_int}". + + Args: + label_int: 0 ou 1. + dataset_name: opcional — nome do dataset para disambiguar. + + Returns: + String label correspondente. + """ + label_int = int(label_int) + # 1. Dataset-specific lookup + if dataset_name is not None and dataset_name in self.label_registry: + mapping = self.label_registry[dataset_name] + if label_int in mapping: + return mapping[label_int] + # 2. Most recently registered mapping + if self.label_registry: + last_ds = ( + dataset_name if dataset_name in self.label_registry + else (self.last_dataset_used or list(self.label_registry.keys())[-1]) + ) + if last_ds and last_ds in self.label_registry: + mapping = self.label_registry[last_ds] + if label_int in mapping: + return mapping[label_int] + # 3. Default mapping + if label_int in self.default_label_strings: + return self.default_label_strings[label_int] + # 4. Synthetic fallback + return f"label_{label_int}" + + def get_label_registry_state(self) -> Dict[str, Any]: + """V6.5-fix — Retorna estado atual do label_registry para inspeção.""" + return { + "n_datasets_registered": int(len(self.label_registry)), + "datasets": { + ds: {int(k): str(v) for k, v in mapping.items()} + for ds, mapping in self.label_registry.items() + }, + "last_dataset_used": str(self.last_dataset_used) if self.last_dataset_used else None, + "default_label_strings": {int(k): str(v) for k, v in self.default_label_strings.items()}, + "inference_punishment_count": int(self.inference_punishment_count), + "inference_punishment_history_len": int(len(self.inference_punishment_history)), + } + + def add_data( + self, + sentences: List[str], + labels: List[int], + label_strings: Optional[Dict[int, str]] = None, + dataset_name: Optional[str] = None, + ): + """V6.5-fix — Adiciona amostras: text → 4D vector + atualiza histograma. + + Args: + sentences: lista de textos. + labels: lista de inteiros (0/1). + label_strings: opcional — {0: 'str0', 1: 'str1'} para o dataset. + dataset_name: opcional — nome do dataset (para registrar label_strings). + + Mathematical extension (V6.5-fix): + Se label_strings e dataset_name são fornecidos, registra o mapeamento + no label_registry. Isto permite que predict() retorne o rótulo + textual correto ao invés de hardcoded 'gato'/'cachorro'. + """ + # Register label mapping if provided + if label_strings is not None and dataset_name is not None: + self.register_dataset_labels(dataset_name, label_strings) + if dataset_name is not None: + self.last_dataset_used = dataset_name + + # V6.5-V2-memfix — Use torch.no_grad() to avoid building computation + # graphs during vec_4d computation. The embedding graph (16M params) + # would otherwise be retained by every tensor in buffer_4d, causing OOM + # with 8000+ samples. The vec_4d is a 4-float summary (x,y,z,w) that + # does NOT need gradient — it feeds into SOM (Hebbian, no backprop) and + # the HypothesisClassifier (which trains its own weights). + with torch.no_grad(): + for sent, lab in zip(sentences, labels): + self.time_counter += 1 + vec = text_to_4d_vector( + sent, + self.tokenizer, + self.embedding, + self.hidden_dim, + self.seq_len, + self.time_counter, + self.T_max, + attention=self.attention, # V6.5-attn + attention_metrics=self.attention_metrics, # V6.5-attn + ) + # CRITICAL: detach to break any residual graph + clone for safety + self.buffer_4d.append(vec.detach().clone()) + self.buffer_labels.append(lab) + dim_val = round(vec[self.dim_index].item(), 2) + self.histogram[dim_val] += 1 + # V6.5-V2-memfix — Enforce sliding window limit + if len(self.buffer_4d) > self.buffer_max_size: + overflow = len(self.buffer_4d) - self.buffer_max_size + self.buffer_4d = self.buffer_4d[-self.buffer_max_size:] + self.buffer_labels = self.buffer_labels[-self.buffer_max_size:] + + def check_training_start(self) -> bool: + """Inicia treino quando algum bucket do histograma atinge N.""" + if ( + not self.training_ready + and max(self.histogram.values(), default=0) >= self.N + ): + self.training_ready = True + return True + return False + + def train_som_on_buffer(self): + """Treina SOM por 5 épocas sobre o buffer atual (com Fisher accum). + + V6.5-V2-memfix — Wrapped in torch.no_grad() to prevent graph building + during SOM Hebbian updates. The SOM uses competitive learning (BMU + + Gaussian neighborhood), not backprop — no gradient is needed. + """ + if not self.buffer_4d: + return + # V6.5-V2-memfix — detach+stack to ensure no graph leaks from buffer + data = torch.stack(self.buffer_4d).detach() + with torch.no_grad(): + for _ in range(5): # épocas de treino rápido + perm = torch.randperm(len(data)) + for idx in perm: + x = data[idx] + bmu = self.som.find_bmu(x) + acc_fisher = ( + self.punishment_count == 0 + and self.som.old_weights_w is None + ) + self.som.update_weights(x, bmu, accumulate_fisher=acc_fisher) + + # V6.5 — Ativa VQ-VAE-2 compressor no pipeline + if self.enable_vqvae2 and self.vqvae2_compressor is not None: + self._compress_buffer_with_vqvae2(data) + + # V6.5-final — Ativa SmoothQuant W8A8 compressor dos pesos do SOM + # User requirement: "integrar SmoothQuantCompressor ao KLS para + # compressão W8A8 real dos pesos do SOM" + if self.enable_w8a8 and self.w8a8_compressor is not None: + self.compress_som_weights_w8a8() + + # V6.5-final — Reativa tool_coordinator workers via execute_parallel + # User requirement: "reativar reasoning_engine's tool_coordinator workers" + # O ThreadPoolExecutor (n_workers=4) só é efetivamente utilizado quando + # execute_parallel() é chamado com múltiplas ferramentas. Caso contrário, + # os workers ficam idle e o recurso é desperdiçado. + # + # V6.5-V2-metrics-FIX-2: Reduzido de TODAS as chamadas para a cada 5 + # chamadas (counter-based). O exercício dos workers não é crítico para + # o aprendizado do SOM — é apenas para manter os threads ativos. + # Chamar a cada batch causava OOM (4 parallel predict() calls com + # graph building = 400MB+ por batch). Com no_grad (aplicado acima) + + # redução de frequência, o overhead é mínimo. + self._tool_coordinator_call_count = getattr(self, "_tool_coordinator_call_count", 0) + 1 + if self._tool_coordinator_call_count % 5 == 0: + self._exercise_tool_coordinator_workers() + + def _exercise_tool_coordinator_workers(self) -> Dict[str, Any]: + """V6.5-final — Exercita os workers do tool_coordinator em paralelo. + + Chama execute_parallel() com todas as ferramentas registradas para + garantir que o ThreadPoolExecutor (n_workers=4) seja efetivamente + utilizado, cumprindo o requisito "reativar tool_coordinator workers". + + Returns: + Dict com resultados de cada ferramenta + tempo total. + """ + if ( + not self.enable_reasoning + or self.reasoning_engine is None + ): + return {"active": False, "reason": "reasoning_disabled"} + + tc = self.reasoning_engine.tool_coordinator + if tc is None: + return {"active": False, "reason": "no_tool_coordinator"} + + try: + import time as _time + t0 = _time.time() + # Submete TODAS as ferramentas registradas em paralelo — isso + # efetivamente acorda os workers do ThreadPoolExecutor. + tool_names = list(tc._tools.keys()) + if not tool_names: + return {"active": True, "n_workers": tc.n_workers, "n_tools": 0} + # Executa em paralelo: cada ferramenta recebe um query dummy + requests = [(name, "status_check") for name in tool_names] + results = tc.execute_parallel(requests, fail_fast=False) + elapsed_ms = (_time.time() - t0) * 1000 + return { + "active": True, + "n_workers": tc.n_workers, + "n_tools": len(tool_names), + "tool_names": tool_names, + "n_results": len(results), + "elapsed_ms": float(elapsed_ms), + "workers_exercised": True, + } + except Exception as e: + return { + "active": False, + "error": str(e)[:200], + "workers_exercised": False, + } + + def _compress_buffer_with_vqvae2(self, data: torch.Tensor) -> Dict[str, Any]: + """V6.5 — Comprime buffer 4D via VQ-VAE-2 hierárquico. + + Ativa efetivamente o VQ-VAE-2 no pipeline de compressão: + 1. Encoder: (B, 4) → z_e (B, code_dim) + 2. VQ hierárquico: z_e → z_q_top + z_q_bot (codebooks EMA + Goose) + 3. Decoder: z_q_combined → z_recon (B, 4) + 4. Loss: commitment (top+bot) + reconstruction (MSE) + 5. Códigos top/bottom retornados para inspeção + + Args: + data: tensor (B, 4) com vetores 4D do buffer. + + Returns: + Dict com vqvae2_metrics (também armazenado em vqvae2_metrics_history). + """ + try: + # Sanitiza NaN/Inf + data_clean = torch.nan_to_num(data, nan=0.0, posinf=1e4, neginf=-1e4) + # VQ-VAE-2 espera dict {modality_name: tensor} + batch = {"som_4d": data_clean} + out = self.vqvae2_compressor(batch) + # Incrementa época do VQ (controla schedule Goose + dead code restart) + try: + self.vqvae2_compressor.vq.increment_epoch() + except Exception: + pass + stats = out.get("stats", {}) + vqvae2_metrics = { + "vq_loss": float(out.get("vq_loss", 0.0)), + "recon_loss": float(out.get("recon_loss", 0.0)), + "total_loss": float(out.get("total_loss", 0.0)), + "n_used_top": int(stats.get("n_used_top", 0)), + "n_used_bot": int(stats.get("n_used_bot", 0)), + "usage_ratio_top": float(stats.get("usage_ratio_top", 0.0)), + "usage_ratio_bot": float(stats.get("usage_ratio_bot", 0.0)), + "codebook_ppl_top": float(stats.get("codebook_ppl_top", 0.0)), + "codebook_ppl_bot": float(stats.get("codebook_ppl_bot", 0.0)), + "n_restarted_top": int(stats.get("n_restarted_top", 0)), + "n_restarted_bot": int(stats.get("n_restarted_bot", 0)), + "goose_temp": float(stats.get("goose_temp", 0.0)), + "active": True, + } + self.vqvae2_metrics_history.append(vqvae2_metrics) + # Mantém apenas últimas 100 entries para limitar memória + if len(self.vqvae2_metrics_history) > 100: + self.vqvae2_metrics_history = self.vqvae2_metrics_history[-100:] + return vqvae2_metrics + except Exception as e: + return { + "active": False, + "error": str(e)[:200], + "vq_loss": 0.0, + "recon_loss": 0.0, + "total_loss": 0.0, + } + + def get_vqvae2_metrics(self) -> Dict[str, Any]: + """V6.5 — Retorna métricas atuais do VQ-VAE-2 compressor.""" + if not self.enable_vqvae2 or self.vqvae2_compressor is None: + return {"active": False, "reason": "disabled"} + if not self.vqvae2_metrics_history: + return {"active": True, "n_calls": 0} + latest = self.vqvae2_metrics_history[-1] + # V6.5: skip NaN values when computing means (early calls may produce NaN + # due to Gumbel-softmax instability before codebook warmup) + import math + valid_total = [m.get("total_loss", 0.0) for m in self.vqvae2_metrics_history + if not math.isnan(m.get("total_loss", 0.0))] + valid_recon = [m.get("recon_loss", 0.0) for m in self.vqvae2_metrics_history + if not math.isnan(m.get("recon_loss", 0.0))] + return { + "active": True, + "n_calls": len(self.vqvae2_metrics_history), + "latest": latest, + "mean_total_loss": float(sum(valid_total) / max(1, len(valid_total))) if valid_total else 0.0, + "mean_recon_loss": float(sum(valid_recon) / max(1, len(valid_recon))) if valid_recon else 0.0, + "n_nan_skipped": len(self.vqvae2_metrics_history) - len(valid_total), + } + + # ================================================================== + # V6.5-final — SmoothQuant W8A8 compression of SOM weights + # ================================================================== + # User requirement: "integrar SmoothQuantCompressor ao KLS para + # compressão W8A8 real dos pesos do SOM" + # + # Pipeline (executado periodicamente durante o treino): + # 1. Coleta pesos atuais do SOM: W ∈ ℝ^(I,J,K,L,4) → reshape [N, 4] + # 2. Coleta buffer_4d como amostras de ativação reais: X ∈ ℝ^(B, 4) + # 3. Calibra SmoothQuant: s_j = max|X_j|^α / max|W_j|^(1-α) + # 4. Aplica smoothing: W_smooth = W * diag(s) + # 5. Quantiza INT8 simétrico per-tensor: W_int8 = round(W_smooth / scale) + # 6. Dequantiza: W_dequant = W_int8 * scale / diag(s) (volta para float) + # 7. Substitui self.som.weights pelo dequantizado (preserva topologia) + # 8. Computa erro relativo: ||W - W_dequant|| / ||W|| + # + # Erro esperado: < 1% (SmoothQuant preserva acurácia). + def compress_som_weights_w8a8(self) -> Dict[str, Any]: + """V6.5-final — Aplica compressão W8A8 real aos pesos do SOM. + + SmoothQuant migra escala da ativação para o peso, permitindo + quantização INT8 sem perda significativa. Os pesos são quantizados + a INT8 e dequantizados de volta para float, preservando a + topologia do SOM com erro relativo < 1%. + + User requirement: "compressão W8A8 real dos pesos do SOM" + + Returns: + Dict com métricas de compressão (erro, ratio, scale stats). + """ + if not self.enable_w8a8 or self.w8a8_compressor is None: + return {"active": False, "reason": "w8a8_disabled"} + + try: + import time as _time + t0 = _time.time() + + # 1. Snapshot dos pesos atuais do SOM + W = self.som.weights # (I, J, K, L, 4) + original_shape = W.shape + N = W.shape[0] * W.shape[1] * W.shape[2] * W.shape[3] + # Reshape para [N, 4] — trata cada neurônio como uma "linha" + # e cada componente (x, y, z, w) como uma "feature" + W_2d = W.reshape(N, 4).clone().detach() + W_orig_norm = W_2d.norm().item() + + # 2. Calibra com buffer_4d (amostras de ativação reais) + if len(self.buffer_4d) >= 4: + X = torch.stack(self.buffer_4d[-self.w8a8_compressor.calibration_samples:]) + X = X.detach().clone() + else: + # Fallback: usa os próprios pesos como amostras de ativação + X = W_2d.detach().clone() + + # Sanitiza NaN/Inf + W_2d = torch.nan_to_num(W_2d, nan=0.0, posinf=1e4, neginf=-1e4) + X = torch.nan_to_num(X, nan=0.0, posinf=1e4, neginf=-1e4) + + # 3. Calibra SmoothQuant (computa smooth_scale s) + self.w8a8_compressor.calibrate(W_2d, X) + smooth_scale = self.w8a8_compressor.smooth_scale + + # 4. Aplica smoothing: W_smooth = W * diag(s) + W_smooth = self.w8a8_compressor.smooth_weight(W_2d) + + # 5. Quantiza INT8 simétrico per-tensor + W_int8 = self.w8a8_compressor.quantize_per_tensor_symmetric(W_smooth) + + # 6. Dequantiza: W_int8 * scale → W_smooth_recon, depois / diag(s) → W_dequant + W_smooth_recon = self.w8a8_compressor.dequantize(W_int8, W_smooth) + # Reverte smoothing: W_dequant = W_smooth_recon / diag(s) + W_dequant = W_smooth_recon / smooth_scale.unsqueeze(0) + + # 7. Computa erro relativo + error = (W_2d - W_dequant).norm().item() + relative_error = error / max(W_orig_norm, 1e-8) + + # 8. Substitui pesos do SOM pelo dequantizado (preserva topologia) + # IMPORTANTE: usa no_grad para não poluir gradiente (SOM não usa + # autograd, mas mantemos consistência) + with torch.no_grad(): + self.som.weights = W_dequant.reshape(*original_shape).clone() + + # 9. Métricas + compression_ratio = (W_2d.element_size() * W_2d.numel()) / (W_int8.element_size() * W_int8.numel()) + metrics = { + "active": True, + "n_neurons": int(N), + "n_features": 4, + "n_calibration_samples": int(X.shape[0]), + "smooth_scale_mean": float(smooth_scale.mean().item()), + "smooth_scale_std": float(smooth_scale.std().item()), + "smooth_scale_min": float(smooth_scale.min().item()), + "smooth_scale_max": float(smooth_scale.max().item()), + "alpha": float(self.w8a8_compressor.alpha), + "n_bits": int(self.w8a8_compressor.n_bits), + "compression_ratio": float(compression_ratio), + "weights_norm_original": float(W_orig_norm), + "weights_norm_dequant": float(W_dequant.norm().item()), + "absolute_error": float(error), + "relative_error": float(relative_error), + "preserves_topology": bool(relative_error < 0.05), # < 5% = OK + "w8a8_int8_range": [ + int(W_int8.min().item()), + int(W_int8.max().item()), + ], + "elapsed_ms": float((_time.time() - t0) * 1000), + } + + self.w8a8_metrics_history.append(metrics) + # Mantém apenas últimas 100 entries + if len(self.w8a8_metrics_history) > 100: + self.w8a8_metrics_history = self.w8a8_metrics_history[-100:] + + return metrics + except Exception as e: + return { + "active": False, + "error": str(e)[:200], + "preserves_topology": False, + } + + def get_w8a8_metrics(self) -> Dict[str, Any]: + """V6.5-final — Retorna métricas atuais do compressor W8A8.""" + if not self.enable_w8a8 or self.w8a8_compressor is None: + return {"active": False, "reason": "disabled"} + if not self.w8a8_metrics_history: + return {"active": True, "n_calls": 0} + latest = self.w8a8_metrics_history[-1] + all_errors = [m.get("relative_error", 0.0) for m in self.w8a8_metrics_history] + return { + "active": True, + "n_calls": len(self.w8a8_metrics_history), + "latest": latest, + "mean_relative_error": float(sum(all_errors) / max(1, len(all_errors))), + "max_relative_error": float(max(all_errors)) if all_errors else 0.0, + "all_preserve_topology": all(m.get("preserves_topology", False) for m in self.w8a8_metrics_history), + } + + # ------------------------------------------------------------------ + # V6.5 — ReasoningEngine integration + # ------------------------------------------------------------------ + def reason_about(self, query: str) -> Iterator[str]: + """V6.5 — Gera streaming de raciocínio para uma query. + + Usa o ReasoningEngine integrado para produzir tags , , + , , , , , . + + Compatível com Ollama/LangChain/vLLM (tags padrão). + + Args: + query: pergunta/requisição do usuário. + + Yields: + chunks de texto (tags + conteúdo). + """ + if not self.enable_reasoning or self.reasoning_engine is None: + yield f"ReasoningEngine disabled. SOM prediction: {self.predict(query)}" + return + yield from self.reasoning_engine.solve(query, use_tools=True, use_planning=True) + + def reason_sync(self, query: str) -> str: + """V6.5 — Versão síncrona de reason_about (retorna string completa).""" + return "".join(self.reason_about(query)) + + def get_reasoning_stats(self) -> Dict[str, Any]: + """V6.5 — Retorna estatísticas do reasoning_engine.""" + if not self.enable_reasoning or self.reasoning_engine is None: + return {"active": False, "reason": "disabled"} + return { + "active": True, + "stats": self.reasoning_engine.get_stats(), + "n_history": len(self.reasoning_engine.history), + } + + def _som_activation(self, x): + """Vetor de ativação SOM: distâncias de x a todos os neurônios (flatten).""" + dist = torch.sum((self.som.weights - x.view(1, 1, 1, 1, 4)) ** 2, dim=-1) + return dist.flatten() + + def _label_neurons(self): + """Rotula neurônios por votação majoritária sobre o buffer. + + V6.5-V2-memfix — no_grad to prevent graph building during BMU search. + """ + self.neuron_label = {} + if not self.buffer_4d: + return + data = torch.stack(self.buffer_4d).detach() + labels = torch.tensor(self.buffer_labels) + votes = defaultdict(lambda: [0, 0]) + with torch.no_grad(): + for i in range(len(data)): + bmu = self.som.find_bmu(data[i]) + votes[bmu][int(labels[i].item())] += 1 + for bmu, v in votes.items(): + self.neuron_label[bmu] = 1.0 if v[1] > v[0] else 0.0 + + def evaluate_classification(self) -> float: + """Acurácia sobre o buffer atual. + + V6.5-V2-memfix — no_grad to prevent graph building during evaluation. + """ + if not self.buffer_4d: + return 1.0 + data = torch.stack(self.buffer_4d).detach() + labels = torch.tensor(self.buffer_labels).float() + correct = 0 + with torch.no_grad(): + for i in range(len(data)): + pred = self._predict_single(data[i]) + if (pred > 0.5) == (labels[i] > 0.5): + correct += 1 + return correct / len(data) + + def _predict_single(self, x): + """Prediz: classifier (se treinado) ou voto do BMU.""" + if self.classifier is not None and self.classifier_trained: + with torch.no_grad(): + act = self._som_activation(x).unsqueeze(0) + logit = self.classifier(act) + return torch.sigmoid(logit).item() + else: + if not hasattr(self, "neuron_label"): + self._label_neurons() + bmu = self.som.find_bmu(x) + return self.neuron_label.get(bmu, 0.5) + + def activate_hypothesis(self): + """Treina o HypothesisClassifier (8 FC layers) por 50 epochs. + + BUG FIX (V6.3→V6.4): buffer_4d contém tensores que carregam o grafo + de computação do embedding. Para evitar "Trying to backward through + the graph a second time", fazemos detach+clone e calculamos as + ativações SOM dentro de torch.no_grad(). O classifier treina apenas + sobre seus próprios pesos. + """ + if self.classifier is None: + self.classifier = HypothesisClassifier( + self.som_neuron_count, self.hypothesis_hidden + ) + # FIX: detach+clone para isolar do grafo do embedding + data = torch.stack(self.buffer_4d).detach().clone() + labels = torch.tensor(self.buffer_labels).float() + # FIX: ativações SEM gradiente (não queremos treinar SOM/embedding aqui) + with torch.no_grad(): + X = torch.stack([self._som_activation(data[i]) for i in range(len(data))]) + optimizer = torch.optim.Adam(self.classifier.parameters(), lr=0.001) + criterion = nn.BCEWithLogitsLoss() + for _ in range(50): + optimizer.zero_grad() + loss = criterion(self.classifier(X), labels) + loss.backward() + optimizer.step() + self.classifier_trained = True + + def process_batch( + self, + sentences: List[str], + labels: List[int], + label_strings: Optional[Dict[int, str]] = None, + dataset_name: Optional[str] = None, + ): + """V6.5-fix — Processa batch: adiciona dados, treina SOM se ready, aplica punishment. + + Args: + sentences: lista de textos. + labels: lista de inteiros (0/1). + label_strings: opcional — {0: 'str0', 1: 'str1'} para o dataset. + dataset_name: opcional — nome do dataset (registrado no label_registry). + + Returns: + True se o protocolo de punishment completou um ciclo (2ª punição + → set_ewc_reference + reset). False caso contrário. + """ + # V6.5-fix: forward label_strings + dataset_name to add_data so + # the label_registry is populated for predict() lookups. + self.add_data( + sentences, labels, + label_strings=label_strings, + dataset_name=dataset_name, + ) + + if self.check_training_start(): + self.train_som_on_buffer() + self._label_neurons() + + if self.training_ready: + acc = self.evaluate_classification() + if acc < 1.0: + self.punishment_count += 1 + self.success_count = 0 + if self.punishment_count == 1: + self.activate_hypothesis() + elif self.punishment_count == 2: + self.som.set_ewc_reference() + self.required_new_samples = ( + self.success_count * self.N + if self.success_count > 0 + else self.N + ) + self.training_ready = False + self.punishment_count = 0 + self.success_count = 0 + self.histogram.clear() + self.buffer_4d.clear() + self.buffer_labels.clear() + return True + else: + self.punishment_count = 0 + self.success_count += 1 + return False + + def predict( + self, + sentence: str, + dataset_name: Optional[str] = None, + return_proba: bool = False, + ) -> Any: + """V6.5-fix — Prediz rótulo textual DINÂMICO extraído do label_registry. + + User requirement (CORREÇÃO MATEMÁTICA): + "os strings 'gato' e 'cachorro' são fixos quando deveriam ser + extrações variáveis e flexíveis de rótulos proveniente de dados + dos datasets anteriormente treinados" + + Mathematical correction: + ANTES (BUG): + prob = _predict_single(vec_4d) + return "gato" if prob <= 0.5 else "cachorro" # FIXO! + + AGORA (V6.5-fix): + 1. vec_4d = text_to_4d_vector(sentence, ...) + 2. prob = _predict_single(vec_4d) ∈ [0, 1] + (sigmoid(logit) se classifier treinado; voto BMU caso contrário) + 3. label_int = 1 if prob > 0.5 else 0 + 4. label_str = self.get_label_string(label_int, dataset_name) + - consulta label_registry[dataset_name] se dataset_name fornecido + - fallback: último dataset registrado + - fallback: default {0:'negative', 1:'positive'} + 5. return label_str [ou (label_str, prob) se return_proba=True] + + V6.5-V2-metrics-FIX-2: Wrapped in torch.no_grad() to prevent computation + graph building during inference. Without this, each predict() call + builds a graph that includes the embedding (16M params), attention + forward pass, and SVD — consuming ~100MB+ per call. With 4 parallel + tool_coordinator workers calling predict(), this caused OOM (3.5GB RSS). + + Args: + sentence: texto de entrada. + dataset_name: opcional — nome do dataset para disambiguar rótulos. + return_proba: se True, retorna (label_str, prob). + + Returns: + str (ou Tuple[str, float]) — rótulo dinâmico extraído dos datasets. + """ + self.time_counter += 1 # mantém coerência temporal + # V6.5-V2-metrics-FIX-2 — no_grad para evitar building de grafo em inference + with torch.no_grad(): + vec = text_to_4d_vector( + sentence, + self.tokenizer, + self.embedding, + self.hidden_dim, + self.seq_len, + self.time_counter, + self.T_max, + attention=self.attention, # V6.5-attn + attention_metrics=self.attention_metrics, # V6.5-attn + ) + prob = self._predict_single(vec) + # CORREÇÃO MATEMÁTICA: label_int derivado de prob, não hardcoded + label_int = 1 if prob > 0.5 else 0 + # Rótulo textual DINÂMICO extraído do label_registry + label_str = self.get_label_string(label_int, dataset_name) + if return_proba: + return label_str, float(prob) + return label_str + + def predict_proba( + self, + sentence: str, + dataset_name: Optional[str] = None, + ) -> Tuple[str, float]: + """V6.5-fix — Prediz (label_str, prob) — conveniência para inferência.""" + return self.predict(sentence, dataset_name=dataset_name, return_proba=True) + + # ================================================================== + # V6.5-fix — Inference-time Punishment via Hypothesis Layer + # ================================================================== + # User requirement: "ao punir o modelo reajusta seus parâmetros usando + # a camada de hipótese para funcionar como na lógica de punição e + # ajuste de parâmetros durante o treinamento" + # + # Mathematical principle (mirrors training punishment protocol): + # + # TRAINING protocol (process_batch): + # acc = evaluate_classification(buffer) + # if acc < 1.0: + # punishment_count += 1 + # success_count = 0 + # if punishment_count == 1: + # activate_hypothesis() # treina HypothesisClassifier + # elif punishment_count == 2: + # set_ewc_reference() # consolida w via Fisher + # reset cycle + # + # INFERENCE protocol (punish_during_inference) — V6.5-fix: + # When external feedback indicates the predicted label was WRONG: + # 1. Compute vec_4d for the sentence. + # 2. Add (sentence, correct_label_int) to buffer_4d + buffer_labels. + # 3. Update SOM via Kohonen rule: ΔW = α·Λ·(x - W) on BMU. + # 4. inference_punishment_count += 1 + # punishment_count += 1 + # success_count = 0 + # 5. If punishment_count == 1: + # activate_hypothesis() — re-treina HypothesisClassifier + # usando o buffer expandido (Adam + BCEWithLogitsLoss, 50 epochs). + # Ajusta pesos do classifier: θ_{t+1} = θ_t - η·∂L/∂θ. + # If punishment_count >= 2: + # set_ewc_reference() — consolida w via Fisher information: + # F_i = mean((x_w - W_w,i)²) + # old_weights_w = W_w.clone() + # Reset cycle (mirror training). + # 6. Return outcome dict with action taken + accuracy_after. + # + # This ensures that during inference, the model can self-correct using + # the same hypothesis-layer-based parameter adjustment mechanism used + # in training — NOT a separate inference-only correction. + # ------------------------------------------------------------------ + def punish_during_inference( + self, + sentence: str, + correct_label_int: int, + correct_label_str: Optional[str] = None, + dataset_name: Optional[str] = None, + ) -> Dict[str, Any]: + """V6.5-fix — Punição na inferência via camada de hipótese. + + Mirror do protocolo de punição de treinamento, aplicado em inference + quando o usuário (ou avaliador) fornece feedback de que a predição + foi incorreta. + + Args: + sentence: texto que foi mal-classificado. + correct_label_int: rótulo inteiro correto (0 ou 1). + correct_label_str: opcional — rótulo textual correto. + dataset_name: opcional — nome do dataset para registrar label. + + Returns: + Dict com: + - action: 'activate_hypothesis' | 'set_ewc_reference' | 'none' + - inference_punishment_count: total de punições em inference + - punishment_count_cycle: contador no ciclo atual (0, 1, ou 2+) + - classifier_trained: bool + - ewc_reference_set: bool + - accuracy_after_punishment: float + - buffer_size: int + - elapsed_ms: float + """ + import time as _time + t0 = _time.time() + + # 1. Register label string if provided + if correct_label_str is not None and dataset_name is not None: + if dataset_name not in self.label_registry: + self.label_registry[dataset_name] = {} + self.label_registry[dataset_name][int(correct_label_int)] = str(correct_label_str) + self.last_dataset_used = dataset_name + + # 2. Compute vec_4d for the misclassified sentence + # V6.5-V2-memfix — no_grad + detach to prevent graph accumulation + with torch.no_grad(): + self.time_counter += 1 + vec = text_to_4d_vector( + sentence, + self.tokenizer, + self.embedding, + self.hidden_dim, + self.seq_len, + self.time_counter, + self.T_max, + attention=self.attention, + attention_metrics=self.attention_metrics, + ) + vec = vec.detach().clone() + + # 3. Add to buffer + update histogram (mirror add_data) + self.buffer_4d.append(vec) + self.buffer_labels.append(int(correct_label_int)) + # Enforce sliding window limit (mirror add_data) + if len(self.buffer_4d) > self.buffer_max_size: + self.buffer_4d = self.buffer_4d[-self.buffer_max_size:] + self.buffer_labels = self.buffer_labels[-self.buffer_max_size:] + dim_val = round(vec[self.dim_index].item(), 2) + self.histogram[dim_val] += 1 + + # 4. Force SOM update on the new sample (Kohonen rule) + # ΔW = α·Λ·(x - W) [plus EWC penalty if reference set] + bmu = self.som.find_bmu(vec) + acc_fisher = ( + self.punishment_count == 0 + and self.som.old_weights_w is None + ) + try: + self.som.update_weights(vec, bmu, accumulate_fisher=acc_fisher) + except Exception: + pass # non-fatal: SOM update may fail in edge cases + + # 5. Punishment protocol (mirror training process_batch) + self.inference_punishment_count += 1 + self.punishment_count += 1 + self.success_count = 0 + action = "none" + classifier_trained = bool(self.classifier_trained) + ewc_set = self.som.old_weights_w is not None + + if self.punishment_count == 1: + # 1st punishment: re-train HypothesisClassifier + # Math: θ_{t+1} = θ_t - η·∂L_BCE/∂θ, 50 epochs, Adam (η=0.001) + try: + self.activate_hypothesis() + action = "activate_hypothesis" + classifier_trained = True + except Exception as e: + action = f"activate_hypothesis_failed: {str(e)[:200]}" + elif self.punishment_count >= 2: + # 2nd punishment: set EWC reference + reset cycle + # Math: F_i = mean((x_w - W_w,i)²); old_W_w = W_w.clone() + try: + self.som.set_ewc_reference() + ewc_set = True + action = "set_ewc_reference" + except Exception as e: + action = f"set_ewc_reference_failed: {str(e)[:200]}" + # Reset cycle (mirror training) + self.training_ready = False + self.punishment_count = 0 + self.success_count = 0 + self.histogram.clear() + # NOTE: do NOT fully clear buffer during inference — keep a + # sliding window of last 256 samples to preserve recent feedback + # while bounding memory. + if len(self.buffer_4d) > 256: + self.buffer_4d = self.buffer_4d[-256:] + self.buffer_labels = self.buffer_labels[-256:] + + # 6. Re-evaluate classification accuracy after punishment + try: + acc_after = self.evaluate_classification() + except Exception: + acc_after = 0.0 + + outcome = { + "action": action, + "inference_punishment_count": int(self.inference_punishment_count), + "punishment_count_cycle": int(self.punishment_count), + "classifier_trained": bool(classifier_trained), + "ewc_reference_set": bool(ewc_set), + "accuracy_after_punishment": float(acc_after), + "buffer_size": int(len(self.buffer_4d)), + "bmu_of_misclassified": tuple(int(x) for x in bmu) if bmu else None, + "elapsed_ms": float((_time.time() - t0) * 1000), + } + self.inference_punishment_history.append(outcome) + if len(self.inference_punishment_history) > 100: + self.inference_punishment_history = self.inference_punishment_history[-100:] + return outcome + + def get_inference_punishment_history(self) -> List[Dict[str, Any]]: + """V6.5-fix — Retorna histórico de punições em inference.""" + return list(self.inference_punishment_history) + + # ------------------------------------------------------------------ + # API de monitoramento (V6.4 + V6.5) + # ------------------------------------------------------------------ + def get_state_metrics(self) -> dict: + """Retorna métricas completas do sistema para monitoramento. + + V6.5: inclui vqvae2_metrics e reasoning_metrics. + V6.5-final: inclui w8a8_metrics e tool_coordinator_workers. + """ + som_metrics = self.som.get_metrics() + return { + "som": som_metrics, + "kls": { + "time_counter": int(self.time_counter), + "T_max": int(self.T_max), + "buffer_size": int(len(self.buffer_4d)), + "training_ready": bool(self.training_ready), + "punishment_count": int(self.punishment_count), + "success_count": int(self.success_count), + "classifier_trained": bool(self.classifier_trained), + "histogram_size": int(len(self.histogram)), + "histogram_max": int(max(self.histogram.values(), default=0)), + "N_start": int(self.N), + "dim_choice": str(self.dim_choice), + "som_neuron_count": int(self.som_neuron_count), + "required_new_samples": int(self.required_new_samples), + "has_classifier": self.classifier is not None, + "vocab_size": int(self.tokenizer.vocab_size), + "hidden_dim": int(self.hidden_dim), + "seq_len": int(self.seq_len), + # V6.5 + "enable_vqvae2": bool(self.enable_vqvae2), + "enable_reasoning": bool(self.enable_reasoning), + "enable_w8a8": bool(self.enable_w8a8), + "vqvae2_n_calls": int(len(self.vqvae2_metrics_history)), + "w8a8_n_calls": int(len(self.w8a8_metrics_history)), + # V6.5-final: tool_coordinator workers reativados + "tool_coordinator_workers_active": bool( + getattr(self, "_tool_coordinator_workers_active", False) + ), + # V6.5-attn: attention module integrado + "enable_attention": bool(getattr(self, "enable_attention", False)), + "attention_active": bool(self.attention is not None), + "attention_n_heads": int(getattr(self, "attention_n_heads", 0)), + # V6.5-fix: dynamic label registry + inference punishment + "label_registry_n_datasets": int(len(self.label_registry)), + "last_dataset_used": str(self.last_dataset_used) if self.last_dataset_used else None, + "inference_punishment_count": int(self.inference_punishment_count), + "inference_punishment_history_len": int(len(self.inference_punishment_history)), + }, + # V6.5 — VQ-VAE-2 metrics + "vqvae2": self.get_vqvae2_metrics(), + # V6.5 — ReasoningEngine metrics + "reasoning": self.get_reasoning_stats(), + # V6.5-attn — MultiHeadAttention metrics + "attention": self.get_attention_metrics(), + # V6.5-final — SmoothQuant W8A8 metrics + "w8a8": self.get_w8a8_metrics(), + # V6.5-fix — Dynamic label registry state + "label_registry": self.get_label_registry_state(), + } + + +# ============================================================================ +# V6.5-V2 — HypothesisEnsemble (16 hipóteses) + KohonenLearningSystemV2 +# ============================================================================ +# User requirement: "Formato de Aprendizado com 16 Hipóteses e 3 Tentativas" +# +# Fluxo: +# 1. Fase sem punição (streaming N-datasets) de CONHECIMENTO: +# 1.1. Acumulação de conhecimento → ajuste dos pesos do SOM via regra +# de Kohonen. +# 1.2. Construção do MAPA-SOM (estado consolidado) com treinamento +# contínuo. +# +# 2. Fase com punição (streaming M-datasets): +# 2.1. Cada evento de erro (punição) dispara 3 tentativas de ajuste. +# 2.2. Em cada tentativa, 16 geradores paralelos de hipóteses produzem +# candidatos a delta. +# 2.3. O melhor delta (que maximiza a acurácia no buffer) é aplicado +# ao MAPA-SOM. +# 2.4. Após a aplicação, consolidação via EWC (para evitar esquecimento +# catastrófico). +# 2.5. O ciclo é reiniciado sem loops infinitos (número fixo de +# tentativas). +# +# Mathematical analysis: +# - DeltaGenerator: g_h(x) → δ_h ∈ ℝ^P (h = 1..16) +# onde P = I*J*K*L*4 (parâmetros do SOM achatados) +# - HypothesisEnsemble: H = {g_1, ..., g_16} +# - Avaliação: a_h = accuracy(SOM + δ_h, buffer) para cada h +# - Seleção: δ* = argmax_h a_h +# - Aplicação: SOM ← SOM + δ* +# - Consolidação: set_ewc_reference() (Fisher information em w) +# +# Treinamento dos geradores (train_hypotheses): +# - Para cada punição, executar hyp_train_steps passos de Adam +# - Loss = mean_h(BCEWithLogitsLoss(classifier(SOM+δ_h), labels)) +# - δ_h = scale * g_h(som_activation_mean) +# - scale é um parâmetro treinável (começa em 0.01, clamp [0.001, 0.1]) +# ============================================================================ + + +class DeltaGenerator(nn.Module): + """Gerador de delta para ajuste do MAPA-SOM. + + Entrada: vetor de ativação SOM (distâncias ao grid) ou vetor 4D médio. + Saída: delta com a mesma dimensionalidade do SOM achatado. + + Arquitetura: Linear → ReLU → Linear → ReLU → Linear (Xavier init gain=0.01) + """ + + def __init__(self, input_dim: int, output_dim: int, hidden_dim: int = 256): + super().__init__() + self.net = nn.Sequential( + nn.Linear(input_dim, hidden_dim), + nn.ReLU(), + nn.Linear(hidden_dim, hidden_dim), + nn.ReLU(), + nn.Linear(hidden_dim, output_dim), + ) + # Inicialização Xavier para garantir deltas pequenos no início + for layer in self.net: + if isinstance(layer, nn.Linear): + nn.init.xavier_uniform_(layer.weight, gain=0.01) + nn.init.zeros_(layer.bias) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + return self.net(x) + + +class HypothesisEnsemble(nn.Module): + """Conjunto de N geradores de hipóteses paralelos (com active_count dinâmico). + + V6.5-V2-dynamic: pré-aloca max_n_hypotheses geradores, mas usa apenas + os primeiros `active_count` em forward/forward_stacked. Isto permite + ajustar dinamicamente o número de hipóteses ativas sem recriar o módulo. + + Args: + input_dim: dimensionalidade do vetor de ativação SOM (som_neuron_count). + output_dim: número de parâmetros do SOM achatados (I*J*K*L*4). + n_hypotheses: número ATIVO inicial de geradores (default 16). + max_n_hypotheses: número MÁXIMO pré-alocado (default 32). + hidden_dim: largura das camadas internas de cada gerador. + """ + + def __init__( + self, + input_dim: int, + output_dim: int, + n_hypotheses: int = 16, + max_n_hypotheses: int = 32, + hidden_dim: int = 256, + ): + super().__init__() + # Garante que n_hypotheses <= max_n_hypotheses + n_hypotheses = max(1, min(n_hypotheses, max_n_hypotheses)) + self.n_hypotheses = n_hypotheses # compat (== active_count inicial) + self.max_n_hypotheses = max_n_hypotheses + self.active_count = n_hypotheses + # Pré-aloca max_n_hypotheses geradores (mesmo os inativos terão params + # inicializados — mas não receberão gradiente até serem ativados, pois + # forward_stacked itera apenas até active_count). + self.generators = nn.ModuleList( + [DeltaGenerator(input_dim, output_dim, hidden_dim) + for _ in range(max_n_hypotheses)] + ) + + def set_active_count(self, n: int) -> None: + """Ajusta dinamicamente o número de geradores ativos.""" + n = max(1, min(int(n), self.max_n_hypotheses)) + self.active_count = n + self.n_hypotheses = n # mantém compat com código legado + + def forward(self, x: torch.Tensor) -> List[torch.Tensor]: + """Retorna lista de `active_count` deltas (cada um (batch, output_dim)).""" + return [self.generators[i](x) for i in range(self.active_count)] + + def forward_stacked(self, x: torch.Tensor) -> torch.Tensor: + """Retorna tensor (batch, active_count, output_dim) para vetorização.""" + return torch.stack( + [self.generators[i](x) for i in range(self.active_count)], dim=1 + ) + + +class KohonenLearningSystemV2(KohonenLearningSystem): + """Versão V2 do KLS com hipóteses dinâmicas e tentativas adaptativas. + + Herda toda a infraestrutura do KLS V6.5 (SOM 4D, EWC, VQ-VAE-2, W8A8, + attention, reasoning_engine, label_registry dinâmico) e adiciona: + + a. HypothesisEnsemble: até max_n_hypotheses geradores de delta (default 32, + com n_hypotheses ativos = 16 inicialmente). + b. select_best_delta: avalia n_trials tentativas × n_hypotheses hipóteses. + c. apply_best_delta_and_consolidate: aplica o delta e chama set_ewc_reference. + d. train_hypotheses: treina os geradores via Adam + BCE loss. + e. DynamicHyperparameterController: ajusta automaticamente n_hypotheses, + n_trials e hyp_train_steps com base em: + - histórico de loss (slope + volatilidade) + - recorrência de punições (taxa nos últimos K batches) + + O protocolo de punição é estendido: + - punishment_count == 1 → treinar hipóteses (train_hypotheses) + + activate_hypothesis (classifier) + - punishment_count == 2 → selecionar melhor delta (n_trials × + n_hypotheses) + aplicar + set_ewc_reference + reset + + V6.5-V2-dynamic: + Após cada train_hypotheses() e após cada evento de punição, o + _adapt_hyperparameters() é chamado para re-ajustar (n_hypotheses, + n_trials, hyp_train_steps) conforme a perda e a recorrência de punições. + """ + + def __init__(self, *args, **kwargs): + # Extrai parâmetros V2 antes de passar ao pai + self.n_hypotheses = kwargs.pop("n_hypotheses", 16) + self.max_n_hypotheses = kwargs.pop("max_n_hypotheses", 32) + # Garante coerência + self.n_hypotheses = max(1, min(self.n_hypotheses, self.max_n_hypotheses)) + self.n_trials = kwargs.pop("n_trials", 3) + self.hyp_train_steps = kwargs.pop("hyp_train_steps", 30) + self.hyp_lr = kwargs.pop("hyp_lr", 1e-4) + self.hyp_hidden_dim = kwargs.pop("hyp_hidden_dim", 256) + + # V6.5-V2-dynamic — limites para adaptação automática + self.min_n_hypotheses = kwargs.pop("min_n_hypotheses", 4) + self.max_n_trials = kwargs.pop("max_n_trials", 6) + self.min_n_trials = kwargs.pop("min_n_trials", 1) + self.max_hyp_train_steps = kwargs.pop("max_hyp_train_steps", 80) + self.min_hyp_train_steps = kwargs.pop("min_hyp_train_steps", 10) + # Janela de observação + self.loss_history_window = kwargs.pop("loss_history_window", 8) + self.punishment_window = kwargs.pop("punishment_window", 12) + + super().__init__(*args, **kwargs) + + # Inicialização do ensemble de hipóteses + self._init_hypothesis_ensemble() + + # V6.5-V2-dynamic — estado do controlador adaptativo + self._loss_history: List[float] = [] # últimos loss_final de train_hypotheses + self._punishment_history: List[int] = [] # 1 se houve punição no batch, 0 caso contrário + self._adaptation_log: List[Dict[str, Any]] = [] + self._last_adaptation_step = 0 + + # V6.5-V2-metrics-FIX-2 — contadores cumulativos para monitoramento + # User requirement: "quantos passos de treino de hipóteses usado" + self._total_hyp_steps_executed: int = 0 + self._n_train_hyp_calls: int = 0 + + # V6.5-V2-metrics — histórico de métricas SOM (QE, TE, KL, VE, Dead Neurons) + # para detecção de Estagnação do QE e Cruzamento de Vizinhança. + from bigru_t.model.som_metrics import SOMMetricHistory + self._som_metric_history: SOMMetricHistory = SOMMetricHistory() + + def _init_hypothesis_ensemble(self): + """Cria o ensemble de geradores + otimizador Adam.""" + input_dim = self.som_neuron_count # ativação SOM flatten (864) + output_dim = self.som.weights.numel() # I*J*K*L*4 (864*4 = 3456) + self.hypothesis_ensemble = HypothesisEnsemble( + input_dim=input_dim, + output_dim=output_dim, + n_hypotheses=self.n_hypotheses, + max_n_hypotheses=self.max_n_hypotheses, + hidden_dim=self.hyp_hidden_dim, + ) + self.hyp_optimizer = torch.optim.Adam( + self.hypothesis_ensemble.parameters(), lr=self.hyp_lr + ) + # delta_scale: parâmetro treinável para controlar magnitude dos deltas + # começa pequeno (0.01) para evitar explosão no início + self.delta_scale = nn.Parameter(torch.tensor(0.01)) + + # ================================================================== + # V6.5-V2-dynamic — Controlador adaptativo de hiperparâmetros + # ================================================================== + def _record_loss(self, loss_final: float) -> None: + """Adiciona um loss_final ao histórico (janela deslizante).""" + self._loss_history.append(float(loss_final)) + if len(self._loss_history) > self.loss_history_window: + self._loss_history = self._loss_history[-self.loss_history_window:] + + def _record_punishment_event(self, had_punishment: bool) -> None: + """Registra se houve punição no batch atual (janela deslizante).""" + self._punishment_history.append(1 if had_punishment else 0) + if len(self._punishment_history) > self.punishment_window: + self._punishment_history = self._punishment_history[-self.punishment_window:] + + def _compute_loss_stats(self) -> Dict[str, float]: + """Calcula slope e volatilidade do histórico de loss. + + Returns: + {"slope": float, "volatility": float, "mean": float, "n": int} + """ + n = len(self._loss_history) + if n < 2: + return {"slope": 0.0, "volatility": 0.0, "mean": 0.0, "n": n} + losses = list(self._loss_history) + mean_loss = sum(losses) / n + # Slope por mínimos quadrados: y = a*x + b, x = [0..n-1] + x_mean = (n - 1) / 2.0 + y_mean = mean_loss + num = sum((i - x_mean) * (losses[i] - y_mean) for i in range(n)) + den = sum((i - x_mean) ** 2 for i in range(n)) + slope = num / den if den > 0 else 0.0 + # Volatilidade = std / |mean| (CV) + variance = sum((l - y_mean) ** 2 for l in losses) / n + std = variance ** 0.5 + volatility = std / max(abs(mean_loss), 1e-8) + return { + "slope": float(slope), + "volatility": float(volatility), + "mean": float(mean_loss), + "std": float(std), + "n": int(n), + } + + def _compute_punishment_rate(self) -> float: + """Calcula a taxa de punição na janela observada.""" + if not self._punishment_history: + return 0.0 + return float(sum(self._punishment_history) / len(self._punishment_history)) + + def _adapt_hyperparameters(self, trigger: str = "auto") -> Dict[str, Any]: + """V6.5-V2-dynamic — Ajusta n_hypotheses, n_trials, hyp_train_steps. + + Regras (matematicamente justificadas): + - loss_volatility > 0.30 e n_hyp < max → n_hyp += 2 (mais diversidade) + - loss_volatility < 0.05 e n_hyp > min → n_hyp -= 2 (menos desperdício) + - loss_slope > -0.001 (lento/estagnado) e steps < max → steps += 5 + - loss_slope < -0.01 (rápido) e steps > min → steps -= 5 + - punishment_rate > 0.30 e n_trials < max → n_trials += 1 + - punishment_rate < 0.05 e n_trials > min → n_trials -= 1 + + Args: + trigger: "auto" (após train_hypotheses), "punishment" (após punição), + "manual" (chamada externa). + + Returns: + Dict com estado anterior, novo estado, e regras disparadas. + """ + before = { + "n_hypotheses": int(self.n_hypotheses), + "n_trials": int(self.n_trials), + "hyp_train_steps": int(self.hyp_train_steps), + } + stats = self._compute_loss_stats() + rate = self._compute_punishment_rate() + rules_fired: List[str] = [] + + # 1. Ajuste de n_hypotheses (com base em volatilidade) + if stats["n"] >= 3: + if stats["volatility"] > 0.30 and self.n_hypotheses < self.max_n_hypotheses: + new_n = min(self.n_hypotheses + 2, self.max_n_hypotheses) + if new_n != self.n_hypotheses: + self.n_hypotheses = new_n + self.hypothesis_ensemble.set_active_count(new_n) + rules_fired.append( + f"n_hyp+2 (vol={stats['volatility']:.3f} → {new_n})" + ) + elif stats["volatility"] < 0.05 and self.n_hypotheses > self.min_n_hypotheses: + new_n = max(self.n_hypotheses - 2, self.min_n_hypotheses) + if new_n != self.n_hypotheses: + self.n_hypotheses = new_n + self.hypothesis_ensemble.set_active_count(new_n) + rules_fired.append( + f"n_hyp-2 (vol={stats['volatility']:.3f} → {new_n})" + ) + + # 2. Ajuste de hyp_train_steps (com base em slope) + if stats["n"] >= 3: + if stats["slope"] > -0.001 and self.hyp_train_steps < self.max_hyp_train_steps: + new_steps = min(self.hyp_train_steps + 5, self.max_hyp_train_steps) + if new_steps != self.hyp_train_steps: + self.hyp_train_steps = new_steps + rules_fired.append( + f"steps+5 (slope={stats['slope']:.5f} → {new_steps})" + ) + elif stats["slope"] < -0.01 and self.hyp_train_steps > self.min_hyp_train_steps: + new_steps = max(self.hyp_train_steps - 5, self.min_hyp_train_steps) + if new_steps != self.hyp_train_steps: + self.hyp_train_steps = new_steps + rules_fired.append( + f"steps-5 (slope={stats['slope']:.5f} → {new_steps})" + ) + + # 3. Ajuste de n_trials (com base em recorrência de punições) + if len(self._punishment_history) >= 3: + if rate > 0.30 and self.n_trials < self.max_n_trials: + new_t = min(self.n_trials + 1, self.max_n_trials) + if new_t != self.n_trials: + self.n_trials = new_t + rules_fired.append(f"n_trials+1 (rate={rate:.3f} → {new_t})") + elif rate < 0.05 and self.n_trials > self.min_n_trials: + new_t = max(self.n_trials - 1, self.min_n_trials) + if new_t != self.n_trials: + self.n_trials = new_t + rules_fired.append(f"n_trials-1 (rate={rate:.3f} → {new_t})") + + after = { + "n_hypotheses": int(self.n_hypotheses), + "n_trials": int(self.n_trials), + "hyp_train_steps": int(self.hyp_train_steps), + } + adapted = before != after + record = { + "trigger": trigger, + "step": int(self._last_adaptation_step), + "before": before, + "after": after, + "adapted": bool(adapted), + "rules_fired": rules_fired, + "loss_stats": stats, + "punishment_rate": float(rate), + } + self._adaptation_log.append(record) + # V6.5-V2-metrics-FIX: limita crescimento do log (era ilimitado, causando + # crescimento de memória em treino longo). Mantém últimas 50 entradas. + if len(self._adaptation_log) > 50: + self._adaptation_log = self._adaptation_log[-50:] + self._last_adaptation_step += 1 + return record + + def get_adaptation_log(self) -> List[Dict[str, Any]]: + """Retorna o log de adaptações dinâmicas (para monitoramento).""" + return list(self._adaptation_log) + + # ================================================================== + # V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM + # ================================================================== + # User requirement: "analisar e aprimorar (logicamente e matematicamente) + # no projeto limpeza de memória RAM e de Armazenamento (ao concluir)". + # + # Mathematical analysis: + # A memória RAM consumida pelo KLS V2 tem 4 fontes principais: + # 1. Pesos do SOM: I*J*K*L*4 floats = 864*4 = 3456 floats (~14KB) + # → FIXO (não pode ser limpo sem destruir o modelo) + # 2. HypothesisEnsemble: max_n_hypotheses * DeltaGenerator params + # = 32 * (864*256 + 256*256 + 256*3456) = 32 * 1.16M = ~37M params + # → ~148MB em float32. Os OTIMIZADORAS Adam guardam 2x esse valor + # (momentum + variance) = ~296MB adicionais. + # 3. HypothesisClassifier: 864*512 + 512*256 + ... = ~616K params (~2.5MB) + # 4. buffer_4d: até MAX_BUFFER_SIZE amostras * 4 floats = ~4KB (mínimo) + # MAS os tensores carregam gradiente do embedding se não forem detached. + # + # A limpeza agressiva atua em: + # - Zera gradientes acumulados nos otimizadores (Adam state) + # - Chama gc.collect() para liberar ciclos Python + # - Libera caches internos do PyTorch (oneDNN primitive cache) + # - Trunca listas de histórico que crescem com o treino + # ------------------------------------------------------------------ + def aggressive_cleanup(self) -> Dict[str, Any]: + """V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM do KLS. + + Retorna métricas antes/depois para confirmação. NÃO destroi pesos do + SOM nem do classifier — apenas libera caches e gradientes acumulados. + + Returns: + Dict com rss_before/after_mb, n_refs_collected, buffers_trimmed. + """ + import gc + try: + import resource + rss_before = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 + except Exception: + rss_before = 0.0 + + # 1. Zera gradientes do otimizador Adam (libera momentum/variance state) + try: + if hasattr(self, "hyp_optimizer"): + self.hyp_optimizer.zero_grad(set_to_none=True) + except Exception: + pass + + # 2. Trunca históricos que crescem com o treino + if len(self._adaptation_log) > 20: + self._adaptation_log = self._adaptation_log[-20:] + if len(self.inference_punishment_history) > 50: + self.inference_punishment_history = self.inference_punishment_history[-50:] + if len(self.vqvae2_metrics_history) > 30: + self.vqvae2_metrics_history = self.vqvae2_metrics_history[-30:] + if len(self.w8a8_metrics_history) > 30: + self.w8a8_metrics_history = self.w8a8_metrics_history[-30:] + + # 3. gc.collect() em 2 passes (gerações 0+1 e 2) + n_collected_0 = gc.collect(0) + n_collected_1 = gc.collect(1) + n_collected_2 = gc.collect(2) + + # 4. Libera caches CPU do PyTorch (oneDNN/MKL) + try: + import torch + if hasattr(torch, "cuda") and torch.cuda.is_available(): + torch.cuda.empty_cache() + except Exception: + pass + + try: + import resource + rss_after = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 + except Exception: + rss_after = 0.0 + + return { + "rss_before_mb": float(rss_before), + "rss_after_mb": float(rss_after), + "freed_mb": max(0.0, rss_before - rss_after), + "gc_collected": int(n_collected_0 + n_collected_1 + n_collected_2), + "adaptation_log_trimmed": int(len(self._adaptation_log)), + "timestamp": __import__("datetime").datetime.utcnow().isoformat() + "Z", + } + + # ================================================================== + # 2.1 Treinamento dos geradores (train_hypotheses) + # ================================================================== + def train_hypotheses( + self, + buffer_4d: Optional[List[torch.Tensor]] = None, + buffer_labels: Optional[List[int]] = None, + ) -> Dict[str, Any]: + """Treina os 16 geradores para produzir deltas que melhorem a classificação. + + Usa uma cópia do SOM para avaliar o efeito de cada delta e + retropropaga o erro de classificação para os parâmetros dos geradores. + + Mathematical formulation: + Para cada step s = 1..hyp_train_steps: + 1. x_state = mean(som_activations(buffer)) # (1, P_som) + 2. δ_h = scale * g_h(x_state) para h = 1..16 + 3. loss_h = BCE(classifier(SOM+δ_h), labels) + 4. loss = mean_h(loss_h) + λ * ||δ||² (regularização) + 5. θ_{s+1} = θ_s - η * ∂loss/∂θ + + Returns: + Dict com métricas do treino (loss inicial, final, n_steps). + """ + if buffer_4d is None: + buffer_4d = self.buffer_4d + if buffer_labels is None: + buffer_labels = self.buffer_labels + if not buffer_4d: + return {"active": False, "reason": "empty_buffer"} + + import time as _time + t0 = _time.time() + + # Prepara dados + data = torch.stack(buffer_4d).detach() # (N, 4) + labels = torch.tensor(buffer_labels, dtype=torch.float, device=data.device) + + # Computa ativações SOM para cada amostra (sem gradiente) + with torch.no_grad(): + som_activations = torch.stack( + [self._som_activation(data[i]) for i in range(len(data))] + ) # (N, P_som) + + # Snapshot dos pesos atuais do SOM (para simular ajustes) + som_weights_flat = self.som.weights.detach().clone().flatten() # (P,) + + self.hypothesis_ensemble.train() + losses = [] + # V6.5-V2-metrics-FIX: pré-computa data_norm_sq uma única vez (N, 1) + # para reuso em todos os steps — evita recomputação redundante. + data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) + labels_expanded = labels # (N,) + + for step in range(self.hyp_train_steps): + self.hyp_optimizer.zero_grad() + + # Ativação média como representação do estado do SOM + x_mean = som_activations.mean(dim=0, keepdim=True) # (1, P_som) + + # Gera deltas: (1, n_hypotheses, P) + deltas_stack = self.hypothesis_ensemble.forward_stacked(x_mean) + deltas_stack = self.delta_scale * deltas_stack # escala + + # V6.5-V2-metrics-FIX: Avaliação VETORIZADA de todas as hipóteses + # em paralelo (substitui loop que materializava 16 cópias do SOM). + # Memória: (n_hyp, P, N) em vez de 16 * (P + 2*P*N). + if self.classifier is not None and self.classifier_trained: + # deltas_stack: (1, n_hyp, P) → (n_hyp, P, 4) reshape + n_hyp = self.n_hypotheses + P_neurons = self.som_neuron_count + # som_weights_flat: (P, 4) detached + W_base = som_weights_flat.reshape(P_neurons, 4) # (P, 4) + # delta_h: (n_hyp, P) → reshape para (n_hyp, P, 4) + deltas_3d = deltas_stack[0].reshape(n_hyp, P_neurons, 4) # (n_hyp, P, 4) + # W_new[h, p, d] = W_base[p, d] + deltas_3d[h, p, d] + W_new = W_base.unsqueeze(0) + deltas_3d # (n_hyp, P, 4) — broadcast + # dist²[h, p, n] = ||W_new[h, p] - data[n]||² + # = ||W_new[h, p]||² + ||data[n]||² - 2*W_new[h, p]·data[n] + W_new_norm_sq = (W_new * W_new).sum(dim=-1) # (n_hyp, P) + # cross[h, p, n] = W_new[h, p] · data[n] + cross = torch.matmul(W_new, data.t()) # (n_hyp, P, N) + dist_sq = ( + W_new_norm_sq.unsqueeze(-1) # (n_hyp, P, 1) + + data_norm_sq # (1, N) → broadcast (n_hyp, P, N) + - 2.0 * cross + ) # (n_hyp, P, N) + dist_sq = torch.clamp(dist_sq, min=0.0) + # activations[h, n, p] = dist_sq[h, p, n] + activations = dist_sq.transpose(1, 2) # (n_hyp, N, P) + # Classifier forward (congelado) + classifier_params_were_grad = [ + p.requires_grad for p in self.classifier.parameters() + ] + for p in self.classifier.parameters(): + p.requires_grad_(False) + try: + logits = self.classifier( + activations.reshape(n_hyp * len(data), P_neurons) + ).reshape(n_hyp, len(data)) # (n_hyp, N) + # BCE por hipótese, depois média + loss_per_hyp = F.binary_cross_entropy_with_logits( + logits, + labels_expanded.unsqueeze(0).expand(n_hyp, -1), + reduction='none', + ).mean(dim=1) # (n_hyp,) + loss_total = loss_per_hyp.mean() + # Regularização L2 sobre os deltas + reg_loss = 0.01 * deltas_stack.norm() + loss_total = loss_total + reg_loss + finally: + for p, was_grad in zip(self.classifier.parameters(), + classifier_params_were_grad): + p.requires_grad_(was_grad) + else: + # Fallback: minimizar norma do delta (regularização pura) + loss_total = deltas_stack.norm() + + loss_total.backward() + self.hyp_optimizer.step() + losses.append(float(loss_total.item())) + + # V6.5-V2-metrics-FIX: libera tensores intermediários explicitamente + # para reduzir pico de memória entre steps. + del loss_total, deltas_stack + if 'W_new' in dir(): + del W_new, cross, dist_sq, activations + + # Atualiza a escala de delta (decai suavemente) + with torch.no_grad(): + self.delta_scale.data = torch.clamp( + self.delta_scale.data * 0.99, 0.001, 0.1 + ) + + self.hypothesis_ensemble.eval() + # V6.5-V2-metrics-FIX: libera som_activations e som_weights_flat + del som_activations, som_weights_flat, data, data_norm_sq + import gc + gc.collect() + + # V6.5-V2-dynamic — registra loss_final no histórico + loss_final_val = float(losses[-1]) if losses else 0.0 + self._record_loss(loss_final_val) + + # V6.5-V2-metrics-FIX-2 — atualiza contadores cumulativos + # User requirement: "quantos passos de treino de hipóteses usado" + self._total_hyp_steps_executed += int(self.hyp_train_steps) + self._n_train_hyp_calls += 1 + + result = { + "active": True, + "n_steps": self.hyp_train_steps, + "n_hypotheses": self.n_hypotheses, + "loss_initial": float(losses[0]) if losses else 0.0, + "loss_final": loss_final_val, + "loss_mean": float(sum(losses) / len(losses)) if losses else 0.0, + "delta_scale_final": float(self.delta_scale.item()), + "elapsed_ms": float((_time.time() - t0) * 1000), + # V6.5-V2-metrics-FIX-2 — contadores cumulativos para monitoramento + "total_hyp_steps_executed": int(self._total_hyp_steps_executed), + "n_train_hyp_calls": int(self._n_train_hyp_calls), + } + + # V6.5-V2-dynamic — dispara adaptação automática de hiperparâmetros + try: + adapt_record = self._adapt_hyperparameters(trigger="auto") + result["adaptation"] = adapt_record + except Exception as adapt_err: + result["adaptation_error"] = str(adapt_err)[:200] + + return result + + def _compute_classification_loss_with_som( + self, som_weights: torch.Tensor, data: torch.Tensor, labels: torch.Tensor + ) -> torch.Tensor: + """V6.5-V2-metrics-FIX — Computa perda de classificação com SOM temporário. + + BUG CORRIGIDO (V6.5-V2-metrics-FIX): + A versão anterior envolvia toda a computação em `torch.no_grad()`, + o que matava o gradiente que deveria fluir dos deltas → SOM → loss. + Resultado: train_hypotheses() não treinava os geradores de fato. + + CORREÇÃO: A computação das ativações SOM (squared distances) é + feita SEM no_grad, permitindo que o gradiente flua dos deltas + (que pertencem ao HypothesisEnsemble) através do SOM temporário + até a loss. Apenas o classifier é congelado (requires_grad=False) + para que seu gradiente não seja atualizado por train_hypotheses. + + Mathematical formulation (squared Euclidean distance, consistente com + _som_activation que usa ||W - x||^2, não ||W - x||): + Para cada neurônio p e amostra n: + dist²[p, n] = ||W[p] - x[n]||² + = ||W[p]||² + ||x[n]||² - 2·W[p]·x[n] + activations = dist².t() → (N, P_neurons) + logits = classifier(activations) → (N,) + loss = BCEWithLogitsLoss(logits, labels) + + Args: + som_weights: tensor (I, J, K, L, 4) com gradiente (dos deltas). + data: tensor (N, 4) — amostras 4D (detached). + labels: tensor (N,) — rótulos binários. + + Returns: + Scalar loss tensor (com gradiente fluindo para som_weights). + """ + if self.classifier is not None and self.classifier_trained: + # NÃO usa torch.no_grad() — gradiente deve fluir dos deltas + # através do SOM temporário até a loss. + W_flat = som_weights.reshape(-1, 4) # (P_neurons, 4) + # Identidade algébrica: ||W - x||^2 = ||W||^2 + ||x||^2 - 2*W·x + # Evita materializar tensor (P, N, 4) — usa (P, N) direto. + W_norm_sq = (W_flat * W_flat).sum(dim=-1, keepdim=True) # (P, 1) + data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) + cross = W_flat @ data.t() # (P, N) — matmul eficiente + dist_sq = W_norm_sq + data_norm_sq - 2.0 * cross # (P, N) + dist_sq = torch.clamp(dist_sq, min=0.0) # estabilidade numérica + activations = dist_sq.t() # (N, P_neurons) — gradiente fluindo! + # Congela classifier durante train_hypotheses (não queremos + # atualizar seus pesos — apenas os deltas dos geradores) + classifier_params_were_grad = [ + p.requires_grad for p in self.classifier.parameters() + ] + for p in self.classifier.parameters(): + p.requires_grad_(False) + try: + logits = self.classifier(activations) # (N,) + loss = F.binary_cross_entropy_with_logits(logits, labels) + finally: + # Restaura requires_grad do classifier + for p, was_grad in zip(self.classifier.parameters(), + classifier_params_were_grad): + p.requires_grad_(was_grad) + return loss + else: + # Fallback: votação BMU (não-diferenciável) + with torch.no_grad(): + if not hasattr(self, "neuron_label"): + self._label_neurons() + correct = 0 + for i in range(len(data)): + bmu = self.som.find_bmu(data[i]) + pred = self.neuron_label.get(bmu, 0.5) + if (pred > 0.5) == (labels[i].item() > 0.5): + correct += 1 + return torch.tensor( + 1.0 - correct / max(1, len(data)), + device=data.device, + dtype=torch.float, + ) + + # ================================================================== + # 2.2 Seleção do melhor delta (3 tentativas × 16 hipóteses) + # ================================================================== + def select_best_delta( + self, + buffer_4d: Optional[List[torch.Tensor]] = None, + buffer_labels: Optional[List[int]] = None, + n_trials: Optional[int] = None, + ) -> Tuple[Optional[torch.Tensor], float, Dict[str, Any]]: + """Avalia 3 tentativas × 16 hipóteses e retorna o melhor delta e sua acurácia. + + Args: + buffer_4d, buffer_labels: opcional — usa self.buffer_* se None. + n_trials: opcional — usa self.n_trials se None. + + Returns: + Tuple (best_delta, best_acc, info_dict): + - best_delta: tensor (P,) ou None se buffer vazio + - best_acc: acurácia obtida com o melhor delta + - info_dict: métricas da seleção (n_trials, n_hypotheses, etc.) + """ + if buffer_4d is None: + buffer_4d = self.buffer_4d + if buffer_labels is None: + buffer_labels = self.buffer_labels + if not buffer_4d: + return None, -1.0, {"active": False, "reason": "empty_buffer"} + + if n_trials is None: + n_trials = self.n_trials + + import time as _time + t0 = _time.time() + + data = torch.stack(buffer_4d).detach() + labels = torch.tensor(buffer_labels, dtype=torch.float, device=data.device) + + # Computa ativação SOM média + with torch.no_grad(): + som_activations = torch.stack( + [self._som_activation(data[i]) for i in range(len(data))] + ) + x_state = som_activations.mean(dim=0, keepdim=True) + + best_delta = None + best_acc = -1.0 + trial_results = [] + + som_flat = self.som.weights.detach().clone().flatten() + + self.hypothesis_ensemble.eval() + for trial in range(n_trials): + # Adiciona ruído à entrada para diversificar tentativas (trial > 0) + if trial > 0: + x_noisy = x_state + 0.01 * torch.randn_like(x_state) + else: + x_noisy = x_state + + with torch.no_grad(): + deltas_stack = self.hypothesis_ensemble.forward_stacked(x_noisy) + deltas_stack = self.delta_scale * deltas_stack + + # Avalia cada uma das 16 hipóteses + trial_accs = [] + for h in range(self.n_hypotheses): + delta_h = deltas_stack[0, h] + som_temp = som_flat + delta_h + som_temp_reshaped = som_temp.view_as(self.som.weights) + acc = self._evaluate_accuracy_with_som( + som_temp_reshaped, data, labels + ) + trial_accs.append(acc) + if acc > best_acc: + best_acc = acc + best_delta = delta_h.clone() + + trial_results.append({ + "trial": trial + 1, + "best_acc_in_trial": float(max(trial_accs)) if trial_accs else 0.0, + "mean_acc_in_trial": float(sum(trial_accs) / len(trial_accs)) if trial_accs else 0.0, + "best_hyp_idx": int(trial_accs.index(max(trial_accs))) if trial_accs else -1, + }) + + info = { + "active": True, + "n_trials": n_trials, + "n_hypotheses": self.n_hypotheses, + "best_acc": float(best_acc), + "delta_norm": float(best_delta.norm().item()) if best_delta is not None else 0.0, + "delta_scale": float(self.delta_scale.item()), + "trials": trial_results, + "elapsed_ms": float((_time.time() - t0) * 1000), + } + return best_delta, best_acc, info + + def _evaluate_accuracy_with_som( + self, som_weights: torch.Tensor, data: torch.Tensor, labels: torch.Tensor + ) -> float: + """V6.5-V2-metrics-FIX — Calcula acurácia com SOM temporário (vetorizado). + + Substitui o loop que chamava _som_activation(data[i]) para cada amostra + por uma única operação matmul. Reduz tempo de O(N*P) para O(P*N) mas + com paralelismo SIMD (single matmul vs N python iterations). + + Para o fallback (votação BMU), ainda precisa trocar self.som.weights + temporariamente pois find_bmu lê self.som.weights diretamente. + """ + if self.classifier is not None and self.classifier_trained: + with torch.no_grad(): + # Vetorizado: dist²[p, n] = ||W[p]||² + ||x[n]||² - 2*W[p]·x[n] + W_flat = som_weights.reshape(-1, 4) # (P, 4) + W_norm_sq = (W_flat * W_flat).sum(dim=-1, keepdim=True) # (P, 1) + data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) + cross = W_flat @ data.t() # (P, N) + dist_sq = W_norm_sq + data_norm_sq - 2.0 * cross # (P, N) + dist_sq = torch.clamp(dist_sq, min=0.0) + activations = dist_sq.t() # (N, P) + logits = self.classifier(activations) # (N,) + preds = (logits > 0.5).float() + acc = (preds == labels).float().mean().item() + return float(acc) + else: + # Fallback: votação BMU — precisa trocar self.som.weights + original_weights = self.som.weights + self.som.weights = som_weights + try: + with torch.no_grad(): + if not hasattr(self, "neuron_label"): + self._label_neurons() + correct = 0 + for i in range(len(data)): + bmu = self.som.find_bmu(data[i]) + pred = self.neuron_label.get(bmu, 0.5) + if (pred > 0.5) == (labels[i].item() > 0.5): + correct += 1 + acc = correct / max(1, len(data)) + return float(acc) + finally: + self.som.weights = original_weights + + # ================================================================== + # 2.3 Aplicação do melhor delta + consolidação via EWC + # ================================================================== + def apply_best_delta_and_consolidate( + self, + buffer_4d: Optional[List[torch.Tensor]] = None, + buffer_labels: Optional[List[int]] = None, + ) -> Dict[str, Any]: + """Seleciona o melhor delta, aplica ao SOM e executa EWC. + + Pipeline: + 1. select_best_delta: avalia 3 tentativas × 16 hipóteses + 2. Aplica δ* ao SOM: SOM ← SOM + δ* + 3. Re-avalia acurácia após aplicação + 4. Consolida via EWC: set_ewc_reference() (Fisher em w) + + Returns: + Dict com métricas (acc_before, acc_after, delta_norm, etc.) + """ + import time as _time + t0 = _time.time() + + # Avalia acurácia ANTES da aplicação + acc_before = self.evaluate_classification() + + # Seleciona melhor delta + best_delta, best_acc, sel_info = self.select_best_delta( + buffer_4d, buffer_labels + ) + + if best_delta is None: + return { + "active": False, + "reason": "no_valid_delta", + "elapsed_ms": float((_time.time() - t0) * 1000), + } + + # V6.5-V2-metrics-FIX-2 — Verifica se best_delta contém NaN/Inf ANTES + # de aplicar ao SOM. User requirement: "investigar QE e KL resultando + # em NAN". Se o delta tem NaN (devido a gradientes explosivos em + # train_hypotheses), aplicar ao SOM contaminaria os pesos e propagaria + # NaN para todas as métricas subsequentes (QE, KL, etc.). + if torch.isnan(best_delta).any() or torch.isinf(best_delta).any(): + n_nan = int(torch.isnan(best_delta).sum().item()) + n_inf = int(torch.isinf(best_delta).sum().item()) + logger = logging.getLogger(__name__) if 'logging' in dir() else None + import logging as _logging + _logging.getLogger(__name__).warning( + f"[V6.5-V2-metrics-FIX-2] best_delta contains NaN/Inf " + f"(nan={n_nan}, inf={n_inf}). SKIPPING application to SOM " + f"to prevent weight contamination. Investigate train_hypotheses " + f"for gradient explosion (consider lowering hyp_lr or adding " + f"gradient clipping)." + ) + return { + "active": False, + "reason": f"best_delta_has_nan (nan={n_nan}, inf={n_inf})", + "delta_norm": float(best_delta.norm().item()), + "elapsed_ms": float((_time.time() - t0) * 1000), + "nan_detected": True, + } + + # Aplica o delta ao SOM + with torch.no_grad(): + som_flat = self.som.weights.flatten() + som_flat_new = som_flat + best_delta + # V6.5-V2-metrics-FIX-2 — verificação pós-aplicação: se os pesos + # do SOM ficaram com NaN (deveria não acontecer após o check acima, + # mas defense in depth), reverte para o estado anterior. + if torch.isnan(som_flat_new).any() or torch.isinf(som_flat_new).any(): + import logging as _logging + _logging.getLogger(__name__).error( + f"[V6.5-V2-metrics-FIX-2] SOM weights became NaN/Inf after " + f"delta application. REVERTING to previous weights." + ) + else: + self.som.weights = som_flat_new.view_as(self.som.weights) + + # Re-avalia acurácia APÓS aplicação + acc_after = self.evaluate_classification() + + # Re-rotula neurônios (BMU labels podem ter mudado) + self._label_neurons() + + # Consolida via EWC (Fisher information em w) + try: + self.som.set_ewc_reference() + ewc_set = True + except Exception as e: + ewc_set = False + sel_info["ewc_error"] = str(e)[:200] + + return { + "active": True, + "acc_before": float(acc_before), + "acc_after": float(acc_after), + "best_acc_during_selection": float(best_acc), + "delta_norm": float(best_delta.norm().item()), + "delta_scale": float(self.delta_scale.item()), + "ewc_reference_set": bool(ewc_set), + "selection_info": sel_info, + "elapsed_ms": float((_time.time() - t0) * 1000), + } + + # ================================================================== + # 2.4 Sobrescrita do process_batch para o novo protocolo V2 + # ================================================================== + def process_batch_v2( + self, + sentences: List[str], + labels: List[int], + label_strings: Optional[Dict[int, str]] = None, + dataset_name: Optional[str] = None, + enable_punishment: bool = True, + ) -> Dict[str, Any]: + """Versão V2 do process_batch com 3 tentativas e 16 hipóteses. + + Protocolo: + - CONHECIMENTO (enable_punishment=False): apenas adiciona dados e + treina SOM (sem aplicar punição). + - TREINAMENTO COM PUNIÇÃO (enable_punishment=True): + * acc < 1.0 → punishment_count += 1 + * punishment_count == 1: treinar 16 hipóteses + activate_hypothesis + * punishment_count == 2: aplicar melhor delta (3×16) + EWC + reset + + Args: + sentences, labels, label_strings, dataset_name: mesmos de process_batch. + enable_punishment: se False, apenas acumula conhecimento (fase 1). + + Returns: + Dict com métricas do batch processado. + """ + import time as _time + t0 = _time.time() + result: Dict[str, Any] = { + "phase": "CONHECIMENTO" if not enable_punishment else "PUNICAO", + "dataset": dataset_name, + "n_samples": len(sentences), + "action": "none", + "cycle_completed": False, + } + + # 1. Adiciona dados ao buffer + self.add_data( + sentences, labels, + label_strings=label_strings, + dataset_name=dataset_name, + ) + + # 2. Verifica início de treino + training_started = self.check_training_start() + if training_started or self.training_ready: + # 3. Treina SOM no buffer (regra de Kohonen) + self.train_som_on_buffer() + self._label_neurons() + result["som_trained"] = True + + if not self.training_ready: + result["elapsed_ms"] = float((_time.time() - t0) * 1000) + return result + + # 4. Fase de CONHECIMENTO: para aqui, sem aplicar punição + if not enable_punishment: + acc = self.evaluate_classification() + result["accuracy"] = float(acc) + result["elapsed_ms"] = float((_time.time() - t0) * 1000) + return result + + # 5. Fase de PUNIÇÃO: avalia e aplica protocolo V2 + acc = self.evaluate_classification() + result["accuracy"] = float(acc) + + if acc < 1.0: + self.punishment_count += 1 + self.success_count = 0 + + # V6.5-V2-dynamic — registra evento de punição + self._record_punishment_event(True) + + if self.punishment_count == 1: + # 1ª punição: treinar hipóteses + ativar classificador + hyp_info = self.train_hypotheses() + if self.classifier is None: + self.activate_hypothesis() + else: + # Re-treina classifier com buffer atual + self.activate_hypothesis() + result["action"] = "train_hypotheses_and_activate" + result["hypotheses_training"] = hyp_info + + elif self.punishment_count >= 2: + # 2ª punição: selecionar melhor delta (n_trials × n_hyp) + aplicar + EWC + apply_info = self.apply_best_delta_and_consolidate() + result["action"] = "apply_best_delta_and_consolidate" + result["delta_application"] = apply_info + + # V6.5-V2-dynamic — adapta hiperparâmetros após punição recorrente + try: + adapt_record = self._adapt_hyperparameters(trigger="punishment") + result["adaptation"] = adapt_record + except Exception as adapt_err: + result["adaptation_error"] = str(adapt_err)[:200] + + # Reset do ciclo (mirror training) + self.training_ready = False + self.punishment_count = 0 + self.success_count = 0 + self.histogram.clear() + # Mantém um sliding window do buffer para preservar contexto + if len(self.buffer_4d) > 256: + self.buffer_4d = self.buffer_4d[-256:] + self.buffer_labels = self.buffer_labels[-256:] + result["cycle_completed"] = True + else: + # acc == 1.0: sucesso + self.punishment_count = 0 + self.success_count += 1 + # V6.5-V2-dynamic — registra sucesso (sem punição) + self._record_punishment_event(False) + result["action"] = "success" + + result["elapsed_ms"] = float((_time.time() - t0) * 1000) + return result + + # ================================================================== + # 2.5 API de monitoramento V2 + # ================================================================== + def get_v2_metrics(self) -> Dict[str, Any]: + """Retorna métricas específicas da versão V2 (inclui adaptação dinâmica).""" + return { + "version": "V2-dynamic", + "n_hypotheses": int(self.n_hypotheses), + "n_hypotheses_active": int(self.hypothesis_ensemble.active_count), + "max_n_hypotheses": int(self.max_n_hypotheses), + "n_trials": int(self.n_trials), + "hyp_train_steps": int(self.hyp_train_steps), + "hyp_lr": float(self.hyp_lr), + "delta_scale": float(self.delta_scale.item()), + "n_generators": int(len(self.hypothesis_ensemble.generators)), + "punishment_count": int(self.punishment_count), + "success_count": int(self.success_count), + "training_ready": bool(self.training_ready), + "classifier_trained": bool(self.classifier_trained), + "ewc_reference_set": bool(self.som.old_weights_w is not None), + "buffer_size": int(len(self.buffer_4d)), + # V6.5-V2-metrics-FIX-2 — contadores cumulativos (user requirement: + # "quantos passos de treino de hipóteses usado") + "total_hyp_steps_executed": int(getattr(self, "_total_hyp_steps_executed", 0)), + "n_train_hyp_calls": int(getattr(self, "_n_train_hyp_calls", 0)), + "dynamic_adaptation": { + "loss_history_len": int(len(self._loss_history)), + "loss_stats": self._compute_loss_stats(), + "punishment_rate": float(self._compute_punishment_rate()), + "punishment_window_size": int(len(self._punishment_history)), + "n_adaptations": int(len(self._adaptation_log)), + "limits": { + "min_n_hypotheses": int(self.min_n_hypotheses), + "max_n_hypotheses": int(self.max_n_hypotheses), + "min_n_trials": int(self.min_n_trials), + "max_n_trials": int(self.max_n_trials), + "min_hyp_train_steps": int(self.min_hyp_train_steps), + "max_hyp_train_steps": int(self.max_hyp_train_steps), + }, + "last_5_adaptations": self._adaptation_log[-5:] if self._adaptation_log else [], + }, + } + + # ================================================================== + # V6.5-V2-metrics-FIX-2 — Detecção de atividade de hipóteses e neurônios + # ================================================================== + # User requirement (latest): "Aprimorar a FASE2 PUNITIVA investigar QE e KL + # resultando em NAN, acrescentar detecção da atividade das layers (quantas + # ativadas) de hipóteses e das ativações dos neurônios, quantos passos de + # treino de hipóteses usado". + # + # Mathematical formulation: + # + # Hipótese ativa: Um gerador g_h (h ∈ {1..n_hypotheses}) é considerado + # "ativo" em um batch se o delta que ele produz tem norma L2 > threshold + # (default 1e-6). Deltas com norma ≈ 0 indicam que o gerador está + # produzindo saída trivial (morto) — não contribui para a diversidade + # do ensemble. + # + # active_h = 1 if ||δ_h||₂ > threshold else 0 + # n_active_hypotheses = Σ_h active_h + # + # Neurônio ativo: Um neurônio (i,j,k,l) é "ativo" se é BMU para pelo + # menos uma amostra do buffer atual. Reuso da lógica de dead_neuron_rate, + # mas exposto como métrica explícita para monitoramento contínuo. + # + # n_active_neurons = |{W_bmu(x_i) : i ∈ buffer}| + # n_dead_neurons = n_total - n_active_neurons + # neuron_activation_rate = n_active_neurons / n_total + # + # Passos de treino de hipóteses: Número total de passos de Adam executados + # em train_hypotheses() desde o início do treino. Cumulativo ao longo de + # todas as chamadas (punições). + # + # total_hyp_train_steps = Σ_call self.hyp_train_steps (na época da call) + # ------------------------------------------------------------------ + def detect_hypothesis_activity( + self, + threshold: float = 1e-6, + ) -> Dict[str, Any]: + """V6.5-V2-metrics-FIX-2 — Detecta quantas hipóteses estão ativas. + + User requirement: "acrescentar detecção da atividade das layers + (quantas ativadas) de hipóteses". + + Executa um forward pass do HypothesisEnsemble sobre a ativação média + do SOM no buffer atual e conta quantos dos active_count geradores + produzem deltas com norma > threshold. + + Args: + threshold: norma L2 mínima para considerar um delta "ativo". + + Returns: + Dict com: + - n_hypotheses_active: int (quantas hipóteses produziram delta ≠ 0) + - n_hypotheses_total: int (active_count atual do ensemble) + - max_n_hypotheses: int (limite máximo pré-alocado) + - delta_norms: list[float] (norma de cada delta, para inspeção) + - active_mask: list[bool] (True para hipóteses ativas) + - activation_rate: float (n_active / n_total) + - mean_delta_norm: float (média das normas) + - max_delta_norm: float (maior norma) + - threshold_used: float + """ + if not self.buffer_4d: + return { + "n_hypotheses_active": 0, + "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), + "max_n_hypotheses": int(self.max_n_hypotheses), + "delta_norms": [], + "active_mask": [], + "activation_rate": 0.0, + "mean_delta_norm": 0.0, + "max_delta_norm": 0.0, + "threshold_used": float(threshold), + "reason": "empty_buffer", + } + try: + import torch as _torch + with _torch.no_grad(): + # Computa ativação SOM média do buffer + data = _torch.stack(self.buffer_4d).detach() + som_activations = _torch.stack( + [self._som_activation(data[i]) for i in range(len(data))] + ) + x_mean = som_activations.mean(dim=0, keepdim=True) # (1, P_som) + + # Forward do ensemble: retorna (1, n_hyp, P) + deltas_stack = self.hypothesis_ensemble.forward_stacked(x_mean) + deltas_stack = self.delta_scale * deltas_stack # aplica escala + + # Norma L2 de cada delta: (n_hyp,) + delta_norms = deltas_stack[0].norm(dim=-1) # (n_hyp,) + delta_norms_list = [float(x) for x in delta_norms] + active_mask = [bool(x > threshold) for x in delta_norms_list] + n_active = sum(active_mask) + + # Libera tensores + del deltas_stack, som_activations, data, x_mean, delta_norms + + return { + "n_hypotheses_active": int(n_active), + "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), + "max_n_hypotheses": int(self.max_n_hypotheses), + "delta_norms": delta_norms_list, + "active_mask": active_mask, + "activation_rate": float(n_active / max(1, self.hypothesis_ensemble.active_count)), + "mean_delta_norm": float(sum(delta_norms_list) / max(1, len(delta_norms_list))), + "max_delta_norm": float(max(delta_norms_list) if delta_norms_list else 0.0), + "threshold_used": float(threshold), + } + except Exception as e: + return { + "n_hypotheses_active": 0, + "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), + "max_n_hypotheses": int(self.max_n_hypotheses), + "delta_norms": [], + "active_mask": [], + "activation_rate": 0.0, + "mean_delta_norm": 0.0, + "max_delta_norm": 0.0, + "threshold_used": float(threshold), + "error": str(e)[:200], + } + + def count_active_neurons(self) -> Dict[str, Any]: + """V6.5-V2-metrics-FIX-2 — Conta neurônios ativos (BMUs) no buffer. + + User requirement: "acrescentar (...) das ativações dos neurônios". + + Para cada amostra no buffer_4d, encontra o BMU. Conta quantos + neurônios únicos são BMUs (n_active_neurons) e quantos nunca são + BMUs (n_dead_neurons = n_total - n_active). + + Returns: + Dict com: + - n_active_neurons: int (neurônios que são BMU para ≥1 amostra) + - n_dead_neurons: int (neurônios que nunca são BMU) + - n_total_neurons: int (total de neurônios no SOM) + - neuron_activation_rate: float (n_active / n_total) + - bmu_distribution: dict {flat_idx: count} (top-20 mais frequentes) + """ + if not self.buffer_4d: + return { + "n_active_neurons": 0, + "n_dead_neurons": int(self.som_neuron_count), + "n_total_neurons": int(self.som_neuron_count), + "neuron_activation_rate": 0.0, + "bmu_distribution": {}, + "reason": "empty_buffer", + } + try: + import torch as _torch + with _torch.no_grad(): + data = _torch.stack(self.buffer_4d).detach() + flat_w = self.som.weights.view(-1, 4).detach() + # (N, M) distâncias + diff = data.unsqueeze(1) - flat_w.unsqueeze(0) + dist_sq = _torch.sum(diff * diff, dim=-1) + _, bmu_flat = _torch.min(dist_sq, dim=1) # (N,) + bmu_list = bmu_flat.tolist() + + unique_bmus = set(bmu_list) + n_active = len(unique_bmus) + n_total = self.som_neuron_count + + # Distribuição dos BMUs (top-20) + from collections import Counter + bmu_counts = Counter(bmu_list) + bmu_distribution = dict(bmu_counts.most_common(20)) + + del data, flat_w, diff, dist_sq, bmu_flat + return { + "n_active_neurons": int(n_active), + "n_dead_neurons": int(n_total - n_active), + "n_total_neurons": int(n_total), + "neuron_activation_rate": float(n_active / max(1, n_total)), + "bmu_distribution": {str(k): int(v) for k, v in bmu_distribution.items()}, + } + except Exception as e: + return { + "n_active_neurons": 0, + "n_dead_neurons": int(self.som_neuron_count), + "n_total_neurons": int(self.som_neuron_count), + "neuron_activation_rate": 0.0, + "bmu_distribution": {}, + "error": str(e)[:200], + } + + def get_hyp_train_steps_used(self) -> Dict[str, Any]: + """V6.5-V2-metrics-FIX-2 — Retorna quantos passos de treino de hipóteses + foram usados até agora. + + User requirement: "quantos passos de treino de hipóteses usado". + + Returns: + Dict com: + - current_hyp_train_steps: int (valor ATUAL de self.hyp_train_steps) + - total_hyp_train_steps_executed: int (cumulativo ao longo do treino) + - n_train_hypotheses_calls: int (número de chamadas a train_hypotheses) + - min_hyp_train_steps: int (limite inferior dinâmico) + - max_hyp_train_steps: int (limite superior dinâmico) + """ + return { + "current_hyp_train_steps": int(self.hyp_train_steps), + "total_hyp_train_steps_executed": int(getattr(self, "_total_hyp_steps_executed", 0)), + "n_train_hypotheses_calls": int(getattr(self, "_n_train_hyp_calls", 0)), + "min_hyp_train_steps": int(self.min_hyp_train_steps), + "max_hyp_train_steps": int(self.max_hyp_train_steps), + } + + # ================================================================== + # V6.5-V2-metrics — Integração das métricas SOM canônicas + # (QE, TE, Kaski-Lagus, Variância Explicada, Dead Neurons, + # Colapso Topológico, Estagnação QE, Cruzamento Vizinhança) + # ================================================================== + def compute_som_metrics(self) -> Dict[str, Any]: + """V6.5-V2-metrics — Computa todas as métricas SOM canônicas. + + User requirement: "ANALISAR matematicamente e logicamente e inserir + melhorias para os scripts das métricas de aprendizado e indicadores + de falha em Mapas Auto-Organizáveis (SOM / Redes de Kohonen) + avaliam a fidelidade de representação dos dados e a preservação da + vizinhança topológica". + + Returns: + Dict com 8 métricas: + 1. quantization_error (QE) + 2. topological_error (TE) + 3. kaski_lagus_error (KL) + 4. explained_variance_share (VE) + 5. topological_collapse (dict) + 6. dead_neuron_rate (dict) + 7. qe_stagnation (dict) — requer histórico + 8. neighborhood_crossing (dict) — requer histórico + Plus: overall_health, failure_indicators + """ + from bigru_t.model.som_metrics import compute_all_metrics + if not self.buffer_4d: + return { + "quantization_error": 0.0, + "topological_error": 0.0, + "kaski_lagus_error": 0.0, + "explained_variance_share": 0.0, + "topological_collapse": {"severity": "none", "effective_rank": 0.0}, + "dead_neuron_rate": {"dead_neuron_rate": 1.0, "n_dead": self.som_neuron_count, + "n_active": 0, "n_total": self.som_neuron_count}, + "qe_stagnation": {"is_stagnant": False, "severity": "none"}, + "neighborhood_crossing": {"detected": False, "severity": "none"}, + "overall_health": "unknown", + "failure_indicators": ["no_buffer_data"], + "n_failure_indicators": 1, + "note": "buffer_4d vazio — métricas indisponíveis", + } + # Stack buffer em tensor (N, 4) + data = torch.stack([ + v.detach().clone() if isinstance(v, torch.Tensor) else torch.tensor(v, dtype=torch.float) + for v in self.buffer_4d + ]).float() + weights = self.som.weights.detach() + metrics = compute_all_metrics(data, weights, self._som_metric_history) + # Registra no histórico + self._som_metric_history.record(metrics) + return metrics + + def get_som_metric_history(self) -> Dict[str, Any]: + """Retorna histórico de métricas SOM (para análise temporal).""" + return self._som_metric_history.to_dict() + + def reset_som_metric_history(self) -> None: + """Reseta o histórico de métricas SOM (início de nova fase).""" + from bigru_t.model.som_metrics import SOMMetricHistory + self._som_metric_history = SOMMetricHistory() + + +__all__ = [ + "SimpleBBPETokenizer", + "positional_encoding", + "text_to_4d_vector", + "KohonenSOM4D", + "HypothesisClassifier", + "KohonenLearningSystem", + "DeltaGenerator", + "HypothesisEnsemble", + "KohonenLearningSystemV2", +] +# V6.5-V2-metrics: som_metrics é importado lazy (dentro de compute_som_metrics) +# para evitar dependência circular e permitir uso standalone.