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,2667 @@ +"""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 = [] + 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.""" + pred = self.predict(query) + bmu_info = "" + if self.buffer_4d: + try: + 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 + + 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 + ) + self.buffer_4d.append(vec) + self.buffer_labels.append(lab) + dim_val = round(vec[self.dim_index].item(), 2) + self.histogram[dim_val] += 1 + + 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).""" + if not self.buffer_4d: + return + data = torch.stack(self.buffer_4d) + 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. + 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.""" + self.neuron_label = {} + if not self.buffer_4d: + return + data = torch.stack(self.buffer_4d) + labels = torch.tensor(self.buffer_labels) + votes = defaultdict(lambda: [0, 0]) + 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.""" + if not self.buffer_4d: + return 1.0 + data = torch.stack(self.buffer_4d) + labels = torch.tensor(self.buffer_labels).float() + correct = 0 + 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] + + 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 + 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 + 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, + ) + + # 3. Add to buffer + update histogram (mirror add_data) + self.buffer_4d.append(vec) + self.buffer_labels.append(int(correct_label_int)) + 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 + + 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) + 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) + + # ================================================================== + # 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 = [] + 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 + + # Avalia cada delta: computa loss de classificação + if self.classifier is not None and self.classifier_trained: + loss_per_hyp = [] + for h in range(self.n_hypotheses): + delta_h = deltas_stack[0, h] # (P,) + som_temp = som_weights_flat + delta_h + som_temp_reshaped = som_temp.view_as(self.som.weights) + loss_h = self._compute_classification_loss_with_som( + som_temp_reshaped, data, labels + ) + loss_per_hyp.append(loss_h) + loss_total = torch.stack(loss_per_hyp).mean() + # Regularização L2 sobre os deltas (evita deltas grandes) + reg_loss = 0.01 * deltas_stack.norm() + loss_total = loss_total + reg_loss + 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())) + + # 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-dynamic — registra loss_final no histórico + loss_final_val = float(losses[-1]) if losses else 0.0 + self._record_loss(loss_final_val) + + 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-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: + """Computa a perda de classificação usando um SOM temporário. + + Substitui temporariamente self.som.weights pelo som_weights fornecido, + computa a loss, e restaura o SOM original. Usa detach no SOM temporário + para evitar retropropagação através dos pesos do SOM (apenas os deltas + dos geradores recebem gradiente). + """ + # Guarda o SOM original + original_weights = self.som.weights + # Substitui temporariamente (sem clone — som_weights já é um tensor + # que carrega gradiente dos deltas) + self.som.weights = som_weights + + try: + if self.classifier is not None and self.classifier_trained: + # Computa ativações SOM (sem treinar SOM/embedding) + with torch.no_grad(): + # Para permitir gradiente fluir pelos deltas, precisamos + # das ativações SOM como função diferenciável de som_weights + # Reimplementamos _som_activation com gradiente + data_expanded = data.unsqueeze(0).unsqueeze(0).unsqueeze(0).unsqueeze(0) + # (1, 1, 1, 1, N, 4) — para broadcast com (I, J, K, L, 4) + # Mas self.som.weights é (I, J, K, L, 4) e data é (N, 4) + # dist[i,j,k,l,n] = ||W[i,j,k,l] - data[n]||² + # reshape: W (I*J*K*L, 4), data (N, 4) → dist (I*J*K*L, N) + W_flat = som_weights.reshape(-1, 4) # (P_neurons, 4) + # W_flat.requires_grad_(True) # já tem gradiente dos deltas + dist = torch.cdist(W_flat, data) # (P_neurons, N) + activations = dist.t() # (N, P_neurons) + # Passa pelo classifier + logits = self.classifier(activations) # (N,) + loss = F.binary_cross_entropy_with_logits(logits, labels) + return loss + else: + # Fallback: votação BMU (não-diferenciável — usa heurística) + with torch.no_grad(): + correct = 0 + for i in range(len(data)): + bmu = self.som.find_bmu(data[i]) + pred = self.neuron_label.get(bmu, 0.5) if hasattr(self, "neuron_label") else 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, + ) + finally: + # Restaura o SOM original + self.som.weights = original_weights + + # ================================================================== + # 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: + """Calcula acurácia com um SOM temporário (sem modificar o original).""" + original_weights = self.som.weights + self.som.weights = som_weights + try: + if self.classifier is not None and self.classifier_trained: + with torch.no_grad(): + activations = torch.stack( + [self._som_activation(data[i]) for i in range(len(data))] + ) + logits = self.classifier(activations) + preds = (logits > 0.5).float() + acc = (preds == labels).float().mean().item() + else: + # Votação BMU + 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), + } + + # Aplica o delta ao SOM + with torch.no_grad(): + som_flat = self.som.weights.flatten() + som_flat_new = som_flat + best_delta + 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)), + "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 [], + }, + } + + +__all__ = [ + "SimpleBBPETokenizer", + "positional_encoding", + "text_to_4d_vector", + "KohonenSOM4D", + "HypothesisClassifier", + "KohonenLearningSystem", + "DeltaGenerator", + "HypothesisEnsemble", + "KohonenLearningSystemV2", +]