diff --git "a/kohonen_learning_system.py" "b/kohonen_learning_system.py" deleted file mode 100644--- "a/kohonen_learning_system.py" +++ /dev/null @@ -1,2960 +0,0 @@ -"""kohonen_learning_system.py — V6.5 (VQ-VAE-2 + reasoning_engine integrados) - -V6.5 — ATIVAR VQ-VAE-2 NO PIPELINE DE COMPRESSÃO + INTEGRAR REASONING_ENGINE - -Implementação CANÔNICA do Sistema de Aprendizado Kohonen 4D fornecida pelo -usuário, com análise matemática formal e 2 correções de bugs críticos. - -============================================================================ -ANÁLISE MATEMÁTICA FORMAL -============================================================================ - -1) text_to_4d_vector(text, ..., time_step, T_max): - ----------------------------------------------- - Tokenização BBPE: ids = BBPE_encode(text, max_len=seq_len) → [seq_len] - Embedding: E = Embedding(ids) → [seq_len, D] - PE sinusoidal: PE = sin/cos(pos · exp(-2k·log(10000)/D)) - Fusão: F = E + PE → [seq_len, D] - Centralização: F_c = F - mean(F, dim=0) → [seq_len, D] - SVD: F_c = U · S · V^T (V: [D, D] ortogonal) - Projeção 3D: coords_3d = F_c · V[:3,:]^T → [seq_len, 3] - Centróide: xyz = mean(coords_3d, dim=0) → [3] - Coordenada temporal: w = time_step / T_max ∈ [0, 1] (LINEAR) - Saída: vec_4d = [x, y, z, w] → [4] - - Nota (V6.4): w é LINEAR no tempo (diferente de sigmoid(||xyz||)). - Isto permite que o SOM organize neurônios ao longo do eixo temporal, - capturando a sequência absoluta de amostras — útil para deteção - de drift e consolidação incremental (EWC). - -2) KohonenSOM4D: - -------------- - Grid: W ∈ ℝ^(I×J×K×L×4), default (6,6,6,4) = 864 neurônios - BMU: bmu = argmin_{i,j,k,l} ||W[i,j,k,l] - x||² (Euclidiana ℝ⁴) - Vizinhança (Gaussiana 4D): - Λ(d, σ) = exp(-d² / (2σ²)) - d² = Δi² + Δj² + Δk² + Δl² (distância quadrada no grid 4D) - Update (Kohonen): - ΔW = α · Λ · (x - W) (competitivo + cooperativo) - Decaimento: - σ_t = σ₀ · exp(-t/1000) (vizinhança encolhe rápido) - α_t = α₀ · exp(-t/2000) (LR decai mais lento) - EWC (apenas na 4ª dimensão w): - L_ewc = (λ/2) · Σ F_i · (W_w,i - W*_w,i)² - ∂L_ewc/∂W_w = λ · F · (W_w - W*_w) - Aplicado como: update[..., 3] -= λ · F · (W_w - W*_w) - Fisher (aproximação como erro quadrado): - F_i = mean((x_w - W_w,i)²) sobre samples pré-punição - Acumulado apenas quando: - (a) punishment_count == 0 - (b) old_weights_w is None (ainda não consolidado) - (c) Λ > 0.1 (neurônios próximos ao BMU) - -3) HypothesisClassifier: - ---------------------- - 8 camadas FC: [in → 512 → 256 → 128 → 64 → 32 → 16 → 8] + ReLU - Output: 8 → 1 (logit) - Loss: BCEWithLogitsLoss - Optimizer: Adam, lr=0.001 - Epochs: 50 (sobre o buffer atual) - Input: vetor de ativação SOM = distâncias ao grid flatten - (864-dim para grid (6,6,6,4)) - -4) Punishment Protocol: - --------------------- - Histograma: bucketiza vec_4d[dim_choice] (default 'y', idx=1) - check_training_start: - max(histogram.values()) >= N_start → ready - Avaliação: acc = correct / len(buffer) (correct = pred==label) - Se acc < 1.0: - punishment_count += 1 - success_count = 0 - Se punishment_count == 1: activate_hypothesis() (treina classifier) - Se punishment_count == 2: - set_ewc_reference() (consolida w via Fisher) - required_new_samples = success_count * N (ou N se success==0) - reset: training_ready=False, punishment=0, success=0, - histogram cleared, buffers cleared - Se acc == 1.0: - punishment_count = 0 - success_count += 1 - -5) PGVector NÃO É MAIS NECESSÁRIO (V6.4): - --------------------------------------- - O método find_bmu realiza a busca nearest-neighbor sobre o grid 4D, - substituindo qualquer lookup pgvector externo. O SOM interno já - armazena todo o conhecimento como pesos 4D, e find_bmu retorna o - neurônio mais próximo em O(I·J·K·L) — equivalente a uma busca - pgvector com indexação flat. - Consequentemente, hyp_t.py NÃO consulta mais pgvector — a decisão - de aplicar punição é delegada ao KohonenLearningSystem. - -============================================================================ -BUGS CORRIGIDOS (V6.3 → mantidos em V6.4) -============================================================================ - -BUG 1 (find_bmu): V6.3 corrigiu return prematuro com Ellipsis. - V6.4: código do usuário já está limpo (unravel manual sem return - prematuro). Mantido as-is. - -BUG 2 (activate_hypothesis): backward tentava retropropagar através - do embedding (via buffer_4d). Causava RuntimeError "Trying to - backward through the graph a second time". - FIX V6.4: detach+clone nos tensores de entrada do classifier, - e cálculo do vetor de ativação SOM dentro de torch.no_grad(). - O classifier treina apenas sobre seus próprios pesos (8 FC layers). - -============================================================================ -V6.5 — VQ-VAE-2 NO PIPELINE DE COMPRESSÃO -============================================================================ -User requirement: "ativar efetivamente o VQ-VAE-2 no pipeline de compressão" - -Integração: o KohonenLearningSystem agora possui um `vqvae2_compressor` -opcional (HierarchicalVQVAE2 do módulo vqvae2_hierarchical_flexnet.py). -Quando ativado: - -1. Após `train_som_on_buffer()`, o buffer_4d (B, 4) é passado ao VQ-VAE-2 - como entrada. O encoder mapeia (B, 4) -> z_e (B, code_dim). -2. O VQ hierárquico produz: - - z_q_top: estrutura global do batch (codebook K_top) - - z_q_bot: detalhes residuais (codebook K_bot) -3. O decoder reconstrói z_recon (B, 4) a partir de z_q_combined. -4. A loss do VQ-VAE-2 (commitment + recon) é computada e retornada - para monitoramento (não adicionada à loss do SOM — são objetivos - ortogonais: SOM aprende topologia, VQ-VAE-2 aprende compressão). -5. Códigos top/bottom podem ser usados como representação compacta - do estado do SOM para armazenamento/transferência. - -Benefícios: -- Compressão neural do espaço 4D do SOM (4 -> code_dim -> 2 códigos) -- Codebook compartilhado entre batches (aprendizado incremental) -- Dead code restart evita colapso do codebook -- Goose VQ (Gumbel-softmax) força uso uniforme do codebook - -============================================================================ -V6.5 — REASONING_ENGINE INTEGRADO -============================================================================ -User requirement: "integrar reasoning_engine ao KohonenLearningSystem" - -Integração: o KohonenLearningSystem agora possui um `reasoning_engine` -opcional (ReasoningEngine do módulo reasoning_engine.py). Quando ativado: - -1. Após `predict()`, o resultado da predição é passado ao reasoning_engine - que gera uma sequência de tags , , , - , , , , . -2. O reasoning_engine pode usar ferramentas registradas (tool_agent) - para consultas externas (ex: calculator, knowledge_base). -3. O streaming de raciocínio é compatível com Ollama/LangChain/vLLM - via tags padrão. -4. Para predições do SOM, o reasoning_engine pode explicar o porquê - do BMU ter sido escolhido (análise de distâncias). - -Métodos adicionados: -- reason_about(query): retorna generator com streaming de raciocínio -- reason_sync(query): retorna string completa com todas as tags -- get_reasoning_stats(): retorna estatísticas do reasoning_engine - -============================================================================ -""" -from __future__ import annotations - -import os -import math -import torch -import torch.nn as nn -import torch.nn.functional as F -from collections import defaultdict, Counter -from typing import List, Tuple, Optional, Dict, Any, Iterator - - -# ============================================================================ -# Configuração da CPU – núcleos físicos -# ============================================================================ -try: - import psutil - N_CORES = psutil.cpu_count(logical=False) -except ImportError: - N_CORES = os.cpu_count() // 2 if os.cpu_count() else 4 -if N_CORES: - try: - torch.set_num_threads(N_CORES) - torch.set_num_interop_threads(N_CORES) - except RuntimeError: - pass # already initialized - - -# ============================================================================ -# Tokenizador BBPE simplificado -# ============================================================================ -class SimpleBBPETokenizer: - """Tokenizador BPE simplificado (nível de palavra). - - Vocabulário especial: pad=0, eos=1, unk=2. - Tokens comuns mapeados a partir do índice 3. - """ - - def __init__(self, vocab_size: int = 16384): - self.vocab_size = vocab_size - self.token_to_id = {} - self.id_to_token = {} - self.pad_token = "" - self.eos_token = "" - self.unk_token = "" - self.pad_id = 0 - self.eos_id = 1 - self.unk_id = 2 - self._init_special_tokens() - - def _init_special_tokens(self): - self.token_to_id[self.pad_token] = self.pad_id - self.token_to_id[self.eos_token] = self.eos_id - self.token_to_id[self.unk_token] = self.unk_id - self.id_to_token[self.pad_id] = self.pad_token - self.id_to_token[self.eos_id] = self.eos_token - self.id_to_token[self.unk_id] = self.unk_token - - def fit(self, texts: List[str]): - """Constrói vocabulário por frequência (top vocab_size-3 palavras).""" - word_counts = Counter() - for text in texts: - words = text.split() - word_counts.update(words) - sorted_words = [w for w, _ in word_counts.most_common(self.vocab_size - 3)] - for idx, word in enumerate(sorted_words, start=3): - self.token_to_id[word] = idx - self.id_to_token[idx] = word - - def encode(self, text: str, max_length: int = 8) -> List[int]: - """Encoda + append EOS + pad/trunca para max_length.""" - words = text.split() - ids = [self.token_to_id.get(w, self.unk_id) for w in words] - ids.append(self.eos_id) - if len(ids) > max_length: - ids = ids[:max_length] - else: - ids += [self.pad_id] * (max_length - len(ids)) - return ids - - -# ============================================================================ -# Codificação posicional e conversão texto → vetor 4D (com w temporal) -# ============================================================================ -def positional_encoding(seq_len: int, hidden_dim: int) -> torch.Tensor: - """PE sinusoidal clássico: PE[pos, 2k] = sin(pos·exp(-2k·log(10000)/D)), - PE[pos, 2k+1] = cos(pos·exp(-2k·log(10000)/D)). - """ - pe = torch.zeros(seq_len, hidden_dim) - position = torch.arange(0, seq_len, dtype=torch.float).unsqueeze(1) - div_term = torch.exp( - torch.arange(0, hidden_dim, 2).float() * (-math.log(10000.0) / hidden_dim) - ) - pe[:, 0::2] = torch.sin(position * div_term) - pe[:, 1::2] = torch.cos(position * div_term) - return pe - - -def text_to_4d_vector( - text: str, - tokenizer: SimpleBBPETokenizer, - embedding: nn.Embedding, - hidden_dim: int, - seq_len: int, - time_step: int, - T_max: int = 10000, - attention: Optional[nn.Module] = None, - attention_metrics: Optional[Dict[str, Any]] = None, -) -> torch.Tensor: - """Converte sentença em vetor 4D (x, y, z, w), onde w = time_step / T_max. - - Pipeline matemático (V6.5-attn: + MultiHeadAttention entre +PE e SVD): - ids → Embedding → +PE → [MultiHeadAttention] → SVD(proj top-3) → centróide xyz - w = time_step / T_max (LINEAR no tempo) - vec_4d = concat(xyz, w) → [4] - - V6.5-attn (user requirement: "verificar se o mecanismo de atenção está ativo - e acessado logicamente funcional"): - Quando `attention` é fornecido (nn.Module), o tensor `fused (L, D)` é - passado pelo attention (self-attention com máscara causal) antes do SVD. - O attention produz um tensor residual (fused + attn_out) preservando a - dimensão. Métricas são coletadas em `attention_metrics` para verificação. - - Args: - attention: Optional[nn.Module] — MultiHeadAttention do attention_multimodal. - attention_metrics: Optional[Dict] — acumula estatísticas do attention. - Se fornecido, é atualizado in-place com: n_calls, last_norm_in, - last_norm_out, last_attn_activated (True se output != input). - """ - ids = tokenizer.encode(text, max_length=seq_len) - input_ids = torch.tensor(ids).unsqueeze(0) - word_emb = embedding(input_ids).squeeze(0) # (L, D) - pe = positional_encoding(seq_len, hidden_dim) - fused = word_emb + pe # (L, D) - - # V6.5-attn — Aplica MultiHeadAttention (self-attention) sobre o fused - # User requirement: "verificar se o mecanismo de atenção está ativo e - # acessado logicamente funcional" - if attention is not None: - try: - # Prepare input: (1, L, D) — batch dim necessária para MHA - fused_in = fused.unsqueeze(0) # (1, L, D) - norm_in = float(fused_in.norm().item()) - # Forward — passa mask=None (sem causal mask; seq_len curto) - attn_out = attention(fused_in, mask=None) # (1, L, D) - # Residual connection (preserva dimensão) - fused_attn = fused_in + attn_out # (1, L, D) - fused = fused_attn.squeeze(0) # (L, D) - norm_out = float(fused.norm().item()) - # Métricas de verificação (in-place se dict fornecido) - if attention_metrics is not None: - attention_metrics["n_calls"] = attention_metrics.get("n_calls", 0) + 1 - attention_metrics["last_norm_in"] = norm_in - attention_metrics["last_norm_out"] = norm_out - # "logicamente funcional" = output é finito e diferente do input - attention_metrics["last_attn_activated"] = bool( - torch.isfinite(attn_out).all().item() - and (attn_out.abs().sum().item() > 1e-8) - ) - attention_metrics["last_attn_diff_norm"] = float( - (attn_out.squeeze(0) - fused_in.squeeze(0)).norm().item() - ) - except Exception as _e_attn: - # Em caso de erro no attention, segue sem attention mas registra - if attention_metrics is not None: - attention_metrics["n_errors"] = attention_metrics.get("n_errors", 0) + 1 - attention_metrics["last_error"] = str(_e_attn)[:200] - - # SVD para 3D - mean_centered = fused - fused.mean(dim=0, keepdim=True) - U, S, V = torch.linalg.svd(mean_centered, full_matrices=False) - coords_3d = torch.mm(mean_centered, V[:3, :].t()) # (L, 3) - - xyz_mean = coords_3d.mean(dim=0) # (3,) - w = torch.tensor(time_step / T_max, dtype=torch.float) # valor temporal - vec_4d = torch.cat([xyz_mean, w.unsqueeze(0)]).contiguous() - return vec_4d - - -# ============================================================================ -# Mapa de Kohonen 4D com EWC (Fisher para w não-nulo) -# ============================================================================ -class KohonenSOM4D: - """Mapa Auto-Organizável 4D com EWC apenas na 4ª dimensão (w temporal). - - Args: - grid_shape: (I, J, K, L) — dimensões do grid 4D. - alpha0: taxa de aprendizado inicial (α₀). - sigma0: largura inicial da vizinhança (σ₀). - lambda_ewc: peso da penalidade EWC (λ). - - Atributos: - weights: W ∈ ℝ^(I×J×K×L×4) — pesos dos neurônios. - old_weights_w: W*_w ∈ ℝ^(I×J×K×L) — referência EWC (apenas w). - fisher_w: F ∈ ℝ^(I×J×K×L) — informação de Fisher por neurônio (w). - fisher_accum / fisher_count: acumuladores para cálculo de F. - """ - - def __init__( - self, - grid_shape: Tuple[int, int, int, int], - alpha0: float = 0.1, - sigma0: float = 1.0, - lambda_ewc: float = 0.01, - ): - self.I, self.J, self.K, self.L = grid_shape - self.alpha0 = alpha0 - self.sigma0 = sigma0 - self.lambda_ewc = lambda_ewc - self.t = 0 - - self.weights = torch.randn(self.I, self.J, self.K, self.L, 4) - self.old_weights_w = None - self.fisher_w = None - self.fisher_accum = torch.zeros(self.I, self.J, self.K, self.L) - self.fisher_count = torch.zeros(self.I, self.J, self.K, self.L) - - def _neighborhood(self, bmu_idx): - """Vizinhança Gaussiana 4D: d² = Δi² + Δj² + Δk² + Δl².""" - i, j, k, l = bmu_idx - II, JJ, KK, LL = torch.meshgrid( - torch.arange(self.I).float(), - torch.arange(self.J).float(), - torch.arange(self.K).float(), - torch.arange(self.L).float(), - indexing="ij", - ) - dist_sq = (II - i) ** 2 + (JJ - j) ** 2 + (KK - k) ** 2 + (LL - l) ** 2 - return dist_sq - - def find_bmu(self, x: torch.Tensor) -> Tuple[int, int, int, int]: - """Best Matching Unit: argmin ||W - x||² em ℝ⁴. - - Substitui pgvector_lookup — busca nearest-neighbor flat sobre o grid. - """ - dist = torch.sum((self.weights - x.view(1, 1, 1, 1, 4)) ** 2, dim=-1) - flat_idx = torch.argmin(dist).item() - i = flat_idx // (self.J * self.K * self.L) - rest = flat_idx % (self.J * self.K * self.L) - j = rest // (self.K * self.L) - rest = rest % (self.K * self.L) - k = rest // self.L - l = rest % self.L - return (i, j, k, l) - - def update_weights(self, x: torch.Tensor, bmu_idx, accumulate_fisher=False): - """Update Kohonen: ΔW = α·Λ·(x - W) + penalidade EWC em w. - - Args: - x: tensor [4] — amostra 4D. - bmu_idx: (i, j, k, l) — índice do BMU. - accumulate_fisher: se True, acumula (x_w - W_w)² nos Fisher accumulators. - """ - dist_sq = self._neighborhood(bmu_idx) - sigma = self.sigma0 * math.exp(-self.t / 1000) - alpha = self.alpha0 * math.exp(-self.t / 2000) - h = torch.exp(-dist_sq / (2 * sigma ** 2)) - - delta = x - self.weights - update = alpha * h.unsqueeze(-1) * delta - - if self.old_weights_w is not None and self.fisher_w is not None: - # Penalidade EWC apenas na 4ª dimensão (w) - # ∂L_ewc/∂W_w = λ · F · (W_w - W*_w) → subtraído do update - ewc_penalty = self.lambda_ewc * self.fisher_w * ( - self.weights[..., 3] - self.old_weights_w - ) - update[..., 3] = update[..., 3] - ewc_penalty - - self.weights = self.weights + update - - if accumulate_fisher: - # Acumula Fisher apenas em neurônios próximos ao BMU (Λ > 0.1) - mask = h > 0.1 - if mask.any(): - diff_sq = (x[3] - self.weights[mask][..., 3]) ** 2 - self.fisher_accum[mask] += diff_sq - self.fisher_count[mask] += 1 - - self.t += 1 - - def finalize_fisher(self): - """Fisher = mean((x_w - W_w)²) sobre samples acumuladas.""" - cnt = self.fisher_count.clamp(min=1e-8) - self.fisher_w = self.fisher_accum / cnt - - def set_ewc_reference(self): - """Consolida W_w como referência EWC e finaliza Fisher.""" - self.old_weights_w = self.weights[..., 3].clone() - self.finalize_fisher() - self.fisher_accum.zero_() - self.fisher_count.zero_() - - # ------------------------------------------------------------------ - # Métricas para monitoramento (V6.4) - # ------------------------------------------------------------------ - def get_metrics(self) -> dict: - """Retorna métricas atuais do SOM para monitoramento.""" - sigma_t = self.sigma0 * math.exp(-self.t / 1000) - alpha_t = self.alpha0 * math.exp(-self.t / 2000) - return { - "t": int(self.t), - "sigma_t": float(sigma_t), - "alpha_t": float(alpha_t), - "sigma0": float(self.sigma0), - "alpha0": float(self.alpha0), - "lambda_ewc": float(self.lambda_ewc), - "grid_shape": [int(self.I), int(self.J), int(self.K), int(self.L)], - "n_neurons": int(self.I * self.J * self.K * self.L), - "has_ewc_reference": self.old_weights_w is not None, - "fisher_w_mean": ( - float(self.fisher_w.mean().item()) - if self.fisher_w is not None - else 0.0 - ), - "fisher_w_max": ( - float(self.fisher_w.max().item()) - if self.fisher_w is not None - else 0.0 - ), - "fisher_accum_count": int(self.fisher_count.sum().item()), - "weights_norm": float(self.weights.norm().item()), - "weights_w_mean": float(self.weights[..., 3].mean().item()), - } - - -# ============================================================================ -# Classificador de hipótese (8 camadas FC) -# ============================================================================ -class HypothesisClassifier(nn.Module): - """Classificador de hipótese: 8 camadas FC + ReLU + output logit. - - Arquitetura: [input → 512 → 256 → 128 → 64 → 32 → 16 → 8] + ReLU - + [8 → 1] (logit) - Loss: BCEWithLogitsLoss - """ - - def __init__(self, input_dim, hidden_dims=[512, 256, 128, 64, 32, 16, 8]): - super().__init__() - layers = [] - prev = input_dim - for h in hidden_dims: - layers.append(nn.Linear(prev, h)) - layers.append(nn.ReLU()) - prev = h - layers.append(nn.Linear(prev, 1)) - self.net = nn.Sequential(*layers) - - def forward(self, x): - return self.net(x).squeeze(-1) - - -# ============================================================================ -# Sistema de aprendizado completo (com w temporal e condição de início por N) -# ============================================================================ -class KohonenLearningSystem: - """Pipeline integrado: tokenizer + embedding + SOM4D + classifier + punishment. - - V6.5: + VQ-VAE-2 compressor (opcional) + reasoning_engine (opcional) - - Args: - vocab_size: tamanho do vocabulário BBPE (default 16384). - hidden_dim: dimensão do embedding (default 1024). - seq_len: comprimento máximo da sequência (default 8). - som_grid: (I, J, K, L) — grid 4D do SOM (default (6, 6, 6, 4) = 864). - alpha0, sigma0: hiperparâmetros do SOM. - lambda_ewc: peso da penalidade EWC. - N_start: threshold do histograma para iniciar treino. - dim_choice: 'x' | 'y' | 'z' — dimensão usada no histograma. - hypothesis_hidden: arquitetura do HypothesisClassifier. - T_max: normalização temporal (w = time_step / T_max). - enable_vqvae2 (V6.5): ativa VQ-VAE-2 compressor no pipeline. - enable_reasoning (V6.5): ativa reasoning_engine integrado. - vqvae2_code_dim (V6.5): dimensão do codebook do VQ-VAE-2. - vqvae2_num_codes (V6.5): tamanho do codebook top+bottom. - """ - - def __init__( - self, - vocab_size=16384, - hidden_dim=1024, - seq_len=8, - som_grid=(6, 6, 6, 4), - alpha0=0.1, - sigma0=1.5, - lambda_ewc=0.02, - N_start=10, - dim_choice="y", - hypothesis_hidden=[512, 256, 128, 64, 32, 16, 8], - T_max=10000, - # V6.5 — VQ-VAE-2 + reasoning_engine - enable_vqvae2: bool = True, - enable_reasoning: bool = True, - vqvae2_code_dim: int = 16, - vqvae2_num_codes_top: int = 64, - vqvae2_num_codes_bot: int = 128, - # V6.5-final — SmoothQuant W8A8 compressor for SOM weights - enable_w8a8: bool = True, - w8a8_alpha: float = 0.5, - w8a8_n_bits: int = 8, - w8a8_calibration_samples: int = 128, - # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector - # User requirement: "verificar se o mecanismo de atenção está ativo - # e acessado logicamente funcional" - enable_attention: bool = True, - attention_n_heads: int = 8, - ): - self.tokenizer = SimpleBBPETokenizer(vocab_size) - self.embedding = nn.Embedding(vocab_size, hidden_dim) - self.hidden_dim = hidden_dim - self.seq_len = seq_len - self.T_max = T_max - self.time_counter = 0 # contador global de amostras processadas - - self.som = KohonenSOM4D(som_grid, alpha0, sigma0, lambda_ewc) - self.som_grid = som_grid - self.som_neuron_count = ( - som_grid[0] * som_grid[1] * som_grid[2] * som_grid[3] - ) - - self.classifier: Optional[HypothesisClassifier] = None - self.hypothesis_hidden = hypothesis_hidden - self.classifier_trained = False - - self.buffer_4d = [] - self.buffer_labels = [] - # V6.5-V2-memfix — Sliding window limit for buffer_4d to prevent OOM. - # User requirement: "resolver falhas de lógica e de bugs que estejam - # causando alto consumo de memória sem distorcer a arquitetura Kohonen". - # The buffer stores vec_4d tensors for SOM training and neuron labeling. - # Without a limit, streaming 8000+ samples accumulates 8000 tensors, - # each retaining the embedding computation graph (16M params), causing - # OOM (3.5GB RSS observed). Sliding window of 512 keeps recent samples - # for SOM updates while bounding memory. The Kohonen architecture (SOM - # grid, BMU, Gaussian neighborhood, EWC) is NOT changed. - self.buffer_max_size = 512 - self.training_ready = False - self.N = N_start - self.dim_choice = dim_choice - self.dim_index = {"x": 0, "y": 1, "z": 2}[dim_choice] - - self.punishment_count = 0 - self.success_count = 0 - self.histogram = Counter() - - self.required_new_samples = 0 - - # ------------------------------------------------------------------ - # V6.5-fix — Dynamic Label Registry (replaces hardcoded 'gato'/'cachorro') - # ------------------------------------------------------------------ - # User requirement: "os strings 'gato' e 'cachorro' são fixos quando - # deveriam ser extrações variáveis e flexíveis de rótulos proveniente - # de dados dos datasets anteriormente treinados" - # - # Mathematical analysis: - # During training, add_data() receives integer labels (0/1) produced - # by make_label(text) — these integers are dataset-agnostic. The - # semantic meaning of "0" vs "1" depends on the dataset: - # - punctuation datasets: 0 = "unpunctuated", 1 = "punctuated" - # - chat datasets: 0 = "user_turn", 1 = "assistant_turn" - # - instruct datasets: 0 = "instruction", 1 = "response" - # - news datasets: 0 = "headline", 1 = "body" - # - corpus datasets: 0 = "raw_text", 1 = "compressed_text" - # - # The label_registry stores, per dataset, the {label_int: label_str} - # mapping. predict() looks up the dynamic string instead of returning - # a hardcoded "gato"/"cachorro". - # - # Math: label_str = registry[dataset_name][label_int] - # where label_int = 1 if sigmoid(logit) > 0.5 else 0 - # - # Fallback chain (when no dataset-specific registry exists): - # 1. Most recently registered dataset mapping. - # 2. Default {0: "negative", 1: "positive"}. - # 3. Synthetic "label_{int}" if all else fails. - self.label_registry: Dict[str, Dict[int, str]] = {} - self.default_label_strings: Dict[int, str] = { - 0: "negative", - 1: "positive", - } - self.last_dataset_used: Optional[str] = None - # V6.5-fix — Inference punishment tracking (mirrors training protocol) - # User requirement: "ao punir o modelo reajusta seus parâmetros usando - # a camada de hipótese para funcionar como na lógica de punição e - # ajuste de parâmetros durante o treinamento" - self.inference_punishment_count: int = 0 - self.inference_punishment_history: List[Dict[str, Any]] = [] - - # ------------------------------------------------------------------ - # V6.5 — VQ-VAE-2 compressor (ativa efetiva no pipeline) - # ------------------------------------------------------------------ - self.enable_vqvae2 = enable_vqvae2 - self.vqvae2_compressor = None - self.vqvae2_metrics_history: List[Dict[str, Any]] = [] - if enable_vqvae2: - try: - from .vqvae2_hierarchical_flexnet import HierarchicalVQVAE2 - # Modalidade única: "som_4d" com input_dim=4 - self.vqvae2_compressor = HierarchicalVQVAE2( - modalities={"som_4d": 4}, - code_dim=vqvae2_code_dim, - num_codes_top=vqvae2_num_codes_top, - num_codes_bot=vqvae2_num_codes_bot, - hidden=32, - beta=0.25, - ema_decay=0.99, - dead_code_threshold=1.0, - dead_code_restart_every=3, - goose_temp_init=2.0, - goose_temp_final=0.5, - goose_schedule="cosine", - total_epochs=25, - norm_type="none", - rmsnorm_in_vq=False, - ) - # Inicia em modo treino para ativar EMA updates - self.vqvae2_compressor.train() - except Exception as e: - # Fallback: desabilita VQ-VAE-2 se houver erro de import - self.enable_vqvae2 = False - self.vqvae2_compressor = None - import warnings - warnings.warn(f"VQ-VAE-2 disabled: {e}") - - # ------------------------------------------------------------------ - # V6.5-final — SmoothQuantCompressor for REAL W8A8 of SOM weights - # ------------------------------------------------------------------ - # User requirement: "integrar SmoothQuantCompressor ao KLS para - # compressão W8A8 real dos pesos do SOM" - # - # O SmoothQuantCompressor é instanciado e calibrado sobre os pesos - # atuais do SOM (self.som.weights reshape para [N_neurons, 4]) usando - # o buffer_4d como amostras de ativação reais. Após calibração, o - # método compress_som_weights_w8a8() aplica quantização INT8 simétrica - # (com smoothing) e dequantização imediata, preservando a topologia - # do SOM com erro relativo < 1%. - self.enable_w8a8 = enable_w8a8 - self.w8a8_compressor = None - self.w8a8_metrics_history: List[Dict[str, Any]] = [] - if enable_w8a8: - try: - from ..quantization.smoothquant_compressor import SmoothQuantCompressor - self.w8a8_compressor = SmoothQuantCompressor( - alpha=w8a8_alpha, - n_bits=w8a8_n_bits, - calibration_samples=w8a8_calibration_samples, - ) - except Exception as e: - self.enable_w8a8 = False - self.w8a8_compressor = None - import warnings - warnings.warn(f"SmoothQuantCompressor disabled: {e}") - - # ------------------------------------------------------------------ - # V6.5 — ReasoningEngine (integração ativa + tool_coordinator workers reativados) - # ------------------------------------------------------------------ - self.enable_reasoning = enable_reasoning - self.reasoning_engine = None - if enable_reasoning: - try: - from ..reasoning.reasoning_engine import ReasoningEngine - # V6.5-final: REATIVAR tool_coordinator workers — explicit n_workers - # User requirement: "reativar reasoning_engine's tool_coordinator workers" - # O ReasoningEngine cria internamente ToolAgentCoordinator(n_workers=4) - # que mantém um ThreadPoolExecutor ativo. Para garantir que os - # workers sejam de fato utilizados (não apenas instanciados), - # o método _compress_buffer_with_vqvae2 abaixo chama - # execute_parallel() periodicamente durante o treino. - self.reasoning_engine = ReasoningEngine( - max_thinking_steps=10, - max_iterations=3, - convergence_threshold=0.9, - verbose=False, - ) - # Verifica que tool_coordinator + workers estão ativos - tc = self.reasoning_engine.tool_coordinator - self._tool_coordinator_workers_active = ( - tc is not None and tc._executor is not None - ) - # Registra uma ferramenta interna: consultar SOM - def som_query_tool(query: str) -> str: - """Ferramenta: consulta o SOM do KLS para responder.""" - pred = self.predict(query) - bmu_info = "" - if self.buffer_4d: - try: - vec = text_to_4d_vector( - query, self.tokenizer, self.embedding, - self.hidden_dim, self.seq_len, - self.time_counter, self.T_max, - ) - bmu = self.som.find_bmu(vec) - bmu_info = f" | BMU={bmu}" - except Exception: - pass - return f"prediction={pred}{bmu_info}" - self.reasoning_engine.register_tool( - "som_query", som_query_tool, - description="Consulta o SOM do KohonenLearningSystem", - timeout_s=10.0, - ) - # V6.5-final: registra segunda ferramenta para forçar uso - # paralelo dos workers via execute_parallel() - def buffer_stats_tool(query: str) -> str: - """Ferramenta: retorna estatísticas do buffer atual.""" - return ( - f"buffer_size={len(self.buffer_4d)}, " - f"punishment_count={self.punishment_count}, " - f"success_count={self.success_count}, " - f"training_ready={self.training_ready}" - ) - self.reasoning_engine.register_tool( - "buffer_stats", buffer_stats_tool, - description="Estatísticas do buffer interno do KLS", - timeout_s=5.0, - ) - except Exception as e: - self.enable_reasoning = False - self.reasoning_engine = None - self._tool_coordinator_workers_active = False - import warnings - warnings.warn(f"ReasoningEngine disabled: {e}") - - # ------------------------------------------------------------------ - # V6.5-attn — MultiHeadAttention integrado ao text_to_4d_vector - # ------------------------------------------------------------------ - # User requirement: "verificar se o mecanismo de atenção está ativo - # e acessado logicamente funcional" - # - # O MultiHeadAttention (de attention_multimodal.py) é instanciado com - # d_model=hidden_dim e n_heads=attention_n_heads. Ele é aplicado em - # text_to_4d_vector entre a fusão (+PE) e o SVD, atuando como - # self-attention residual sobre a sequência embedded. - # - # Métricas (n_calls, last_norm_in/out, last_attn_activated) são - # acumuladas em self.attention_metrics e expostas via - # get_attention_metrics() para verificação de "ativo e funcional". - self.enable_attention = enable_attention - self.attention = None - self.attention_metrics: Dict[str, Any] = { - "n_calls": 0, - "n_errors": 0, - "last_norm_in": 0.0, - "last_norm_out": 0.0, - "last_attn_activated": False, - "last_attn_diff_norm": 0.0, - } - if enable_attention: - try: - from .attention_multimodal import MultiHeadAttention - # d_model = hidden_dim, n_heads = attention_n_heads - # assert hidden_dim % n_heads == 0 - n_heads = attention_n_heads - if hidden_dim % n_heads != 0: - # ajusta para divisor válido mais próximo - for cand in [8, 4, 2, 1]: - if hidden_dim % cand == 0: - n_heads = cand - break - self.attention = MultiHeadAttention( - d_model=hidden_dim, - n_heads=n_heads, - dropout=0.0, - max_chunk=4096, - ) - # Em eval mode para determinismo (não treinamos o attention separadamente) - self.attention.eval() - self.attention_n_heads = n_heads - except Exception as e: - self.enable_attention = False - self.attention = None - import warnings - warnings.warn(f"MultiHeadAttention disabled: {e}") - - def get_attention_metrics(self) -> Dict[str, Any]: - """V6.5-attn — Retorna métricas do MultiHeadAttention integrado. - - User requirement: "verificar se o mecanismo de atenção está ativo - e acessado logicamente funcional" - - Returns: - Dict com: - - active: bool — attention module está instanciado - - n_calls: int — número de chamadas forward - - n_errors: int — número de erros silenciados - - last_norm_in / last_norm_out: normas do input/output - - last_attn_activated: bool — output é finito e diferente do input - - last_attn_diff_norm: float — ||attn_out - input|| - - logic_functional: bool — n_calls > 0 AND n_errors == 0 - AND last_attn_activated == True - """ - m = self.attention_metrics - return { - "active": bool(self.attention is not None), - "n_calls": int(m.get("n_calls", 0)), - "n_errors": int(m.get("n_errors", 0)), - "last_norm_in": float(m.get("last_norm_in", 0.0)), - "last_norm_out": float(m.get("last_norm_out", 0.0)), - "last_attn_activated": bool(m.get("last_attn_activated", False)), - "last_attn_diff_norm": float(m.get("last_attn_diff_norm", 0.0)), - "n_heads": int(getattr(self, "attention_n_heads", 0)), - # "logicamente funcional" = foi chamado, sem erros, e última chamada - # produziu output finito e diferente do input - "logic_functional": bool( - self.attention is not None - and m.get("n_calls", 0) > 0 - and m.get("n_errors", 0) == 0 - and m.get("last_attn_activated", False) - ), - } - - # ================================================================== - # V6.5-fix — Dynamic Label Registry API - # ================================================================== - # User requirement: "os strings 'gato' e 'cachorro' são fixos quando - # deveriam ser extrações variáveis e flexíveis de rótulos proveniente - # de dados dos datasets anteriormente treinados" - # - # Mathematical formalization: - # Let D = {d_1, d_2, ..., d_K} be the set of datasets seen during - # training. For each d_k, the training script provides a mapping - # μ_k : {0, 1} → Σ (where Σ is the string-label alphabet). - # - # The label_registry stores R = {(d_k, μ_k) : k = 1..K}. - # - # At inference time, given a query sentence s: - # 1. vec_4d = text_to_4d_vector(s, ...) - # 2. p = _predict_single(vec_4d) ∈ [0, 1] - # 3. label_int = 1 if p > 0.5 else 0 - # 4. label_str = μ_last(label_int) [fallback chain if unknown] - # - # If dataset_name is provided at inference, uses μ_dataset_name; - # otherwise uses the most recently registered μ (assumes temporal - # coherence: latest trained dataset is most relevant). - # ------------------------------------------------------------------ - def register_dataset_labels( - self, - dataset_name: str, - label_strings: Dict[int, str], - ) -> None: - """V6.5-fix — Registra mapeamento {label_int: label_str} para um dataset. - - Args: - dataset_name: nome do dataset (ex: 'dominguesm/restore-punctuation-ptbr-dataset'). - label_strings: dict {0: 'unpunctuated', 1: 'punctuated'}. - """ - if not isinstance(label_strings, dict): - raise TypeError(f"label_strings must be dict, got {type(label_strings)}") - # Sanitize keys to int, values to str - sanitized = {int(k): str(v) for k, v in label_strings.items()} - self.label_registry[dataset_name] = sanitized - self.last_dataset_used = dataset_name - - def get_label_string( - self, - label_int: int, - dataset_name: Optional[str] = None, - ) -> str: - """V6.5-fix — Resolve label_int → label_str dinamicamente. - - Fallback chain (mathematically well-defined): - 1. If dataset_name is registered, use registry[dataset_name][label_int]. - 2. Else if any dataset is registered, use the most recent one. - 3. Else use default_label_strings. - 4. Final fallback: synthetic f"label_{label_int}". - - Args: - label_int: 0 ou 1. - dataset_name: opcional — nome do dataset para disambiguar. - - Returns: - String label correspondente. - """ - label_int = int(label_int) - # 1. Dataset-specific lookup - if dataset_name is not None and dataset_name in self.label_registry: - mapping = self.label_registry[dataset_name] - if label_int in mapping: - return mapping[label_int] - # 2. Most recently registered mapping - if self.label_registry: - last_ds = ( - dataset_name if dataset_name in self.label_registry - else (self.last_dataset_used or list(self.label_registry.keys())[-1]) - ) - if last_ds and last_ds in self.label_registry: - mapping = self.label_registry[last_ds] - if label_int in mapping: - return mapping[label_int] - # 3. Default mapping - if label_int in self.default_label_strings: - return self.default_label_strings[label_int] - # 4. Synthetic fallback - return f"label_{label_int}" - - def get_label_registry_state(self) -> Dict[str, Any]: - """V6.5-fix — Retorna estado atual do label_registry para inspeção.""" - return { - "n_datasets_registered": int(len(self.label_registry)), - "datasets": { - ds: {int(k): str(v) for k, v in mapping.items()} - for ds, mapping in self.label_registry.items() - }, - "last_dataset_used": str(self.last_dataset_used) if self.last_dataset_used else None, - "default_label_strings": {int(k): str(v) for k, v in self.default_label_strings.items()}, - "inference_punishment_count": int(self.inference_punishment_count), - "inference_punishment_history_len": int(len(self.inference_punishment_history)), - } - - def add_data( - self, - sentences: List[str], - labels: List[int], - label_strings: Optional[Dict[int, str]] = None, - dataset_name: Optional[str] = None, - ): - """V6.5-fix — Adiciona amostras: text → 4D vector + atualiza histograma. - - Args: - sentences: lista de textos. - labels: lista de inteiros (0/1). - label_strings: opcional — {0: 'str0', 1: 'str1'} para o dataset. - dataset_name: opcional — nome do dataset (para registrar label_strings). - - Mathematical extension (V6.5-fix): - Se label_strings e dataset_name são fornecidos, registra o mapeamento - no label_registry. Isto permite que predict() retorne o rótulo - textual correto ao invés de hardcoded 'gato'/'cachorro'. - """ - # Register label mapping if provided - if label_strings is not None and dataset_name is not None: - self.register_dataset_labels(dataset_name, label_strings) - if dataset_name is not None: - self.last_dataset_used = dataset_name - - # V6.5-V2-memfix — Use torch.no_grad() to avoid building computation - # graphs during vec_4d computation. The embedding graph (16M params) - # would otherwise be retained by every tensor in buffer_4d, causing OOM - # with 8000+ samples. The vec_4d is a 4-float summary (x,y,z,w) that - # does NOT need gradient — it feeds into SOM (Hebbian, no backprop) and - # the HypothesisClassifier (which trains its own weights). - with torch.no_grad(): - for sent, lab in zip(sentences, labels): - self.time_counter += 1 - vec = text_to_4d_vector( - sent, - self.tokenizer, - self.embedding, - self.hidden_dim, - self.seq_len, - self.time_counter, - self.T_max, - attention=self.attention, # V6.5-attn - attention_metrics=self.attention_metrics, # V6.5-attn - ) - # CRITICAL: detach to break any residual graph + clone for safety - self.buffer_4d.append(vec.detach().clone()) - self.buffer_labels.append(lab) - dim_val = round(vec[self.dim_index].item(), 2) - self.histogram[dim_val] += 1 - # V6.5-V2-memfix — Enforce sliding window limit - if len(self.buffer_4d) > self.buffer_max_size: - overflow = len(self.buffer_4d) - self.buffer_max_size - self.buffer_4d = self.buffer_4d[-self.buffer_max_size:] - self.buffer_labels = self.buffer_labels[-self.buffer_max_size:] - - def check_training_start(self) -> bool: - """Inicia treino quando algum bucket do histograma atinge N.""" - if ( - not self.training_ready - and max(self.histogram.values(), default=0) >= self.N - ): - self.training_ready = True - return True - return False - - def train_som_on_buffer(self): - """Treina SOM por 5 épocas sobre o buffer atual (com Fisher accum). - - V6.5-V2-memfix — Wrapped in torch.no_grad() to prevent graph building - during SOM Hebbian updates. The SOM uses competitive learning (BMU + - Gaussian neighborhood), not backprop — no gradient is needed. - """ - if not self.buffer_4d: - return - # V6.5-V2-memfix — detach+stack to ensure no graph leaks from buffer - data = torch.stack(self.buffer_4d).detach() - with torch.no_grad(): - for _ in range(5): # épocas de treino rápido - perm = torch.randperm(len(data)) - for idx in perm: - x = data[idx] - bmu = self.som.find_bmu(x) - acc_fisher = ( - self.punishment_count == 0 - and self.som.old_weights_w is None - ) - self.som.update_weights(x, bmu, accumulate_fisher=acc_fisher) - - # V6.5 — Ativa VQ-VAE-2 compressor no pipeline - if self.enable_vqvae2 and self.vqvae2_compressor is not None: - self._compress_buffer_with_vqvae2(data) - - # V6.5-final — Ativa SmoothQuant W8A8 compressor dos pesos do SOM - # User requirement: "integrar SmoothQuantCompressor ao KLS para - # compressão W8A8 real dos pesos do SOM" - if self.enable_w8a8 and self.w8a8_compressor is not None: - self.compress_som_weights_w8a8() - - # V6.5-final — Reativa tool_coordinator workers via execute_parallel - # User requirement: "reativar reasoning_engine's tool_coordinator workers" - # O ThreadPoolExecutor (n_workers=4) só é efetivamente utilizado quando - # execute_parallel() é chamado com múltiplas ferramentas. Caso contrário, - # os workers ficam idle e o recurso é desperdiçado. - 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] - - Args: - sentence: texto de entrada. - dataset_name: opcional — nome do dataset para disambiguar rótulos. - return_proba: se True, retorna (label_str, prob). - - Returns: - str (ou Tuple[str, float]) — rótulo dinâmico extraído dos datasets. - """ - self.time_counter += 1 # mantém coerência temporal - vec = text_to_4d_vector( - sentence, - self.tokenizer, - self.embedding, - self.hidden_dim, - self.seq_len, - self.time_counter, - self.T_max, - attention=self.attention, # V6.5-attn - attention_metrics=self.attention_metrics, # V6.5-attn - ) - prob = self._predict_single(vec) - # CORREÇÃO MATEMÁTICA: label_int derivado de prob, não hardcoded - label_int = 1 if prob > 0.5 else 0 - # Rótulo textual DINÂMICO extraído do label_registry - label_str = self.get_label_string(label_int, dataset_name) - if return_proba: - return label_str, float(prob) - return label_str - - def predict_proba( - self, - sentence: str, - dataset_name: Optional[str] = None, - ) -> Tuple[str, float]: - """V6.5-fix — Prediz (label_str, prob) — conveniência para inferência.""" - return self.predict(sentence, dataset_name=dataset_name, return_proba=True) - - # ================================================================== - # V6.5-fix — Inference-time Punishment via Hypothesis Layer - # ================================================================== - # User requirement: "ao punir o modelo reajusta seus parâmetros usando - # a camada de hipótese para funcionar como na lógica de punição e - # ajuste de parâmetros durante o treinamento" - # - # Mathematical principle (mirrors training punishment protocol): - # - # TRAINING protocol (process_batch): - # acc = evaluate_classification(buffer) - # if acc < 1.0: - # punishment_count += 1 - # success_count = 0 - # if punishment_count == 1: - # activate_hypothesis() # treina HypothesisClassifier - # elif punishment_count == 2: - # set_ewc_reference() # consolida w via Fisher - # reset cycle - # - # INFERENCE protocol (punish_during_inference) — V6.5-fix: - # When external feedback indicates the predicted label was WRONG: - # 1. Compute vec_4d for the sentence. - # 2. Add (sentence, correct_label_int) to buffer_4d + buffer_labels. - # 3. Update SOM via Kohonen rule: ΔW = α·Λ·(x - W) on BMU. - # 4. inference_punishment_count += 1 - # punishment_count += 1 - # success_count = 0 - # 5. If punishment_count == 1: - # activate_hypothesis() — re-treina HypothesisClassifier - # usando o buffer expandido (Adam + BCEWithLogitsLoss, 50 epochs). - # Ajusta pesos do classifier: θ_{t+1} = θ_t - η·∂L/∂θ. - # If punishment_count >= 2: - # set_ewc_reference() — consolida w via Fisher information: - # F_i = mean((x_w - W_w,i)²) - # old_weights_w = W_w.clone() - # Reset cycle (mirror training). - # 6. Return outcome dict with action taken + accuracy_after. - # - # This ensures that during inference, the model can self-correct using - # the same hypothesis-layer-based parameter adjustment mechanism used - # in training — NOT a separate inference-only correction. - # ------------------------------------------------------------------ - def punish_during_inference( - self, - sentence: str, - correct_label_int: int, - correct_label_str: Optional[str] = None, - dataset_name: Optional[str] = None, - ) -> Dict[str, Any]: - """V6.5-fix — Punição na inferência via camada de hipótese. - - Mirror do protocolo de punição de treinamento, aplicado em inference - quando o usuário (ou avaliador) fornece feedback de que a predição - foi incorreta. - - Args: - sentence: texto que foi mal-classificado. - correct_label_int: rótulo inteiro correto (0 ou 1). - correct_label_str: opcional — rótulo textual correto. - dataset_name: opcional — nome do dataset para registrar label. - - Returns: - Dict com: - - action: 'activate_hypothesis' | 'set_ewc_reference' | 'none' - - inference_punishment_count: total de punições em inference - - punishment_count_cycle: contador no ciclo atual (0, 1, ou 2+) - - classifier_trained: bool - - ewc_reference_set: bool - - accuracy_after_punishment: float - - buffer_size: int - - elapsed_ms: float - """ - import time as _time - t0 = _time.time() - - # 1. Register label string if provided - if correct_label_str is not None and dataset_name is not None: - if dataset_name not in self.label_registry: - self.label_registry[dataset_name] = {} - self.label_registry[dataset_name][int(correct_label_int)] = str(correct_label_str) - self.last_dataset_used = dataset_name - - # 2. Compute vec_4d for the misclassified sentence - # 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 — histórico de métricas SOM (QE, TE, KL, VE, Dead Neurons) - # para detecção de Estagnação do QE e Cruzamento de Vizinhança. - from bigru_t.model.som_metrics import SOMMetricHistory - self._som_metric_history: SOMMetricHistory = SOMMetricHistory() - - def _init_hypothesis_ensemble(self): - """Cria o ensemble de geradores + otimizador Adam.""" - input_dim = self.som_neuron_count # ativação SOM flatten (864) - output_dim = self.som.weights.numel() # I*J*K*L*4 (864*4 = 3456) - self.hypothesis_ensemble = HypothesisEnsemble( - input_dim=input_dim, - output_dim=output_dim, - n_hypotheses=self.n_hypotheses, - max_n_hypotheses=self.max_n_hypotheses, - hidden_dim=self.hyp_hidden_dim, - ) - self.hyp_optimizer = torch.optim.Adam( - self.hypothesis_ensemble.parameters(), lr=self.hyp_lr - ) - # delta_scale: parâmetro treinável para controlar magnitude dos deltas - # começa pequeno (0.01) para evitar explosão no início - self.delta_scale = nn.Parameter(torch.tensor(0.01)) - - # ================================================================== - # V6.5-V2-dynamic — Controlador adaptativo de hiperparâmetros - # ================================================================== - def _record_loss(self, loss_final: float) -> None: - """Adiciona um loss_final ao histórico (janela deslizante).""" - self._loss_history.append(float(loss_final)) - if len(self._loss_history) > self.loss_history_window: - self._loss_history = self._loss_history[-self.loss_history_window:] - - def _record_punishment_event(self, had_punishment: bool) -> None: - """Registra se houve punição no batch atual (janela deslizante).""" - self._punishment_history.append(1 if had_punishment else 0) - if len(self._punishment_history) > self.punishment_window: - self._punishment_history = self._punishment_history[-self.punishment_window:] - - def _compute_loss_stats(self) -> Dict[str, float]: - """Calcula slope e volatilidade do histórico de loss. - - Returns: - {"slope": float, "volatility": float, "mean": float, "n": int} - """ - n = len(self._loss_history) - if n < 2: - return {"slope": 0.0, "volatility": 0.0, "mean": 0.0, "n": n} - losses = list(self._loss_history) - mean_loss = sum(losses) / n - # Slope por mínimos quadrados: y = a*x + b, x = [0..n-1] - x_mean = (n - 1) / 2.0 - y_mean = mean_loss - num = sum((i - x_mean) * (losses[i] - y_mean) for i in range(n)) - den = sum((i - x_mean) ** 2 for i in range(n)) - slope = num / den if den > 0 else 0.0 - # Volatilidade = std / |mean| (CV) - variance = sum((l - y_mean) ** 2 for l in losses) / n - std = variance ** 0.5 - volatility = std / max(abs(mean_loss), 1e-8) - return { - "slope": float(slope), - "volatility": float(volatility), - "mean": float(mean_loss), - "std": float(std), - "n": int(n), - } - - def _compute_punishment_rate(self) -> float: - """Calcula a taxa de punição na janela observada.""" - if not self._punishment_history: - return 0.0 - return float(sum(self._punishment_history) / len(self._punishment_history)) - - def _adapt_hyperparameters(self, trigger: str = "auto") -> Dict[str, Any]: - """V6.5-V2-dynamic — Ajusta n_hypotheses, n_trials, hyp_train_steps. - - Regras (matematicamente justificadas): - - loss_volatility > 0.30 e n_hyp < max → n_hyp += 2 (mais diversidade) - - loss_volatility < 0.05 e n_hyp > min → n_hyp -= 2 (menos desperdício) - - loss_slope > -0.001 (lento/estagnado) e steps < max → steps += 5 - - loss_slope < -0.01 (rápido) e steps > min → steps -= 5 - - punishment_rate > 0.30 e n_trials < max → n_trials += 1 - - punishment_rate < 0.05 e n_trials > min → n_trials -= 1 - - Args: - trigger: "auto" (após train_hypotheses), "punishment" (após punição), - "manual" (chamada externa). - - Returns: - Dict com estado anterior, novo estado, e regras disparadas. - """ - before = { - "n_hypotheses": int(self.n_hypotheses), - "n_trials": int(self.n_trials), - "hyp_train_steps": int(self.hyp_train_steps), - } - stats = self._compute_loss_stats() - rate = self._compute_punishment_rate() - rules_fired: List[str] = [] - - # 1. Ajuste de n_hypotheses (com base em volatilidade) - if stats["n"] >= 3: - if stats["volatility"] > 0.30 and self.n_hypotheses < self.max_n_hypotheses: - new_n = min(self.n_hypotheses + 2, self.max_n_hypotheses) - if new_n != self.n_hypotheses: - self.n_hypotheses = new_n - self.hypothesis_ensemble.set_active_count(new_n) - rules_fired.append( - f"n_hyp+2 (vol={stats['volatility']:.3f} → {new_n})" - ) - elif stats["volatility"] < 0.05 and self.n_hypotheses > self.min_n_hypotheses: - new_n = max(self.n_hypotheses - 2, self.min_n_hypotheses) - if new_n != self.n_hypotheses: - self.n_hypotheses = new_n - self.hypothesis_ensemble.set_active_count(new_n) - rules_fired.append( - f"n_hyp-2 (vol={stats['volatility']:.3f} → {new_n})" - ) - - # 2. Ajuste de hyp_train_steps (com base em slope) - if stats["n"] >= 3: - if stats["slope"] > -0.001 and self.hyp_train_steps < self.max_hyp_train_steps: - new_steps = min(self.hyp_train_steps + 5, self.max_hyp_train_steps) - if new_steps != self.hyp_train_steps: - self.hyp_train_steps = new_steps - rules_fired.append( - f"steps+5 (slope={stats['slope']:.5f} → {new_steps})" - ) - elif stats["slope"] < -0.01 and self.hyp_train_steps > self.min_hyp_train_steps: - new_steps = max(self.hyp_train_steps - 5, self.min_hyp_train_steps) - if new_steps != self.hyp_train_steps: - self.hyp_train_steps = new_steps - rules_fired.append( - f"steps-5 (slope={stats['slope']:.5f} → {new_steps})" - ) - - # 3. Ajuste de n_trials (com base em recorrência de punições) - if len(self._punishment_history) >= 3: - if rate > 0.30 and self.n_trials < self.max_n_trials: - new_t = min(self.n_trials + 1, self.max_n_trials) - if new_t != self.n_trials: - self.n_trials = new_t - rules_fired.append(f"n_trials+1 (rate={rate:.3f} → {new_t})") - elif rate < 0.05 and self.n_trials > self.min_n_trials: - new_t = max(self.n_trials - 1, self.min_n_trials) - if new_t != self.n_trials: - self.n_trials = new_t - rules_fired.append(f"n_trials-1 (rate={rate:.3f} → {new_t})") - - after = { - "n_hypotheses": int(self.n_hypotheses), - "n_trials": int(self.n_trials), - "hyp_train_steps": int(self.hyp_train_steps), - } - adapted = before != after - record = { - "trigger": trigger, - "step": int(self._last_adaptation_step), - "before": before, - "after": after, - "adapted": bool(adapted), - "rules_fired": rules_fired, - "loss_stats": stats, - "punishment_rate": float(rate), - } - self._adaptation_log.append(record) - # V6.5-V2-metrics-FIX: limita crescimento do log (era ilimitado, causando - # crescimento de memória em treino longo). Mantém últimas 50 entradas. - if len(self._adaptation_log) > 50: - self._adaptation_log = self._adaptation_log[-50:] - self._last_adaptation_step += 1 - return record - - def get_adaptation_log(self) -> List[Dict[str, Any]]: - """Retorna o log de adaptações dinâmicas (para monitoramento).""" - return list(self._adaptation_log) - - # ================================================================== - # V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM - # ================================================================== - # User requirement: "analisar e aprimorar (logicamente e matematicamente) - # no projeto limpeza de memória RAM e de Armazenamento (ao concluir)". - # - # Mathematical analysis: - # A memória RAM consumida pelo KLS V2 tem 4 fontes principais: - # 1. Pesos do SOM: I*J*K*L*4 floats = 864*4 = 3456 floats (~14KB) - # → FIXO (não pode ser limpo sem destruir o modelo) - # 2. HypothesisEnsemble: max_n_hypotheses * DeltaGenerator params - # = 32 * (864*256 + 256*256 + 256*3456) = 32 * 1.16M = ~37M params - # → ~148MB em float32. Os OTIMIZADORAS Adam guardam 2x esse valor - # (momentum + variance) = ~296MB adicionais. - # 3. HypothesisClassifier: 864*512 + 512*256 + ... = ~616K params (~2.5MB) - # 4. buffer_4d: até MAX_BUFFER_SIZE amostras * 4 floats = ~4KB (mínimo) - # MAS os tensores carregam gradiente do embedding se não forem detached. - # - # A limpeza agressiva atua em: - # - Zera gradientes acumulados nos otimizadores (Adam state) - # - Chama gc.collect() para liberar ciclos Python - # - Libera caches internos do PyTorch (oneDNN primitive cache) - # - Trunca listas de histórico que crescem com o treino - # ------------------------------------------------------------------ - def aggressive_cleanup(self) -> Dict[str, Any]: - """V6.5-V2-metrics-FIX — Limpeza agressiva de memória RAM do KLS. - - Retorna métricas antes/depois para confirmação. NÃO destroi pesos do - SOM nem do classifier — apenas libera caches e gradientes acumulados. - - Returns: - Dict com rss_before/after_mb, n_refs_collected, buffers_trimmed. - """ - import gc - try: - import resource - rss_before = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 - except Exception: - rss_before = 0.0 - - # 1. Zera gradientes do otimizador Adam (libera momentum/variance state) - try: - if hasattr(self, "hyp_optimizer"): - self.hyp_optimizer.zero_grad(set_to_none=True) - except Exception: - pass - - # 2. Trunca históricos que crescem com o treino - if len(self._adaptation_log) > 20: - self._adaptation_log = self._adaptation_log[-20:] - if len(self.inference_punishment_history) > 50: - self.inference_punishment_history = self.inference_punishment_history[-50:] - if len(self.vqvae2_metrics_history) > 30: - self.vqvae2_metrics_history = self.vqvae2_metrics_history[-30:] - if len(self.w8a8_metrics_history) > 30: - self.w8a8_metrics_history = self.w8a8_metrics_history[-30:] - - # 3. gc.collect() em 2 passes (gerações 0+1 e 2) - n_collected_0 = gc.collect(0) - n_collected_1 = gc.collect(1) - n_collected_2 = gc.collect(2) - - # 4. Libera caches CPU do PyTorch (oneDNN/MKL) - try: - import torch - if hasattr(torch, "cuda") and torch.cuda.is_available(): - torch.cuda.empty_cache() - except Exception: - pass - - try: - import resource - rss_after = float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0 - except Exception: - rss_after = 0.0 - - return { - "rss_before_mb": float(rss_before), - "rss_after_mb": float(rss_after), - "freed_mb": max(0.0, rss_before - rss_after), - "gc_collected": int(n_collected_0 + n_collected_1 + n_collected_2), - "adaptation_log_trimmed": int(len(self._adaptation_log)), - "timestamp": __import__("datetime").datetime.utcnow().isoformat() + "Z", - } - - # ================================================================== - # 2.1 Treinamento dos geradores (train_hypotheses) - # ================================================================== - def train_hypotheses( - self, - buffer_4d: Optional[List[torch.Tensor]] = None, - buffer_labels: Optional[List[int]] = None, - ) -> Dict[str, Any]: - """Treina os 16 geradores para produzir deltas que melhorem a classificação. - - Usa uma cópia do SOM para avaliar o efeito de cada delta e - retropropaga o erro de classificação para os parâmetros dos geradores. - - Mathematical formulation: - Para cada step s = 1..hyp_train_steps: - 1. x_state = mean(som_activations(buffer)) # (1, P_som) - 2. δ_h = scale * g_h(x_state) para h = 1..16 - 3. loss_h = BCE(classifier(SOM+δ_h), labels) - 4. loss = mean_h(loss_h) + λ * ||δ||² (regularização) - 5. θ_{s+1} = θ_s - η * ∂loss/∂θ - - Returns: - Dict com métricas do treino (loss inicial, final, n_steps). - """ - if buffer_4d is None: - buffer_4d = self.buffer_4d - if buffer_labels is None: - buffer_labels = self.buffer_labels - if not buffer_4d: - return {"active": False, "reason": "empty_buffer"} - - import time as _time - t0 = _time.time() - - # Prepara dados - data = torch.stack(buffer_4d).detach() # (N, 4) - labels = torch.tensor(buffer_labels, dtype=torch.float, device=data.device) - - # Computa ativações SOM para cada amostra (sem gradiente) - with torch.no_grad(): - som_activations = torch.stack( - [self._som_activation(data[i]) for i in range(len(data))] - ) # (N, P_som) - - # Snapshot dos pesos atuais do SOM (para simular ajustes) - som_weights_flat = self.som.weights.detach().clone().flatten() # (P,) - - self.hypothesis_ensemble.train() - losses = [] - # V6.5-V2-metrics-FIX: pré-computa data_norm_sq uma única vez (N, 1) - # para reuso em todos os steps — evita recomputação redundante. - data_norm_sq = (data * data).sum(dim=-1, keepdim=True).t() # (1, N) - labels_expanded = labels # (N,) - - for step in range(self.hyp_train_steps): - self.hyp_optimizer.zero_grad() - - # Ativação média como representação do estado do SOM - x_mean = som_activations.mean(dim=0, keepdim=True) # (1, P_som) - - # Gera deltas: (1, n_hypotheses, P) - deltas_stack = self.hypothesis_ensemble.forward_stacked(x_mean) - deltas_stack = self.delta_scale * deltas_stack # escala - - # V6.5-V2-metrics-FIX: Avaliação VETORIZADA de todas as hipóteses - # em paralelo (substitui loop que materializava 16 cópias do SOM). - # Memória: (n_hyp, P, N) em vez de 16 * (P + 2*P*N). - if self.classifier is not None and self.classifier_trained: - # deltas_stack: (1, n_hyp, P) → (n_hyp, P, 4) reshape - n_hyp = self.n_hypotheses - P_neurons = self.som_neuron_count - # som_weights_flat: (P, 4) detached - W_base = som_weights_flat.reshape(P_neurons, 4) # (P, 4) - # delta_h: (n_hyp, P) → reshape para (n_hyp, P, 4) - deltas_3d = deltas_stack[0].reshape(n_hyp, P_neurons, 4) # (n_hyp, P, 4) - # W_new[h, p, d] = W_base[p, d] + deltas_3d[h, p, d] - W_new = W_base.unsqueeze(0) + deltas_3d # (n_hyp, P, 4) — broadcast - # dist²[h, p, n] = ||W_new[h, p] - data[n]||² - # = ||W_new[h, p]||² + ||data[n]||² - 2*W_new[h, p]·data[n] - W_new_norm_sq = (W_new * W_new).sum(dim=-1) # (n_hyp, P) - # cross[h, p, n] = W_new[h, p] · data[n] - cross = torch.matmul(W_new, data.t()) # (n_hyp, P, N) - dist_sq = ( - W_new_norm_sq.unsqueeze(-1) # (n_hyp, P, 1) - + data_norm_sq # (1, N) → broadcast (n_hyp, P, N) - - 2.0 * cross - ) # (n_hyp, P, N) - dist_sq = torch.clamp(dist_sq, min=0.0) - # activations[h, n, p] = dist_sq[h, p, n] - activations = dist_sq.transpose(1, 2) # (n_hyp, N, P) - # Classifier forward (congelado) - classifier_params_were_grad = [ - p.requires_grad for p in self.classifier.parameters() - ] - for p in self.classifier.parameters(): - p.requires_grad_(False) - try: - logits = self.classifier( - activations.reshape(n_hyp * len(data), P_neurons) - ).reshape(n_hyp, len(data)) # (n_hyp, N) - # BCE por hipótese, depois média - loss_per_hyp = F.binary_cross_entropy_with_logits( - logits, - labels_expanded.unsqueeze(0).expand(n_hyp, -1), - reduction='none', - ).mean(dim=1) # (n_hyp,) - loss_total = loss_per_hyp.mean() - # Regularização L2 sobre os deltas - reg_loss = 0.01 * deltas_stack.norm() - loss_total = loss_total + reg_loss - finally: - for p, was_grad in zip(self.classifier.parameters(), - classifier_params_were_grad): - p.requires_grad_(was_grad) - else: - # Fallback: minimizar norma do delta (regularização pura) - loss_total = deltas_stack.norm() - - loss_total.backward() - self.hyp_optimizer.step() - losses.append(float(loss_total.item())) - - # V6.5-V2-metrics-FIX: libera tensores intermediários explicitamente - # para reduzir pico de memória entre steps. - del loss_total, deltas_stack - if 'W_new' in dir(): - del W_new, cross, dist_sq, activations - - # Atualiza a escala de delta (decai suavemente) - with torch.no_grad(): - self.delta_scale.data = torch.clamp( - self.delta_scale.data * 0.99, 0.001, 0.1 - ) - - self.hypothesis_ensemble.eval() - # V6.5-V2-metrics-FIX: libera som_activations e som_weights_flat - del som_activations, som_weights_flat, data, data_norm_sq - import gc - gc.collect() - - # V6.5-V2-dynamic — registra loss_final no histórico - loss_final_val = float(losses[-1]) if losses else 0.0 - self._record_loss(loss_final_val) - - result = { - "active": True, - "n_steps": self.hyp_train_steps, - "n_hypotheses": self.n_hypotheses, - "loss_initial": float(losses[0]) if losses else 0.0, - "loss_final": loss_final_val, - "loss_mean": float(sum(losses) / len(losses)) if losses else 0.0, - "delta_scale_final": float(self.delta_scale.item()), - "elapsed_ms": float((_time.time() - t0) * 1000), - } - - # V6.5-V2-dynamic — dispara adaptação automática de hiperparâmetros - try: - adapt_record = self._adapt_hyperparameters(trigger="auto") - result["adaptation"] = adapt_record - except Exception as adapt_err: - result["adaptation_error"] = str(adapt_err)[:200] - - return result - - def _compute_classification_loss_with_som( - self, som_weights: torch.Tensor, data: torch.Tensor, labels: torch.Tensor - ) -> torch.Tensor: - """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), - } - - # Aplica o delta ao SOM - with torch.no_grad(): - som_flat = self.som.weights.flatten() - som_flat_new = som_flat + best_delta - self.som.weights = som_flat_new.view_as(self.som.weights) - - # Re-avalia acurácia APÓS aplicação - acc_after = self.evaluate_classification() - - # Re-rotula neurônios (BMU labels podem ter mudado) - self._label_neurons() - - # Consolida via EWC (Fisher information em w) - try: - self.som.set_ewc_reference() - ewc_set = True - except Exception as e: - ewc_set = False - sel_info["ewc_error"] = str(e)[:200] - - return { - "active": True, - "acc_before": float(acc_before), - "acc_after": float(acc_after), - "best_acc_during_selection": float(best_acc), - "delta_norm": float(best_delta.norm().item()), - "delta_scale": float(self.delta_scale.item()), - "ewc_reference_set": bool(ewc_set), - "selection_info": sel_info, - "elapsed_ms": float((_time.time() - t0) * 1000), - } - - # ================================================================== - # 2.4 Sobrescrita do process_batch para o novo protocolo V2 - # ================================================================== - def process_batch_v2( - self, - sentences: List[str], - labels: List[int], - label_strings: Optional[Dict[int, str]] = None, - dataset_name: Optional[str] = None, - enable_punishment: bool = True, - ) -> Dict[str, Any]: - """Versão V2 do process_batch com 3 tentativas e 16 hipóteses. - - Protocolo: - - CONHECIMENTO (enable_punishment=False): apenas adiciona dados e - treina SOM (sem aplicar punição). - - TREINAMENTO COM PUNIÇÃO (enable_punishment=True): - * acc < 1.0 → punishment_count += 1 - * punishment_count == 1: treinar 16 hipóteses + activate_hypothesis - * punishment_count == 2: aplicar melhor delta (3×16) + EWC + reset - - Args: - sentences, labels, label_strings, dataset_name: mesmos de process_batch. - enable_punishment: se False, apenas acumula conhecimento (fase 1). - - Returns: - Dict com métricas do batch processado. - """ - import time as _time - t0 = _time.time() - result: Dict[str, Any] = { - "phase": "CONHECIMENTO" if not enable_punishment else "PUNICAO", - "dataset": dataset_name, - "n_samples": len(sentences), - "action": "none", - "cycle_completed": False, - } - - # 1. Adiciona dados ao buffer - self.add_data( - sentences, labels, - label_strings=label_strings, - dataset_name=dataset_name, - ) - - # 2. Verifica início de treino - training_started = self.check_training_start() - if training_started or self.training_ready: - # 3. Treina SOM no buffer (regra de Kohonen) - self.train_som_on_buffer() - self._label_neurons() - result["som_trained"] = True - - if not self.training_ready: - result["elapsed_ms"] = float((_time.time() - t0) * 1000) - return result - - # 4. Fase de CONHECIMENTO: para aqui, sem aplicar punição - if not enable_punishment: - acc = self.evaluate_classification() - result["accuracy"] = float(acc) - result["elapsed_ms"] = float((_time.time() - t0) * 1000) - return result - - # 5. Fase de PUNIÇÃO: avalia e aplica protocolo V2 - acc = self.evaluate_classification() - result["accuracy"] = float(acc) - - if acc < 1.0: - self.punishment_count += 1 - self.success_count = 0 - - # V6.5-V2-dynamic — registra evento de punição - self._record_punishment_event(True) - - if self.punishment_count == 1: - # 1ª punição: treinar hipóteses + ativar classificador - hyp_info = self.train_hypotheses() - if self.classifier is None: - self.activate_hypothesis() - else: - # Re-treina classifier com buffer atual - self.activate_hypothesis() - result["action"] = "train_hypotheses_and_activate" - result["hypotheses_training"] = hyp_info - - elif self.punishment_count >= 2: - # 2ª punição: selecionar melhor delta (n_trials × n_hyp) + aplicar + EWC - apply_info = self.apply_best_delta_and_consolidate() - result["action"] = "apply_best_delta_and_consolidate" - result["delta_application"] = apply_info - - # V6.5-V2-dynamic — adapta hiperparâmetros após punição recorrente - try: - adapt_record = self._adapt_hyperparameters(trigger="punishment") - result["adaptation"] = adapt_record - except Exception as adapt_err: - result["adaptation_error"] = str(adapt_err)[:200] - - # Reset do ciclo (mirror training) - self.training_ready = False - self.punishment_count = 0 - self.success_count = 0 - self.histogram.clear() - # Mantém um sliding window do buffer para preservar contexto - if len(self.buffer_4d) > 256: - self.buffer_4d = self.buffer_4d[-256:] - self.buffer_labels = self.buffer_labels[-256:] - result["cycle_completed"] = True - else: - # acc == 1.0: sucesso - self.punishment_count = 0 - self.success_count += 1 - # V6.5-V2-dynamic — registra sucesso (sem punição) - self._record_punishment_event(False) - result["action"] = "success" - - result["elapsed_ms"] = float((_time.time() - t0) * 1000) - return result - - # ================================================================== - # 2.5 API de monitoramento V2 - # ================================================================== - def get_v2_metrics(self) -> Dict[str, Any]: - """Retorna métricas específicas da versão V2 (inclui adaptação dinâmica).""" - return { - "version": "V2-dynamic", - "n_hypotheses": int(self.n_hypotheses), - "n_hypotheses_active": int(self.hypothesis_ensemble.active_count), - "max_n_hypotheses": int(self.max_n_hypotheses), - "n_trials": int(self.n_trials), - "hyp_train_steps": int(self.hyp_train_steps), - "hyp_lr": float(self.hyp_lr), - "delta_scale": float(self.delta_scale.item()), - "n_generators": int(len(self.hypothesis_ensemble.generators)), - "punishment_count": int(self.punishment_count), - "success_count": int(self.success_count), - "training_ready": bool(self.training_ready), - "classifier_trained": bool(self.classifier_trained), - "ewc_reference_set": bool(self.som.old_weights_w is not None), - "buffer_size": int(len(self.buffer_4d)), - "dynamic_adaptation": { - "loss_history_len": int(len(self._loss_history)), - "loss_stats": self._compute_loss_stats(), - "punishment_rate": float(self._compute_punishment_rate()), - "punishment_window_size": int(len(self._punishment_history)), - "n_adaptations": int(len(self._adaptation_log)), - "limits": { - "min_n_hypotheses": int(self.min_n_hypotheses), - "max_n_hypotheses": int(self.max_n_hypotheses), - "min_n_trials": int(self.min_n_trials), - "max_n_trials": int(self.max_n_trials), - "min_hyp_train_steps": int(self.min_hyp_train_steps), - "max_hyp_train_steps": int(self.max_hyp_train_steps), - }, - "last_5_adaptations": self._adaptation_log[-5:] if self._adaptation_log else [], - }, - } - - # ================================================================== - # 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.