Delete cnn_bigru/tests/test_50_samples.py with huggingface_hub

cnn_bigru/tests/test_50_samples.py

@@ -1,513 +0,0 @@
-"""test_50_samples.py — Teste de 50 amostras para erros lógicos ou falhas.
-
-Executa o pipeline completo:
-    1. Inicializa xeon_runtime + memory_optimizer
-    2. Treina BBPE tokenizer em corpus sintético
-    3. Cria dataset streaming multimodal (50 amostras)
-    4. Instancia modelo multimodal + generator + verifier + anti-hallucination
-    5. Executa treinamento cooperativo com:
-       - Synergy search (N tentativas)
-       - Hypothesis controller (ativa em punições)
-       - Auto-learner (ajuste dinâmico de LR + spectral norm)
-    6. Executa inferência com sampling
-    7. Avalia perplexidade
-    8. Reporta erros lógicos/falhas encontradas
-
-Usage:
-    python -m cnn_bigru.tests.test_50_samples
-"""
-from __future__ import annotations
-
-import logging
-import os
-import sys
-import time
-import traceback
-from pathlib import Path
-from typing import Dict, List
-
-# Setup paths (must come before torch import for xeon_runtime)
-PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
-sys.path.insert(0, str(PROJECT_ROOT))
-
-# Ativa Xeon runtime ANTES de importar torch
-from cnn_bigru.utils.xeon_runtime import optimize_xeon_environment
-N_CORES = optimize_xeon_environment()
-
-import numpy as np
-import torch
-from torch.utils.data import DataLoader
-
-from cnn_bigru.tokenizer.bbpe_tokenizer import BBPETokenizer
-from cnn_bigru.data.streaming_dataset import (
-    MultimodalStreamingDataset,
-    collate_multimodal,
-)
-from cnn_bigru.models.multimodal_model import MultimodalCNNBiGRU
-from cnn_bigru.models.generator_verifier import (
-    GeneratorCNNBiGRU,
-    VerifierCNNBiGRU,
-    AntiHallucinationLayer,
-)
-from cnn_bigru.losses.losses import LossConfig, MultiLoss
-from cnn_bigru.training.trainer import TrainerConfig, CooperativeTrainer
-from cnn_bigru.training.auto_learner import (
-    AutoLearnConfig,
-    orthogonal_init_model,
-)
-from cnn_bigru.training.hypothesis_controller import (
-    HypothesisConfig,
-    HypothesisController,
-)
-from cnn_bigru.inference.inference import (
-    generate_with_sampling,
-    evaluate_perplexity,
-)
-from cnn_bigru.utils.memory_optimizer import MemoryOptimizer
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
-    datefmt="%H:%M:%S",
-)
-logger = logging.getLogger("test_50")
-
-
-# ============================================================================
-# Corpus para treino do tokenizer
-# ============================================================================
-
-CORPUS_TEXTS = [
-    "o modelo cooperativo cnn-bigru combina fluxos paralelos",
-    "a atencao cruzada troca informacoes entre redes a e b",
-    "gru bidirecional captura dependencias temporais em ambas direcoes",
-    "a porta de atenuacao controla o fluxo de informacao entre celulas",
-    "a camada anti-alucinacao usa logica fuzzy de lukasiewicz",
-    "o verificador classifica passos com sigmoid binaria",
-    "penalidades linear e exponencial reduzem o erro grave",
-    "ajuste dinamico de lr estabiliza o treinamento cooperativo",
-    "normalizacao espectral limita a norma dos pesos",
-    "inicializacao ortogonal estabiliza matrizes recorrentes",
-    "o gerador decodificador usa atencao bahdanau sobre o encoder",
-    "a fusao multimodal combina texto imagem e audio",
-    "hipoteses sao ativadas quando o verificador pune o passo",
-    "synergy search tenta n configuracoes e escolhe a melhor",
-    "a perplexidade mede a confusao do modelo na previsao",
-    "top-k e top-p filtram a distribuicao de probabilidade",
-    "temperature ajusta a entropia das previsoes do modelo",
-    "presence penalty pune tokens ja aparecidos na geracao",
-    "frequency penalty pune proporcionalmente a frequencia do token",
-    "streaming dataset carrega amostras sem materializacao completa",
-    "byte bpe tokeniza qualquer string utf-8 sem unk",
-    "o otimizador adamw combina momentum e weight decay",
-    "gradient clipping previne explosao de gradiente",
-    "amp reduz vram com precisao mista bfloat16",
-    "memory optimizer limpa cache entre batches",
-] * 4  # ~100 amostras para treino BBPE
-
-
-def step_01_init_runtime() -> Dict:
-    """Inicializa runtime e reporta configurações."""
-    logger.info("=" * 70)
-    logger.info("PASSO 1: Inicialização do runtime Xeon + memory optimizer")
-    logger.info("=" * 70)
-    from cnn_bigru.utils.xeon_runtime import get_runtime_info
-    info = get_runtime_info()
-    for k, v in info.items():
-        logger.info("  %s: %s", k, v)
-    mem = MemoryOptimizer(enable_amp=False)
-    mem.configure()
-    mem_info = mem.get_memory_mb()
-    logger.info("  Memória: %s", mem_info)
-    return {"runtime": info, "memory": mem_info}
-
-
-def step_02_train_tokenizer() -> BBPETokenizer:
-    """Treina BBPE tokenizer em corpus sintético."""
-    logger.info("=" * 70)
-    logger.info("PASSO 2: Treino do BBPE tokenizer")
-    logger.info("=" * 70)
-    t0 = time.time()
-    tok = BBPETokenizer.train_from_texts(
-        CORPUS_TEXTS,
-        vocab_size=2000,
-        min_frequency=1,
-    )
-    elapsed = time.time() - t0
-    logger.info("  Vocab size: %d", tok.vocab_size)
-    logger.info("  BOS/PAD/EOS IDs: %d/%d/%d", tok.bos_id, tok.pad_id, tok.eos_id)
-    logger.info("  Tempo: %.2fs", elapsed)
-
-    # Validação roundtrip
-    test_strs = CORPUS_TEXTS[:5]
-    roundtrip = tok.validate_roundtrip(test_strs)
-    logger.info("  Roundtrip accuracy: %.2f%%", roundtrip * 100)
-
-    if roundtrip < 0.8:
-        logger.warning("  Roundtrip baixo — possível problema no tokenizer")
-    return tok, roundtrip
-
-
-def step_03_create_dataset(tokenizer: BBPETokenizer, n_samples: int = 50) -> DataLoader:
-    """Cria dataset streaming multimodal com N amostras."""
-    logger.info("=" * 70)
-    logger.info("PASSO 3: Criação do dataset streaming multimodal (%d amostras)", n_samples)
-    logger.info("=" * 70)
-
-    # Para teste determinístico, usamos fallback sintético
-    dataset = MultimodalStreamingDataset(
-        n_samples=n_samples,
-        hf_datasets=[],  # skip HF forçando sintético
-        use_synthetic_fallback=True,
-        seed=42,
-        image_size=(28, 28, 1),
-        audio_shape=(32, 40),
-    )
-
-    # Coleta todas as amostras em uma lista (para DataLoader iterável)
-    samples = list(dataset)
-    logger.info("  Amostras coletadas: %d", len(samples))
-    assert len(samples) == n_samples, f"Esperado {n_samples}, obtido {len(samples)}"
-
-    # Cria DataLoader com collate_fn
-    from torch.utils.data import DataLoader
-    loader = DataLoader(
-        samples,
-        batch_size=8,
-        shuffle=False,
-        collate_fn=lambda b: collate_multimodal(b, tokenizer, max_len=32),
-    )
-    logger.info("  DataLoader criado: batch_size=8, max_len=32")
-    return loader
-
-
-def step_04_init_model(tokenizer: BBPETokenizer, device: str = "cpu") -> Dict:
-    """Instancia todos os componentes do modelo."""
-    logger.info("=" * 70)
-    logger.info("PASSO 4: Instanciação dos modelos")
-    logger.info("=" * 70)
-    V = tokenizer.vocab_size
-
-    # Modelo multimodal principal
-    model = MultimodalCNNBiGRU(
-        vocab_size=V,
-        num_classes=3,
-        embedding_dim=32,
-        cnn_filters=32,
-        gru_hidden=32,
-        n_heads=4,
-        dropout=0.1,
-        pad_idx=tokenizer.pad_id,
-        img_channels=1,
-        img_hidden=16,
-        img_out_dim=32,
-        audio_freq=40,
-        audio_hidden=16,
-        audio_out_dim=32,
-        fusion_dim=64,
-        use_spectral_norm=False,
-    )
-
-    # Aplica inicialização ortogonal
-    n_init = orthogonal_init_model(model)
-    logger.info("  Modelo multimodal: %d params, %d camadas ortogonalizadas",
-                sum(p.numel() for p in model.parameters()), n_init)
-
-    # Generator (encoder + decoder)
-    generator = GeneratorCNNBiGRU(
-        vocab_size=V,
-        embedding_dim=32,
-        cnn_filters=32,
-        gru_hidden=32,
-        n_heads=4,
-        dropout=0.1,
-        pad_idx=tokenizer.pad_id,
-        max_proof_len=16,
-    )
-    orthogonal_init_model(generator)
-
-    # Verifier
-    verifier = VerifierCNNBiGRU(
-        vocab_size=V,
-        embedding_dim=32,
-        cnn_filters=32,
-        gru_hidden=32,
-        n_heads=4,
-        dropout=0.1,
-        pad_idx=tokenizer.pad_id,
-    )
-    orthogonal_init_model(verifier)
-
-    # Anti-hallucination
-    anti_hall = AntiHallucinationLayer(vocab_size=V, embed_dim=16)
-
-    logger.info("  Generator params: %d", sum(p.numel() for p in generator.parameters()))
-    logger.info("  Verifier params: %d", sum(p.numel() for p in verifier.parameters()))
-    logger.info("  Anti-hall params: %d", sum(p.numel() for p in anti_hall.parameters()))
-
-    return {
-        "model": model,
-        "generator": generator,
-        "verifier": verifier,
-        "anti_hall": anti_hall,
-    }
-
-
-def step_05_train(
-    models: Dict,
-    tokenizer: BBPETokenizer,
-    dataloader: DataLoader,
-    device: str = "cpu",
-) -> Dict:
-    """Executa treinamento cooperativo."""
-    logger.info("=" * 70)
-    logger.info("PASSO 5: Treinamento cooperativo (synergy + hipóteses + auto-learn)")
-    logger.info("=" * 70)
-
-    cfg = TrainerConfig(
-        num_epochs=2,
-        max_batches_per_epoch=6,
-        batch_size=8,
-        n_synergy_attempts=3,
-        use_synergy_search=True,
-        use_hypotheses=True,
-        n_hypotheses=4,
-        loss_config=LossConfig(
-            alpha=1.0, beta=0.5, gamma_loss=0.3, delta=0.01,
-            lambda_penal=0.1, mu_exp_penal=0.05,
-            gamma_exp=1.0, threshold_err=0.5,
-            l2_reg=1e-5, use_curvature=True, curvature_eps=1e-3,
-        ),
-        auto_config=AutoLearnConfig(
-            kappa_curv=0.01, grad_clip=1.0, spectral_radius=1.0,
-            lr_min=1e-6, lr_max=1e-2,
-            initial_lr_G=1e-3, initial_lr_V=1e-3,
-            use_spectral_norm=True, apply_after_step=True,
-        ),
-        device=device,
-        log_every=1,
-    )
-
-    trainer = CooperativeTrainer(
-        model=models["model"],
-        tokenizer=tokenizer,
-        config=cfg,
-        generator=models["generator"],
-        verifier=models["verifier"],
-        anti_hallucination=models["anti_hall"],
-    )
-
-    try:
-        result = trainer.train(dataloader)
-        logger.info("  Treinamento OK | final_loss=%.4f | final_ppl=%.2f | elapsed=%.1fs",
-                    result["final_loss"], result["final_ppl"], result["elapsed_s"])
-        return result
-    except Exception as e:
-        logger.error("  Treinamento FALHOU: %s", e)
-        logger.error(traceback.format_exc())
-        raise
-
-
-def step_06_inference(
-    model: MultimodalCNNBiGRU,
-    tokenizer: BBPETokenizer,
-    device: str = "cpu",
-) -> Dict:
-    """Executa inferência com sampling."""
-    logger.info("=" * 70)
-    logger.info("PASSO 6: Inferência com Temperatura + Top-K + Top-P + Penalidades")
-    logger.info("=" * 70)
-
-    prompts = [
-        ("o modelo coopera entre", "fluxos paralelos"),
-        ("atencao cruzada troca", "informacoes entre redes"),
-        ("gru bidirecional captura", "dependencias temporais"),
-    ]
-
-    results = []
-    for prompt_a, prompt_b in prompts:
-        try:
-            result = generate_with_sampling(
-                model=model,
-                tokenizer=tokenizer,
-                prompt_a=prompt_a,
-                prompt_b=prompt_b,
-                max_new_tokens=16,
-                temperature=0.7,
-                top_k=20,
-                top_p=0.9,
-                presence_penalty=0.3,
-                frequency_penalty=0.3,
-                device=device,
-            )
-            logger.info("  Prompt A: %s", prompt_a)
-            logger.info("  Prompt B: %s", prompt_b)
-            logger.info("  Gerado: %s", result["text"][:100])
-            logger.info("  Metrics: %s", result["metrics"])
-            results.append({"prompt_a": prompt_a, "prompt_b": prompt_b, **result})
-        except Exception as e:
-            logger.error("  Inferência falhou para (%s, %s): %s", prompt_a, prompt_b, e)
-            results.append({"prompt_a": prompt_a, "prompt_b": prompt_b, "error": str(e)})
-
-    return {"results": results}
-
-
-def step_07_eval_ppl(
-    model: MultimodalCNNBiGRU,
-    tokenizer: BBPETokenizer,
-    dataloader: DataLoader,
-    device: str = "cpu",
-) -> Dict:
-    """Avalia perplexidade."""
-    logger.info("=" * 70)
-    logger.info("PASSO 7: Avaliação de Perplexidade (PPL)")
-    logger.info("=" * 70)
-    try:
-        result = evaluate_perplexity(
-            model=model,
-            dataloader=dataloader,
-            tokenizer=tokenizer,
-            device=device,
-            max_batches=5,
-        )
-        logger.info("  Loss: %.4f | PPL: %.2f | batches: %d",
-                    result["loss"], result["ppl"], result["n_batches"])
-        return result
-    except Exception as e:
-        logger.error("  PPL falhou: %s", e)
-        logger.error(traceback.format_exc())
-        return {"error": str(e)}
-
-
-def step_08_report_errors(errors: List[str], warnings: List[str]) -> None:
-    """Reporta erros lógicos ou falhas encontradas."""
-    logger.info("=" * 70)
-    logger.info("PASSO 8: Relatório de erros lógicos ou falhas")
-    logger.info("=" * 70)
-
-    if not errors:
-        logger.info("  ✓ NENHUM ERRO CRÍTICO encontrado")
-    else:
-        logger.error("  ✗ %d ERRO(S) CRÍTICO(S) encontrado(s):", len(errors))
-        for e in errors:
-            logger.error("    - %s", e)
-
-    if not warnings:
-        logger.info("  ✓ Nenhum warning relevante")
-    else:
-        logger.warning("  ⚠ %d warning(s):", len(warnings))
-        for w in warnings:
-            logger.warning("    - %s", w)
-
-    logger.info("=" * 70)
-
-
-def main():
-    """Executa o teste completo de 50 amostras."""
-    logger.info("INICIANDO TESTE DE 50 AMOSTRAS — CNN-BiGRU MULTIMODAL COOPERATIVO")
-    logger.info("Project root: %s", PROJECT_ROOT)
-    logger.info("Device: %s", "cpu")
-    logger.info("Cores: %d", N_CORES)
-
-    errors: List[str] = []
-    warnings: List[str] = []
-
-    try:
-        # Step 1
-        step_01_init_runtime()
-
-        # Step 2
-        tokenizer, roundtrip = step_02_train_tokenizer()
-        if roundtrip < 0.8:
-            warnings.append(f"BBPE roundtrip accuracy = {roundtrip:.2%} (esperado >= 80%)")
-
-        # Step 3
-        dataloader = step_03_create_dataset(tokenizer, n_samples=50)
-
-        # Step 4
-        models = step_04_init_model(tokenizer, device="cpu")
-
-        # Verificação: forward pass simples antes do treino
-        try:
-            batch = next(iter(dataloader))
-            with torch.no_grad():
-                out = models["model"](
-                    batch["input_ids_a"], batch["input_ids_b"],
-                    images=batch["images"], audios=batch["audios"],
-                    mode="classify",
-                )
-            logger.info("  Forward pass OK | logits shape: %s", out["logits"].shape)
-            assert out["logits"].shape == (batch["input_ids_a"].size(0), 3), \
-                f"Shape inesperado: {out['logits'].shape}"
-        except Exception as e:
-            errors.append(f"Forward pass inicial falhou: {e}")
-            logger.error(traceback.format_exc())
-
-        # Step 5
-        if not errors:
-            try:
-                train_result = step_05_train(models, tokenizer, dataloader, device="cpu")
-            except Exception as e:
-                errors.append(f"Treinamento falhou: {e}")
-                train_result = None
-        else:
-            train_result = None
-
-        # Step 6
-        if not errors:
-            try:
-                step_06_inference(models["model"], tokenizer, device="cpu")
-            except Exception as e:
-                errors.append(f"Inferência falhou: {e}")
-                logger.error(traceback.format_exc())
-
-        # Step 7
-        if not errors:
-            try:
-                step_07_eval_ppl(models["model"], tokenizer, dataloader, device="cpu")
-            except Exception as e:
-                warnings.append(f"PPL falhou (não crítico): {e}")
-
-        # Step 8
-        step_08_report_errors(errors, warnings)
-
-        # Resumo final
-        logger.info("=" * 70)
-        logger.info("RESUMO FINAL DO TESTE DE 50 AMOSTRAS")
-        logger.info("=" * 70)
-        logger.info("  Amostras processadas: 50")
-        logger.info("  Erros críticos: %d", len(errors))
-        logger.info("  Warnings: %d", len(warnings))
-        if train_result:
-            logger.info("  Loss final: %.4f", train_result["final_loss"])
-            logger.info("  PPL final: %.2f", train_result["final_ppl"])
-            logger.info("  Tempo: %.1fs", train_result["elapsed_s"])
-            logger.info("  Synergy attempts: %d", len(train_result.get("synergy_history", [])))
-            logger.info("  Hypothesis activations: %d",
-                        sum(h.get("hypothesis_activations", 0) for h in train_result.get("history", [])))
-
-        if errors:
-            logger.error("  STATUS: FALHA — %d erro(s)", len(errors))
-            return 1
-        else:
-            logger.info("  STATUS: SUCESSO")
-            return 0
-
-    except Exception as e:
-        logger.error("ERRO FATAL: %s", e)
-        logger.error(traceback.format_exc())
-        return 2
-
-
-if __name__ == "__main__":
-    try:
-        rc = main()
-    except SystemExit:
-        raise
-    except Exception as e:
-        logger.error("Unhandled exception: %s", e)
-        rc = 2
-    # Evita o "Fatal Python error: PyGILState_Release" no shutdown causado
-    # por threads do tokenizers/datasets que ainda estão ativas.
-    import os
-    os._exit(rc)