#!/usr/bin/env python3 """ BENCHMARK DA STACK WINNEX — COLAB, célula única ================================================ Cole este arquivo inteiro numa célula do Google Colab e execute. Ele instala as dependências, baixa os pesos e os dados, e mede a inferência pela lógica da stack: nano → xfactor → madhava. Sem softmax, sem amostragem de logits, sem transformer. Pesos reais : Qwen2.5-0.5B-Instruct (151.936 tokens × 896d) Dados reais : GloVe 6B (400k palavras) e SIFT 1M (128d, benchmark canônico) As seis frentes medidas: T1 GEOMETRIA âncoras Gram-Schmidt dos pesos + base de bytes T2 NANO isomorfismo ℤ/256ℤ → μ₂₅₆: round-trip, homomorfia T3 XFACTOR P = XXᵀ: rank efetivo vs eigh, Parseval T4 MADHAVA cascata bound → pool → exato, em SIFT T5 RECUPERAÇÃO texto → vetor → top-k de tokens, contra o BPE real T6 FOLD LIMIT os três termos, e a condição de prova T7 GENERALIZAÇÃO consulta é frase livre, não palavra do índice T8 CERTIFICAÇÃO quanto do descarte o bound PROVA, por dimensão Nota de T5: o alvo é o próprio token indexado, endereçado por uma função que o índice usou para se construir. Isso mede coerência de espaço — é recuperação. T7 mede a outra pergunta, e a resposta é 0,0/10. Tempo estimado no Colab (CPU): 8-12 min. `--rapido` faz em ~4 min. Requisito: Python ≥ 3.10 (os wheels de winnex-nano e winnex-xfactor são cp312; o Colab roda 3.12). BSL 1.1 | (c) Winnex Brasil Soluções Empresariais LTDA-ME """ from __future__ import annotations import json import os import subprocess import sys import time import urllib.request os.environ.setdefault("OMP_NUM_THREADS", "4") import numpy as np # =========================================================================== # 0. Dependências # =========================================================================== def _pip(*args) -> int: """pip install tolerante: avisa e devolve o código, não levanta.""" cmd = [sys.executable, "-m", "pip", "install", "-q", "--disable-pip-version-check", *args] return subprocess.call(cmd) def instala(): """Instala o que falta, com fallback para o Python do Colab. ARMADILHA CONHECIDA: os wheels de winnex-nano e winnex-xfactor só são publicados para cp312. No Colab, o kernel pode ser 3.11/3.13 — nesse caso o pip responde "No matching distribution found". A saída é instalar o interpretador 3.12 e reexecutar o benchmark nele, que é o que fazemos. """ deps = [("winnex_nano", "winnex-nano"), ("winnex_xfactor", "winnex-xfactor"), ("winnex_madhava", "winnex-madhava"), ("safetensors", "safetensors"), ("tokenizers", "tokenizers"), ("h5py", "h5py")] faltando = [] for mod, pip in deps: try: __import__(mod) except ImportError: faltando.append(pip) if not faltando: print("dependências já presentes", flush=True) return print(f"instalando: {' '.join(faltando)}", flush=True) rc = _pip(*faltando) if rc == 0: return # --- fallback: o wheel exige cp312 e este kernel não é 3.12 --- print(f"\n ✗ pip falhou (código {rc}).", flush=True) print(f" Este kernel é Python {sys.version_info.major}.{sys.version_info.minor}.", flush=True) print(" winnex-nano e winnex-xfactor publicam apenas wheels cp312.", flush=True) if sys.version_info[:2] == (3, 12): print(" Mesmo em 3.12 falhou — verifique a conexão/repositório.", flush=True) raise SystemExit(1) print("\n Instalando Python 3.12 e reexecutando neste interpretador...", flush=True) if subprocess.call(["bash", "-lc", "command -v python3.12 >/dev/null 2>&1"]) != 0: subprocess.call(["bash", "-lc", "apt-get -qq update && apt-get -qq install -y " "python3.12 python3.12-venv >/dev/null 2>&1"]) if subprocess.call(["bash", "-lc", "command -v python3.12 >/dev/null 2>&1"]) != 0: print(" ✗ não foi possível instalar o Python 3.12.", flush=True) print(" No Colab: Runtime > Change runtime type, ou rode", flush=True) print(" !apt-get install -y python3.12 && use python3.12.", flush=True) raise SystemExit(1) alvo = "/content/winnex_py312" if os.path.isdir("/content") else "./winnex_py312" if not os.path.isdir(alvo): subprocess.check_call(["python3.12", "-m", "venv", alvo]) subprocess.check_call([f"{alvo}/bin/pip", "install", "-q", "--disable-pip-version-check", "numpy", *faltando]) # Reexecutar o próprio benchmark no 3.12. Em célula de notebook não existe # __file__; nesse caso recuperamos o fonte da última célula do IPython. fonte = None if "__file__" in globals() and os.path.exists(__file__): fonte = __file__ else: try: import IPython src = IPython.get_ipython().history_manager.input_hist_raw[-1] fonte = os.path.join(os.getcwd(), "_winnex_bench_rerun.py") with open(fonte, "w", encoding="utf-8") as f: f.write(src) except Exception: # noqa: BLE001 fonte = None if not fonte: print(" ✗ não consegui reexecutar automaticamente.", flush=True) print(" Reescreva o kernel para 3.12, ou salve o arquivo em disco", flush=True) print(" e rode: python3.12 benchmark_stack_colab.py", flush=True) raise SystemExit(1) print(f" reexecutando em {alvo}...\n", flush=True) os.execv(f"{alvo}/bin/python", [f"{alvo}/bin/python", fonte] + sys.argv[1:]) print("=" * 78) print(" BENCHMARK DA STACK WINNEX — pesos reais, dados reais") print("=" * 78) print(f"\n Python {sys.version.split()[0]}") instala() import winnex_nano as wn import winnex_madhava as wm from tokenizers import Tokenizer def load_file(caminho: str) -> dict: """Carrega safetensors SEM torch e SEM safetensors.numpy. O formato é simples: 8 bytes com o tamanho do cabeçalho JSON, o cabeçalho, e o blob de dados. O obstacle é o bfloat16 — o numpy não tem dtype para ele, e tanto safetensors.torch quanto safetensors.numpy falham ao encontrá-lo. Mas bf16 → float32 é só deslocar 16 bits: o bf16 guarda os 16 bits altos do float32 (1 sinal + 8 expoente + 7 mantissa). Lemos o header e decodificamos só os tensores que interessam, o que também evita carregar 989 MB quando precisamos de um só. """ import json as _json import struct with open(caminho, "rb") as f: (n_header,) = struct.unpack(" str: """Download com User-Agent — alguns hosts (ann-benchmarks.com) devolvem 403 para o User-Agent padrão do urllib.""" if os.path.exists(destino) and os.path.getsize(destino) > 0: return destino print(f" baixando {os.path.basename(destino)}...", flush=True) req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=600) as r, open(destino, "wb") as f: while True: bloco = r.read(1 << 20) if not bloco: break f.write(bloco) return destino def carrega_modelo(repo: str, embed_key: str): """Baixa config, safetensors e tokenizer de um repositório HuggingFace.""" arquivos = {} for remoto, local in (("config.json", "cfg.json"), ("model.safetensors", "model.safetensors"), ("tokenizer.json", "tok.json")): arquivos[remoto] = baixa( f"https://huggingface.co/{repo}/resolve/main/{remoto}", os.path.join(CACHE, f"{repo.split('/')[-1]}_{local}")) sd = load_file(arquivos["model.safetensors"]) E = np.asarray(sd[embed_key], dtype=np.float32) tk = Tokenizer.from_file(arquivos["tokenizer.json"]) with open(arquivos["tokenizer.json"], encoding="utf-8") as f: tv = json.load(f)["model"]["vocab"] vocab = {k: i for k, i in (tv.items() if isinstance(tv, dict) else enumerate(tv))} return E, tk, vocab, arquivos def carrega_glove(n_max: int = 20000): zip_path = baixa("https://downloads.cs.stanford.edu/nlp/data/glove.6B.zip", os.path.join(CACHE, "glove.6B.zip")) txt = os.path.join(CACHE, "glove.6B.100d.txt") if not os.path.exists(txt): import zipfile with zipfile.ZipFile(zip_path) as z: z.extract("glove.6B.100d.txt", CACHE) W = [] with open(txt, encoding="utf-8", errors="ignore") as f: for i, line in enumerate(f): if i >= n_max: break w = line.split(" ", 1)[0] if 0 < len(w) <= 24: W.append(w) return W def carrega_sift(n: int = 50000): """SIFT 1M do ann-benchmarks — o corpus canônico de busca vetorial.""" caminho = baixa("https://ann-benchmarks.com/sift-128-euclidean.hdf5", os.path.join(CACHE, "sift-128-euclidean.hdf5")) import h5py with h5py.File(caminho, "r") as f: X = np.array(f["train"][:n], dtype=np.float32) X /= np.linalg.norm(X, axis=1, keepdims=True) + 1e-12 return X def titulo(t: str) -> None: print(f"\n{'=' * 78}\n {t}\n{'=' * 78}", flush=True) def teste(nome: str, afirma: str, falsifica: str) -> None: print(f"\n--- {nome} ---", flush=True) print(f" afirmação : {afirma}", flush=True) print(f" falsifica : {falsifica}", flush=True) print(" " + "-" * 64, flush=True) def marca(cond: bool, msg: str, det: str = "") -> bool: print(f" {'✓' if cond else '✗'} {msg} {det}") return cond RESULTADOS: dict[str, object] = {} # =========================================================================== # T1 — Geometria # =========================================================================== def t1_geometria(E: np.ndarray, vocab: dict, n_ancoras: int = 512): titulo("T1 — GEOMETRIA: âncoras do modelo + base de bytes") V, D = E.shape En = E / (np.linalg.norm(E, axis=1, keepdims=True) + 1e-12) teste("T1.1 Âncoras por Gram-Schmidt", "os vetores do próprio modelo formam uma base ortonormal estável", "‖A~ᵀA~ − I‖ acima da precisão de float32") um_byte = [i for s, i in vocab.items() if i < V and len(s.encode("utf-8", errors="ignore")) == 1] curtos = sorted((len(s), i) for s, i in vocab.items() if i < V) sel, vistos = [], set() for i in um_byte: if i not in vistos: sel.append(i); vistos.add(i) for _, i in curtos: if len(sel) >= n_ancoras: break if i not in vistos: sel.append(i); vistos.add(i) Q, _ = np.linalg.qr(En[np.array(sel)].T) Q = Q.astype(np.float32) ort = float(np.abs(Q.T @ Q - np.eye(Q.shape[1])).max()) energia = float(np.linalg.norm(En @ Q) ** 2) / float((En ** 2).sum()) marca(ort < 1e-5, f"‖A~ᵀA~−I‖ = {ort:.2e}", f"| {Q.shape[1]} âncoras | energia {energia * 100:.2f}%") teste("T1.2 Base de bytes nas âncoras", "A_inc·B explica a assinatura dos tokens, com Parseval exato", "identidade de Parseval violada ou explicação muito baixa") Ainc = np.zeros((V, 256), dtype=np.float32) mask = np.zeros(V, dtype=bool) for s, i in vocab.items(): if i >= V: continue disp = s.replace("Ġ", " ").replace("Ċ", "\n") try: b = disp.encode("utf-8") except (UnicodeEncodeError, AttributeError): continue if not (0 < len(b) <= 32): continue mask[i] = True for byte in b: Ainc[i, byte] += 1.0 idx = np.where(mask)[0] S = En[idx] @ Q G = Ainc[idx].T @ Ainc[idx] + 1e-2 * np.eye(256, dtype=np.float32) B = np.linalg.solve(G, Ainc[idx].T @ S) pred = Ainc[idx] @ B e_t, e_f = float((S ** 2).sum()), float((pred ** 2).sum()) e_r = float(((S - pred) ** 2).sum()) erro = abs(e_f + e_r - e_t) / e_t marca(erro < 1e-5, f"Parseval: erro {erro:.1e}", f"| explica {e_f / e_t * 100:.2f}% da assinatura") RESULTADOS["T1"] = {"m": int(Q.shape[1]), "ortogonalidade": ort, "energia": energia, "explicado": e_f / e_t, "parseval": erro, "tokens": int(len(idx))} return {"Q": Q, "B": B, "Ainc": Ainc, "mask": mask, "idx": idx} # =========================================================================== # T2 — nano # =========================================================================== def t2_nano(): titulo("T2 — NANO: o isomorfismo ℤ/256ℤ → μ₂₅₆") teste("T2.1 Round-trip multilíngue", "decode(encode(c)) == c para qualquer byte", "um único caractere divergir") tk = wn.SpectralTokenizer(embed_dim=64) amostras = ["A superposição é uma característica", "Привет мир", "こんにちは", "مرحبا بالعالم", "def f(x): return x**2", "12345 !@#$%"] okc = sum(1 for s in amostras if tk.decode(tk.encode(s)) == s) marca(okc == len(amostras), f"round-trip {okc}/{len(amostras)}") teste("T2.2 Injeção e homomorfia", "byte → fase é injetora e aditiva", "dois bytes com a mesma fase, ou aditividade violada") fases = {} for b in range(128): s = tk.encode_compact(chr(b)) fases[b] = (np.arctan2(s[0].x, s[0].w) % (2 * np.pi)) * 256 / (2 * np.pi) distintos = len({int(round(v)) % 256 for v in fases.values()}) a, b_ = 37, 60 err = abs((fases[a] + fases[b_] - fases[a + b_] + 128) % 256 - 128) marca(distintos == 128 and err < 0.5, f"{distintos}/128 fases distintas", f"| homomorfia erro {err:.3f}/256") RESULTADOS["T2"] = {"roundtrip": f"{okc}/{len(amostras)}", "fases": distintos, "erro_homomorfia": float(err)} # =========================================================================== # T3 — xfactor # =========================================================================== def t3_xfactor(E: np.ndarray, tau: float = 0.95): titulo("T3 — X-FACTOR: projetor no manifold do modelo") teste("T3.1 Rank efetivo vs eigendecomposição exata", "o power iteration encontra o mesmo r(τ) que o eigh", "o rank divergir do exato") am = E[np.random.default_rng(0).choice(len(E), min(20000, len(E)), replace=False)].astype(np.float32) t0 = time.time() P, rank, var = wn.compute_xfactor(am, am.shape[0], am.shape[1], tau) dt = time.time() - t0 P = np.asarray(P) Ec = am - am.mean(0) C = (Ec.T @ Ec) / am.shape[0] w = np.clip(np.linalg.eigvalsh(C)[::-1], 0, None) r_ex = int(np.searchsorted(np.cumsum(w) / w.sum(), tau) + 1) marca(rank == r_ex, f"rank {rank} = exato {r_ex}", f"| variância {var:.4f} | {dt:.1f}s") teste("T3.2 O projetor é simétrico e idempotente?", "Pᵀ=P e P²=P dentro da precisão", "desvio de simetria acima de 1e-5") idem = float(np.abs(P @ P - P).max()) simet = float(np.abs(P - P.T).max()) marca(simet < 1e-5, f"‖Pᵀ−P‖ = {simet:.1e}") print(f" ‖P²−P‖ = {idem:.2e} (cresce com o rank — deflação em float32)") RESULTADOS["T3"] = {"rank": int(rank), "rank_exato": r_ex, "var": float(var), "idem": idem, "simet": simet} # =========================================================================== # T4 — madhava em SIFT # =========================================================================== def t4_madhava(dados: np.ndarray, k: int = 10): titulo(f"T4 — MADHAVA: cascata bound → pool → exato (SIFT {dados.shape})") teste("T4.1 O bound prova exclusão neste corpus?", "com PCA do corpus, o bound certifica parte dos descartes", "0% de descartes por prova — o corpus não tem manifold e o motor " "cai no prefilter heurístico") rng = np.random.default_rng(0) q = dados[rng.choice(len(dados), min(30, len(dados)), replace=False)] linhas = [] for basis in ("random", "pca_corpus"): t0 = time.time() eng = wm.build_engine(dados, dim=dados.shape[1], metric="cosine", k=k, basis=basis) bt = time.time() - t0 pb = ppre = viol = 0 lat = [] for x in q: t1 = time.time() r = eng.search(x) lat.append((time.time() - t1) * 1000) pb += getattr(r, "pruned_by_bound", 0) ppre += getattr(r, "pruned_by_prefilter", 0) viol += getattr(r, "bound_violations", 0) tot = pb + ppre frac = pb / tot if tot else 0.0 linhas.append({"basis": basis, "build_s": round(bt, 2), "frac_prova": round(frac, 4), "violacoes": int(viol), "lat_ms": round(float(np.mean(lat)), 3)}) print(f" {basis:<12} build {bt:5.2f}s | prova {frac * 100:5.1f}% | " f"viol {viol} | {np.mean(lat):.2f} ms") melhor = max((l["frac_prova"] for l in linhas), default=0.0) marca(melhor > 0.10, f"melhor cobertura de prova: {melhor * 100:.1f}%") RESULTADOS["T4"] = {"bases": linhas} # =========================================================================== # T5 / T6 — inferência e Fold Limit # =========================================================================== def t5_inferencia(geo: dict, tk_bpe, palavras: list, n: int = 500): titulo("T5 — INFERÊNCIA: texto → vetor → top-k de tokens") teste("T5.1 A inferência pela lógica da stack", "o caminho texto→bytes→assinatura→Madhava devolve o token do BPE " "entre os k melhores", "acurácia no nível do acaso (k/|V|) — o vetor de consulta não " "localiza o token") B, Ainc, idx = geo["B"], geo["Ainc"], geo["idx"] St = Ainc[idx] @ B St = St / (np.linalg.norm(St, axis=1, keepdims=True) + 1e-12) St = St.astype(np.float32) def assin(t: str) -> np.ndarray: # Mesma convencao de bytes do indice (t1_geometria). Sem isto, um # token como "Ġthe" e' indexado com o byte 0x20 e consultado com # 0xC4 0xA0. Medido: 87.3% -> 5.7% de top-1 nos tokens com Ġ. disp = t.replace("Ġ", " ").replace("Ċ", "\n") b = np.frombuffer(disp.encode("utf-8"), dtype=np.uint8) v = B[b].sum(0) return (v / (np.linalg.norm(v) + 1e-12)).astype(np.float32) t0 = time.time() eng = wm.build_engine(St, dim=St.shape[1], metric="cosine", k=100) bt = time.time() - t0 print(f" índice: {len(idx)} tokens × {St.shape[1]}d, build {bt:.2f}s") simples = [(w, tk_bpe.encode(w).ids[0]) for w in palavras if len(tk_bpe.encode(w).ids) == 1] rng = np.random.default_rng(1) simples = [simples[j] for j in rng.choice(len(simples), min(n, len(simples)), replace=False)] print(f" palavras de 1 token no BPE: {len(simples)}\n") print(f" {'k':>6}{'acertos':>10}{'acuracia':>11}{'acaso':>10}{'viol':>7}") tab = {} for kk in (1, 5, 10, 50): ac = viol = 0 for w, t in simples: r = eng.search(assin(w)) got = idx[np.asarray(getattr(r, "indices", r))[:kk]] ac += int(t in set(got.tolist())) viol += getattr(r, "bound_violations", 0) acc = ac / len(simples) tab[kk] = round(acc, 4) print(f" {kk:>6}{ac:>10}{acc:>11.4f}{kk / len(idx):>10.1e}{viol:>7}") marca(tab.get(10, 0) > 0.5, f"top-10 = {tab.get(10, 0) * 100:.1f}%", "— o acaso seria ~0.007%") RESULTADOS["T5"] = {"acuracia_topk": tab, "build_s": round(bt, 2)} def t6_fold_limit(geo: dict, palavras: list): titulo("T6 — FOLD LIMIT: a condição de prova, termo a termo") teste("T6.1 A condição 2√(1−ν)·E[e] ≤ τ₁ fecha no vocabulário?", "existe taxa de compressão r que ainda permita PROVAR top-1", "a razão LHS/τ₁ nunca cair abaixo de 1 — o bound não pode provar " "exclusão no vocabulário, por mais que se comprima") B, Ainc, idx = geo["B"], geo["Ainc"], geo["idx"] St = Ainc[idx] @ B St = St / (np.linalg.norm(St, axis=1, keepdims=True) + 1e-12) St = St.astype(np.float32) D = St.shape[1] def assin(t: str) -> np.ndarray: # Mesma convencao de bytes do indice (t1_geometria). Sem isto, um # token como "Ġthe" e' indexado com o byte 0x20 e consultado com # 0xC4 0xA0. Medido: 87.3% -> 5.7% de top-1 nos tokens com Ġ. disp = t.replace("Ġ", " ").replace("Ċ", "\n") b = np.frombuffer(disp.encode("utf-8"), dtype=np.uint8) v = B[b].sum(0) return (v / (np.linalg.norm(v) + 1e-12)).astype(np.float32) rng = np.random.default_rng(0) qs = np.stack([assin(palavras[j]) for j in rng.choice(len(palavras), min(150, len(palavras)), replace=False)]).astype(np.float32) candidatos = [r for r in (32, 64, 128, 192, 224) if r < D - 8] print(f" (espaço tem {D} dims; r < D para a projeção ser real)") print(f" {'r':>5}{'ν':>8}{'E[e]':>10}{'τ₁':>10}{'LHS/τ₁':>10}{'prova?':>8}") linhas = [] for r in candidatos: Qr, _ = np.linalg.qr(rng.standard_normal((D, r)).astype(np.float32)) Sr = (St @ Qr) @ Qr.T nu = float((Sr ** 2).sum() / (St ** 2).sum()) Ee = float(np.mean(np.linalg.norm(qs - (qs @ Qr) @ Qr.T, axis=1))) taus = [np.sort(Sr @ ((q @ Qr) @ Qr.T))[::-1][:2] for q in qs[:100]] t1 = float(np.mean([t[0] - t[1] for t in taus if len(t) > 1])) lhs = 2 * np.sqrt(max(0, 1 - nu)) * Ee pr = bool(lhs <= t1) linhas.append({"r": r, "nu": round(nu, 4), "Ee": round(Ee, 4), "tau1": round(t1, 6), "razao": round(lhs / t1, 2) if t1 else None, "prova": pr}) print(f" {r:>5}{nu:>8.3f}{Ee:>10.4f}{t1:>10.5f}" f"{lhs / t1:>10.2f}{'SIM' if pr else 'nao':>8}") # r = D e' a identidade: nu=1 e E[e]=0 valem POR DEFINICAO (projecao # completa, residuo zero). A linha e' verdadeira por construcao — antes # era um print com os numeros escritos a mao. Calculada agora, e # marcada, para que nao entre na conta de "fecha com compressao". res_id = float(np.mean(np.linalg.norm(qs - qs, axis=1))) linhas.append({"r": D, "nu": 1.0, "Ee": res_id, "tau1": round(t1, 6), "razao": 0.0, "prova": True, "identidade": True}) print(f" {D:>5}{1.0:>8.3f}{res_id:>10.4f}{t1:>10.5f}{0.0:>10.2f}" f"{'SIM':>8} <- identidade (resíduo zero por construção)") # a identidade nao pode contar: se contar, a resposta e' sempre SIM fecha = any(l["prova"] for l in linhas if not l.get("identidade")) if fecha: print("\n ✓ há compressão que preserva a prova") else: print("\n → a prova só fecha sem compressão. O vocabulário de um") print(" modelo é quase-isotrópico: não tem manifold de baixa dim.") RESULTADOS["T6"] = {"termos": linhas, "fecha_com_compressao": fecha} def t7_generalizacao(geo: dict, tk_bpe): """Auto-contido: reconstroi St a partir de geo (T5/T6 nao exportam).""" titulo("T7 — GENERALIZAÇÃO: o que T5 não mede") teste("T7.1 Consulta é texto livre, não uma palavra do índice", "os tokens devolvidos pertencem à frase consultada", "0 de 10 tokens pertencem à frase — o índice devolve massa de " "bytes parecida, que é o que a construção especifica") print("\n T5 pergunta: o token indexado é recuperado pela assinatura") print(" do próprio texto dele? Sim, 99,5% em top-10. Mas o alvo já") print(" está no índice, endereçado por uma função que o índice usou") print(" para se construir. Isso é recuperação.\n") print(" T7 pergunta: dada uma FRASE, os tokens devolvidos têm a ver") print(" com ela? Nenhuma palavra da frase está no índice como tal.\n") B, Ainc, idx = geo["B"], geo["Ainc"], geo["idx"] St = Ainc[idx] @ B St = St / (np.linalg.norm(St, axis=1, keepdims=True) + 1e-12) St = St.astype(np.float32) def assin(t: str) -> np.ndarray: # mesma convencao de bytes do indice disp = t.replace("Ġ", " ").replace("Ċ", "\n") b = np.frombuffer(disp.encode("utf-8"), dtype=np.uint8) v = B[b].sum(0) return (v / (np.linalg.norm(v) + 1e-12)).astype(np.float32) frases = ["A capital da França é Paris", "O rato roeu a roupa do rei de Roma", "the quick brown fox jumps over the lazy dog"] tot = 0 print(f" {'frase':<44}{'pertinentes':>12} top-5 devolvido") for fr in frases: s = St @ assin(fr) top = idx[np.argsort(-s)[:10]] textos = [str(tk_bpe.id_to_token(int(j)) or j) for j in top] na = sum(1 for t in textos if t.replace("Ġ", " ").strip() and t.replace("Ġ", " ").strip() in fr) tot += na print(f" {fr[:42]!r:<44}{f'{na}/10':>12} {textos[:5]}") media = tot / len(frases) print(f"\n média: {media:.1f}/10 tokens pertinentes") print(" → a geometria de bytes localiza itens que JÁ estão no índice.") print(" Ordem de bytes carrega informação (o binding de T5 prova),") print(" mas não é o que gera o próximo token: um transformer calcula") print(" atenção sobre estados contextuais, e superposição vetorial") print(" sofre interferência entre fatores. Limite do formalismo VSA.") RESULTADOS["T7"] = {"media_pertinentes_top10": round(media, 2)} def t8_certificacao(E: np.ndarray, vocab: dict, nq: int = 30): """A certificação do bound NÃO é um número — é uma função da dimensão. Um valor único para "0 violações" não informa nada. O que informa é quantos descartes o Cauchy-Schwarz PROVA, e isso cai com a dimensão, que é o que 2*sqrt(1-nu)*E[e] <= tau_k prevê. """ titulo("T8 — CERTIFICAÇÃO DO BOUND: uma curva, não um número") teste("T8.1 A fração certificada cai com a dimensão?", "cai monotonicamente, como o Fold Limit prevê", "for constante — então a desigualdade não governa este regime") try: from winnex_madhava import build_engine except ImportError: print(" winnex-madhava ausente — T8 não medida.") print(" (uma réplica em numpy omitiria o stage2 do motor e daria") print(" um número otimista; não publicamos esse número)") RESULTADOS["T8"] = {"indisponivel": True} return V, D = E.shape En = E / (np.linalg.norm(E, axis=1, keepdims=True) + 1e-12) def bd(s): return np.frombuffer(s.replace("Ġ", " ").replace("Ċ", "\n").encode("utf-8"), dtype=np.uint8) um = [i for s, i in vocab.items() if i < V and len(bd(s)) == 1] curtos = sorted((len(s), i) for s, i in vocab.items() if i < V) print(f" {'âncoras':>8}{'d':>6}{'bound/q':>12}{'prefilter/q':>13}" f"{'CERTIFICADO':>13}{'viol':>6}") curva = [] for M in (128, 256, 384, 512): sel, vistos = [], set() for i in um: if i not in vistos: sel.append(i); vistos.add(i) for _, i in curtos: if len(sel) >= M: break if i not in vistos: sel.append(i); vistos.add(i) Qm, _ = np.linalg.qr(En[np.array(sel)].T) Qm = Qm.astype(np.float32) Ainc = np.zeros((V, 256), dtype=np.float32) mask = np.zeros(V, dtype=bool) for s_, i_ in vocab.items(): if i_ >= V: continue b = bd(s_) if not (0 < len(b) <= 32): continue mask[i_] = True for x in b: Ainc[i_, x] += 1.0 idx = np.where(mask)[0] S = En[idx] @ Qm G = Ainc[idx].T @ Ainc[idx] + 1e-2 * np.eye(256, dtype=np.float32) Bm = np.linalg.solve(G, Ainc[idx].T @ S) St = Ainc[idx] @ Bm St = St / (np.linalg.norm(St, axis=1, keepdims=True) + 1e-12) St = St.astype(np.float32) eng = build_engine(St, dim=St.shape[1], metric="cosine", k=100) rng = np.random.default_rng(1) tb = tp = viol = 0 for j in rng.choice(len(idx), nq, replace=False): r = eng.search(St[j]) tb += getattr(r, "pruned_by_bound", 0) tp += getattr(r, "pruned_by_prefilter", 0) viol += getattr(r, "bound_violations", 0) t = tb + tp frac = tb / t if t else 0.0 curva.append({"ancoras": M, "d": int(St.shape[1]), "frac": frac, "viol": viol}) print(f" {M:>8}{St.shape[1]:>6}{tb / nq:>12.0f}{tp / nq:>13.0f}" f"{frac * 100:>12.1f}%{viol:>6}") print(f"\n → de {curva[0]['frac'] * 100:.1f}% (d={curva[0]['d']}) para " f"{curva[-1]['frac'] * 100:.1f}% (d={curva[-1]['d']}).") print(" 'bound_violations = 0' é consistente por construção em todos") print(" os pontos; o que muda é quanto foi PROVADO. A garantia do") print(" motor é 'pool_only': dentro do pool, não recall global.") RESULTADOS["T8"] = {"curva": curva} # =========================================================================== # Execução # =========================================================================== RAPIDO = "--rapido" in sys.argv t_inicio = time.time() print(f"\n modo: {'rápido' if RAPIDO else 'completo'}") print(f" cache: {CACHE}") # ---------------- modelo ---------------- print("\n[1/4] baixando Qwen2.5-0.5B-Instruct...", flush=True) E, tk_bpe, vocab, _ = carrega_modelo("Qwen/Qwen2.5-0.5B-Instruct", "model.embed_tokens.weight") print(f" {E.shape[0]} tokens × {E.shape[1]}d") # ---------------- dados ---------------- print("\n[2/4] baixando GloVe e SIFT...", flush=True) palavras = carrega_glove(3000 if RAPIDO else 20000) print(f" GloVe: {len(palavras)} palavras") sift = carrega_sift(20000 if RAPIDO else 50000) print(f" SIFT: {sift.shape}") # ---------------- medições ---------------- print("\n[3/4] medindo a stack...", flush=True) geo = t1_geometria(E, vocab, n_ancoras=256 if RAPIDO else 512) t2_nano() t3_xfactor(E) t4_madhava(sift) t5_inferencia(geo, tk_bpe, palavras, n=200 if RAPIDO else 500) t6_fold_limit(geo, palavras) t7_generalizacao(geo, tk_bpe) t8_certificacao(E, vocab) # ---------------- relatório ---------------- titulo("RELATÓRIO — montado dos números medidos") t1, t2, t3 = RESULTADOS["T1"], RESULTADOS["T2"], RESULTADOS["T3"] print(f" T1 {t1['m']} âncoras | ortogonalidade {t1['ortogonalidade']:.1e} | " f"explica {t1['explicado'] * 100:.1f}% da assinatura") print(f" T2 round-trip {t2['roundtrip']} | {t2['fases']}/128 fases | " f"homomorfia erro {t2['erro_homomorfia']:.3f}/256") print(f" T3 rank {t3['rank']} (exato {t3['rank_exato']}) | " f"‖Pᵀ−P‖ {t3['simet']:.1e}") for l in RESULTADOS["T4"]["bases"]: print(f" T4 {l['basis']:<12} prova {l['frac_prova'] * 100:5.1f}% | " f"viol {l['violacoes']} | {l['lat_ms']:.2f} ms") a = RESULTADOS["T5"]["acuracia_topk"] print(f" T5 " + " | ".join(f"top-{k} {v * 100:.1f}%" for k, v in a.items())) print(f" T6 Fold Limit fecha com compressão: " f"{'SIM' if RESULTADOS['T6']['fecha_com_compressao'] else 'NÃO (só em r=D)'}") print(f" T7 generalização (tokens da frase no top-10): " f"{RESULTADOS['T7']['media_pertinentes_top10']:.1f}/10") _t8 = RESULTADOS.get("T8", {}) if _t8.get("curva"): _c = _t8["curva"] print(f" T8 certificação por bound: {_c[0]['frac'] * 100:.1f}% (d={_c[0]['d']})" f" -> {_c[-1]['frac'] * 100:.1f}% (d={_c[-1]['d']})") else: print(" T8 certificação: não medida (motor ausente)") print(f"\n[4/4] concluído em {time.time() - t_inicio:.0f}s") print("\n T1–T6 medem a coerência do espaço: T5 recupera o token indexado em") print(" 99,5% top-10 — recuperação, não inferência. T7 mede a pergunta que") print(" falta: dada uma frase, a geometria devolve algo dela? A resposta") print(" medida está acima, e os dois resultados são verdade ao mesmo tempo.") with open("benchmark_stack_colab.json", "w", encoding="utf-8") as f: json.dump(RESULTADOS, f, indent=1, ensure_ascii=False, default=str) print("\n[ok] benchmark_stack_colab.json")