"""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 — DELEGADO ao módulo canônico src/bigru_t/tokenizer/bbpe_tokenizer.py # ============================================================================ # User requirement: "na subpasta 'src/bigru_t/tokenizer' está o tokenizer # bbpe_tokenizer.py que deve ser usado portanto REMOVER tokenizadores # implementados em outros módulos". # # V7-tokenizer-cleanup: a classe SimpleBBPETokenizer local foi REMOVIDA # COMPLETAMENTE. Não há mais fallback word-level duplicado nesta classe — # o BBPETokenizer canônico JÁ possui seu próprio fallback word-level interno # (`_wordlevel_fallback`) e OOM-safe serial training (`_train_serial_inprocess`). # Manter um segundo fallback aqui era duplicação de código e confundia a # auditoria (qual tokenizer está realmente ativo?). # # SimpleBBPETokenizer agora é APENAS um alias direto para BBPETokenizer. # Imports existentes (ex: `from bigru_t.model.kohonen_learning_system import # SimpleBBPETokenizer`) continuam funcionando, mas não há mais nenhuma # implementação de tokenizer neste arquivo. try: from bigru_t.tokenizer.bbpe_tokenizer import BBPETokenizer as SimpleBBPETokenizer _HAS_CANONICAL_TOKENIZER = True except Exception as _tok_err: _HAS_CANONICAL_TOKENIZER = False # Fallback final: define um stub mínimo para que o import não quebre # (apenas para testes isolados sem o pacote bigru_t instalado). # Em produção, o pacote bigru_t está sempre instalado. class SimpleBBPETokenizer: # type: ignore[no-redef] """Stub fallback — NÃO é uma implementação de tokenizer. Apenas evita ImportError quando bigru_t.tokenizer não está disponível. Em produção, use BBPETokenizer de src/bigru_t/tokenizer/bbpe_tokenizer.py. """ def __init__(self, vocab_size: int = 16384): self.vocab_size = vocab_size raise RuntimeError( "BBPETokenizer canônico não disponível. Instale o pacote bigru_t " "ou adicione src/ ao PYTHONPATH." ) # ============================================================================ # 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) # V6.5-V2-metrics-FIX-3 — SVD com fallback robusto: se a matriz for # rank-deficiente (ex: todos os tokens são → todas as linhas # idênticas), torch.linalg.svd pode produzir NaN ou levantar RuntimeError. # Fallback 1: driver 'gesvd' (mais estável para matrizes singulares). # Fallback 2: zeros (vetor 4D neutro, não contamina o SOM). try: U, S, V = torch.linalg.svd(mean_centered, full_matrices=False) except Exception: try: U, S, V = torch.linalg.svd(mean_centered, full_matrices=False, driver='gesvd') except Exception: S = torch.zeros(min(mean_centered.shape), device=mean_centered.device) V = torch.eye(mean_centered.shape[1], device=mean_centered.device) # V6.5-V2-metrics-FIX-3 — sanitiza NaN do SVD (rank-deficient → S pode ter NaN) if torch.isnan(S).any() or torch.isinf(S).any(): S = torch.nan_to_num(S, nan=0.0, posinf=0.0, neginf=0.0) V = torch.nan_to_num(V, nan=0.0, posinf=0.0, neginf=0.0) 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 # V6.5-V2-auto-conscience-v2 — Normalização SEPARADA de xyz e w. # ANTES: vetores 4D tinham magnitude ~1e-4 (xyz) e ~1e-4 (w). Após L2 norm, # w dominava (≈1.0) e xyz virava ~1e-3. Todos os vetores apontavam para # [0,0,0,1] com pequenas perturbações, fazendo SOM mapear tudo para o # mesmo neurônio (neurons_active=1/864). # # AGORA: normalizamos xyz e w INDEPENDENTEMENTE para [-1, 1]: # - xyz: L2-normalizado (preserva direção semântica do texto) # - w: mapeado para [-1, 1] via (2w - 1) = 2·(t/T_max) - 1 # Isto garante que ambas as componentes têm peso comparável no SOM, # permitindo que diferentes textos mapeiem para diferentes neurônios. xyz_norm = xyz_mean.norm() if xyz_norm > 1e-8: xyz_normalized = xyz_mean / xyz_norm # unit vector in xyz subspace else: # Texto degenerado — usa direção aleatória reproducível baseada em time_step torch.manual_seed(int(time_step) % 1000) xyz_normalized = torch.randn(3) xyz_normalized = xyz_normalized / xyz_normalized.norm() # w mapeado de [0, 1] para [-1, 1] (melhor distribuição no espaço 4D) w_normalized = 2.0 * w - 1.0 vec_4d = torch.cat([xyz_normalized, w_normalized.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.5, sigma0: float = 3.0, lambda_ewc: float = 0.01, # V6.5-V2-auto-conscience-v2 — initial γ raised from 0.1 to 0.5 to # ensure strong conscience bias from the start (prevents 2/864 bug). conscience_gamma: float = 0.5, conscience_beta: float = 0.1, ): # V6.5-V2-metrics-FIX-3 — α₀=0.5 e σ₀=3.0 conforme especificação # canônica para SOM 4D (Kohonen classic): # - α₀ ∈ [0.5, 1.0] para fase de ordenação (rough training) # - σ₀ = metade da maior dimensão da grade = max(6,6,6,4)/2 = 3.0 # V6.5-V2-metrics-FIX-3 — floors explícitos no update_weights garantem # que α_t e σ_t nunca decaiam abaixo de 0.001 e 0.1 respectivamente # (sem isso, após ~7000 updates σ→0 e o SOM degenera em k-means puro, # perdendo preservação topológica). # # V6.5-V2-metrics-FIX-4 (Conscience Mechanism — DeSieno 1988): # Corrige o problema neurons_active=2/864 relatado pelo usuário. # Cada neurônio i mantém uma frequência de vitória p_i ∈ [0,1]. # BMU selection: bmu = argmin_i ( ||W_i - x||² - b_i ) # onde b_i = γ · (1/N - p_i) é o bias de consciência # γ = 0.1 (default), N = total de neurônios (864) # Update da frequência (EMA): p_i ← (1-β)·p_i + β·𝟙[i==bmu], β=0.1 # Quando um neurônio ganha demais (p_i > 1/N), b_i fica negativo # (penaliza), e quando nunca ganha (p_i ≈ 0), b_i fica positivo # (empurra para ganhar). Isto força distribuição uniforme de BMU. self.I, self.J, self.K, self.L = grid_shape self.alpha0 = float(alpha0) self.sigma0 = float(sigma0) self.lambda_ewc = lambda_ewc self.t = 0 # V6.5-V2-auto-conscience-v2 — Inicialização diversa para evitar # neurônios com pesos idênticos (que causaria todos os inputs a # mapearem para o mesmo BMU). Usa grid coordinates como offset # inicial + ruído aleatório, garantindo que cada neurônio comece # em uma região diferente do espaço 4D. # User requirement: "distribuindo o processamento paralelamente". # Cada uma das 4 dimensões do peso recebe uma coordenada de grid # diferente (i, j, k, l), normalizada para [-1, 1]. grid_i = torch.arange(self.I).float().view(self.I, 1, 1, 1) / max(1, self.I - 1) grid_j = torch.arange(self.J).float().view(1, self.J, 1, 1) / max(1, self.J - 1) grid_k = torch.arange(self.K).float().view(1, 1, self.K, 1) / max(1, self.K - 1) grid_l = torch.arange(self.L).float().view(1, 1, 1, self.L) / max(1, self.L - 1) # Broadcast para (I, J, K, L) — cada coordenada ocupa seu próprio eixo grid_bc = grid_i * grid_j * grid_k * grid_l # (I, J, K, L) com valores ∈ [0,1] # Repete para 4 dimensões de peso, aplicando offset diferente para cada # dim para que neurônios vizinhos tenham pesos diferentes. w_dim_0 = (grid_i.expand(self.I, self.J, self.K, self.L) - 0.5) * 2.0 # [-1,1] em I w_dim_1 = (grid_j.expand(self.I, self.J, self.K, self.L) - 0.5) * 2.0 # [-1,1] em J w_dim_2 = (grid_k.expand(self.I, self.J, self.K, self.L) - 0.5) * 2.0 # [-1,1] em K w_dim_3 = (grid_l.expand(self.I, self.J, self.K, self.L) - 0.5) * 2.0 # [-1,1] em L # Stack na última dimensão → (I, J, K, L, 4) grid_coords = torch.stack([w_dim_0, w_dim_1, w_dim_2, w_dim_3], dim=-1) # Combina grid coords (determinístico) com ruído (estocástico) self.weights = grid_coords + 0.3 * 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) # V6.6 — Fisher accumulation CONTÍNUA (online EWC pattern) # Quando True, Fisher é acumulado em TODOS os updates (não só # antes da primeira consolidação). Após set_ewc_reference(), # Fisher é resetado e começa a acumular novamente para a próxima # tarefa. A consolidada é preservada via EMA em finalize_fisher(). self.fisher_continuous: bool = True # Histórico de Fisher consolidations (uma entrada por set_ewc_reference) self.fisher_consolidation_count: int = 0 # V6.5-V2-metrics-FIX-4 — Conscience mechanism (DeSieno 1988) # Inicializa p_i = 1/N (uniforme) — sem viés inicial. # b_i = γ · (1/N - p_i) começa em 0 (sem bias). self.n_neurons = self.I * self.J * self.K * self.L self.conscience_gamma = float(conscience_gamma) self.conscience_beta = float(conscience_beta) target_p = 1.0 / self.n_neurons self.win_frequency = torch.full( (self.I, self.J, self.K, self.L), target_p, dtype=torch.float ) # Contador absoluto de vitórias por neurônio (para auditoria) self.bmu_win_count = torch.zeros( self.I, self.J, self.K, self.L, dtype=torch.long ) # Histórico dos últimos K BMUs (para revive_dead_neurons) self._recent_bmu_flat: List[int] = [] # ------------------------------------------------------------------ # V6.5-V2-auto-conscience — Estado para ajuste adaptativo de γ. # User requirement: "aprimorar conscience_gamma para ajuste # automático conforme a quantidade de neurônios esteja caindo". # # Math (DeSieno 1988 + adaptação adaptativa): # r = n_active / n_total (∈ [0,1]) # γ_base = γ_max · (1 - r)^2 (intervenção cresce # com colapso) # drop = max(0, r_prev - r) (tendência de queda) # γ_boost = 0.3 · drop (reforço antecipado) # γ_target = clip(γ_base + γ_boost, γ_min, γ_max) # γ_new = (1 - η) · γ_old + η · γ_target (EMA suave) # # Comportamento esperado: # r = 1.0 → γ_base = 0.00 (data distribution domina) # r = 0.5 → γ_base = 0.25 (intervenção moderada) # r = 0.1 → γ_base = 0.81 (intervenção forte) # r = 0.0 → γ_base = 1.00 (intervenção máxima) # r caindo 1.0→0.5 → +0.15 boost (antecipa colapso) # ------------------------------------------------------------------ self._conscience_gamma_min = 0.01 self._conscience_gamma_max = 1.0 self._conscience_eta = 0.05 # EMA smoothing (normal regime) self._prev_n_active = -1 self._prev_n_active_rate = -1.0 self._last_gamma_target = float(conscience_gamma) self._last_gamma_drop = 0.0 self._last_activation_rate = 1.0 # V6.5-V2-auto-conscience-v2 — reduzido de 50 → 10 para detectar # colapso de neurônios mais cedo e ajustar γ a tempo. self._auto_adjust_interval = 10 # ajusta γ a cada 10 updates # V6.5-V2-kmeans-pp — flag para inicialização k-means++ deferida. # User requirement: "inicialização com k-means++ sobre embeddings em # vez de grid coords+ruído dos embeddings que saem do tokenizador # para ambas as FASE1 e FASE2". Como no momento do __init__ ainda # não temos dados, a init k-means++ é DEFERIDA: ela acontece na # primeira chamada de init_weights_kmeans_pp(data_buffer) feita # pelo KLS quando o buffer_4d tem >= 64 amostras. self._kmeans_pp_initialized = False self._kmeans_pp_min_samples = 64 # ====================================================================== # V6.5-V2-kmeans-pp — Inicialização k-means++ sobre embeddings # User requirement: "inicialização com k-means++ sobre embeddings em # vez de grid coords+ruído dos embeddings que saem do tokenizador # para ambas as FASE1 e FASE2". # Math (Arthur & Vassilvitskii 2007): # 1. c_1 = data[índice aleatório] # 2. Para c_2..c_N: escolhe c_i = data[j] com prob ∝ D(j)^2 # onde D(j) = min_k ||x_j - c_k||^2 # 3. Após seeding, roda n_iter iterações de Lloyd para refinar centros. # Vantagem: distribui pesos nas regiões de ALTA densidade dos dados, # reduzindo dead neuron rate inicial. # ====================================================================== def init_weights_kmeans_pp( self, data_buffer, n_iter: int = 5, random_seed=None, ): """Inicializa pesos do SOM via k-means++ (Arthur & Vassilvitskii 2007). Args: data_buffer: lista de tensores [4] — amostras do buffer_4d do KLS. n_iter: número de iterações de Lloyd (default 5). random_seed: seed para reprodutibilidade. Returns: Dict com: initialized, n_samples_used, n_neurons_assigned, inertia (soma das dist^2 ao centro mais próximo). """ if not data_buffer or len(data_buffer) < 8: return { "initialized": False, "reason": f"insufficient_data ({len(data_buffer) if data_buffer else 0} < 8)", } if random_seed is not None: g = torch.Generator().manual_seed(int(random_seed)) else: g = None with torch.no_grad(): try: data = torch.stack( [v.detach().clone().float() if isinstance(v, torch.Tensor) else torch.tensor(v, dtype=torch.float) for v in data_buffer] ).float() except Exception: return {"initialized": False, "reason": "stack_failed"} data = torch.nan_to_num(data, nan=0.0, posinf=1.0, neginf=-1.0) N = data.shape[0] P = self.n_neurons if N < P: repeats = (P // N) + 1 data = data.repeat(repeats, 1)[:P * 4] N = data.shape[0] # ---- k-means++ seeding ---- if g is not None: first_idx = int(torch.randint(0, N, (1,), generator=g).item()) else: first_idx = int(torch.randint(0, N, (1,)).item()) centers = [data[first_idx].clone()] for k_idx in range(1, P): C = torch.stack(centers) data_sq = (data * data).sum(dim=-1, keepdim=True) C_sq = (C * C).sum(dim=-1, keepdim=True).t() cross = data @ C.t() dist_sq = data_sq + C_sq - 2.0 * cross dist_sq = torch.clamp(dist_sq, min=0.0) D_sq_min = dist_sq.min(dim=-1).values D_sum = D_sq_min.sum().clamp(min=1e-12) probs = D_sq_min / D_sum if g is not None: new_idx = int(torch.multinomial(probs, 1, generator=g).item()) else: new_idx = int(torch.multinomial(probs, 1).item()) centers.append(data[new_idx].clone()) centers_tensor = torch.stack(centers) centers_tensor = centers_tensor + 0.02 * torch.randn(P, 4) # ---- Lloyd iterations (refinamento) ---- for it in range(n_iter): data_sq = (data * data).sum(dim=-1, keepdim=True) C_sq = (centers_tensor * centers_tensor).sum(dim=-1, keepdim=True).t() cross = data @ centers_tensor.t() dist_sq = data_sq + C_sq - 2.0 * cross dist_sq = torch.clamp(dist_sq, min=0.0) nearest = dist_sq.argmin(dim=-1) for p in range(P): mask = (nearest == p) if mask.any(): centers_tensor[p] = data[mask].mean(dim=0) data_sq = (data * data).sum(dim=-1, keepdim=True) C_sq = (centers_tensor * centers_tensor).sum(dim=-1, keepdim=True).t() cross = data @ centers_tensor.t() dist_sq = data_sq + C_sq - 2.0 * cross dist_sq = torch.clamp(dist_sq, min=0.0) inertia = float(dist_sq.min(dim=-1).values.sum().item()) perm = torch.randperm(P, generator=g) centers_shuffled = centers_tensor[perm] new_weights = centers_shuffled.view(self.I, self.J, self.K, self.L, 4).clone() new_weights = torch.clamp(new_weights, -10.0, 10.0) self.weights = new_weights self._kmeans_pp_initialized = True return { "initialized": True, "n_samples_used": int(N), "n_neurons_assigned": int(P), "n_lloyd_iters": int(n_iter), "inertia": float(inertia), "method": "kmeans++ (Arthur & Vassilvitskii 2007)", } 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, use_conscience: bool = True) -> Tuple[int, int, int, int]: """Best Matching Unit: argmin ||W - x||² em ℝ⁴, com conscience bias opcional. Substitui pgvector_lookup — busca nearest-neighbor flat sobre o grid. V6.5-V2-auto-conscience-v2 (FIX math asymmetry): A fórmula anterior b_i = γ·(1/N - p_i) dava bias quase zero para neurônios mortos (γ/N ≈ 0.001 para N=864), enquanto penalizava fortemente neurônios vencedores (γ·p_i ≈ 0.3γ). Isto causava neurons_active=2/864 — a vantagem dada a neurônios mortos era insuficiente para superar a diferença natural de distância. Nova fórmula (simétrica, DeSieno 1988 corrigida): p_norm_i = bmu_win_count[i] / max(1, max(bmu_win_count)) bias_i = γ · (0.5 - p_norm_i) (simétrica em [-0.5γ, +0.5γ]) dist_eff = dist² - bias_i Comportamento: - Neurônio dominante (p_norm = 1.0): bias = -0.5γ → dist_eff = dist² + 0.5γ (PENALIZADO) - Neurônio nunca-vencedor (p_norm = 0): bias = +0.5γ → dist_eff = dist² - 0.5γ (FAVORECIDO) - Magnitudes iguais → distribuição efetiva do BMU entre os 864 neurônios. V6.5-V2-auto-conscience-v2 (forced dead-neuron exploration): Para neurônios com bmu_win_count == 0, aplica um boost adicional de -γ·0.5 (reduz distância efetiva em 0.5γ). Isto garante que neurônios mortos sejam explorados ativamente, distribuindo o processamento paralelamente (user requirement). Args: x: tensor [4] — vetor de entrada. use_conscience: se True, aplica bias de consciência (default True). """ # Sanitiza x antes de computar distância if torch.isnan(x).any() or torch.isinf(x).any(): x = torch.nan_to_num(x, nan=0.0, posinf=1e4, neginf=-1e4) dist = torch.sum((self.weights - x.view(1, 1, 1, 1, 4)) ** 2, dim=-1) if use_conscience: # V6.5-V2-auto-conscience-v2 — bias simétrico baseado em win_count normalizado with torch.no_grad(): win_count_flat = self.bmu_win_count.flatten().float() max_win = win_count_flat.max().clamp(min=1.0) p_norm = win_count_flat / max_win # ∈ [0, 1] # Bias simétrico: +0.5γ para mortos, -0.5γ para dominante bias = self.conscience_gamma * (0.5 - p_norm) # Boost adicional para neurônios nunca-vencedores (win_count == 0) dead_mask = (win_count_flat == 0).float() dead_boost = self.conscience_gamma * 0.5 * dead_mask # -0.5γ na distância # dist_effective = dist² - bias - dead_boost # (bias positivo reduz distância → favorece; bias negativo aumenta → penaliza) bias_flat = bias.flatten() + dead_boost dist_flat = dist.flatten() - bias_flat flat_idx = int(torch.argmin(dist_flat).item()) else: flat_idx = int(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_win_frequency(self, bmu_idx: Tuple[int, int, int, int]) -> None: """V6.5-V2-metrics-FIX-4 — Atualiza frequência de vitória (EMA). p_i ← (1-β)·p_i + β·𝟙[i==bmu], β = conscience_beta (default 0.1) Também mantém contador absoluto bmu_win_count para auditoria. Deve ser chamado APÓS find_bmu e APÓS update_weights. Args: bmu_idx: (i, j, k, l) — índice do BMU selecionado. """ with torch.no_grad(): # EMA update self.win_frequency = ( (1.0 - self.conscience_beta) * self.win_frequency ) self.win_frequency[bmu_idx] += self.conscience_beta # Contador absoluto self.bmu_win_count[bmu_idx] += 1 # Histórico recente (para revive_dead_neurons) flat = ( bmu_idx[0] * (self.J * self.K * self.L) + bmu_idx[1] * (self.K * self.L) + bmu_idx[2] * self.L + bmu_idx[3] ) self._recent_bmu_flat.append(flat) # Mantém últimos 200 BMUs if len(self._recent_bmu_flat) > 200: self._recent_bmu_flat = self._recent_bmu_flat[-200:] def revive_dead_neurons( self, data_buffer: Optional[List[torch.Tensor]] = None, dead_threshold: int = 0, ) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-4 — Revive neurônios mortos reinicializando pesos. User requirement: "ANALISAR matematicamente e logicamente a ativação e uso e acesso dos neurônios (apenas dois estão sendo ativados: neurons_active=2/864) distribuindo o processamento paralelamente". Um neurônio é considerado "morto" se bmu_win_count <= dead_threshold (nunca ou raramente foi BMU). Para cada neurônio morto: 1. Se data_buffer fornecido: amostra um vetor aleatório do buffer e atribui aos pesos do neurônio (reinicialização data-driven). 2. Se buffer vazio: reinicializa com ruído gaussiano pequeno (N(0, 0.1)). 3. Reseta win_frequency para 1/N (sem bias) e bmu_win_count para 0. Isto garante que TODOS os 864 neurônios sejam utilizados, distribuindo o processamento paralelo do SOM conforme solicitado. Args: data_buffer: lista de tensores [4] — amostras do buffer_4d do KLS. dead_threshold: neurônios com win_count <= threshold são revividos. Returns: Dict com: n_revived, n_total, revived_indices, dead_rate_before, dead_rate_after. """ with torch.no_grad(): dead_mask = self.bmu_win_count <= dead_threshold n_dead = int(dead_mask.sum().item()) n_total = self.n_neurons dead_rate_before = float(n_dead / n_total) revived_indices: List[Tuple[int, int, int, int]] = [] if n_dead == 0: return { "n_revived": 0, "n_total": n_total, "revived_indices": [], "dead_rate_before": dead_rate_before, "dead_rate_after": dead_rate_before, "action": "none", } # Amostra pontos do buffer se disponível buffer_tensor = None if data_buffer and len(data_buffer) > 0: try: buffer_tensor = torch.stack( [v.detach().clone() if isinstance(v, torch.Tensor) else torch.tensor(v, dtype=torch.float) for v in data_buffer] ).float() except Exception: buffer_tensor = None # V6.5-V2-kmeans-pp-revive — Reinicializa neurônios mortos usando # estratégia k-means++ (max-min selection): em vez de amostrar um # ponto aleatório do buffer, escolhe o ponto do buffer que está # MAIS DISTANTE de todos os neurônios ativos atuais. Isto garante # que os pesos revividos cubram regiões do espaço de entrada que # não estavam sendo representadas, maximizando a diversidade # topológica do SOM. # User requirement: "reforçar auto_revive para reinicializar pesos # dos neurônios mortos (não só boostar γ)". dead_indices = dead_mask.nonzero(as_tuple=False) # Identifica neurônios ATIVOS (win_count > 0) para cálculo de distância active_mask = self.bmu_win_count > 0 active_indices = active_mask.nonzero(as_tuple=False) # Stack pesos ativos (centros atuais do SOM) if active_indices.shape[0] > 0: active_weights = self.weights[active_mask] # (n_active, 4) else: active_weights = None for idx_tensor in dead_indices: i, j, k, l = idx_tensor.tolist() if buffer_tensor is not None and buffer_tensor.shape[0] > 0: if active_weights is not None and active_weights.shape[0] > 0: # k-means++ selection: ponto mais distante dos ativos # dist²[j, k] = ||buffer[j] - active[k]||² buf_sq = (buffer_tensor * buffer_tensor).sum(dim=-1, keepdim=True) # (B, 1) act_sq = (active_weights * active_weights).sum(dim=-1, keepdim=True).t() # (1, A) cross = buffer_tensor @ active_weights.t() # (B, A) dist_sq = buf_sq + act_sq - 2.0 * cross dist_sq = torch.clamp(dist_sq, min=0.0) # D²(j) = min_k ||buf[j] - active[k]||² D_sq_min = dist_sq.min(dim=-1).values # (B,) # Escolhe o ponto com maior D² (mais distante) sample_idx = int(torch.argmax(D_sq_min).item()) new_w = buffer_tensor[sample_idx].clone() # Pequeno ruído para evitar duplicação exata new_w = new_w + 0.05 * torch.randn(4) # Adiciona o novo peso à lista de ativos para próxima iteração active_weights = torch.cat([active_weights, new_w.unsqueeze(0)], dim=0) else: # Sem ativos: amostra aleatória simples sample_idx = torch.randint(0, buffer_tensor.shape[0], (1,)).item() new_w = buffer_tensor[sample_idx].clone() new_w = new_w + 0.05 * torch.randn(4) # Inicia lista de ativos active_weights = new_w.unsqueeze(0).clone() else: # Reinicialização gaussiana pequena new_w = 0.1 * torch.randn(4) new_w = torch.clamp(new_w, -10.0, 10.0) self.weights[i, j, k, l] = new_w # Reset counters self.win_frequency[i, j, k, l] = 1.0 / n_total self.bmu_win_count[i, j, k, l] = 0 revived_indices.append((i, j, k, l)) # Recalcula dead rate após revival new_dead_mask = self.bmu_win_count <= dead_threshold n_dead_after = int(new_dead_mask.sum().item()) dead_rate_after = float(n_dead_after / n_total) return { "n_revived": len(revived_indices), "n_total": n_total, "revived_indices": revived_indices[:50], # top 50 para log "n_dead_before": n_dead, "n_dead_after": n_dead_after, "dead_rate_before": dead_rate_before, "dead_rate_after": dead_rate_after, "action": "revived" if revived_indices else "none", "used_buffer": buffer_tensor is not None, } def parallel_neuron_activation_report(self) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-4 — Relatório estruturado de ativação dos 864 neurônios. User requirement: "distribuindo o processamento paralelamente" + o exemplo de código fornecido mostra estruturação vetorizada para análise estatística e auditoria do modelo. Este método produz um relatório análogo ao DataFrame do exemplo, mas otimizado para o grid 4D (6,6,6,4) com 864 neurônios: Returns: Dict com: - n_total_neurons: int (864) - n_active_neurons: int (vitória em ≥1 amostra histórica) - n_dead_neurons: int (nunca foi BMU) - neuron_activation_rate: float - max_win_count: int (neurônio mais ativo) - min_win_count: int (neurônio menos ativo) - mean_win_count: float - std_win_count: float - max_win_frequency: float - min_win_frequency: float - bmu_distribution_top20: dict {flat_idx: count} - bmu_distribution_bottom20: dict {flat_idx: count} (mortos) - conscience_bias_mean: float (deve tender a 0 se uniforme) - conscience_bias_std: float (deve tender a 0 se uniforme) - uniformity_score: float (1 - CV da win_frequency, ∈ [0,1]) """ with torch.no_grad(): win_counts_flat = self.bmu_win_count.flatten().float() win_freq_flat = self.win_frequency.flatten() n_total = self.n_neurons n_active = int((self.bmu_win_count > 0).sum().item()) n_dead = n_total - n_active # Estatísticas if n_total > 0: max_wc = float(win_counts_flat.max().item()) min_wc = float(win_counts_flat.min().item()) mean_wc = float(win_counts_flat.mean().item()) std_wc = float(win_counts_flat.std().item()) max_wf = float(win_freq_flat.max().item()) min_wf = float(win_freq_flat.min().item()) # Uniformidade: 1 - CV (coeficiente de variação) cv = float(std_wc / max(mean_wc, 1e-8)) uniformity = max(0.0, 1.0 - cv) else: max_wc = min_wc = mean_wc = std_wc = 0.0 max_wf = min_wf = 0.0 uniformity = 0.0 # Bias de consciência: b_i = γ · (1/N - p_i) target_p = 1.0 / n_total bias_flat = self.conscience_gamma * (target_p - win_freq_flat) bias_mean = float(bias_flat.mean().item()) bias_std = float(bias_flat.std().item()) # Top-20 BMUs mais frequentes from collections import Counter recent_counter = Counter(self._recent_bmu_flat) top20 = dict(recent_counter.most_common(20)) # Bottom-20 (mortos ou raros) — últimos em vitórias absolutas sorted_indices = torch.argsort(win_counts_flat) bottom20_idx = sorted_indices[:20].tolist() bottom20 = { int(idx): int(self.bmu_win_count.flatten()[idx].item()) for idx in bottom20_idx } return { "n_total_neurons": int(n_total), "n_active_neurons": int(n_active), "n_dead_neurons": int(n_dead), "neuron_activation_rate": float(n_active / max(1, n_total)), "max_win_count": max_wc, "min_win_count": min_wc, "mean_win_count": mean_wc, "std_win_count": std_wc, "max_win_frequency": max_wf, "min_win_frequency": min_wf, "bmu_distribution_top20": {str(k): int(v) for k, v in top20.items()}, "bmu_distribution_bottom20": {str(k): int(v) for k, v in bottom20.items()}, "conscience_bias_mean": bias_mean, "conscience_bias_std": bias_std, "uniformity_score": uniformity, "conscience_gamma": float(self.conscience_gamma), "conscience_beta": float(self.conscience_beta), # V6.5-V2-auto-conscience — telemetria do ajuste adaptativo "conscience_gamma_min": float(self._conscience_gamma_min), "conscience_gamma_max": float(self._conscience_gamma_max), "conscience_gamma_target": float(self._last_gamma_target), "conscience_gamma_drop": float(self._last_gamma_drop), "activation_rate_at_last_adjust": float(self._last_activation_rate), "prev_n_active_at_last_adjust": int(self._prev_n_active), "n_recent_bmus_tracked": int(len(self._recent_bmu_flat)), } # ================================================================== # V6.5-V2-parallel — Vectorial parallel SOM (ParallelSOM4D pattern) # User requirement: "Para elevar o desempenho ao nível máximo, a arquitetura # com multiprocessing (baseada em CPU) que usamos anteriormente não é a # ideal para cenários de alta dimensão. O custo de copiar os dados entre # múltiplos processos do Python (overhead de IPC) anula o ganho da # computação paralela. Para um ajuste paralelo verdadeiro e massivo de uma # SOM de 4 dimensões, a abordagem ideal é o paralelismo vetorial em GPU # utilizando PyTorch." # # Math: todas as operações são feitas via broadcasting PyTorch: # 1. distances = sum((flat_weights - sample)^2, dim=-1) # (N_neurons,) # 2. bmu_idx = argmin(distances) # 3. grid_dists_sq = sum((grid_coords - bmu_coord)^2, dim=-1) # (N_neurons,) # 4. influence = exp(-grid_dists_sq / (2 * sigma^2)) # (N_neurons,) # 5. weights += alpha * influence * (sample - weights) # broadcast (N, 4) # ================================================================== def _precompute_grid_coords_flat(self) -> torch.Tensor: """Pré-computa coordenadas 4D da grade planificadas (N_neurons, 4). V6.5-V2-parallel — pré-computação evita recriar meshgrid a cada update. Usado por train_batch_parallel e reorganize_map_by_density. """ if not hasattr(self, "_grid_coords_flat_cache") or self._grid_coords_flat_cache is None: 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", ) grid_coords = torch.stack([II, JJ, KK, LL], dim=-1) # (I, J, K, L, 4) self._grid_coords_flat_cache = grid_coords.view(-1, 4) # (N, 4) return self._grid_coords_flat_cache def train_batch_parallel( self, samples: torch.Tensor, use_conscience: bool = True, accumulate_fisher: bool = False, ) -> Dict[str, Any]: """V6.5-V2-parallel — Treina SOM com paralelismo vetorial PyTorch. User requirement (ParallelSOM4D pattern): processa amostras usando operações matriciais PyTorch ao invés de loops Python. Cada amostra ainda é processada sequencialmente (online learning — Kohonen canônico), mas todas as operações internas (distância, BMU, vizinhança, update) são feitas via broadcasting sobre os 864 neurônios simultaneamente. Vantagens: 1. Zero IPC overhead (sem multiprocessing) 2. Broadcast de tensores — uma instrução matricial atualiza todos os pesos da hipergrade 4D simultaneamente 3. Compatível com AVX512_VNNI / AMX_INT8 via xeon_runtime 4. Memória eficiente: apenas 2 tensores temporários (dist, influence) Args: samples: tensor (B, 4) — batch de amostras 4D. use_conscience: se True, aplica bias de consciência (default True). accumulate_fisher: se True, acumula Fisher info em w. Returns: Dict com: n_processed, n_bmu_unique, dead_rate, alpha_t, sigma_t. """ if not isinstance(samples, torch.Tensor): samples = torch.tensor(samples, dtype=torch.float) samples = samples.float() if samples.dim() == 1: samples = samples.unsqueeze(0) # Sanitiza NaN/Inf samples = torch.nan_to_num(samples, nan=0.0, posinf=1e4, neginf=-1e4) n_processed = 0 bmu_indices: List[Tuple[int, int, int, int]] = [] grid_coords_flat = self._precompute_grid_coords_flat() with torch.no_grad(): for sample_idx in range(samples.shape[0]): x = samples[sample_idx] # (4,) # 1. PARALELISMO MATRICIAL: distância de 1 amostra vs TODOS os neurônios flat_weights = self.weights.view(-1, 4) # (N, 4) dist = torch.sum((flat_weights - x.unsqueeze(0)) ** 2, dim=-1) # (N,) # 2. BMU selection (com conscience opcional) if use_conscience: win_count_flat = self.bmu_win_count.flatten().float() max_win = win_count_flat.max().clamp(min=1.0) p_norm = win_count_flat / max_win bias = self.conscience_gamma * (0.5 - p_norm) dead_mask = (win_count_flat == 0).float() dead_boost = self.conscience_gamma * 0.5 * dead_mask bias_flat = bias + dead_boost dist_eff = dist - bias_flat flat_idx = int(torch.argmin(dist_eff).item()) else: flat_idx = int(torch.argmin(dist).item()) # Recupera coordenada 4D da BMU 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 bmu_idx = (i, j, k, l) bmu_indices.append(bmu_idx) # 3. Vizinhança 4D paralela — distância na grade bmu_coord = grid_coords_flat[flat_idx] # (4,) grid_dists_sq = torch.sum((grid_coords_flat - bmu_coord.unsqueeze(0)) ** 2, dim=-1) # (N,) # 4. Hiperparâmetros com floors sigma = max(self.sigma0 * math.exp(-self.t / 1000), 0.1) alpha = max(self.alpha0 * math.exp(-self.t / 2000), 0.001) # 5. Influência Gaussiana influence = torch.exp(-grid_dists_sq / (2 * sigma ** 2)) # (N,) influence = influence.view(self.I, self.J, self.K, self.L, 1) # (I,J,K,L,1) # 6. Update em massa — uma operação vetorial delta = x.view(1, 1, 1, 1, 4) - self.weights # (I,J,K,L,4) update = alpha * influence * delta # EWC penalty na 4ª dimensão if self.old_weights_w is not None and self.fisher_w is not None: 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 self.weights = torch.clamp(self.weights, -100.0, 100.0) if torch.isnan(self.weights).any() or torch.isinf(self.weights).any(): self.weights = torch.nan_to_num( self.weights, nan=0.0, posinf=100.0, neginf=-100.0 ) # 7. Fisher accumulation if accumulate_fisher: influence_flat = influence.view(-1) mask = influence_flat > 0.1 if mask.any(): diff_sq = (x[3] - self.weights.view(-1, 4)[mask, 3]) ** 2 self.fisher_accum.view(-1)[mask] += diff_sq self.fisher_count.view(-1)[mask] += 1 self.t += 1 n_processed += 1 # 8. Atualiza conscience (win frequency) try: self.update_win_frequency(bmu_idx) except Exception: pass # 9. Auto-ajuste de γ (throttled) try: if self.t % self._auto_adjust_interval == 0: self.auto_adjust_conscience_gamma() except Exception: pass # Estatísticas pós-batch n_unique_bmus = len(set(bmu_indices)) n_active = int((self.bmu_win_count > 0).sum().item()) dead_rate = float(1.0 - n_active / max(1, self.n_neurons)) sigma_t = max(self.sigma0 * math.exp(-self.t / 1000), 0.1) alpha_t = max(self.alpha0 * math.exp(-self.t / 2000), 0.001) return { "n_processed": n_processed, "n_bmu_unique": n_unique_bmus, "dead_rate": dead_rate, "n_active": n_active, "n_total": self.n_neurons, "alpha_t": float(alpha_t), "sigma_t": float(sigma_t), "som_t": int(self.t), "conscience_gamma": float(self.conscience_gamma), "method": "vectorial_parallel_pytorch", } # ================================================================== # V6.5-V2-reorganize — Density-based map reorganization # (ReorganizingSOM4D pattern) # User requirement: "Para reorganizar o mapa SOM dinamicamente durante o # treinamento, a técnica matemática mais robusta e paralelizável é a # Filtragem de Ativação por Convolução Baseada em Densidade. Em vez de # alterar a topologia de forma sequencial (o que quebraria o paralelismo), # nós calculamos a frequência de ativação de cada neurônio (densidade de # uso) diretamente na GPU/CPU e aplicamos um filtro gaussiano # multidimensional. Os neurônios pouco ativados (isolados) sofrem uma força # de atração em direção aos agrupamentos de alta densidade, reorganizando # geometricamente o hipercubo 4D sem perder o alinhamento matricial." # # Math: # density_i = bmu_win_count[i] / sum(bmu_win_count) ∈ [0, 1] # density_center = sum(weights * density, dim=0) (centro de massa ponderado) # pull_force_i = (1 - density_i) * regularization_factor # new_weights_i = weights_i + pull_force_i * (density_center - weights_i) # # Comportamento: # - Neurônios com alta densidade (muito ativos): pull_force ≈ 0 → ficam parados # - Neurônios com baixa densidade (mortos): pull_force ≈ 1 → puxados para o centro # - Centro de massa é dominado pelos neurônios ativos (ponderação por densidade) # ================================================================== def reorganize_map_by_density( self, regularization_factor: float = 0.05, min_dead_rate_for_trigger: float = 0.3, ) -> Dict[str, Any]: """V6.5-V2-reorganize — Reorganização topológica paralela baseada em densidade. Implementa o padrão ReorganizingSOM4D: identifica zonas frias (neurônios mortos) via hit_map e as desloca matematicamente em direção ao centro de massa ponderado pelos neurônios ativos. Vantagens sobre revive_dead_neurons puro: 1. Não zera bmu_win_count (preserva histórico de ativação) 2. Pulso suave e contínuo (não reinicialização abrupta) 3. Compatível com paralelismo matricial — uma única operação vetorial 4. Decaimento temporal: regularization_factor *= decay (early epochs reorganizam drasticamente, late epochs estabilizam) 5. Eliminação efetiva de neurônios mortos sem quebrar a topologia Args: regularization_factor: magnitude do pull (default 0.05 — 5% do deslocamento em direção ao centro de massa). min_dead_rate_for_trigger: só reorganiza se dead_rate > threshold (default 0.3 = 30% mortos). Evita reorganização desnecessária. Returns: Dict com: reorganized, dead_rate_before, dead_rate_after (estimado), n_dead_pulled, density_center_norm, regularization_factor. """ with torch.no_grad(): # Computa dead rate atual n_total = self.n_neurons n_active = int((self.bmu_win_count > 0).sum().item()) dead_rate = float(1.0 - n_active / max(1, n_total)) if dead_rate < min_dead_rate_for_trigger: return { "reorganized": False, "reason": f"dead_rate={dead_rate:.3f} < threshold={min_dead_rate_for_trigger}", "dead_rate": dead_rate, "n_active": n_active, "n_total": n_total, } # Normaliza hit_map para distribuição de densidade total_hits = float(self.bmu_win_count.sum().item()) if total_hits < 1e-8: # Sem hits — não há centro de massa para puxar return { "reorganized": False, "reason": "no_hits_in_buffer", "dead_rate": dead_rate, "n_active": n_active, "n_total": n_total, } density = self.bmu_win_count.float() / total_hits # (I, J, K, L) # Planifica para operações matriciais flat_weights = self.weights.view(-1, 4) # (N, 4) flat_density = density.view(-1, 1) # (N, 1) # Centro de massa ponderado por densidade (âncora global) density_center = torch.sum(flat_weights * flat_density, dim=0, keepdim=True) # (1, 4) density_center_norm = float(torch.norm(density_center).item()) # Decaimento temporal: reorganização forte em épocas iniciais, # estabiliza em épocas finais (decay = exp(-t / 5000)) decay = float(math.exp(-self.t / 5000.0)) effective_factor = float(regularization_factor * decay) # Força de atração: neurônios com baixa densidade sofrem maior deslocamento pull_force = (1.0 - flat_density) * effective_factor # (N, 1) # Aplica reorganização em uma operação vetorial new_flat_weights = flat_weights + pull_force * (density_center - flat_weights) # Restaura shape original self.weights = new_flat_weights.view(self.I, self.J, self.K, self.L, 4).clone() self.weights = torch.clamp(self.weights, -100.0, 100.0) # Sanitiza NaN/Inf if torch.isnan(self.weights).any() or torch.isinf(self.weights).any(): self.weights = torch.nan_to_num( self.weights, nan=0.0, posinf=100.0, neginf=-100.0 ) # Conta quantos neurônios mortos foram efetivamente deslocados n_dead_pulled = int((pull_force.view(-1) > 1e-6).sum().item()) return { "reorganized": True, "dead_rate_before": dead_rate, "n_dead_pulled": n_dead_pulled, "density_center_norm": density_center_norm, "regularization_factor": effective_factor, "decay": decay, "n_active": n_active, "n_total": n_total, "method": "density_based_reorganization", } def auto_adjust_conscience_gamma(self) -> Dict[str, Any]: """V6.5-V2-auto-conscience-v2 — Ajusta γ dinamicamente conforme ativação. User requirement: "aprimorar conscience_gamma para ajuste automático conforme a quantidade de neurônios esteja caindo". V6.5-V2-auto-conscience-v2 (FIX slow EMA convergence): A versão anterior usava η=0.05 (EMA suave) que levava 50+ updates para convergir ao target. Como FASE1 tem apenas ~30-50 updates por chunk, o γ ficava preso em 0.1 (valor inicial) e nunca atingia o γ_target=1.0 necessário para reviver neurônios mortos. Nova estratégia: - Se r < 0.05 (colapso crítico): salta diretamente para γ_max (intervenção máxima imediata — sem EMA, bypass) - Se r < 0.20 (degradação severa): η_eff = 0.5 (convergência rápida) - Caso contrário: η_eff = η_default = 0.05 (suavização normal) Math (DeSieno 1988 + adaptação adaptativa): r = n_active / n_total (∈ [0,1]) γ_base = γ_max · (1 - r)^2 (intervenção cresce com colapso) drop = max(0, r_prev - r) (tendência de queda) γ_boost = 0.3 · drop (reforço antecipado) γ_target = clip(γ_base + γ_boost, γ_min, γ_max) γ_new = (1 - η_eff) · γ_old + η_eff · γ_target (EMA adaptativo) Comportamento esperado: r = 1.0 (saudável) → γ_base = 0.00 (data distribution domina) r = 0.5 (degradação) → γ_base = 0.25 (intervenção moderada) r = 0.1 (colapso severo) → γ_base = 0.81 (intervenção forte, η=0.5) r < 0.05 (crítico) → γ = γ_max = 1.0 (salto direto, sem EMA) Returns: Dict com: n_active, n_total, activation_rate, prev_activation_rate, drop, gamma_target, gamma_old, gamma_new, gamma_min, gamma_max, eta. """ with torch.no_grad(): n_total = self.n_neurons n_active = int((self.bmu_win_count > 0).sum().item()) r = float(n_active / max(1, n_total)) # Tendência: se _prev_n_active_rate < 0, ainda não houve ajuste r_prev = ( self._prev_n_active_rate if self._prev_n_active_rate >= 0 else r ) drop = max(0.0, r_prev - r) # γ alvo: intervenção quadrática no colapso + boost antecipado gamma_base = self._conscience_gamma_max * (1.0 - r) ** 2 gamma_trend_boost = 0.3 * drop gamma_target = max( self._conscience_gamma_min, min( self._conscience_gamma_max, gamma_base + gamma_trend_boost, ), ) # V6.5-V2-auto-conscience-v2 — adaptive EMA rate # Critical collapse: jump directly to γ_max (bypass EMA) # Severe degradation: fast EMA (η=0.5) # Normal: slow EMA (η=0.05, default) gamma_old = float(self.conscience_gamma) if r < 0.05: # CRITICAL: bypass EMA, jump directly to target eta_eff = 1.0 gamma_new = float(gamma_target) elif r < 0.20: # SEVERE: fast EMA convergence eta_eff = 0.5 gamma_new = (1.0 - eta_eff) * gamma_old + eta_eff * float(gamma_target) else: # NORMAL: slow EMA eta_eff = self._conscience_eta gamma_new = (1.0 - eta_eff) * gamma_old + eta_eff * float(gamma_target) gamma_new = float(max( self._conscience_gamma_min, min(self._conscience_gamma_max, gamma_new), )) # Persiste estado self.conscience_gamma = gamma_new self._prev_n_active = n_active self._prev_n_active_rate = r self._last_gamma_target = float(gamma_target) self._last_gamma_drop = float(drop) self._last_activation_rate = r return { "n_active": n_active, "n_total": n_total, "activation_rate": r, "prev_activation_rate": r_prev, "drop": drop, "gamma_target": float(gamma_target), "gamma_old": gamma_old, "gamma_new": gamma_new, "gamma_min": float(self._conscience_gamma_min), "gamma_max": float(self._conscience_gamma_max), "eta": float(eta_eff), "bypass_ema": bool(r < 0.05), } def update_weights(self, x: torch.Tensor, bmu_idx, accumulate_fisher=False): """Update Kohonen: ΔW = α·Λ·(x - W) + penalidade EWC em w. V6.5-V2-metrics-FIX-4 — agora chama update_win_frequency automaticamente após o update, garantindo que o conscience mechanism seja atualizado a cada amostra processada (sem necessidade de chamada externa). V6.5-V2-metrics-FIX-3 — Correções matemáticas: 1. Floors explícitos em α_t e σ_t (previnem colapso topológico após muitas épocas, quando σ_t→0 degenera o SOM em k-means). 2. Sanitização de NaN/Inf em x ANTES do update (uma única amostra NaN não deve contaminar permanentemente os pesos). 3. Clamp dos pesos após update (evita explosão numérica). V6.6 — Fisher accumulation CONTÍNUA (online EWC pattern): Quando self.fisher_continuous=True (default), acumula Fisher em TODOS os updates, não apenas quando accumulate_fisher=True. Após set_ewc_reference(), Fisher é resetado e começa a acumular novamente para a próxima tarefa. Isto suporta online EWC (Schwarz et al. 2018) onde múltiplas tarefas são consolidadas sequencialmente sem catastrophic forgetting. Args: x: tensor [4] — amostra 4D. bmu_idx: (i, j, k, l) — índice do BMU. accumulate_fisher: se True (ou fisher_continuous=True), acumula (x_w - W_w)² nos Fisher accumulators. """ # V6.5-V2-metrics-FIX-3 — sanitiza x: substitui NaN/Inf por 0 # (uma amostra degenerada não deve contaminar todo o SOM). if torch.isnan(x).any() or torch.isinf(x).any(): x = torch.nan_to_num(x, nan=0.0, posinf=1e4, neginf=-1e4) dist_sq = self._neighborhood(bmu_idx) # V6.5-V2-metrics-FIX-3 — floors explícitos (previnem σ_t→0) sigma = max(self.sigma0 * math.exp(-self.t / 1000), 0.1) alpha = max(self.alpha0 * math.exp(-self.t / 2000), 0.001) 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 # V6.5-V2-metrics-FIX-3 — clamp pós-update para evitar explosão # (pesos Kohonen são tipicamente O(1); permitir valores > 100 indica # divergência e propaga NaN para todas as métricas subsequentes). self.weights = torch.clamp(self.weights, -100.0, 100.0) # V6.5-V2-metrics-FIX-3 — se ainda assim surgir NaN, reverte este update if torch.isnan(self.weights).any() or torch.isinf(self.weights).any(): self.weights = torch.nan_to_num(self.weights, nan=0.0, posinf=100.0, neginf=-100.0) # V6.6 — Fisher accumulation contínua (online EWC) # Acumula SEMPRE que: accumulate_fisher=True OU fisher_continuous=True should_accumulate = accumulate_fisher or getattr(self, 'fisher_continuous', False) if should_accumulate: # 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] = self.fisher_accum[mask] + diff_sq self.fisher_count[mask] = self.fisher_count[mask] + 1 self.t += 1 # V6.5-V2-metrics-FIX-4 — atualiza conscience mechanism (win frequency) # automaticamente após cada update. Isto garante que o bias de consciência # seja aplicado corretamente no próximo find_bmu, forçando distribuição # uniforme de BMU entre os 864 neurônios. try: self.update_win_frequency(bmu_idx) except Exception: # Não deixa falha no conscience quebrar o treino principal pass # V6.5-V2-auto-conscience — ajuste adaptativo de γ (throttled). # Roda a cada _auto_adjust_interval updates (default 50) para # evitar custo computacional em cada amostra e suavizar o ajuste. try: if self.t % self._auto_adjust_interval == 0: self.auto_adjust_conscience_gamma() except Exception: # Não deixa falha no auto-ajuste quebrar o treino principal pass def finalize_fisher(self): """Fisher = mean((x_w - W_w)²) sobre samples acumuladas. V6.6 — se fisher_continuous=True E já existe fisher_w (consolidação anterior), faz EMA (exponential moving average) com a Fisher anterior em vez de substituir — isto preserva informação de tarefas passadas (online EWC pattern, cf. Schwarz et al. 2018). """ cnt = self.fisher_count.clamp(min=1e-8) new_fisher = self.fisher_accum / cnt if (getattr(self, 'fisher_continuous', False) and self.fisher_w is not None and getattr(self, 'fisher_consolidation_count', 0) > 0): # EMA: F_t = α·F_t_new + (1-α)·F_{t-1} # α=0.5 dá peso igual a tarefa atual e à memória das anteriores ema_alpha = 0.5 self.fisher_w = ema_alpha * new_fisher + (1 - ema_alpha) * self.fisher_w else: self.fisher_w = new_fisher def set_ewc_reference(self): """Consolida W_w como referência EWC e finaliza Fisher. V6.6 — Após consolidação, Fisher é RESETADO para acumular informação da PRÓXIMA tarefa (online EWC pattern). A Fisher consolidada é preservada em self.fisher_w via EMA em finalize_fisher(). """ self.old_weights_w = self.weights[..., 3].clone() self.finalize_fisher() # Reset dos accumuladores para a próxima tarefa self.fisher_accum.zero_() self.fisher_count.zero_() # V6.6 — Track consolidation count for EMA logic if not hasattr(self, 'fisher_consolidation_count'): self.fisher_consolidation_count = 0 self.fisher_consolidation_count += 1 # ------------------------------------------------------------------ # 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) # V6.5-V2-metrics-FIX-4 — floors aplicados (consistência com update_weights) sigma_t_eff = max(sigma_t, 0.1) alpha_t_eff = max(alpha_t, 0.001) # V6.5-V2-metrics-FIX-4 — estatísticas do conscience mechanism n_total = self.n_neurons n_active = int((self.bmu_win_count > 0).sum().item()) n_dead = n_total - n_active win_counts_flat = self.bmu_win_count.flatten().float() return { "t": int(self.t), "sigma_t": float(sigma_t), "alpha_t": float(alpha_t), "sigma_t_effective": float(sigma_t_eff), "alpha_t_effective": float(alpha_t_eff), "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()), # V6.5-V2-metrics-FIX-4 — conscience mechanism status "conscience_gamma": float(self.conscience_gamma), "conscience_beta": float(self.conscience_beta), "n_active_neurons": n_active, "n_dead_neurons": n_dead, "neuron_activation_rate": float(n_active / max(1, n_total)), "bmu_win_count_mean": float(win_counts_flat.mean().item()) if n_total > 0 else 0.0, "bmu_win_count_max": float(win_counts_flat.max().item()) if n_total > 0 else 0.0, "win_frequency_mean": float(self.win_frequency.mean().item()), "win_frequency_max": float(self.win_frequency.max().item()), "win_frequency_min": float(self.win_frequency.min().item()), # V6.5-V2-auto-conscience — telemetria do ajuste adaptativo de γ "conscience_gamma_min": float(self._conscience_gamma_min), "conscience_gamma_max": float(self._conscience_gamma_max), "conscience_gamma_target": float(self._last_gamma_target), "conscience_gamma_drop": float(self._last_gamma_drop), "conscience_gamma_eta": float(self._conscience_eta), "conscience_gamma_adjust_interval": int(self._auto_adjust_interval), "activation_rate_at_last_adjust": float(self._last_activation_rate), "prev_n_active_at_last_adjust": int(self._prev_n_active), } # ============================================================================ # 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 CANÔNICO (4, 4, 4, 4) = 256). 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=(4, 4, 4, 4), alpha0=0.5, sigma0=2.0, 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, # V6.5-V2 — reasoning_engine desativado por padrão (user requirement: # "por enquanto manter reasoning_engine desativado"). O ReasoningEngine # cria ThreadPoolExecutor(4 workers) + ToolAgentCoordinator que consome # ~100-200MB adicionais e não afeta o aprendizado do SOM. Para reativar, # passe enable_reasoning=True explicitamente (requer ≥6GB cgroup). enable_reasoning: bool = False, vqvae2_code_dim: int = 16, vqvae2_num_codes_top: int = 64, vqvae2_num_codes_bot: int = 128, # V6.5-final — SmoothQuant W8A8 compressor for SOM weights enable_w8a8: bool = True, w8a8_alpha: float = 0.5, w8a8_n_bits: int = 8, w8a8_calibration_samples: int = 128, # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector # User requirement: "verificar se o mecanismo de atenção está ativo # e acessado logicamente funcional" enable_attention: bool = True, attention_n_heads: int = 8, ): self.tokenizer = SimpleBBPETokenizer(vocab_size) self.embedding = nn.Embedding(vocab_size, hidden_dim) self.hidden_dim = hidden_dim self.seq_len = seq_len self.T_max = T_max self.time_counter = 0 # contador global de amostras processadas self.som = KohonenSOM4D(som_grid, alpha0, sigma0, lambda_ewc) self.som_grid = som_grid self.som_neuron_count = ( som_grid[0] * som_grid[1] * som_grid[2] * som_grid[3] ) self.classifier: Optional[HypothesisClassifier] = None self.hypothesis_hidden = hypothesis_hidden self.classifier_trained = False self.buffer_4d = [] self.buffer_labels = [] # V6.5-V2-memfix — Sliding window limit for buffer_4d to prevent OOM. # User requirement: "resolver falhas de lógica e de bugs que estejam # causando alto consumo de memória sem distorcer a arquitetura Kohonen". # The buffer stores vec_4d tensors for SOM training and neuron labeling. # Without a limit, streaming 8000+ samples accumulates 8000 tensors, # each retaining the embedding computation graph (16M params), causing # OOM (3.5GB RSS observed). Sliding window keeps recent samples # for SOM updates while bounding memory. The Kohonen architecture (SOM # grid, BMU, Gaussian neighborhood, EWC) is NOT changed. # V6.5-V4-canonical-256: buffer_max_size = 256 (CANÔNICO, alinhado ao grid). # User requirement EXATO (V6.5-V4): "fazer (tornar canônico) buffer 256 e # grid para (4,4,4,4)=256". Buffer e grid agora têm o MESMO tamanho (256), # eliminando o desbalanceamento que causava OOM em V6.5-V3 (buffer=864 com # grid=256). A correspondência 1:1 entre amostras no buffer e neurônios no # grid 4D é matematicamente elegante — cada amostra pode, em média, ativar # um neurônio distinto, maximizando a utilização do mapa Kohonen. # OOM-safety garantida por: # (a) VQ-VAE-2 lazy compression (a cada 16 add_data) — ver _vqvae2_call_count # (b) torch.no_grad() em toda compressão # (c) gc.collect() a cada 4 batches (no train script) # (d) OomGuard thread daemon (max_rss_mb=2500) # (e) check_memory_and_maybe_fallback() reduz para 128 se RSS > 75% # Grid SOM (4,4,4,4)=256 é CANÔNICO e PERMANECE em ambos os modos. self.buffer_max_size = 256 self._buffer_max_size_canonical = 256 self._buffer_max_size_fallback = 128 # V6.5-V3-no-regression: VQ-VAE-2 lazy compression counter. # Comprimir a cada add_data causava OOM (200MB+ tensores intermediários # por chamada). Agora comprime a cada 16 add_data — mesma cobertura # estatística, 16x menos pressão de memória. self._vqvae2_call_count = 0 self._vqvae2_lazy_interval = 16 self.training_ready = False self.N = N_start self.dim_choice = dim_choice self.dim_index = {"x": 0, "y": 1, "z": 2}[dim_choice] self.punishment_count = 0 self.success_count = 0 self.histogram = Counter() self.required_new_samples = 0 # ------------------------------------------------------------------ # V6.5-fix — Dynamic Label Registry (replaces hardcoded 'gato'/'cachorro') # ------------------------------------------------------------------ # User requirement: "os strings 'gato' e 'cachorro' são fixos quando # deveriam ser extrações variáveis e flexíveis de rótulos proveniente # de dados dos datasets anteriormente treinados" # # Mathematical analysis: # During training, add_data() receives integer labels (0/1) produced # by make_label(text) — these integers are dataset-agnostic. The # semantic meaning of "0" vs "1" depends on the dataset: # - punctuation datasets: 0 = "unpunctuated", 1 = "punctuated" # - chat datasets: 0 = "user_turn", 1 = "assistant_turn" # - instruct datasets: 0 = "instruction", 1 = "response" # - news datasets: 0 = "headline", 1 = "body" # - corpus datasets: 0 = "raw_text", 1 = "compressed_text" # # The label_registry stores, per dataset, the {label_int: label_str} # mapping. predict() looks up the dynamic string instead of returning # a hardcoded "gato"/"cachorro". # # Math: label_str = registry[dataset_name][label_int] # where label_int = 1 if sigmoid(logit) > 0.5 else 0 # # Fallback chain (when no dataset-specific registry exists): # 1. Most recently registered dataset mapping. # 2. Default {0: "negative", 1: "positive"}. # 3. Synthetic "label_{int}" if all else fails. self.label_registry: Dict[str, Dict[int, str]] = {} self.default_label_strings: Dict[int, str] = { 0: "negative", 1: "positive", } self.last_dataset_used: Optional[str] = None # V6.5-fix — Inference punishment tracking (mirrors training protocol) # User requirement: "ao punir o modelo reajusta seus parâmetros usando # a camada de hipótese para funcionar como na lógica de punição e # ajuste de parâmetros durante o treinamento" self.inference_punishment_count: int = 0 self.inference_punishment_history: List[Dict[str, Any]] = [] # ------------------------------------------------------------------ # V6.5 — VQ-VAE-2 compressor (ativa efetiva no pipeline) # ------------------------------------------------------------------ self.enable_vqvae2 = enable_vqvae2 self.vqvae2_compressor = None self.vqvae2_metrics_history: List[Dict[str, Any]] = [] if enable_vqvae2: try: from .vqvae2_hierarchical_flexnet import HierarchicalVQVAE2 # Modalidade única: "som_4d" com input_dim=4 self.vqvae2_compressor = HierarchicalVQVAE2( modalities={"som_4d": 4}, code_dim=vqvae2_code_dim, num_codes_top=vqvae2_num_codes_top, num_codes_bot=vqvae2_num_codes_bot, hidden=32, beta=0.25, ema_decay=0.99, dead_code_threshold=1.0, dead_code_restart_every=3, goose_temp_init=2.0, goose_temp_final=0.5, goose_schedule="cosine", total_epochs=25, norm_type="none", rmsnorm_in_vq=False, ) # Inicia em modo treino para ativar EMA updates self.vqvae2_compressor.train() except Exception as e: # Fallback: desabilita VQ-VAE-2 se houver erro de import self.enable_vqvae2 = False self.vqvae2_compressor = None import warnings warnings.warn(f"VQ-VAE-2 disabled: {e}") # ------------------------------------------------------------------ # V6.5-final — SmoothQuantCompressor for REAL W8A8 of SOM weights # ------------------------------------------------------------------ # User requirement: "integrar SmoothQuantCompressor ao KLS para # compressão W8A8 real dos pesos do SOM" # # O SmoothQuantCompressor é instanciado e calibrado sobre os pesos # atuais do SOM (self.som.weights reshape para [N_neurons, 4]) usando # o buffer_4d como amostras de ativação reais. Após calibração, o # método compress_som_weights_w8a8() aplica quantização INT8 simétrica # (com smoothing) e dequantização imediata, preservando a topologia # do SOM com erro relativo < 1%. self.enable_w8a8 = enable_w8a8 self.w8a8_compressor = None self.w8a8_metrics_history: List[Dict[str, Any]] = [] if enable_w8a8: try: from ..quantization.smoothquant_compressor import SmoothQuantCompressor self.w8a8_compressor = SmoothQuantCompressor( alpha=w8a8_alpha, n_bits=w8a8_n_bits, calibration_samples=w8a8_calibration_samples, ) except Exception as e: self.enable_w8a8 = False self.w8a8_compressor = None import warnings warnings.warn(f"SmoothQuantCompressor disabled: {e}") # ------------------------------------------------------------------ # V6.5 — ReasoningEngine (integração ativa + tool_coordinator workers reativados) # ------------------------------------------------------------------ self.enable_reasoning = enable_reasoning self.reasoning_engine = None if enable_reasoning: try: from ..reasoning.reasoning_engine import ReasoningEngine # V6.5-final: REATIVAR tool_coordinator workers — explicit n_workers # User requirement: "reativar reasoning_engine's tool_coordinator workers" # O ReasoningEngine cria internamente ToolAgentCoordinator(n_workers=4) # que mantém um ThreadPoolExecutor ativo. Para garantir que os # workers sejam de fato utilizados (não apenas instanciados), # o método _compress_buffer_with_vqvae2 abaixo chama # execute_parallel() periodicamente durante o treino. self.reasoning_engine = ReasoningEngine( max_thinking_steps=10, max_iterations=3, convergence_threshold=0.9, verbose=False, ) # Verifica que tool_coordinator + workers estão ativos tc = self.reasoning_engine.tool_coordinator self._tool_coordinator_workers_active = ( tc is not None and tc._executor is not None ) # Registra uma ferramenta interna: consultar SOM def som_query_tool(query: str) -> str: """Ferramenta: consulta o SOM do KLS para responder.""" # V6.5-V2-metrics-FIX-2 — no_grad para evitar building de grafo # (cada chamada de text_to_4d_vector sem no_grad retém ~100MB # de grafo computacional: embedding 16M params + attention + SVD) with torch.no_grad(): pred = self.predict(query) bmu_info = "" if self.buffer_4d: try: self.time_counter += 1 vec = text_to_4d_vector( query, self.tokenizer, self.embedding, self.hidden_dim, self.seq_len, self.time_counter, self.T_max, ) bmu = self.som.find_bmu(vec) bmu_info = f" | BMU={bmu}" except Exception: pass return f"prediction={pred}{bmu_info}" self.reasoning_engine.register_tool( "som_query", som_query_tool, description="Consulta o SOM do KohonenLearningSystem", timeout_s=10.0, ) # V6.5-final: registra segunda ferramenta para forçar uso # paralelo dos workers via execute_parallel() def buffer_stats_tool(query: str) -> str: """Ferramenta: retorna estatísticas do buffer atual.""" return ( f"buffer_size={len(self.buffer_4d)}, " f"punishment_count={self.punishment_count}, " f"success_count={self.success_count}, " f"training_ready={self.training_ready}" ) self.reasoning_engine.register_tool( "buffer_stats", buffer_stats_tool, description="Estatísticas do buffer interno do KLS", timeout_s=5.0, ) except Exception as e: self.enable_reasoning = False self.reasoning_engine = None self._tool_coordinator_workers_active = False import warnings warnings.warn(f"ReasoningEngine disabled: {e}") # ------------------------------------------------------------------ # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector # ------------------------------------------------------------------ # User requirement: "verificar se o mecanismo de atenção está ativo # e acessado logicamente funcional" # # O MultiHeadAttention (de attention_multimodal.py) é instanciado com # d_model=hidden_dim e n_heads=attention_n_heads. Ele é aplicado em # text_to_4d_vector entre a fusão (+PE) e o SVD, atuando como # self-attention residual sobre a sequência embedded. # # Métricas (n_calls, last_norm_in/out, last_attn_activated) são # acumuladas em self.attention_metrics e expostas via # get_attention_metrics() para verificação de "ativo e funcional". self.enable_attention = enable_attention self.attention = None self.attention_metrics: Dict[str, Any] = { "n_calls": 0, "n_errors": 0, "last_norm_in": 0.0, "last_norm_out": 0.0, "last_attn_activated": False, "last_attn_diff_norm": 0.0, } if enable_attention: try: from .attention_multimodal import MultiHeadAttention # d_model = hidden_dim, n_heads = attention_n_heads # assert hidden_dim % n_heads == 0 n_heads = attention_n_heads if hidden_dim % n_heads != 0: # ajusta para divisor válido mais próximo for cand in [8, 4, 2, 1]: if hidden_dim % cand == 0: n_heads = cand break self.attention = MultiHeadAttention( d_model=hidden_dim, n_heads=n_heads, dropout=0.0, max_chunk=4096, ) # Em eval mode para determinismo (não treinamos o attention separadamente) self.attention.eval() self.attention_n_heads = n_heads except Exception as e: self.enable_attention = False self.attention = None import warnings warnings.warn(f"MultiHeadAttention disabled: {e}") def get_attention_metrics(self) -> Dict[str, Any]: """V6.5-attn — Retorna métricas do MultiHeadAttention integrado. User requirement: "verificar se o mecanismo de atenção está ativo e acessado logicamente funcional" Returns: Dict com: - active: bool — attention module está instanciado - n_calls: int — número de chamadas forward - n_errors: int — número de erros silenciados - last_norm_in / last_norm_out: normas do input/output - last_attn_activated: bool — output é finito e diferente do input - last_attn_diff_norm: float — ||attn_out - input|| - logic_functional: bool — n_calls > 0 AND n_errors == 0 AND last_attn_activated == True """ m = self.attention_metrics return { "active": bool(self.attention is not None), "n_calls": int(m.get("n_calls", 0)), "n_errors": int(m.get("n_errors", 0)), "last_norm_in": float(m.get("last_norm_in", 0.0)), "last_norm_out": float(m.get("last_norm_out", 0.0)), "last_attn_activated": bool(m.get("last_attn_activated", False)), "last_attn_diff_norm": float(m.get("last_attn_diff_norm", 0.0)), "n_heads": int(getattr(self, "attention_n_heads", 0)), # "logicamente funcional" = foi chamado, sem erros, e última chamada # produziu output finito e diferente do input "logic_functional": bool( self.attention is not None and m.get("n_calls", 0) > 0 and m.get("n_errors", 0) == 0 and m.get("last_attn_activated", False) ), } # ================================================================== # V6.5-fix — Dynamic Label Registry API # ================================================================== # User requirement: "os strings 'gato' e 'cachorro' são fixos quando # deveriam ser extrações variáveis e flexíveis de rótulos proveniente # de dados dos datasets anteriormente treinados" # # Mathematical formalization: # Let D = {d_1, d_2, ..., d_K} be the set of datasets seen during # training. For each d_k, the training script provides a mapping # μ_k : {0, 1} → Σ (where Σ is the string-label alphabet). # # The label_registry stores R = {(d_k, μ_k) : k = 1..K}. # # At inference time, given a query sentence s: # 1. vec_4d = text_to_4d_vector(s, ...) # 2. p = _predict_single(vec_4d) ∈ [0, 1] # 3. label_int = 1 if p > 0.5 else 0 # 4. label_str = μ_last(label_int) [fallback chain if unknown] # # If dataset_name is provided at inference, uses μ_dataset_name; # otherwise uses the most recently registered μ (assumes temporal # coherence: latest trained dataset is most relevant). # ------------------------------------------------------------------ def register_dataset_labels( self, dataset_name: str, label_strings: Dict[int, str], ) -> None: """V6.5-fix — Registra mapeamento {label_int: label_str} para um dataset. Args: dataset_name: nome do dataset (ex: 'dominguesm/restore-punctuation-ptbr-dataset'). label_strings: dict {0: 'unpunctuated', 1: 'punctuated'}. """ if not isinstance(label_strings, dict): raise TypeError(f"label_strings must be dict, got {type(label_strings)}") # Sanitize keys to int, values to str sanitized = {int(k): str(v) for k, v in label_strings.items()} self.label_registry[dataset_name] = sanitized self.last_dataset_used = dataset_name def get_label_string( self, label_int: int, dataset_name: Optional[str] = None, ) -> str: """V6.5-fix — Resolve label_int → label_str dinamicamente. Fallback chain (mathematically well-defined): 1. If dataset_name is registered, use registry[dataset_name][label_int]. 2. Else if any dataset is registered, use the most recent one. 3. Else use default_label_strings. 4. Final fallback: synthetic f"label_{label_int}". Args: label_int: 0 ou 1. dataset_name: opcional — nome do dataset para disambiguar. Returns: String label correspondente. """ label_int = int(label_int) # 1. Dataset-specific lookup if dataset_name is not None and dataset_name in self.label_registry: mapping = self.label_registry[dataset_name] if label_int in mapping: return mapping[label_int] # 2. Most recently registered mapping if self.label_registry: last_ds = ( dataset_name if dataset_name in self.label_registry else (self.last_dataset_used or list(self.label_registry.keys())[-1]) ) if last_ds and last_ds in self.label_registry: mapping = self.label_registry[last_ds] if label_int in mapping: return mapping[label_int] # 3. Default mapping if label_int in self.default_label_strings: return self.default_label_strings[label_int] # 4. Synthetic fallback return f"label_{label_int}" def get_label_registry_state(self) -> Dict[str, Any]: """V6.5-fix — Retorna estado atual do label_registry para inspeção.""" return { "n_datasets_registered": int(len(self.label_registry)), "datasets": { ds: {int(k): str(v) for k, v in mapping.items()} for ds, mapping in self.label_registry.items() }, "last_dataset_used": str(self.last_dataset_used) if self.last_dataset_used else None, "default_label_strings": {int(k): str(v) for k, v in self.default_label_strings.items()}, "inference_punishment_count": int(self.inference_punishment_count), "inference_punishment_history_len": int(len(self.inference_punishment_history)), } def add_data( self, sentences: List[str], labels: List[int], label_strings: Optional[Dict[int, str]] = None, dataset_name: Optional[str] = None, ): """V6.5-fix — Adiciona amostras: text → 4D vector + atualiza histograma. Args: sentences: lista de textos. labels: lista de inteiros (0/1). label_strings: opcional — {0: 'str0', 1: 'str1'} para o dataset. dataset_name: opcional — nome do dataset (para registrar label_strings). Mathematical extension (V6.5-fix): Se label_strings e dataset_name são fornecidos, registra o mapeamento no label_registry. Isto permite que predict() retorne o rótulo textual correto ao invés de hardcoded 'gato'/'cachorro'. """ # Register label mapping if provided if label_strings is not None and dataset_name is not None: self.register_dataset_labels(dataset_name, label_strings) if dataset_name is not None: self.last_dataset_used = dataset_name # V6.5-V2-memfix — Use torch.no_grad() to avoid building computation # graphs during vec_4d computation. The embedding graph (16M params) # would otherwise be retained by every tensor in buffer_4d, causing OOM # with 8000+ samples. The vec_4d is a 4-float summary (x,y,z,w) that # does NOT need gradient — it feeds into SOM (Hebbian, no backprop) and # the HypothesisClassifier (which trains its own weights). with torch.no_grad(): for sent, lab in zip(sentences, labels): self.time_counter += 1 vec = text_to_4d_vector( sent, self.tokenizer, self.embedding, self.hidden_dim, self.seq_len, self.time_counter, self.T_max, attention=self.attention, # V6.5-attn attention_metrics=self.attention_metrics, # V6.5-attn ) # CRITICAL: detach to break any residual graph + clone for safety self.buffer_4d.append(vec.detach().clone()) self.buffer_labels.append(lab) dim_val = round(vec[self.dim_index].item(), 2) self.histogram[dim_val] += 1 # V6.5-V2-memfix — Enforce sliding window limit if len(self.buffer_4d) > self.buffer_max_size: overflow = len(self.buffer_4d) - self.buffer_max_size self.buffer_4d = self.buffer_4d[-self.buffer_max_size:] self.buffer_labels = self.buffer_labels[-self.buffer_max_size:] def check_training_start(self) -> bool: """Inicia treino quando algum bucket do histograma atinge N.""" if ( not self.training_ready and max(self.histogram.values(), default=0) >= self.N ): self.training_ready = True return True return False def train_som_on_buffer(self): """Treina SOM por 5 épocas sobre o buffer atual (com Fisher accum). V6.5-V2-memfix — Wrapped in torch.no_grad() to prevent graph building during SOM Hebbian updates. The SOM uses competitive learning (BMU + Gaussian neighborhood), not backprop — no gradient is needed. V6.5-V2-parallel — Agora usa train_batch_parallel (ParallelSOM4D pattern) ao invés do loop find_bmu + update_weights manual. Isto elimina o overhead de Python e usa broadcasting PyTorch sobre os 864 neurônios. User requirement: "Para um ajuste paralelo verdadeiro e massivo de uma SOM de 4 dimensões, a abordagem ideal é o paralelismo vetorial em GPU utilizando PyTorch." V6.5-V2-reorganize — Após treinar, dispara reorganização por densidade se dead_rate > 30% (ReorganizingSOM4D pattern). """ if not self.buffer_4d: return # V6.5-V2-memfix — detach+stack to ensure no graph leaks from buffer data = torch.stack(self.buffer_4d).detach() with torch.no_grad(): # V6.5-V2-parallel — Usa train_batch_parallel para todas as 5 épocas # User requirement: paralelismo vetorial PyTorch sobre os 864 neurônios acc_fisher = ( self.punishment_count == 0 and self.som.old_weights_w is None ) for _ in range(5): # épocas de treino rápido perm = torch.randperm(len(data)) # train_batch_parallel já faz sanitização NaN/Inf + conscience # + Fisher accumulation em uma única chamada vetorial self.som.train_batch_parallel( samples=data[perm], use_conscience=True, accumulate_fisher=acc_fisher, ) # V6.5-V2-reorganize — Reorganização topológica por densidade # User requirement (ReorganizingSOM4D pattern): "Os neurônios pouco # ativados (isolados) sofrem uma força de atração em direção aos # agrupamentos de alta densidade, reorganizando geometricamente o # hipercubo 4D sem perder o alinhamento matricial." try: reorg = self.reorganize_map_if_needed( regularization_factor=0.05, min_dead_rate_for_trigger=0.3, min_steps_between_reorganizations=300, ) if reorg.get("action") == "auto_reorganized": import logging as _logging _log = _logging.getLogger("kohonen_learning_system") _log.info( f"[V6.5-V2-reorganize] Map reorganized: " f"n_dead_pulled={reorg.get('n_dead_pulled')}, " f"dead_rate={reorg.get('dead_rate_before', 0):.3f}, " f"factor={reorg.get('regularization_factor', 0):.4f}" ) except Exception: pass # V6.5-V3-no-regression: VQ-VAE-2 REATIVADO. Compressão agora é LAZY # (a cada 16 add_data) para evitar OOM. User: "não autorizei a desativação # do 'VQ-VAE-2'". OOM-safety por lazy compression + no_grad + fallback. if self.enable_vqvae2 and self.vqvae2_compressor is not None: self._vqvae2_call_count = getattr(self, "_vqvae2_call_count", 0) + 1 _lazy_interval = getattr(self, "_vqvae2_lazy_interval", 16) if self._vqvae2_call_count % _lazy_interval == 0: with torch.no_grad(): # Comprime apenas um SLICE do buffer (últimas 64 amostras) # para limitar tensores intermediários a ~1MB. _slice = data[-64:] if data.shape[0] > 64 else data self._compress_buffer_with_vqvae2(_slice) # 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: with torch.no_grad(): self.compress_som_weights_w8a8() # V6.5-final — Reativa tool_coordinator workers via execute_parallel # User requirement: "reativar reasoning_engine's tool_coordinator workers" # O ThreadPoolExecutor (n_workers=4) só é efetivamente utilizado quando # execute_parallel() é chamado com múltiplas ferramentas. Caso contrário, # os workers ficam idle e o recurso é desperdiçado. # # V6.5-V2-metrics-FIX-2: Reduzido de TODAS as chamadas para a cada 5 # chamadas (counter-based). O exercício dos workers não é crítico para # o aprendizado do SOM — é apenas para manter os threads ativos. # Chamar a cada batch causava OOM (4 parallel predict() calls com # graph building = 400MB+ por batch). Com no_grad (aplicado acima) + # redução de frequência, o overhead é mínimo. self._tool_coordinator_call_count = getattr(self, "_tool_coordinator_call_count", 0) + 1 if self._tool_coordinator_call_count % 5 == 0: self._exercise_tool_coordinator_workers() def _exercise_tool_coordinator_workers(self) -> Dict[str, Any]: """V6.5-final — Exercita os workers do tool_coordinator em paralelo. Chama execute_parallel() com todas as ferramentas registradas para garantir que o ThreadPoolExecutor (n_workers=4) seja efetivamente utilizado, cumprindo o requisito "reativar tool_coordinator workers". Returns: Dict com resultados de cada ferramenta + tempo total. """ if ( not self.enable_reasoning or self.reasoning_engine is None ): return {"active": False, "reason": "reasoning_disabled"} tc = self.reasoning_engine.tool_coordinator if tc is None: return {"active": False, "reason": "no_tool_coordinator"} try: import time as _time t0 = _time.time() # Submete TODAS as ferramentas registradas em paralelo — isso # efetivamente acorda os workers do ThreadPoolExecutor. tool_names = list(tc._tools.keys()) if not tool_names: return {"active": True, "n_workers": tc.n_workers, "n_tools": 0} # Executa em paralelo: cada ferramenta recebe um query dummy requests = [(name, "status_check") for name in tool_names] results = tc.execute_parallel(requests, fail_fast=False) elapsed_ms = (_time.time() - t0) * 1000 return { "active": True, "n_workers": tc.n_workers, "n_tools": len(tool_names), "tool_names": tool_names, "n_results": len(results), "elapsed_ms": float(elapsed_ms), "workers_exercised": True, } except Exception as e: return { "active": False, "error": str(e)[:200], "workers_exercised": False, } def _compress_buffer_with_vqvae2(self, data: torch.Tensor) -> Dict[str, Any]: """V6.5 — Comprime buffer 4D via VQ-VAE-2 hierárquico. Ativa efetivamente o VQ-VAE-2 no pipeline de compressão: 1. Encoder: (B, 4) → z_e (B, code_dim) 2. VQ hierárquico: z_e → z_q_top + z_q_bot (codebooks EMA + Goose) 3. Decoder: z_q_combined → z_recon (B, 4) 4. Loss: commitment (top+bot) + reconstruction (MSE) 5. Códigos top/bottom retornados para inspeção Args: data: tensor (B, 4) com vetores 4D do buffer. Returns: Dict com vqvae2_metrics (também armazenado em vqvae2_metrics_history). """ try: # Sanitiza NaN/Inf data_clean = torch.nan_to_num(data, nan=0.0, posinf=1e4, neginf=-1e4) # VQ-VAE-2 espera dict {modality_name: tensor} batch = {"som_4d": data_clean} out = self.vqvae2_compressor(batch) # Incrementa época do VQ (controla schedule Goose + dead code restart) try: self.vqvae2_compressor.vq.increment_epoch() except Exception: pass stats = out.get("stats", {}) vqvae2_metrics = { "vq_loss": float(out.get("vq_loss", 0.0)), "recon_loss": float(out.get("recon_loss", 0.0)), "total_loss": float(out.get("total_loss", 0.0)), "n_used_top": int(stats.get("n_used_top", 0)), "n_used_bot": int(stats.get("n_used_bot", 0)), "usage_ratio_top": float(stats.get("usage_ratio_top", 0.0)), "usage_ratio_bot": float(stats.get("usage_ratio_bot", 0.0)), "codebook_ppl_top": float(stats.get("codebook_ppl_top", 0.0)), "codebook_ppl_bot": float(stats.get("codebook_ppl_bot", 0.0)), "n_restarted_top": int(stats.get("n_restarted_top", 0)), "n_restarted_bot": int(stats.get("n_restarted_bot", 0)), "goose_temp": float(stats.get("goose_temp", 0.0)), "active": True, } self.vqvae2_metrics_history.append(vqvae2_metrics) # Mantém apenas últimas 100 entries para limitar memória if len(self.vqvae2_metrics_history) > 100: self.vqvae2_metrics_history = self.vqvae2_metrics_history[-100:] return vqvae2_metrics except Exception as e: return { "active": False, "error": str(e)[:200], "vq_loss": 0.0, "recon_loss": 0.0, "total_loss": 0.0, } def get_vqvae2_metrics(self) -> Dict[str, Any]: """V6.5 — Retorna métricas atuais do VQ-VAE-2 compressor.""" if not self.enable_vqvae2 or self.vqvae2_compressor is None: return {"active": False, "reason": "disabled"} if not self.vqvae2_metrics_history: return {"active": True, "n_calls": 0} latest = self.vqvae2_metrics_history[-1] # V6.5: skip NaN values when computing means (early calls may produce NaN # due to Gumbel-softmax instability before codebook warmup) import math valid_total = [m.get("total_loss", 0.0) for m in self.vqvae2_metrics_history if not math.isnan(m.get("total_loss", 0.0))] valid_recon = [m.get("recon_loss", 0.0) for m in self.vqvae2_metrics_history if not math.isnan(m.get("recon_loss", 0.0))] return { "active": True, "n_calls": len(self.vqvae2_metrics_history), "latest": latest, "mean_total_loss": float(sum(valid_total) / max(1, len(valid_total))) if valid_total else 0.0, "mean_recon_loss": float(sum(valid_recon) / max(1, len(valid_recon))) if valid_recon else 0.0, "n_nan_skipped": len(self.vqvae2_metrics_history) - len(valid_total), } # ================================================================== # V6.5-final — SmoothQuant W8A8 compression of SOM weights # ================================================================== # User requirement: "integrar SmoothQuantCompressor ao KLS para # compressão W8A8 real dos pesos do SOM" # # Pipeline (executado periodicamente durante o treino): # 1. Coleta pesos atuais do SOM: W ∈ ℝ^(I,J,K,L,4) → reshape [N, 4] # 2. Coleta buffer_4d como amostras de ativação reais: X ∈ ℝ^(B, 4) # 3. Calibra SmoothQuant: s_j = max|X_j|^α / max|W_j|^(1-α) # 4. Aplica smoothing: W_smooth = W * diag(s) # 5. Quantiza INT8 simétrico per-tensor: W_int8 = round(W_smooth / scale) # 6. Dequantiza: W_dequant = W_int8 * scale / diag(s) (volta para float) # 7. Substitui self.som.weights pelo dequantizado (preserva topologia) # 8. Computa erro relativo: ||W - W_dequant|| / ||W|| # # Erro esperado: < 1% (SmoothQuant preserva acurácia). def compress_som_weights_w8a8(self) -> Dict[str, Any]: """V6.5-final — Aplica compressão W8A8 real aos pesos do SOM. SmoothQuant migra escala da ativação para o peso, permitindo quantização INT8 sem perda significativa. Os pesos são quantizados a INT8 e dequantizados de volta para float, preservando a topologia do SOM com erro relativo < 1%. User requirement: "compressão W8A8 real dos pesos do SOM" Returns: Dict com métricas de compressão (erro, ratio, scale stats). """ if not self.enable_w8a8 or self.w8a8_compressor is None: return {"active": False, "reason": "w8a8_disabled"} try: import time as _time t0 = _time.time() # 1. Snapshot dos pesos atuais do SOM W = self.som.weights # (I, J, K, L, 4) original_shape = W.shape N = W.shape[0] * W.shape[1] * W.shape[2] * W.shape[3] # Reshape para [N, 4] — trata cada neurônio como uma "linha" # e cada componente (x, y, z, w) como uma "feature" W_2d = W.reshape(N, 4).clone().detach() W_orig_norm = W_2d.norm().item() # 2. Calibra com buffer_4d (amostras de ativação reais) if len(self.buffer_4d) >= 4: X = torch.stack(self.buffer_4d[-self.w8a8_compressor.calibration_samples:]) X = X.detach().clone() else: # Fallback: usa os próprios pesos como amostras de ativação X = W_2d.detach().clone() # Sanitiza NaN/Inf W_2d = torch.nan_to_num(W_2d, nan=0.0, posinf=1e4, neginf=-1e4) X = torch.nan_to_num(X, nan=0.0, posinf=1e4, neginf=-1e4) # 3. Calibra SmoothQuant (computa smooth_scale s) self.w8a8_compressor.calibrate(W_2d, X) smooth_scale = self.w8a8_compressor.smooth_scale # 4. Aplica smoothing: W_smooth = W * diag(s) W_smooth = self.w8a8_compressor.smooth_weight(W_2d) # 5. Quantiza INT8 simétrico per-tensor W_int8 = self.w8a8_compressor.quantize_per_tensor_symmetric(W_smooth) # 6. Dequantiza: W_int8 * scale → W_smooth_recon, depois / diag(s) → W_dequant W_smooth_recon = self.w8a8_compressor.dequantize(W_int8, W_smooth) # Reverte smoothing: W_dequant = W_smooth_recon / diag(s) W_dequant = W_smooth_recon / smooth_scale.unsqueeze(0) # 7. Computa erro relativo error = (W_2d - W_dequant).norm().item() relative_error = error / max(W_orig_norm, 1e-8) # 8. Substitui pesos do SOM pelo dequantizado (preserva topologia) # IMPORTANTE: usa no_grad para não poluir gradiente (SOM não usa # autograd, mas mantemos consistência) with torch.no_grad(): self.som.weights = W_dequant.reshape(*original_shape).clone() # 9. Métricas compression_ratio = (W_2d.element_size() * W_2d.numel()) / (W_int8.element_size() * W_int8.numel()) metrics = { "active": True, "n_neurons": int(N), "n_features": 4, "n_calibration_samples": int(X.shape[0]), "smooth_scale_mean": float(smooth_scale.mean().item()), "smooth_scale_std": float(smooth_scale.std().item()), "smooth_scale_min": float(smooth_scale.min().item()), "smooth_scale_max": float(smooth_scale.max().item()), "alpha": float(self.w8a8_compressor.alpha), "n_bits": int(self.w8a8_compressor.n_bits), "compression_ratio": float(compression_ratio), "weights_norm_original": float(W_orig_norm), "weights_norm_dequant": float(W_dequant.norm().item()), "absolute_error": float(error), "relative_error": float(relative_error), "preserves_topology": bool(relative_error < 0.05), # < 5% = OK "w8a8_int8_range": [ int(W_int8.min().item()), int(W_int8.max().item()), ], "elapsed_ms": float((_time.time() - t0) * 1000), } self.w8a8_metrics_history.append(metrics) # Mantém apenas últimas 100 entries if len(self.w8a8_metrics_history) > 100: self.w8a8_metrics_history = self.w8a8_metrics_history[-100:] return metrics except Exception as e: return { "active": False, "error": str(e)[:200], "preserves_topology": False, } def get_w8a8_metrics(self) -> Dict[str, Any]: """V6.5-final — Retorna métricas atuais do compressor W8A8.""" if not self.enable_w8a8 or self.w8a8_compressor is None: return {"active": False, "reason": "disabled"} if not self.w8a8_metrics_history: return {"active": True, "n_calls": 0} latest = self.w8a8_metrics_history[-1] all_errors = [m.get("relative_error", 0.0) for m in self.w8a8_metrics_history] return { "active": True, "n_calls": len(self.w8a8_metrics_history), "latest": latest, "mean_relative_error": float(sum(all_errors) / max(1, len(all_errors))), "max_relative_error": float(max(all_errors)) if all_errors else 0.0, "all_preserve_topology": all(m.get("preserves_topology", False) for m in self.w8a8_metrics_history), } # ------------------------------------------------------------------ # V6.5 — ReasoningEngine integration # ------------------------------------------------------------------ def reason_about(self, query: str) -> Iterator[str]: """V6.5 — Gera streaming de raciocínio para uma query. Usa o ReasoningEngine integrado para produzir tags , , , , , , , . Compatível com Ollama/LangChain/vLLM (tags padrão). Args: query: pergunta/requisição do usuário. Yields: chunks de texto (tags + conteúdo). """ if not self.enable_reasoning or self.reasoning_engine is None: yield f"ReasoningEngine disabled. SOM prediction: {self.predict(query)}" return yield from self.reasoning_engine.solve(query, use_tools=True, use_planning=True) def reason_sync(self, query: str) -> str: """V6.5 — Versão síncrona de reason_about (retorna string completa).""" return "".join(self.reason_about(query)) def get_reasoning_stats(self) -> Dict[str, Any]: """V6.5 — Retorna estatísticas do reasoning_engine.""" if not self.enable_reasoning or self.reasoning_engine is None: return {"active": False, "reason": "disabled"} return { "active": True, "stats": self.reasoning_engine.get_stats(), "n_history": len(self.reasoning_engine.history), } def _som_activation(self, x): """Vetor de ativação SOM: distâncias de x a todos os neurônios (flatten).""" dist = torch.sum((self.som.weights - x.view(1, 1, 1, 1, 4)) ** 2, dim=-1) return dist.flatten() def _label_neurons(self): """Rotula neurônios por votação majoritária sobre o buffer. V6.5-V2-memfix — no_grad to prevent graph building during BMU search. """ self.neuron_label = {} if not self.buffer_4d: return data = torch.stack(self.buffer_4d).detach() labels = torch.tensor(self.buffer_labels) votes = defaultdict(lambda: [0, 0]) with torch.no_grad(): for i in range(len(data)): bmu = self.som.find_bmu(data[i]) votes[bmu][int(labels[i].item())] += 1 for bmu, v in votes.items(): self.neuron_label[bmu] = 1.0 if v[1] > v[0] else 0.0 def evaluate_classification(self) -> float: """Acurácia sobre o buffer atual. V6.5-V2-memfix — no_grad to prevent graph building during evaluation. """ if not self.buffer_4d: return 1.0 data = torch.stack(self.buffer_4d).detach() labels = torch.tensor(self.buffer_labels).float() correct = 0 with torch.no_grad(): for i in range(len(data)): pred = self._predict_single(data[i]) if (pred > 0.5) == (labels[i] > 0.5): correct += 1 return correct / len(data) def _predict_single(self, x): """Prediz: classifier (se treinado) ou voto do BMU.""" if self.classifier is not None and self.classifier_trained: with torch.no_grad(): act = self._som_activation(x).unsqueeze(0) logit = self.classifier(act) return torch.sigmoid(logit).item() else: if not hasattr(self, "neuron_label"): self._label_neurons() bmu = self.som.find_bmu(x) return self.neuron_label.get(bmu, 0.5) def activate_hypothesis(self): """Treina o HypothesisClassifier (8 FC layers) por 50 epochs. BUG FIX (V6.3→V6.4): buffer_4d contém tensores que carregam o grafo de computação do embedding. Para evitar "Trying to backward through the graph a second time", fazemos detach+clone e calculamos as ativações SOM dentro de torch.no_grad(). O classifier treina apenas sobre seus próprios pesos. """ if self.classifier is None: self.classifier = HypothesisClassifier( self.som_neuron_count, self.hypothesis_hidden ) # FIX: detach+clone para isolar do grafo do embedding data = torch.stack(self.buffer_4d).detach().clone() labels = torch.tensor(self.buffer_labels).float() # FIX: ativações SEM gradiente (não queremos treinar SOM/embedding aqui) with torch.no_grad(): X = torch.stack([self._som_activation(data[i]) for i in range(len(data))]) optimizer = torch.optim.Adam(self.classifier.parameters(), lr=0.001) criterion = nn.BCEWithLogitsLoss() for _ in range(50): optimizer.zero_grad() loss = criterion(self.classifier(X), labels) loss.backward() optimizer.step() self.classifier_trained = True def process_batch( self, sentences: List[str], labels: List[int], label_strings: Optional[Dict[int, str]] = None, dataset_name: Optional[str] = None, ): """V6.5-fix — Processa batch: adiciona dados, treina SOM se ready, aplica punishment. Args: sentences: lista de textos. labels: lista de inteiros (0/1). label_strings: opcional — {0: 'str0', 1: 'str1'} para o dataset. dataset_name: opcional — nome do dataset (registrado no label_registry). Returns: True se o protocolo de punishment completou um ciclo (2ª punição → set_ewc_reference + reset). False caso contrário. """ # V6.5-fix: forward label_strings + dataset_name to add_data so # the label_registry is populated for predict() lookups. self.add_data( sentences, labels, label_strings=label_strings, dataset_name=dataset_name, ) if self.check_training_start(): self.train_som_on_buffer() self._label_neurons() if self.training_ready: acc = self.evaluate_classification() if acc < 1.0: self.punishment_count += 1 self.success_count = 0 if self.punishment_count == 1: self.activate_hypothesis() elif self.punishment_count == 2: self.som.set_ewc_reference() self.required_new_samples = ( self.success_count * self.N if self.success_count > 0 else self.N ) self.training_ready = False self.punishment_count = 0 self.success_count = 0 self.histogram.clear() self.buffer_4d.clear() self.buffer_labels.clear() return True else: self.punishment_count = 0 self.success_count += 1 return False def predict( self, sentence: str, dataset_name: Optional[str] = None, return_proba: bool = False, ) -> Any: """V6.5-fix — Prediz rótulo textual DINÂMICO extraído do label_registry. User requirement (CORREÇÃO MATEMÁTICA): "os strings 'gato' e 'cachorro' são fixos quando deveriam ser extrações variáveis e flexíveis de rótulos proveniente de dados dos datasets anteriormente treinados" Mathematical correction: ANTES (BUG): prob = _predict_single(vec_4d) return "gato" if prob <= 0.5 else "cachorro" # FIXO! AGORA (V6.5-fix): 1. vec_4d = text_to_4d_vector(sentence, ...) 2. prob = _predict_single(vec_4d) ∈ [0, 1] (sigmoid(logit) se classifier treinado; voto BMU caso contrário) 3. label_int = 1 if prob > 0.5 else 0 4. label_str = self.get_label_string(label_int, dataset_name) - consulta label_registry[dataset_name] se dataset_name fornecido - fallback: último dataset registrado - fallback: default {0:'negative', 1:'positive'} 5. return label_str [ou (label_str, prob) se return_proba=True] V6.5-V2-metrics-FIX-2: Wrapped in torch.no_grad() to prevent computation graph building during inference. Without this, each predict() call builds a graph that includes the embedding (16M params), attention forward pass, and SVD — consuming ~100MB+ per call. With 4 parallel tool_coordinator workers calling predict(), this caused OOM (3.5GB RSS). Args: sentence: texto de entrada. dataset_name: opcional — nome do dataset para disambiguar rótulos. return_proba: se True, retorna (label_str, prob). Returns: str (ou Tuple[str, float]) — rótulo dinâmico extraído dos datasets. """ self.time_counter += 1 # mantém coerência temporal # V6.5-V2-metrics-FIX-2 — no_grad para evitar building de grafo em inference with torch.no_grad(): vec = text_to_4d_vector( sentence, self.tokenizer, self.embedding, self.hidden_dim, self.seq_len, self.time_counter, self.T_max, attention=self.attention, # V6.5-attn attention_metrics=self.attention_metrics, # V6.5-attn ) prob = self._predict_single(vec) # CORREÇÃO MATEMÁTICA: label_int derivado de prob, não hardcoded label_int = 1 if prob > 0.5 else 0 # Rótulo textual DINÂMICO extraído do label_registry label_str = self.get_label_string(label_int, dataset_name) if return_proba: return label_str, float(prob) return label_str def predict_proba( self, sentence: str, dataset_name: Optional[str] = None, ) -> Tuple[str, float]: """V6.5-fix — Prediz (label_str, prob) — conveniência para inferência.""" return self.predict(sentence, dataset_name=dataset_name, return_proba=True) # ================================================================== # V6.5-fix — Inference-time Punishment via Hypothesis Layer # ================================================================== # User requirement: "ao punir o modelo reajusta seus parâmetros usando # a camada de hipótese para funcionar como na lógica de punição e # ajuste de parâmetros durante o treinamento" # # Mathematical principle (mirrors training punishment protocol): # # TRAINING protocol (process_batch): # acc = evaluate_classification(buffer) # if acc < 1.0: # punishment_count += 1 # success_count = 0 # if punishment_count == 1: # activate_hypothesis() # treina HypothesisClassifier # elif punishment_count == 2: # set_ewc_reference() # consolida w via Fisher # reset cycle # # INFERENCE protocol (punish_during_inference) — V6.5-fix: # When external feedback indicates the predicted label was WRONG: # 1. Compute vec_4d for the sentence. # 2. Add (sentence, correct_label_int) to buffer_4d + buffer_labels. # 3. Update SOM via Kohonen rule: ΔW = α·Λ·(x - W) on BMU. # 4. inference_punishment_count += 1 # punishment_count += 1 # success_count = 0 # 5. If punishment_count == 1: # activate_hypothesis() — re-treina HypothesisClassifier # usando o buffer expandido (Adam + BCEWithLogitsLoss, 50 epochs). # Ajusta pesos do classifier: θ_{t+1} = θ_t - η·∂L/∂θ. # If punishment_count >= 2: # set_ewc_reference() — consolida w via Fisher information: # F_i = mean((x_w - W_w,i)²) # old_weights_w = W_w.clone() # Reset cycle (mirror training). # 6. Return outcome dict with action taken + accuracy_after. # # This ensures that during inference, the model can self-correct using # the same hypothesis-layer-based parameter adjustment mechanism used # in training — NOT a separate inference-only correction. # ------------------------------------------------------------------ def punish_during_inference( self, sentence: str, correct_label_int: int, correct_label_str: Optional[str] = None, dataset_name: Optional[str] = None, ) -> Dict[str, Any]: """V6.5-fix — Punição na inferência via camada de hipótese. Mirror do protocolo de punição de treinamento, aplicado em inference quando o usuário (ou avaliador) fornece feedback de que a predição foi incorreta. Args: sentence: texto que foi mal-classificado. correct_label_int: rótulo inteiro correto (0 ou 1). correct_label_str: opcional — rótulo textual correto. dataset_name: opcional — nome do dataset para registrar label. Returns: Dict com: - action: 'activate_hypothesis' | 'set_ewc_reference' | 'none' - inference_punishment_count: total de punições em inference - punishment_count_cycle: contador no ciclo atual (0, 1, ou 2+) - classifier_trained: bool - ewc_reference_set: bool - accuracy_after_punishment: float - buffer_size: int - elapsed_ms: float """ import time as _time t0 = _time.time() # 1. Register label string if provided if correct_label_str is not None and dataset_name is not None: if dataset_name not in self.label_registry: self.label_registry[dataset_name] = {} self.label_registry[dataset_name][int(correct_label_int)] = str(correct_label_str) self.last_dataset_used = dataset_name # 2. Compute vec_4d for the misclassified sentence # V6.5-V2-memfix — no_grad + detach to prevent graph accumulation with torch.no_grad(): self.time_counter += 1 vec = text_to_4d_vector( sentence, self.tokenizer, self.embedding, self.hidden_dim, self.seq_len, self.time_counter, self.T_max, attention=self.attention, attention_metrics=self.attention_metrics, ) vec = vec.detach().clone() # 3. Add to buffer + update histogram (mirror add_data) self.buffer_4d.append(vec) self.buffer_labels.append(int(correct_label_int)) # Enforce sliding window limit (mirror add_data) if len(self.buffer_4d) > self.buffer_max_size: self.buffer_4d = self.buffer_4d[-self.buffer_max_size:] self.buffer_labels = self.buffer_labels[-self.buffer_max_size:] dim_val = round(vec[self.dim_index].item(), 2) self.histogram[dim_val] += 1 # 4. Force SOM update on the new sample (Kohonen rule) # ΔW = α·Λ·(x - W) [plus EWC penalty if reference set] bmu = self.som.find_bmu(vec) acc_fisher = ( self.punishment_count == 0 and self.som.old_weights_w is None ) try: self.som.update_weights(vec, bmu, accumulate_fisher=acc_fisher) except Exception: pass # non-fatal: SOM update may fail in edge cases # 5. Punishment protocol (mirror training process_batch) self.inference_punishment_count += 1 self.punishment_count += 1 self.success_count = 0 action = "none" classifier_trained = bool(self.classifier_trained) ewc_set = self.som.old_weights_w is not None if self.punishment_count == 1: # 1st punishment: re-train HypothesisClassifier # Math: θ_{t+1} = θ_t - η·∂L_BCE/∂θ, 50 epochs, Adam (η=0.001) try: self.activate_hypothesis() action = "activate_hypothesis" classifier_trained = True except Exception as e: action = f"activate_hypothesis_failed: {str(e)[:200]}" elif self.punishment_count >= 2: # 2nd punishment: set EWC reference + reset cycle # Math: F_i = mean((x_w - W_w,i)²); old_W_w = W_w.clone() try: self.som.set_ewc_reference() ewc_set = True action = "set_ewc_reference" except Exception as e: action = f"set_ewc_reference_failed: {str(e)[:200]}" # Reset cycle (mirror training) self.training_ready = False self.punishment_count = 0 self.success_count = 0 self.histogram.clear() # NOTE: do NOT fully clear buffer during inference — keep a # sliding window of last 256 samples to preserve recent feedback # while bounding memory. if len(self.buffer_4d) > 256: self.buffer_4d = self.buffer_4d[-256:] self.buffer_labels = self.buffer_labels[-256:] # 6. Re-evaluate classification accuracy after punishment try: acc_after = self.evaluate_classification() except Exception: acc_after = 0.0 outcome = { "action": action, "inference_punishment_count": int(self.inference_punishment_count), "punishment_count_cycle": int(self.punishment_count), "classifier_trained": bool(classifier_trained), "ewc_reference_set": bool(ewc_set), "accuracy_after_punishment": float(acc_after), "buffer_size": int(len(self.buffer_4d)), "bmu_of_misclassified": tuple(int(x) for x in bmu) if bmu else None, "elapsed_ms": float((_time.time() - t0) * 1000), } self.inference_punishment_history.append(outcome) if len(self.inference_punishment_history) > 100: self.inference_punishment_history = self.inference_punishment_history[-100:] return outcome def get_inference_punishment_history(self) -> List[Dict[str, Any]]: """V6.5-fix — Retorna histórico de punições em inference.""" return list(self.inference_punishment_history) # ------------------------------------------------------------------ # API de monitoramento (V6.4 + V6.5) # ------------------------------------------------------------------ def get_state_metrics(self) -> dict: """Retorna métricas completas do sistema para monitoramento. V6.5: inclui vqvae2_metrics e reasoning_metrics. V6.5-final: inclui w8a8_metrics e tool_coordinator_workers. """ som_metrics = self.som.get_metrics() return { "som": som_metrics, "kls": { "time_counter": int(self.time_counter), "T_max": int(self.T_max), "buffer_size": int(len(self.buffer_4d)), "training_ready": bool(self.training_ready), "punishment_count": int(self.punishment_count), "success_count": int(self.success_count), "classifier_trained": bool(self.classifier_trained), "histogram_size": int(len(self.histogram)), "histogram_max": int(max(self.histogram.values(), default=0)), "N_start": int(self.N), "dim_choice": str(self.dim_choice), "som_neuron_count": int(self.som_neuron_count), "required_new_samples": int(self.required_new_samples), "has_classifier": self.classifier is not None, "vocab_size": int(self.tokenizer.vocab_size), "hidden_dim": int(self.hidden_dim), "seq_len": int(self.seq_len), # V6.5 "enable_vqvae2": bool(self.enable_vqvae2), "enable_reasoning": bool(self.enable_reasoning), "enable_w8a8": bool(self.enable_w8a8), "vqvae2_n_calls": int(len(self.vqvae2_metrics_history)), "w8a8_n_calls": int(len(self.w8a8_metrics_history)), # V6.5-final: tool_coordinator workers reativados "tool_coordinator_workers_active": bool( getattr(self, "_tool_coordinator_workers_active", False) ), # V6.5-attn: attention module integrado "enable_attention": bool(getattr(self, "enable_attention", False)), "attention_active": bool(self.attention is not None), "attention_n_heads": int(getattr(self, "attention_n_heads", 0)), # V6.5-fix: dynamic label registry + inference punishment "label_registry_n_datasets": int(len(self.label_registry)), "last_dataset_used": str(self.last_dataset_used) if self.last_dataset_used else None, "inference_punishment_count": int(self.inference_punishment_count), "inference_punishment_history_len": int(len(self.inference_punishment_history)), }, # V6.5 — VQ-VAE-2 metrics "vqvae2": self.get_vqvae2_metrics(), # V6.5 — ReasoningEngine metrics "reasoning": self.get_reasoning_stats(), # V6.5-attn — MultiHeadAttention metrics "attention": self.get_attention_metrics(), # V6.5-final — SmoothQuant W8A8 metrics "w8a8": self.get_w8a8_metrics(), # V6.5-fix — Dynamic label registry state "label_registry": self.get_label_registry_state(), } # ============================================================================ # V6.5-V2 — HypothesisEnsemble (16 hipóteses) + KohonenLearningSystemV2 # ============================================================================ # User requirement: "Formato de Aprendizado com 16 Hipóteses e 3 Tentativas" # # Fluxo: # 1. Fase sem punição (streaming N-datasets) de CONHECIMENTO: # 1.1. Acumulação de conhecimento → ajuste dos pesos do SOM via regra # de Kohonen. # 1.2. Construção do MAPA-SOM (estado consolidado) com treinamento # contínuo. # # 2. Fase com punição (streaming M-datasets): # 2.1. Cada evento de erro (punição) dispara 3 tentativas de ajuste. # 2.2. Em cada tentativa, 16 geradores paralelos de hipóteses produzem # candidatos a delta. # 2.3. O melhor delta (que maximiza a acurácia no buffer) é aplicado # ao MAPA-SOM. # 2.4. Após a aplicação, consolidação via EWC (para evitar esquecimento # catastrófico). # 2.5. O ciclo é reiniciado sem loops infinitos (número fixo de # tentativas). # # Mathematical analysis: # - DeltaGenerator: g_h(x) → δ_h ∈ ℝ^P (h = 1..16) # onde P = I*J*K*L*4 (parâmetros do SOM achatados) # - HypothesisEnsemble: H = {g_1, ..., g_16} # - Avaliação: a_h = accuracy(SOM + δ_h, buffer) para cada h # - Seleção: δ* = argmax_h a_h # - Aplicação: SOM ← SOM + δ* # - Consolidação: set_ewc_reference() (Fisher information em w) # # Treinamento dos geradores (train_hypotheses): # - Para cada punição, executar hyp_train_steps passos de Adam # - Loss = mean_h(BCEWithLogitsLoss(classifier(SOM+δ_h), labels)) # - δ_h = scale * g_h(som_activation_mean) # - scale é um parâmetro treinável (começa em 0.01, clamp [0.001, 0.1]) # ============================================================================ class DeltaGenerator(nn.Module): """Gerador de delta para ajuste do MAPA-SOM. Entrada: vetor de ativação SOM (distâncias ao grid) ou vetor 4D médio. Saída: delta com a mesma dimensionalidade do SOM achatado. Arquitetura: Linear → ReLU → Linear → ReLU → Linear (Xavier init gain=0.01) """ def __init__(self, input_dim: int, output_dim: int, hidden_dim: int = 256): super().__init__() self.net = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim), ) # Inicialização Xavier para garantir deltas pequenos no início for layer in self.net: if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight, gain=0.01) nn.init.zeros_(layer.bias) def forward(self, x: torch.Tensor) -> torch.Tensor: return self.net(x) class HypothesisEnsemble(nn.Module): """Conjunto de N geradores de hipóteses paralelos (com active_count dinâmico). V6.5-V2-dynamic: pré-aloca max_n_hypotheses geradores, mas usa apenas os primeiros `active_count` em forward/forward_stacked. Isto permite ajustar dinamicamente o número de hipóteses ativas sem recriar o módulo. Args: input_dim: dimensionalidade do vetor de ativação SOM (som_neuron_count). output_dim: número de parâmetros do SOM achatados (I*J*K*L*4). n_hypotheses: número ATIVO inicial de geradores (default 16). max_n_hypotheses: número MÁXIMO pré-alocado (default 32). hidden_dim: largura das camadas internas de cada gerador. """ def __init__( self, input_dim: int, output_dim: int, n_hypotheses: int = 16, max_n_hypotheses: int = 32, hidden_dim: int = 256, ): super().__init__() # Garante que n_hypotheses <= max_n_hypotheses n_hypotheses = max(1, min(n_hypotheses, max_n_hypotheses)) self.n_hypotheses = n_hypotheses # compat (== active_count inicial) self.max_n_hypotheses = max_n_hypotheses self.active_count = n_hypotheses # Pré-aloca max_n_hypotheses geradores (mesmo os inativos terão params # inicializados — mas não receberão gradiente até serem ativados, pois # forward_stacked itera apenas até active_count). self.generators = nn.ModuleList( [DeltaGenerator(input_dim, output_dim, hidden_dim) for _ in range(max_n_hypotheses)] ) def set_active_count(self, n: int) -> None: """Ajusta dinamicamente o número de geradores ativos.""" n = max(1, min(int(n), self.max_n_hypotheses)) self.active_count = n self.n_hypotheses = n # mantém compat com código legado def forward(self, x: torch.Tensor) -> List[torch.Tensor]: """Retorna lista de `active_count` deltas (cada um (batch, output_dim)).""" return [self.generators[i](x) for i in range(self.active_count)] def forward_stacked(self, x: torch.Tensor) -> torch.Tensor: """Retorna tensor (batch, active_count, output_dim) para vetorização.""" return torch.stack( [self.generators[i](x) for i in range(self.active_count)], dim=1 ) class KohonenLearningSystemV2(KohonenLearningSystem): """Versão V2 do KLS com hipóteses dinâmicas e tentativas adaptativas. Herda toda a infraestrutura do KLS V6.5 (SOM 4D, EWC, VQ-VAE-2, W8A8, attention, reasoning_engine, label_registry dinâmico) e adiciona: a. HypothesisEnsemble: até max_n_hypotheses geradores de delta (default 32, com n_hypotheses ativos = 16 inicialmente). b. select_best_delta: avalia n_trials tentativas × n_hypotheses hipóteses. c. apply_best_delta_and_consolidate: aplica o delta e chama set_ewc_reference. d. train_hypotheses: treina os geradores via Adam + BCE loss. e. DynamicHyperparameterController: ajusta automaticamente n_hypotheses, n_trials e hyp_train_steps com base em: - histórico de loss (slope + volatilidade) - recorrência de punições (taxa nos últimos K batches) O protocolo de punição é estendido: - punishment_count == 1 → treinar hipóteses (train_hypotheses) + activate_hypothesis (classifier) - punishment_count == 2 → selecionar melhor delta (n_trials × n_hypotheses) + aplicar + set_ewc_reference + reset V6.5-V2-dynamic: Após cada train_hypotheses() e após cada evento de punição, o _adapt_hyperparameters() é chamado para re-ajustar (n_hypotheses, n_trials, hyp_train_steps) conforme a perda e a recorrência de punições. """ def __init__(self, *args, **kwargs): # Extrai parâmetros V2 antes de passar ao pai self.n_hypotheses = kwargs.pop("n_hypotheses", 16) self.max_n_hypotheses = kwargs.pop("max_n_hypotheses", 32) # Garante coerência self.n_hypotheses = max(1, min(self.n_hypotheses, self.max_n_hypotheses)) self.n_trials = kwargs.pop("n_trials", 3) self.hyp_train_steps = kwargs.pop("hyp_train_steps", 30) self.hyp_lr = kwargs.pop("hyp_lr", 1e-4) self.hyp_hidden_dim = kwargs.pop("hyp_hidden_dim", 256) # V6.5-V2-dynamic — limites para adaptação automática self.min_n_hypotheses = kwargs.pop("min_n_hypotheses", 4) self.max_n_trials = kwargs.pop("max_n_trials", 6) self.min_n_trials = kwargs.pop("min_n_trials", 1) self.max_hyp_train_steps = kwargs.pop("max_hyp_train_steps", 80) self.min_hyp_train_steps = kwargs.pop("min_hyp_train_steps", 10) # Janela de observação self.loss_history_window = kwargs.pop("loss_history_window", 8) self.punishment_window = kwargs.pop("punishment_window", 12) super().__init__(*args, **kwargs) # Inicialização do ensemble de hipóteses self._init_hypothesis_ensemble() # V6.5-V2-dynamic — estado do controlador adaptativo self._loss_history: List[float] = [] # últimos loss_final de train_hypotheses self._punishment_history: List[int] = [] # 1 se houve punição no batch, 0 caso contrário self._adaptation_log: List[Dict[str, Any]] = [] self._last_adaptation_step = 0 # V6.5-V2-metrics-FIX-2 — contadores cumulativos para monitoramento # User requirement: "quantos passos de treino de hipóteses usado" self._total_hyp_steps_executed: int = 0 self._n_train_hyp_calls: int = 0 # V6.5-V2-metrics — histórico de métricas SOM (QE, TE, KL, VE, Dead Neurons) # para detecção de Estagnação do QE e Cruzamento de Vizinhança. from bigru_t.model.som_metrics import SOMMetricHistory self._som_metric_history: SOMMetricHistory = SOMMetricHistory() def _init_hypothesis_ensemble(self): """Cria o ensemble de geradores + otimizador Adam.""" input_dim = self.som_neuron_count # ativação SOM flatten (256 para grid (4,4,4,4)) output_dim = self.som.weights.numel() # I*J*K*L*4 (256*4 = 1024) self.hypothesis_ensemble = HypothesisEnsemble( input_dim=input_dim, output_dim=output_dim, n_hypotheses=self.n_hypotheses, max_n_hypotheses=self.max_n_hypotheses, hidden_dim=self.hyp_hidden_dim, ) self.hyp_optimizer = torch.optim.Adam( self.hypothesis_ensemble.parameters(), lr=self.hyp_lr ) # delta_scale: parâmetro treinável para controlar magnitude dos deltas # começa pequeno (0.01) para evitar explosão no início self.delta_scale = nn.Parameter(torch.tensor(0.01)) # ================================================================== # V6.5-V2-dynamic — Controlador adaptativo de hiperparâmetros # ================================================================== def _record_loss(self, loss_final: float) -> None: """Adiciona um loss_final ao histórico (janela deslizante).""" self._loss_history.append(float(loss_final)) if len(self._loss_history) > self.loss_history_window: self._loss_history = self._loss_history[-self.loss_history_window:] def _record_punishment_event(self, had_punishment: bool) -> None: """Registra se houve punição no batch atual (janela deslizante).""" self._punishment_history.append(1 if had_punishment else 0) if len(self._punishment_history) > self.punishment_window: self._punishment_history = self._punishment_history[-self.punishment_window:] def _compute_loss_stats(self) -> Dict[str, float]: """Calcula slope e volatilidade do histórico de loss. Returns: {"slope": float, "volatility": float, "mean": float, "n": int} """ n = len(self._loss_history) if n < 2: return {"slope": 0.0, "volatility": 0.0, "mean": 0.0, "n": n} losses = list(self._loss_history) mean_loss = sum(losses) / n # Slope por mínimos quadrados: y = a*x + b, x = [0..n-1] x_mean = (n - 1) / 2.0 y_mean = mean_loss num = sum((i - x_mean) * (losses[i] - y_mean) for i in range(n)) den = sum((i - x_mean) ** 2 for i in range(n)) slope = num / den if den > 0 else 0.0 # Volatilidade = std / |mean| (CV) variance = sum((l - y_mean) ** 2 for l in losses) / n std = variance ** 0.5 volatility = std / max(abs(mean_loss), 1e-8) return { "slope": float(slope), "volatility": float(volatility), "mean": float(mean_loss), "std": float(std), "n": int(n), } def _compute_punishment_rate(self) -> float: """Calcula a taxa de punição na janela observada.""" if not self._punishment_history: return 0.0 return float(sum(self._punishment_history) / len(self._punishment_history)) def _adapt_hyperparameters(self, trigger: str = "auto") -> Dict[str, Any]: """V6.5-V2-dynamic — Ajusta n_hypotheses, n_trials, hyp_train_steps. Regras (matematicamente justificadas): - loss_volatility > 0.30 e n_hyp < max → n_hyp += 2 (mais diversidade) - loss_volatility < 0.05 e n_hyp > min → n_hyp -= 2 (menos desperdício) - loss_slope > -0.001 (lento/estagnado) e steps < max → steps += 5 - loss_slope < -0.01 (rápido) e steps > min → steps -= 5 - punishment_rate > 0.30 e n_trials < max → n_trials += 1 - punishment_rate < 0.05 e n_trials > min → n_trials -= 1 Args: trigger: "auto" (após train_hypotheses), "punishment" (após punição), "manual" (chamada externa). Returns: Dict com estado anterior, novo estado, e regras disparadas. """ before = { "n_hypotheses": int(self.n_hypotheses), "n_trials": int(self.n_trials), "hyp_train_steps": int(self.hyp_train_steps), } stats = self._compute_loss_stats() rate = self._compute_punishment_rate() rules_fired: List[str] = [] # 1. Ajuste de n_hypotheses (com base em volatilidade) if stats["n"] >= 3: if stats["volatility"] > 0.30 and self.n_hypotheses < self.max_n_hypotheses: new_n = min(self.n_hypotheses + 2, self.max_n_hypotheses) if new_n != self.n_hypotheses: self.n_hypotheses = new_n self.hypothesis_ensemble.set_active_count(new_n) rules_fired.append( f"n_hyp+2 (vol={stats['volatility']:.3f} → {new_n})" ) elif stats["volatility"] < 0.05 and self.n_hypotheses > self.min_n_hypotheses: new_n = max(self.n_hypotheses - 2, self.min_n_hypotheses) if new_n != self.n_hypotheses: self.n_hypotheses = new_n self.hypothesis_ensemble.set_active_count(new_n) rules_fired.append( f"n_hyp-2 (vol={stats['volatility']:.3f} → {new_n})" ) # 2. Ajuste de hyp_train_steps (com base em slope) if stats["n"] >= 3: if stats["slope"] > -0.001 and self.hyp_train_steps < self.max_hyp_train_steps: new_steps = min(self.hyp_train_steps + 5, self.max_hyp_train_steps) if new_steps != self.hyp_train_steps: self.hyp_train_steps = new_steps rules_fired.append( f"steps+5 (slope={stats['slope']:.5f} → {new_steps})" ) elif stats["slope"] < -0.01 and self.hyp_train_steps > self.min_hyp_train_steps: new_steps = max(self.hyp_train_steps - 5, self.min_hyp_train_steps) if new_steps != self.hyp_train_steps: self.hyp_train_steps = new_steps rules_fired.append( f"steps-5 (slope={stats['slope']:.5f} → {new_steps})" ) # 3. Ajuste de n_trials (com base em recorrência de punições) if len(self._punishment_history) >= 3: if rate > 0.30 and self.n_trials < self.max_n_trials: new_t = min(self.n_trials + 1, self.max_n_trials) if new_t != self.n_trials: self.n_trials = new_t rules_fired.append(f"n_trials+1 (rate={rate:.3f} → {new_t})") elif rate < 0.05 and self.n_trials > self.min_n_trials: new_t = max(self.n_trials - 1, self.min_n_trials) if new_t != self.n_trials: self.n_trials = new_t rules_fired.append(f"n_trials-1 (rate={rate:.3f} → {new_t})") after = { "n_hypotheses": int(self.n_hypotheses), "n_trials": int(self.n_trials), "hyp_train_steps": int(self.hyp_train_steps), } adapted = before != after record = { "trigger": trigger, "step": int(self._last_adaptation_step), "before": before, "after": after, "adapted": bool(adapted), "rules_fired": rules_fired, "loss_stats": stats, "punishment_rate": float(rate), } self._adaptation_log.append(record) # V6.5-V2-metrics-FIX: limita crescimento do log (era ilimitado, causando # crescimento de memória em treino longo). Mantém últimas 50 entradas. if len(self._adaptation_log) > 50: self._adaptation_log = self._adaptation_log[-50:] self._last_adaptation_step += 1 return record def get_adaptation_log(self) -> List[Dict[str, Any]]: """Retorna o log de adaptações dinâmicas (para monitoramento).""" return list(self._adaptation_log) # ================================================================== # V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM # ================================================================== # User requirement: "analisar e aprimorar (logicamente e matematicamente) # no projeto limpeza de memória RAM e de Armazenamento (ao concluir)". # # Mathematical analysis: # A memória RAM consumida pelo KLS V2 tem 4 fontes principais: # 1. Pesos do SOM: I*J*K*L*4 floats = 864*4 = 3456 floats (~14KB) # → FIXO (não pode ser limpo sem destruir o modelo) # 2. HypothesisEnsemble: max_n_hypotheses * DeltaGenerator params # = 32 * (864*256 + 256*256 + 256*3456) = 32 * 1.16M = ~37M params # → ~148MB em float32. Os OTIMIZADORAS Adam guardam 2x esse valor # (momentum + variance) = ~296MB adicionais. # 3. HypothesisClassifier: 864*512 + 512*256 + ... = ~616K params (~2.5MB) # 4. buffer_4d: até MAX_BUFFER_SIZE amostras * 4 floats = ~4KB (mínimo) # MAS os tensores carregam gradiente do embedding se não forem detached. # # A limpeza agressiva atua em: # - Zera gradientes acumulados nos otimizadores (Adam state) # - Chama gc.collect() para liberar ciclos Python # - Libera caches internos do PyTorch (oneDNN primitive cache) # - Trunca listas de histórico que crescem com o treino # ------------------------------------------------------------------ def aggressive_cleanup(self) -> Dict[str, Any]: """V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM do KLS. Retorna métricas antes/depois para confirmação. NÃO destroi pesos do SOM nem do classifier — apenas libera caches e gradientes acumulados. Returns: Dict com rss_before/after_mb, n_refs_collected, buffers_trimmed. """ import gc try: import resource rss_before = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 except Exception: rss_before = 0.0 # 1. Zera gradientes do otimizador Adam (libera momentum/variance state) try: if hasattr(self, "hyp_optimizer"): self.hyp_optimizer.zero_grad(set_to_none=True) except Exception: pass # 2. Trunca históricos que crescem com o treino if len(self._adaptation_log) > 20: self._adaptation_log = self._adaptation_log[-20:] if len(self.inference_punishment_history) > 50: self.inference_punishment_history = self.inference_punishment_history[-50:] if len(self.vqvae2_metrics_history) > 30: self.vqvae2_metrics_history = self.vqvae2_metrics_history[-30:] if len(self.w8a8_metrics_history) > 30: self.w8a8_metrics_history = self.w8a8_metrics_history[-30:] # 3. gc.collect() em 2 passes (gerações 0+1 e 2) n_collected_0 = gc.collect(0) n_collected_1 = gc.collect(1) n_collected_2 = gc.collect(2) # 4. Libera caches CPU do PyTorch (oneDNN/MKL) try: import torch if hasattr(torch, "cuda") and torch.cuda.is_available(): torch.cuda.empty_cache() except Exception: pass try: import resource rss_after = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 except Exception: rss_after = 0.0 return { "rss_before_mb": float(rss_before), "rss_after_mb": float(rss_after), "freed_mb": max(0.0, rss_before - rss_after), "gc_collected": int(n_collected_0 + n_collected_1 + n_collected_2), "adaptation_log_trimmed": int(len(self._adaptation_log)), "timestamp": __import__("datetime").datetime.utcnow().isoformat() + "Z", } # ================================================================== # 2.1 Treinamento dos geradores (train_hypotheses) # ================================================================== def train_hypotheses( self, buffer_4d: Optional[List[torch.Tensor]] = None, buffer_labels: Optional[List[int]] = None, ) -> Dict[str, Any]: """Treina os 16 geradores para produzir deltas que melhorem a classificação. Usa uma cópia do SOM para avaliar o efeito de cada delta e retropropaga o erro de classificação para os parâmetros dos geradores. Mathematical formulation: Para cada step s = 1..hyp_train_steps: 1. x_state = mean(som_activations(buffer)) # (1, P_som) 2. δ_h = scale * g_h(x_state) para h = 1..16 3. loss_h = BCE(classifier(SOM+δ_h), labels) 4. loss = mean_h(loss_h) + λ * ||δ||² (regularização) 5. θ_{s+1} = θ_s - η * ∂loss/∂θ Returns: Dict com métricas do treino (loss inicial, final, n_steps). """ if buffer_4d is None: buffer_4d = self.buffer_4d if buffer_labels is None: buffer_labels = self.buffer_labels if not buffer_4d: return {"active": False, "reason": "empty_buffer"} import time as _time import gc as _gc t0 = _time.time() # V6.5-V2-oom-guard — Pré-check de memória antes de alocar tensores # grandes (n_hyp * P * N * 4 bytes). Se memória livre < 2x estimado, # trunca o buffer para evitar OOM-killer. # User requirement: "o processo vem sendo morto OOM-kiler". try: import os as _os # Estima uso: n_hyp * P * N * 4 bytes (float32) + overhead 50% _est_bytes = int(self.n_hypotheses * self.som_neuron_count * len(buffer_4d) * 4 * 1.5) _est_mb = _est_bytes / 1e6 # Se estimativa > 800MB, trunca buffer para últimos 64 samples if _est_mb > 800 and len(buffer_4d) > 64: _orig_n = len(buffer_4d) buffer_4d = buffer_4d[-64:] buffer_labels = buffer_labels[-64:] # Log silently (não podemos usar logger aqui para evitar circular import) print(f"[V6.5-V2-oom-guard] train_hypotheses: buffer truncated " f"{_orig_n}→64 (est. {_est_mb:.0f}MB > 800MB)") except Exception: pass # Nunca deixa o pré-check quebrar o treino # 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 = [] _oom_truncated = False # V6.5-V2-oom-guard: flag setada em caso de OOM _oom_step = -1 # V6.5-V2-oom-guard: step onde OOM ocorreu # V6.5-V2-metrics-FIX: pré-computa data_norm_sq uma única vez (N, 1) # para reuso em todos os steps — evita recomputação redundante. data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) labels_expanded = labels # (N,) for step in range(self.hyp_train_steps): # V6.5-V2-oom-guard — captura OOM/RuntimeError por step. # Em caso de OOM, interrompe o treino de hipóteses graciosamente. try: self.hyp_optimizer.zero_grad() # Ativação média como representação do estado do SOM x_mean = som_activations.mean(dim=0, keepdim=True) # (1, P_som) # Gera deltas: (1, n_hypotheses, P) deltas_stack = self.hypothesis_ensemble.forward_stacked(x_mean) deltas_stack = self.delta_scale * deltas_stack # escala # V6.5-V2-metrics-FIX: Avaliação VETORIZADA de todas as hipóteses # em paralelo (substitui loop que materializava 16 cópias do SOM). # Memória: (n_hyp, P, N) em vez de 16 * (P + 2*P*N). if self.classifier is not None and self.classifier_trained: # deltas_stack: (1, n_hyp, P) → (n_hyp, P, 4) reshape n_hyp = self.n_hypotheses P_neurons = self.som_neuron_count # som_weights_flat: (P, 4) detached W_base = som_weights_flat.reshape(P_neurons, 4) # (P, 4) # delta_h: (n_hyp, P) → reshape para (n_hyp, P, 4) deltas_3d = deltas_stack[0].reshape(n_hyp, P_neurons, 4) # (n_hyp, P, 4) # W_new[h, p, d] = W_base[p, d] + deltas_3d[h, p, d] W_new = W_base.unsqueeze(0) + deltas_3d # (n_hyp, P, 4) — broadcast # dist²[h, p, n] = ||W_new[h, p] - data[n]||² # = ||W_new[h, p]||² + ||data[n]||² - 2*W_new[h, p]·data[n] W_new_norm_sq = (W_new * W_new).sum(dim=-1) # (n_hyp, P) # cross[h, p, n] = W_new[h, p] · data[n] cross = torch.matmul(W_new, data.t()) # (n_hyp, P, N) dist_sq = ( W_new_norm_sq.unsqueeze(-1) # (n_hyp, P, 1) + data_norm_sq # (1, N) → broadcast (n_hyp, P, N) - 2.0 * cross ) # (n_hyp, P, N) dist_sq = torch.clamp(dist_sq, min=0.0) # activations[h, n, p] = dist_sq[h, p, n] activations = dist_sq.transpose(1, 2) # (n_hyp, N, P) # Classifier forward (congelado) classifier_params_were_grad = [ p.requires_grad for p in self.classifier.parameters() ] for p in self.classifier.parameters(): p.requires_grad_(False) try: logits = self.classifier( activations.reshape(n_hyp * len(data), P_neurons) ).reshape(n_hyp, len(data)) # (n_hyp, N) # BCE por hipótese, depois média loss_per_hyp = F.binary_cross_entropy_with_logits( logits, labels_expanded.unsqueeze(0).expand(n_hyp, -1), reduction='none', ).mean(dim=1) # (n_hyp,) loss_total = loss_per_hyp.mean() # Regularização L2 sobre os deltas reg_loss = 0.01 * deltas_stack.norm() loss_total = loss_total + reg_loss finally: for p, was_grad in zip(self.classifier.parameters(), classifier_params_were_grad): p.requires_grad_(was_grad) else: # Fallback: minimizar norma do delta (regularização pura) loss_total = deltas_stack.norm() loss_total.backward() # V6.5-V2-metrics-FIX-3 — gradient clipping no treino de hipóteses # previne explosão de gradientes em batches degenerados (ex: todos # os labels iguais → BCE produz gradientes grandes). max_norm=1.0 # é o valor canônico recomendado pela literatura para BCE heads. torch.nn.utils.clip_grad_norm_( self.hypothesis_ensemble.parameters(), max_norm=1.0 ) # V6.5-V2-metrics-FIX-3 — zera grad do delta_scale (que não está # no optimizer mas aparece no grafo de forward, aculumando .grad # silenciosamente a cada step). if self.delta_scale.grad is not None: self.delta_scale.grad = None self.hyp_optimizer.step() losses.append(float(loss_total.item())) # V6.5-V2-metrics-FIX: libera tensores intermediários explicitamente # para reduzir pico de memória entre steps. del loss_total, deltas_stack if 'W_new' in dir(): del W_new, cross, dist_sq, activations # Atualiza a escala de delta (decai suavemente) with torch.no_grad(): self.delta_scale.data = torch.clamp( self.delta_scale.data * 0.99, 0.001, 0.1 ) except (RuntimeError, MemoryError) as _oom_err: # V6.5-V2-oom-guard — OOM detectado durante o step. # Interrompe o treino de hipóteses e retorna partial results. _oom_truncated = True _oom_step = step try: _gc.collect() except Exception: pass break self.hypothesis_ensemble.eval() # V6.5-V2-metrics-FIX: libera som_activations e som_weights_flat del som_activations, som_weights_flat, data, data_norm_sq import gc gc.collect() # V6.5-V2-dynamic — registra loss_final no histórico loss_final_val = float(losses[-1]) if losses else 0.0 self._record_loss(loss_final_val) # V6.5-V2-metrics-FIX-2 — atualiza contadores cumulativos # User requirement: "quantos passos de treino de hipóteses usado" self._total_hyp_steps_executed += int(self.hyp_train_steps) self._n_train_hyp_calls += 1 result = { "active": True, "n_steps": self.hyp_train_steps, "n_steps_executed": len(losses), "oom_truncated": bool(_oom_truncated), "oom_step": int(_oom_step), "n_hypotheses": self.n_hypotheses, "loss_initial": float(losses[0]) if losses else 0.0, "loss_final": loss_final_val, "loss_mean": float(sum(losses) / len(losses)) if losses else 0.0, "delta_scale_final": float(self.delta_scale.item()), "elapsed_ms": float((_time.time() - t0) * 1000), # V6.5-V2-metrics-FIX-2 — contadores cumulativos para monitoramento "total_hyp_steps_executed": int(self._total_hyp_steps_executed), "n_train_hyp_calls": int(self._n_train_hyp_calls), } # V6.5-V2-dynamic — dispara adaptação automática de hiperparâmetros try: adapt_record = self._adapt_hyperparameters(trigger="auto") result["adaptation"] = adapt_record except Exception as adapt_err: result["adaptation_error"] = str(adapt_err)[:200] return result def _compute_classification_loss_with_som( self, som_weights: torch.Tensor, data: torch.Tensor, labels: torch.Tensor ) -> torch.Tensor: """V6.5-V2-metrics-FIX — Computa perda de classificação com SOM temporário. BUG CORRIGIDO (V6.5-V2-metrics-FIX): A versão anterior envolvia toda a computação em `torch.no_grad()`, o que matava o gradiente que deveria fluir dos deltas → SOM → loss. Resultado: train_hypotheses() não treinava os geradores de fato. CORREÇÃO: A computação das ativações SOM (squared distances) é feita SEM no_grad, permitindo que o gradiente flua dos deltas (que pertencem ao HypothesisEnsemble) através do SOM temporário até a loss. Apenas o classifier é congelado (requires_grad=False) para que seu gradiente não seja atualizado por train_hypotheses. Mathematical formulation (squared Euclidean distance, consistente com _som_activation que usa ||W - x||^2, não ||W - x||): Para cada neurônio p e amostra n: dist²[p, n] = ||W[p] - x[n]||² = ||W[p]||² + ||x[n]||² - 2·W[p]·x[n] activations = dist².t() → (N, P_neurons) logits = classifier(activations) → (N,) loss = BCEWithLogitsLoss(logits, labels) Args: som_weights: tensor (I, J, K, L, 4) com gradiente (dos deltas). data: tensor (N, 4) — amostras 4D (detached). labels: tensor (N,) — rótulos binários. Returns: Scalar loss tensor (com gradiente fluindo para som_weights). """ if self.classifier is not None and self.classifier_trained: # NÃO usa torch.no_grad() — gradiente deve fluir dos deltas # através do SOM temporário até a loss. W_flat = som_weights.reshape(-1, 4) # (P_neurons, 4) # Identidade algébrica: ||W - x||^2 = ||W||^2 + ||x||^2 - 2*W·x # Evita materializar tensor (P, N, 4) — usa (P, N) direto. W_norm_sq = (W_flat * W_flat).sum(dim=-1, keepdim=True) # (P, 1) data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) cross = W_flat @ data.t() # (P, N) — matmul eficiente dist_sq = W_norm_sq + data_norm_sq - 2.0 * cross # (P, N) dist_sq = torch.clamp(dist_sq, min=0.0) # estabilidade numérica activations = dist_sq.t() # (N, P_neurons) — gradiente fluindo! # Congela classifier durante train_hypotheses (não queremos # atualizar seus pesos — apenas os deltas dos geradores) classifier_params_were_grad = [ p.requires_grad for p in self.classifier.parameters() ] for p in self.classifier.parameters(): p.requires_grad_(False) try: logits = self.classifier(activations) # (N,) loss = F.binary_cross_entropy_with_logits(logits, labels) finally: # Restaura requires_grad do classifier for p, was_grad in zip(self.classifier.parameters(), classifier_params_were_grad): p.requires_grad_(was_grad) return loss else: # Fallback: votação BMU (não-diferenciável) with torch.no_grad(): if not hasattr(self, "neuron_label"): self._label_neurons() correct = 0 for i in range(len(data)): bmu = self.som.find_bmu(data[i]) pred = self.neuron_label.get(bmu, 0.5) if (pred > 0.5) == (labels[i].item() > 0.5): correct += 1 return torch.tensor( 1.0 - correct / max(1, len(data)), device=data.device, dtype=torch.float, ) # ================================================================== # 2.2 Seleção do melhor delta (3 tentativas × 16 hipóteses) # ================================================================== def select_best_delta( self, buffer_4d: Optional[List[torch.Tensor]] = None, buffer_labels: Optional[List[int]] = None, n_trials: Optional[int] = None, ) -> Tuple[Optional[torch.Tensor], float, Dict[str, Any]]: """Avalia 3 tentativas × 16 hipóteses e retorna o melhor delta e sua acurácia. Args: buffer_4d, buffer_labels: opcional — usa self.buffer_* se None. n_trials: opcional — usa self.n_trials se None. Returns: Tuple (best_delta, best_acc, info_dict): - best_delta: tensor (P,) ou None se buffer vazio - best_acc: acurácia obtida com o melhor delta - info_dict: métricas da seleção (n_trials, n_hypotheses, etc.) """ if buffer_4d is None: buffer_4d = self.buffer_4d if buffer_labels is None: buffer_labels = self.buffer_labels if not buffer_4d: return None, -1.0, {"active": False, "reason": "empty_buffer"} if n_trials is None: n_trials = self.n_trials import time as _time t0 = _time.time() data = torch.stack(buffer_4d).detach() labels = torch.tensor(buffer_labels, dtype=torch.float, device=data.device) # Computa ativação SOM média with torch.no_grad(): som_activations = torch.stack( [self._som_activation(data[i]) for i in range(len(data))] ) x_state = som_activations.mean(dim=0, keepdim=True) best_delta = None best_acc = -1.0 trial_results = [] som_flat = self.som.weights.detach().clone().flatten() self.hypothesis_ensemble.eval() for trial in range(n_trials): # Adiciona ruído à entrada para diversificar tentativas (trial > 0) if trial > 0: x_noisy = x_state + 0.01 * torch.randn_like(x_state) else: x_noisy = x_state with torch.no_grad(): deltas_stack = self.hypothesis_ensemble.forward_stacked(x_noisy) deltas_stack = self.delta_scale * deltas_stack # Avalia cada uma das 16 hipóteses trial_accs = [] for h in range(self.n_hypotheses): delta_h = deltas_stack[0, h] som_temp = som_flat + delta_h som_temp_reshaped = som_temp.view_as(self.som.weights) acc = self._evaluate_accuracy_with_som( som_temp_reshaped, data, labels ) trial_accs.append(acc) if acc > best_acc: best_acc = acc best_delta = delta_h.clone() trial_results.append({ "trial": trial + 1, "best_acc_in_trial": float(max(trial_accs)) if trial_accs else 0.0, "mean_acc_in_trial": float(sum(trial_accs) / len(trial_accs)) if trial_accs else 0.0, "best_hyp_idx": int(trial_accs.index(max(trial_accs))) if trial_accs else -1, }) info = { "active": True, "n_trials": n_trials, "n_hypotheses": self.n_hypotheses, "best_acc": float(best_acc), "delta_norm": float(best_delta.norm().item()) if best_delta is not None else 0.0, "delta_scale": float(self.delta_scale.item()), "trials": trial_results, "elapsed_ms": float((_time.time() - t0) * 1000), } return best_delta, best_acc, info def _evaluate_accuracy_with_som( self, som_weights: torch.Tensor, data: torch.Tensor, labels: torch.Tensor ) -> float: """V6.5-V2-metrics-FIX — Calcula acurácia com SOM temporário (vetorizado). Substitui o loop que chamava _som_activation(data[i]) para cada amostra por uma única operação matmul. Reduz tempo de O(N*P) para O(P*N) mas com paralelismo SIMD (single matmul vs N python iterations). Para o fallback (votação BMU), ainda precisa trocar self.som.weights temporariamente pois find_bmu lê self.som.weights diretamente. """ if self.classifier is not None and self.classifier_trained: with torch.no_grad(): # Vetorizado: dist²[p, n] = ||W[p]||² + ||x[n]||² - 2*W[p]·x[n] W_flat = som_weights.reshape(-1, 4) # (P, 4) W_norm_sq = (W_flat * W_flat).sum(dim=-1, keepdim=True) # (P, 1) data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) cross = W_flat @ data.t() # (P, N) dist_sq = W_norm_sq + data_norm_sq - 2.0 * cross # (P, N) dist_sq = torch.clamp(dist_sq, min=0.0) activations = dist_sq.t() # (N, P) logits = self.classifier(activations) # (N,) preds = (logits > 0.5).float() acc = (preds == labels).float().mean().item() return float(acc) else: # Fallback: votação BMU — precisa trocar self.som.weights original_weights = self.som.weights self.som.weights = som_weights try: with torch.no_grad(): if not hasattr(self, "neuron_label"): self._label_neurons() correct = 0 for i in range(len(data)): bmu = self.som.find_bmu(data[i]) pred = self.neuron_label.get(bmu, 0.5) if (pred > 0.5) == (labels[i].item() > 0.5): correct += 1 acc = correct / max(1, len(data)) return float(acc) finally: self.som.weights = original_weights # ================================================================== # 2.3 Aplicação do melhor delta + consolidação via EWC # ================================================================== def apply_best_delta_and_consolidate( self, buffer_4d: Optional[List[torch.Tensor]] = None, buffer_labels: Optional[List[int]] = None, ) -> Dict[str, Any]: """Seleciona o melhor delta, aplica ao SOM e executa EWC. Pipeline: 1. select_best_delta: avalia 3 tentativas × 16 hipóteses 2. Aplica δ* ao SOM: SOM ← SOM + δ* 3. Re-avalia acurácia após aplicação 4. Consolida via EWC: set_ewc_reference() (Fisher em w) Returns: Dict com métricas (acc_before, acc_after, delta_norm, etc.) """ import time as _time t0 = _time.time() # Avalia acurácia ANTES da aplicação acc_before = self.evaluate_classification() # Seleciona melhor delta best_delta, best_acc, sel_info = self.select_best_delta( buffer_4d, buffer_labels ) if best_delta is None: return { "active": False, "reason": "no_valid_delta", "elapsed_ms": float((_time.time() - t0) * 1000), } # V6.5-V2-metrics-FIX-2 — Verifica se best_delta contém NaN/Inf ANTES # de aplicar ao SOM. User requirement: "investigar QE e KL resultando # em NAN". Se o delta tem NaN (devido a gradientes explosivos em # train_hypotheses), aplicar ao SOM contaminaria os pesos e propagaria # NaN para todas as métricas subsequentes (QE, KL, etc.). if torch.isnan(best_delta).any() or torch.isinf(best_delta).any(): n_nan = int(torch.isnan(best_delta).sum().item()) n_inf = int(torch.isinf(best_delta).sum().item()) logger = logging.getLogger(__name__) if 'logging' in dir() else None import logging as _logging _logging.getLogger(__name__).warning( f"[V6.5-V2-metrics-FIX-2] best_delta contains NaN/Inf " f"(nan={n_nan}, inf={n_inf}). SKIPPING application to SOM " f"to prevent weight contamination. Investigate train_hypotheses " f"for gradient explosion (consider lowering hyp_lr or adding " f"gradient clipping)." ) return { "active": False, "reason": f"best_delta_has_nan (nan={n_nan}, inf={n_inf})", "delta_norm": float(best_delta.norm().item()), "elapsed_ms": float((_time.time() - t0) * 1000), "nan_detected": True, } # Aplica o delta ao SOM with torch.no_grad(): som_flat = self.som.weights.flatten() som_flat_new = som_flat + best_delta # V6.5-V2-metrics-FIX-2 — verificação pós-aplicação: se os pesos # do SOM ficaram com NaN (deveria não acontecer após o check acima, # mas defense in depth), reverte para o estado anterior. if torch.isnan(som_flat_new).any() or torch.isinf(som_flat_new).any(): import logging as _logging _logging.getLogger(__name__).error( f"[V6.5-V2-metrics-FIX-2] SOM weights became NaN/Inf after " f"delta application. REVERTING to previous weights." ) else: self.som.weights = som_flat_new.view_as(self.som.weights) # Re-avalia acurácia APÓS aplicação acc_after = self.evaluate_classification() # Re-rotula neurônios (BMU labels podem ter mudado) self._label_neurons() # V6.5-V2-metrics-FIX-3 — se best_delta tinha NaN (detectado acima e # retornado early), NÃO chegamos aqui. Mas se chegamos via caminho # else (pesos ficaram NaN após aplicação e foram revertidos), NÃO # devemos consolidar via EWC — consolidar pesos potencialmente # corruptos como referência EWC propagaria o problema para o futuro. # Verificação: se os pesos atuais têm NaN, pula set_ewc_reference. weights_corrupt = torch.isnan(self.som.weights).any() or torch.isinf(self.som.weights).any() if weights_corrupt: import logging as _logging _logging.getLogger(__name__).error( f"[V6.5-V2-metrics-FIX-3] SOM weights contain NaN/Inf after " f"delta application — SKIPPING set_ewc_reference to avoid " f"propagating corruption. Weights will be sanitized on next " f"update_weights call." ) # Sanitiza imediatamente para permitir continuidade do treino self.som.weights = torch.nan_to_num( self.som.weights, nan=0.0, posinf=100.0, neginf=-100.0 ) ewc_set = False sel_info["ewc_skipped"] = "weights_corrupt_after_delta" else: # Consolida via EWC (Fisher information em w) try: self.som.set_ewc_reference() ewc_set = True except Exception as e: ewc_set = False sel_info["ewc_error"] = str(e)[:200] return { "active": True, "acc_before": float(acc_before), "acc_after": float(acc_after), "best_acc_during_selection": float(best_acc), "delta_norm": float(best_delta.norm().item()), "delta_scale": float(self.delta_scale.item()), "ewc_reference_set": bool(ewc_set), "selection_info": sel_info, "elapsed_ms": float((_time.time() - t0) * 1000), } # ================================================================== # 2.4 Sobrescrita do process_batch para o novo protocolo V2 # ================================================================== def process_batch_v2( self, sentences: List[str], labels: List[int], label_strings: Optional[Dict[int, str]] = None, dataset_name: Optional[str] = None, enable_punishment: bool = True, ) -> Dict[str, Any]: """Versão V2 do process_batch com 3 tentativas e 16 hipóteses. Protocolo: - CONHECIMENTO (enable_punishment=False): apenas adiciona dados e treina SOM (sem aplicar punição). - TREINAMENTO COM PUNIÇÃO (enable_punishment=True): * acc < 1.0 → punishment_count += 1 * punishment_count == 1: treinar 16 hipóteses + activate_hypothesis * punishment_count == 2: aplicar melhor delta (3×16) + EWC + reset Args: sentences, labels, label_strings, dataset_name: mesmos de process_batch. enable_punishment: se False, apenas acumula conhecimento (fase 1). Returns: Dict com métricas do batch processado. """ import time as _time t0 = _time.time() result: Dict[str, Any] = { "phase": "CONHECIMENTO" if not enable_punishment else "PUNICAO", "dataset": dataset_name, "n_samples": len(sentences), "action": "none", "cycle_completed": False, } # 1. Adiciona dados ao buffer self.add_data( sentences, labels, label_strings=label_strings, dataset_name=dataset_name, ) # 2. Verifica início de treino training_started = self.check_training_start() if training_started or self.training_ready: # 3. Treina SOM no buffer (regra de Kohonen) self.train_som_on_buffer() self._label_neurons() result["som_trained"] = True if not self.training_ready: result["elapsed_ms"] = float((_time.time() - t0) * 1000) return result # 4. Fase de CONHECIMENTO: para aqui, sem aplicar punição if not enable_punishment: acc = self.evaluate_classification() result["accuracy"] = float(acc) result["elapsed_ms"] = float((_time.time() - t0) * 1000) return result # 5. Fase de PUNIÇÃO: avalia e aplica protocolo V2 acc = self.evaluate_classification() result["accuracy"] = float(acc) if acc < 1.0: self.punishment_count += 1 self.success_count = 0 # V6.5-V2-dynamic — registra evento de punição self._record_punishment_event(True) if self.punishment_count == 1: # 1ª punição: treinar hipóteses + ativar classificador hyp_info = self.train_hypotheses() if self.classifier is None: self.activate_hypothesis() else: # Re-treina classifier com buffer atual self.activate_hypothesis() result["action"] = "train_hypotheses_and_activate" result["hypotheses_training"] = hyp_info elif self.punishment_count >= 2: # 2ª punição: selecionar melhor delta (n_trials × n_hyp) + aplicar + EWC apply_info = self.apply_best_delta_and_consolidate() result["action"] = "apply_best_delta_and_consolidate" result["delta_application"] = apply_info # V6.5-V2-dynamic — adapta hiperparâmetros após punição recorrente try: adapt_record = self._adapt_hyperparameters(trigger="punishment") result["adaptation"] = adapt_record except Exception as adapt_err: result["adaptation_error"] = str(adapt_err)[:200] # Reset do ciclo (mirror training) self.training_ready = False self.punishment_count = 0 self.success_count = 0 self.histogram.clear() # Mantém um sliding window do buffer para preservar contexto if len(self.buffer_4d) > 256: self.buffer_4d = self.buffer_4d[-256:] self.buffer_labels = self.buffer_labels[-256:] result["cycle_completed"] = True else: # acc == 1.0: sucesso self.punishment_count = 0 self.success_count += 1 # V6.5-V2-dynamic — registra sucesso (sem punição) self._record_punishment_event(False) result["action"] = "success" result["elapsed_ms"] = float((_time.time() - t0) * 1000) return result # ================================================================== # 2.5 API de monitoramento V2 # ================================================================== def get_v2_metrics(self) -> Dict[str, Any]: """Retorna métricas específicas da versão V2 (inclui adaptação dinâmica).""" return { "version": "V2-dynamic", "n_hypotheses": int(self.n_hypotheses), "n_hypotheses_active": int(self.hypothesis_ensemble.active_count), "max_n_hypotheses": int(self.max_n_hypotheses), "n_trials": int(self.n_trials), "hyp_train_steps": int(self.hyp_train_steps), "hyp_lr": float(self.hyp_lr), "delta_scale": float(self.delta_scale.item()), "n_generators": int(len(self.hypothesis_ensemble.generators)), "punishment_count": int(self.punishment_count), "success_count": int(self.success_count), "training_ready": bool(self.training_ready), "classifier_trained": bool(self.classifier_trained), "ewc_reference_set": bool(self.som.old_weights_w is not None), "buffer_size": int(len(self.buffer_4d)), # V6.5-V2-metrics-FIX-2 — contadores cumulativos (user requirement: # "quantos passos de treino de hipóteses usado") "total_hyp_steps_executed": int(getattr(self, "_total_hyp_steps_executed", 0)), "n_train_hyp_calls": int(getattr(self, "_n_train_hyp_calls", 0)), "dynamic_adaptation": { "loss_history_len": int(len(self._loss_history)), "loss_stats": self._compute_loss_stats(), "punishment_rate": float(self._compute_punishment_rate()), "punishment_window_size": int(len(self._punishment_history)), "n_adaptations": int(len(self._adaptation_log)), "limits": { "min_n_hypotheses": int(self.min_n_hypotheses), "max_n_hypotheses": int(self.max_n_hypotheses), "min_n_trials": int(self.min_n_trials), "max_n_trials": int(self.max_n_trials), "min_hyp_train_steps": int(self.min_hyp_train_steps), "max_hyp_train_steps": int(self.max_hyp_train_steps), }, "last_5_adaptations": self._adaptation_log[-5:] if self._adaptation_log else [], }, } # ================================================================== # V6.5-V2-metrics-FIX-2 — Detecção de atividade de hipóteses e neurônios # ================================================================== # User requirement (latest): "Aprimorar a FASE2 PUNITIVA investigar QE e KL # resultando em NAN, acrescentar detecção da atividade das layers (quantas # ativadas) de hipóteses e das ativações dos neurônios, quantos passos de # treino de hipóteses usado". # # Mathematical formulation: # # Hipótese ativa: Um gerador g_h (h ∈ {1..n_hypotheses}) é considerado # "ativo" em um batch se o delta que ele produz tem norma L2 > threshold # (default 1e-6). Deltas com norma ≈ 0 indicam que o gerador está # produzindo saída trivial (morto) — não contribui para a diversidade # do ensemble. # # active_h = 1 if ||δ_h||₂ > threshold else 0 # n_active_hypotheses = Σ_h active_h # # Neurônio ativo: Um neurônio (i,j,k,l) é "ativo" se é BMU para pelo # menos uma amostra do buffer atual. Reuso da lógica de dead_neuron_rate, # mas exposto como métrica explícita para monitoramento contínuo. # # n_active_neurons = |{W_bmu(x_i) : i ∈ buffer}| # n_dead_neurons = n_total - n_active_neurons # neuron_activation_rate = n_active_neurons / n_total # # Passos de treino de hipóteses: Número total de passos de Adam executados # em train_hypotheses() desde o início do treino. Cumulativo ao longo de # todas as chamadas (punições). # # total_hyp_train_steps = Σ_call self.hyp_train_steps (na época da call) # ------------------------------------------------------------------ def detect_hypothesis_activity( self, threshold: float = 1e-6, ) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-2 — Detecta quantas hipóteses estão ativas. User requirement: "acrescentar detecção da atividade das layers (quantas ativadas) de hipóteses". Executa um forward pass do HypothesisEnsemble sobre a ativação média do SOM no buffer atual e conta quantos dos active_count geradores produzem deltas com norma > threshold. Args: threshold: norma L2 mínima para considerar um delta "ativo". Returns: Dict com: - n_hypotheses_active: int (quantas hipóteses produziram delta ≠ 0) - n_hypotheses_total: int (active_count atual do ensemble) - max_n_hypotheses: int (limite máximo pré-alocado) - delta_norms: list[float] (norma de cada delta, para inspeção) - active_mask: list[bool] (True para hipóteses ativas) - activation_rate: float (n_active / n_total) - mean_delta_norm: float (média das normas) - max_delta_norm: float (maior norma) - threshold_used: float """ if not self.buffer_4d: return { "n_hypotheses_active": 0, "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), "max_n_hypotheses": int(self.max_n_hypotheses), "delta_norms": [], "active_mask": [], "activation_rate": 0.0, "mean_delta_norm": 0.0, "max_delta_norm": 0.0, "threshold_used": float(threshold), "reason": "empty_buffer", } try: import torch as _torch with _torch.no_grad(): # Computa ativação SOM média do buffer data = _torch.stack(self.buffer_4d).detach() som_activations = _torch.stack( [self._som_activation(data[i]) for i in range(len(data))] ) x_mean = som_activations.mean(dim=0, keepdim=True) # (1, P_som) # Forward do ensemble: retorna (1, n_hyp, P) deltas_stack = self.hypothesis_ensemble.forward_stacked(x_mean) deltas_stack = self.delta_scale * deltas_stack # aplica escala # Norma L2 de cada delta: (n_hyp,) delta_norms = deltas_stack[0].norm(dim=-1) # (n_hyp,) delta_norms_list = [float(x) for x in delta_norms] active_mask = [bool(x > threshold) for x in delta_norms_list] n_active = sum(active_mask) # Libera tensores del deltas_stack, som_activations, data, x_mean, delta_norms return { "n_hypotheses_active": int(n_active), "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), "max_n_hypotheses": int(self.max_n_hypotheses), "delta_norms": delta_norms_list, "active_mask": active_mask, "activation_rate": float(n_active / max(1, self.hypothesis_ensemble.active_count)), "mean_delta_norm": float(sum(delta_norms_list) / max(1, len(delta_norms_list))), "max_delta_norm": float(max(delta_norms_list) if delta_norms_list else 0.0), "threshold_used": float(threshold), } except Exception as e: return { "n_hypotheses_active": 0, "n_hypotheses_total": int(self.hypothesis_ensemble.active_count), "max_n_hypotheses": int(self.max_n_hypotheses), "delta_norms": [], "active_mask": [], "activation_rate": 0.0, "mean_delta_norm": 0.0, "max_delta_norm": 0.0, "threshold_used": float(threshold), "error": str(e)[:200], } def count_active_neurons(self) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-2 — Conta neurônios ativos (BMUs) no buffer. User requirement: "acrescentar (...) das ativações dos neurônios". Para cada amostra no buffer_4d, encontra o BMU. Conta quantos neurônios únicos são BMUs (n_active_neurons) e quantos nunca são BMUs (n_dead_neurons = n_total - n_active). Returns: Dict com: - n_active_neurons: int (neurônios que são BMU para ≥1 amostra) - n_dead_neurons: int (neurônios que nunca são BMU) - n_total_neurons: int (total de neurônios no SOM) - neuron_activation_rate: float (n_active / n_total) - bmu_distribution: dict {flat_idx: count} (top-20 mais frequentes) """ if not self.buffer_4d: return { "n_active_neurons": 0, "n_dead_neurons": int(self.som_neuron_count), "n_total_neurons": int(self.som_neuron_count), "neuron_activation_rate": 0.0, "bmu_distribution": {}, "reason": "empty_buffer", } try: import torch as _torch with _torch.no_grad(): data = _torch.stack(self.buffer_4d).detach() flat_w = self.som.weights.view(-1, 4).detach() # (N, M) distâncias diff = data.unsqueeze(1) - flat_w.unsqueeze(0) dist_sq = _torch.sum(diff * diff, dim=-1) _, bmu_flat = _torch.min(dist_sq, dim=1) # (N,) bmu_list = bmu_flat.tolist() unique_bmus = set(bmu_list) n_active = len(unique_bmus) n_total = self.som_neuron_count # Distribuição dos BMUs (top-20) from collections import Counter bmu_counts = Counter(bmu_list) bmu_distribution = dict(bmu_counts.most_common(20)) del data, flat_w, diff, dist_sq, bmu_flat return { "n_active_neurons": int(n_active), "n_dead_neurons": int(n_total - n_active), "n_total_neurons": int(n_total), "neuron_activation_rate": float(n_active / max(1, n_total)), "bmu_distribution": {str(k): int(v) for k, v in bmu_distribution.items()}, } except Exception as e: return { "n_active_neurons": 0, "n_dead_neurons": int(self.som_neuron_count), "n_total_neurons": int(self.som_neuron_count), "neuron_activation_rate": 0.0, "bmu_distribution": {}, "error": str(e)[:200], } def get_hyp_train_steps_used(self) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-2 — Retorna quantos passos de treino de hipóteses foram usados até agora. User requirement: "quantos passos de treino de hipóteses usado". Returns: Dict com: - current_hyp_train_steps: int (valor ATUAL de self.hyp_train_steps) - total_hyp_train_steps_executed: int (cumulativo ao longo do treino) - n_train_hypotheses_calls: int (número de chamadas a train_hypotheses) - min_hyp_train_steps: int (limite inferior dinâmico) - max_hyp_train_steps: int (limite superior dinâmico) """ return { "current_hyp_train_steps": int(self.hyp_train_steps), "total_hyp_train_steps_executed": int(getattr(self, "_total_hyp_steps_executed", 0)), "n_train_hypotheses_calls": int(getattr(self, "_n_train_hyp_calls", 0)), "min_hyp_train_steps": int(self.min_hyp_train_steps), "max_hyp_train_steps": int(self.max_hyp_train_steps), } # ================================================================== # V6.5-V2-metrics-FIX-4 — Conscience mechanism + Dead neuron revival # ================================================================== # User requirement (latest): "APRIMORAR (em ambas as FASE1 e FASE2): # analisar matematicamente e logicamente a ativação e uso e acesso dos # neurônios (apenas dois estão sendo ativados: neurons_active=2/864) # distribuindo o processamento paralelamente". # # Mathematical formulation (Conscience Mechanism — DeSieno 1988): # # Cada neurônio i mantém win_frequency p_i ∈ [0,1] (EMA, β=0.1). # BMU selection: bmu = argmin_i ( ||W_i - x||² - b_i ) # onde b_i = γ · (1/N - p_i) é o conscience bias # γ = 0.1 (default), N = total de neurônios (864) # # Quando p_i > 1/N (neurônio ganha demais): b_i < 0 → distância # efetiva AUMENTA → neurônio é penalizado. # Quando p_i < 1/N (neurônio nunca ganha): b_i > 0 → distância # efetiva DIMINUI → neurônio é favorecido. # # Convergência: p_i → 1/N para todo i (distribuição uniforme de BMU), # garantindo que TODOS os 864 neurônios sejam utilizados. # # Dead neuron revival (complementar): # Se após K amostras um neurônio ainda tem win_count = 0, ele é # reinicializado para uma amostra aleatória do buffer (data-driven) # ou para N(0, 0.1) se buffer vazio. Isto acelera a diversificação # quando o conscience mechanism sozinho não basta. # ------------------------------------------------------------------ def init_weights_kmeans_pp_if_ready(self): """V6.5-V2-kmeans-pp - Chama init_weights_kmeans_pp no SOM se buffer >= 64. Tenta uma unica vez durante o treino (idempotente via flag _kmeans_pp_init_attempted). Se falhar (buffer pequeno, erro), retorna silently e o SOM mantem a inicializacao grid coords+ruido original. Returns: Dict com status da inicializacao. """ if getattr(self, "_kmeans_pp_init_attempted", False): return {"initialized": False, "reason": "already_attempted"} self._kmeans_pp_init_attempted = True if len(self.buffer_4d) < self.som._kmeans_pp_min_samples: return { "initialized": False, "reason": f"buffer_too_small ({len(self.buffer_4d)} < {self.som._kmeans_pp_min_samples})", } try: result = self.som.init_weights_kmeans_pp( data_buffer=self.buffer_4d, n_iter=5, random_seed=42, ) if result.get("initialized"): self.som.bmu_win_count.zero_() target_p = 1.0 / self.som.n_neurons self.som.win_frequency.fill_(target_p) self.som._recent_bmu_flat.clear() return result except Exception as e: return {"initialized": False, "reason": f"error: {e}"} def revive_dead_neurons( self, dead_threshold: int = 0, use_buffer: bool = True, ) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-4 — Wrapper KLS para som.revive_dead_neurons. Usa o buffer_4d atual do KLS como fonte de dados para reinicialização data-driven dos neurônios mortos. Args: dead_threshold: neurônios com win_count <= threshold são revividos. use_buffer: se True, usa buffer_4d do KLS como fonte. Returns: Dict com status do revival (n_revived, dead_rate_before/after, etc). """ data_buffer = self.buffer_4d if use_buffer else None return self.som.revive_dead_neurons( data_buffer=data_buffer, dead_threshold=dead_threshold, ) def parallel_neuron_activation_report(self) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-4 — Wrapper KLS para som.parallel_neuron_activation_report. Retorna relatório estruturado de ativação dos 864 neurônios, análogo ao DataFrame do exemplo fornecido pelo usuário, mas otimizado para o grid 4D do SOM. Returns: Dict com estatísticas detalhadas (n_active, n_dead, win_count distribution, conscience_bias stats, uniformity_score, etc). """ return self.som.parallel_neuron_activation_report() # ================================================================== # V6.5-V2-parallel — Wrappers para novos métodos parallel SOM # ================================================================== def train_som_batch_parallel( self, samples: torch.Tensor, use_conscience: bool = True, accumulate_fisher: bool = False, ) -> Dict[str, Any]: """V6.5-V2-parallel — Wrapper KLS para som.train_batch_parallel. User requirement (ParallelSOM4D pattern): usa paralelismo vetorial PyTorch ao invés de loops Python + multiprocessing. Todas as operações internas (distância, BMU, vizinhança, update) são feitas via broadcasting sobre os 864 neurônios simultaneamente. Args: samples: tensor (B, 4) — batch de amostras 4D. use_conscience: se True, aplica bias de consciência. accumulate_fisher: se True, acumula Fisher info em w. Returns: Dict com: n_processed, n_bmu_unique, dead_rate, alpha_t, sigma_t. """ return self.som.train_batch_parallel( samples=samples, use_conscience=use_conscience, accumulate_fisher=accumulate_fisher, ) def reorganize_som_map_by_density( self, regularization_factor: float = 0.05, min_dead_rate_for_trigger: float = 0.3, ) -> Dict[str, Any]: """V6.5-V2-reorganize — Wrapper KLS para som.reorganize_map_by_density. User requirement (ReorganizingSOM4D pattern): reorganiza topologia do SOM dinamicamente durante o treinamento via filtragem de ativação por densidade. Neurônios pouco ativados (isolados) sofrem força de atração em direção aos agrupamentos de alta densidade, reorganizando geometricamente o hipercubo 4D sem perder o alinhamento matricial. Args: regularization_factor: magnitude do pull (default 0.05). min_dead_rate_for_trigger: só reorganiza se dead_rate > threshold. Returns: Dict com: reorganized, dead_rate_before, n_dead_pulled, etc. """ return self.som.reorganize_map_by_density( regularization_factor=regularization_factor, min_dead_rate_for_trigger=min_dead_rate_for_trigger, ) def reorganize_map_if_needed( self, regularization_factor: float = 0.05, min_dead_rate_for_trigger: float = 0.3, min_steps_between_reorganizations: int = 300, ) -> Dict[str, Any]: """V6.5-V2-reorganize — Auto-dispara reorganização por densidade se necessário. Verifica o dead_rate atual e reorganiza o mapa SOM se: - dead_rate > min_dead_rate_for_trigger (default 0.3 = 30% mortos) - pelo menos min_steps_between_reorganizations desde a última reorganização (cooldown) Complementa auto_revive_if_needed: enquanto auto_revive faz reinicialização abrupta (k-means++ selection), reorganize_map faz pulso suave e contínuo (gravitational pull). Ambos podem coexistir. Args: regularization_factor: magnitude do pull (default 0.05). min_dead_rate_for_trigger: limite para disparar (default 0.3). min_steps_between_reorganizations: cooldown em steps (default 300). Returns: Dict com status da reorganização (ou action="skipped" se não disparou). """ last_reorg_step = getattr( self, "_last_reorganize_step", -min_steps_between_reorganizations ) current_step = self.som.t steps_since_last = current_step - last_reorg_step # Computa dead rate atual n_total = self.som.n_neurons n_active = int((self.som.bmu_win_count > 0).sum().item()) dead_rate = float(1.0 - n_active / max(1, n_total)) if dead_rate <= min_dead_rate_for_trigger: return { "action": "skipped", "reason": f"dead_rate={dead_rate:.3f} <= threshold={min_dead_rate_for_trigger}", "dead_rate": dead_rate, "n_active": n_active, "n_total": n_total, } if steps_since_last < min_steps_between_reorganizations: return { "action": "skipped", "reason": ( f"cooldown: only {steps_since_last} steps since last reorg " f"(need {min_steps_between_reorganizations})" ), "dead_rate": dead_rate, "n_active": n_active, "n_total": n_total, } # Dispara reorganização reorg = self.reorganize_som_map_by_density( regularization_factor=regularization_factor, min_dead_rate_for_trigger=min_dead_rate_for_trigger, ) self._last_reorganize_step = current_step reorg["action"] = "auto_reorganized" reorg["trigger_dead_rate"] = dead_rate reorg["steps_since_last_reorg"] = steps_since_last return reorg def auto_revive_if_needed( self, dead_rate_threshold: float = 0.5, min_steps_between_revivals: int = 200, ) -> Dict[str, Any]: """V6.5-V2-metrics-FIX-4 — Revive neurônios automaticamente se dead_rate alto. Verifica o dead_rate atual e revive neurônios mortos se: - dead_rate > dead_rate_threshold (default 0.5 = 50% mortos) - pelo menos min_steps_between_revivals desde o último revival Isto é chamado automaticamente pelo treinador após cada chunk, garantindo que o SOM mantenha distribuição uniforme de BMU ao longo do treino sem intervenção manual. Args: dead_rate_threshold: limite para disparar revival (default 0.5). min_steps_between_revivals: cooldown em steps (default 200). Returns: Dict com status do revival (ou action="skipped" se não disparou). """ last_revival_step = getattr(self, "_last_revival_step", -min_steps_between_revivals) current_step = self.som.t steps_since_last = current_step - last_revival_step # Computa dead rate atual report = self.som.parallel_neuron_activation_report() dead_rate = 1.0 - report["neuron_activation_rate"] if dead_rate <= dead_rate_threshold: return { "action": "skipped", "reason": f"dead_rate={dead_rate:.3f} <= threshold={dead_rate_threshold}", "dead_rate": dead_rate, "n_active": report["n_active_neurons"], "n_total": report["n_total_neurons"], } if steps_since_last < min_steps_between_revivals: return { "action": "skipped", "reason": f"cooldown: only {steps_since_last} steps since last revival " f"(need {min_steps_between_revivals})", "dead_rate": dead_rate, "n_active": report["n_active_neurons"], "n_total": report["n_total_neurons"], } # Dispara revival revival = self.revive_dead_neurons( dead_threshold=0, use_buffer=True, ) self._last_revival_step = current_step revival["action"] = "auto_revived" revival["trigger_dead_rate"] = dead_rate revival["steps_since_last_revival"] = steps_since_last return revival # ================================================================== # V6.5-V2-metrics — Integração das métricas SOM canônicas # (QE, TE, Kaski-Lagus, Variância Explicada, Dead Neurons, # Colapso Topológico, Estagnação QE, Cruzamento Vizinhança) # ================================================================== def compute_som_metrics(self) -> Dict[str, Any]: """V6.5-V2-metrics — Computa todas as métricas SOM canônicas. User requirement: "ANALISAR matematicamente e logicamente e inserir melhorias para os scripts das métricas de aprendizado e indicadores de falha em Mapas Auto-Organizáveis (SOM / Redes de Kohonen) avaliam a fidelidade de representação dos dados e a preservação da vizinhança topológica". Returns: Dict com 8 métricas: 1. quantization_error (QE) 2. topological_error (TE) 3. kaski_lagus_error (KL) 4. explained_variance_share (VE) 5. topological_collapse (dict) 6. dead_neuron_rate (dict) 7. qe_stagnation (dict) — requer histórico 8. neighborhood_crossing (dict) — requer histórico Plus: overall_health, failure_indicators """ from bigru_t.model.som_metrics import compute_all_metrics if not self.buffer_4d: return { "quantization_error": 0.0, "topological_error": 0.0, "kaski_lagus_error": 0.0, "explained_variance_share": 0.0, "topological_collapse": {"severity": "none", "effective_rank": 0.0}, "dead_neuron_rate": {"dead_neuron_rate": 1.0, "n_dead": self.som_neuron_count, "n_active": 0, "n_total": self.som_neuron_count}, "qe_stagnation": {"is_stagnant": False, "severity": "none"}, "neighborhood_crossing": {"detected": False, "severity": "none"}, "overall_health": "unknown", "failure_indicators": ["no_buffer_data"], "n_failure_indicators": 1, "note": "buffer_4d vazio — métricas indisponíveis", } # Stack buffer em tensor (N, 4) data = torch.stack([ v.detach().clone() if isinstance(v, torch.Tensor) else torch.tensor(v, dtype=torch.float) for v in self.buffer_4d ]).float() weights = self.som.weights.detach() metrics = compute_all_metrics(data, weights, self._som_metric_history) # Registra no histórico self._som_metric_history.record(metrics) return metrics def get_som_metric_history(self) -> Dict[str, Any]: """Retorna histórico de métricas SOM (para análise temporal).""" return self._som_metric_history.to_dict() def reset_som_metric_history(self) -> None: """Reseta o histórico de métricas SOM (início de nova fase).""" from bigru_t.model.som_metrics import SOMMetricHistory self._som_metric_history = SOMMetricHistory() __all__ = [ "SimpleBBPETokenizer", "positional_encoding", "text_to_4d_vector", "KohonenSOM4D", "HypothesisClassifier", "KohonenLearningSystem", "DeltaGenerator", "HypothesisEnsemble", "KohonenLearningSystemV2", ] # V6.5-V2-metrics: som_metrics é importado lazy (dentro de compute_som_metrics) # para evitar dependência circular e permitir uso standalone.