v0.6.0: verificador NCM calibrado (sem oráculo, V-1..V-4) + treino conjunto Pensamento↔backbone (fase C, PCGrad, acoplador duplex AC) + MoEs redirecionáveis c/ atenção própria (M-R1..M-R5, JEPA+SOM) + autoajuste por telemetria (A-1..A-5, UCB1) — 46/46 gates, provas r-23..r-27
bb2363d verified | #!/usr/bin/env python3 | |
| """58_autoajuste_iterativo.py — v0.6.0: LOOP DE APRIMORAMENTO contínuo | |
| (análise de telemetria → autoajuste de parâmetros → seleção de arquitetura | |
| por UCB1 → repetição até convergência marginal; provas r-26). | |
| Mecanismo (dados REAIS da Wikipédia PT-BR): | |
| - BRAÇOS de arquitetura: {mlp-pequena, moe-e4, moe-e8-redirect} — o | |
| redirecionável troca papéis cod↔dec a cada rodada (M-R1/M-R4); | |
| - UCB1 (A-4) escolhe o braço: r̄_a + √(2 ln T / n_a) — arrependimento | |
| sublinear; recompensa = melhora relativa da perda preditiva − custo de | |
| tempo normalizado; | |
| - AUTOAJUSTE por telemetria (A-1..A-5): veredito Theil–Sen da série do | |
| braço regula a fração de máscara (estagnado ⇒ explora mais; | |
| divergindo ⇒ explora menos) — parâmetro REGULADO por sinal medido; | |
| - PARADA MARGINAL (A-2 no nível meta): para quando a melhora marginal do | |
| melhor braço em 3 avaliações < 2%; | |
| - AUDITORIA: cada decisão (escolha UCB1, regulação, parada) registrada. | |
| Saída: resultados/autoajuste_iterativo_v060.json | |
| """ | |
| import json, math, os, time | |
| def ucb1_escolher(stats, t): | |
| """A-4: braço com maior r̄ + √(2 ln T / n); braços não-jogados primeiro.""" | |
| for a, (n, _) in stats.items(): | |
| if n == 0: | |
| return a | |
| ln_t = math.log(max(t, 2)) | |
| melhor, melhor_v = None, -1e18 | |
| for a, (n, soma) in stats.items(): | |
| v = soma / n + math.sqrt(2.0 * ln_t / n) | |
| if v > melhor_v: | |
| melhor_v, melhor = v, a | |
| return melhor | |
| def theil_sen(ys): | |
| if len(ys) < 3: | |
| return 0.0, "amostra_insuficiente" | |
| incl = [] | |
| for i in range(len(ys)): | |
| for j in range(i + 1, len(ys)): | |
| incl.append((ys[j] - ys[i]) / (j - i)) | |
| incl.sort() | |
| med = incl[len(incl) // 2] | |
| m = sum(ys) / len(ys) | |
| rel = med / max(abs(m), 1e-9) | |
| return med, ("aprendendo" if rel < -1e-4 else ("divergindo" if rel > 1e-4 else "estagnado")) | |
| def main(): | |
| import khtst_rust.nucleo_rust as nr | |
| pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| res = {"versao": "0.6.0", "modulo": nr.versao(), "auditoria": [], "rodadas": []} | |
| # ---------- dados REAIS ---------- | |
| with open(os.path.join(pasta, "dados", "shard_validacao.jsonl")) as f: | |
| textos = [json.loads(l)["texto"] for l in f][:3] | |
| serie = [] | |
| for t in textos: | |
| serie += [min(ord(c), 255) / 255.0 for c in t[:1600] if c.strip()] | |
| patches = nr.patches_serie(serie, 1, len(serie), 16) | |
| print(f"dados reais: {len(patches)} patches ({len(serie)} caracteres, 3 artigos)", flush=True) | |
| res["dados_reais"] = {"n_patches": len(patches), "fonte": "wikimedia/wikipedia 20231101.pt"} | |
| # ---------- estado do loop ---------- | |
| bracos = { | |
| "mlp-pequena": {"modo": "pequena", "frac": 0.45, "serie_perda": [], "tempo_ref": None}, | |
| "moe-e4": {"modo": "moe", "frac": 0.45, "serie_perda": [], "tempo_ref": None, "nota": "4 experts/papel"}, | |
| "moe-e8-redirect": {"modo": "moe", "frac": 0.45, "serie_perda": [], "tempo_ref": None, "nota": "8 experts + redirect por rodada"}, | |
| } | |
| stats = {a: [0, 0.0] for a in bracos} # n, soma_recompensa | |
| t_global = 0 | |
| melhor_melhora, melhor_braco = 0.0, None | |
| teto_rodadas = 14 | |
| parou_por_margem = False | |
| print("\n=== LOOP: UCB1 (A-4) + regulação por telemetria (Theil–Sen) + parada marginal (A-2) ===", flush=True) | |
| for rodada in range(teto_rodadas): | |
| braco = ucb1_escolher({a: tuple(v) for a, v in stats.items()}, t_global) | |
| est = bracos[braco] | |
| # regulação pré-rodada: veredito Theil–Sen (≥3 pontos) ou variação | |
| # local (2 pontos) da SÉRIE do braço — parâmetro REGULADO por sinal | |
| reg = None | |
| if len(est["serie_perda"]) >= 3: | |
| _, vered = theil_sen(est["serie_perda"]) | |
| if vered == "estagnado" and est["frac"] < 0.75: | |
| est["frac"] = min(0.75, est["frac"] + 0.10) | |
| reg = f"estagnado ⇒ frac_mascara ↑ para {est['frac']:.2f} (explora mais)" | |
| elif vered == "divergindo" and est["frac"] > 0.25: | |
| est["frac"] = max(0.25, est["frac"] - 0.10) | |
| reg = f"divergindo ⇒ frac_mascara ↓ para {est['frac']:.2f} (estabiliza)" | |
| elif len(est["serie_perda"]) == 2: | |
| a_, b_ = est["serie_perda"] | |
| if b_ > a_ * 1.02 and est["frac"] > 0.25: | |
| est["frac"] = max(0.25, est["frac"] - 0.10) | |
| reg = f"piora local ({100*(b_/a_-1):.1f}%) ⇒ frac_mascara ↓ para {est['frac']:.2f}" | |
| elif abs(b_ - a_) < 0.01 * abs(a_) and est["frac"] < 0.75: | |
| est["frac"] = min(0.75, est["frac"] + 0.10) | |
| reg = f"estagnação local ⇒ frac_mascara ↑ para {est['frac']:.2f}" | |
| if reg: | |
| res["auditoria"].append({"rodada": rodada, "tipo": "regulacao", "braco": braco, "acao": reg}) | |
| # treina o braço (passos moderados p/ orçamento de tempo) | |
| t0 = time.time() | |
| r = nr.jepa_com_moe(patches, est["modo"], passos=70, frac_mascara=est["frac"], semente=1000 + rodada) | |
| dt = time.time() - t0 | |
| melhora = (r["perda_pred_inicial"] - r["perda_pred_final"]) / max(r["perda_pred_inicial"], 1e-9) | |
| if est["tempo_ref"] is None: | |
| est["tempo_ref"] = dt | |
| custo = 0.5 * (dt / max(est["tempo_ref"], 1e-9) - 1.0) # custo RELATIVO ao 1º do braço | |
| recompensa = max(0.0, melhora - max(0.0, custo)) | |
| stats[braco][0] += 1 | |
| stats[braco][1] += recompensa | |
| t_global += 1 | |
| est["serie_perda"].append(r["perda_pred_final"]) | |
| if melhora > melhor_melhora: | |
| melhor_melhora, melhor_braco = melhora, braco | |
| res["rodadas"].append({ | |
| "rodada": rodada, "braco": braco, "modo": est["modo"], "frac_mascara": est["frac"], | |
| "perda_inicial": r["perda_pred_inicial"], "perda_final": r["perda_pred_final"], | |
| "melhora_rel": melhora, "recompensa": recompensa, "tempo_s": dt, | |
| "std_emb": r["std_emb"], "regulacao": reg, | |
| }) | |
| print(f" r{rodada:02d} {braco:16s} frac={est['frac']:.2f} perda {r['perda_pred_inicial']:.4f} -> " | |
| f"{r['perda_pred_final']:.4f} (melhora {100*melhora:.1f}%, {dt:.1f}s)" | |
| + (f" | REG: {reg}" if reg else ""), flush=True) | |
| # parada marginal (A-2 meta): só após 10 jogadas globais (a regulação | |
| # precisa de espaço p/ agir); melhor braço sem melhora marginal em 3 avaliações | |
| serie_mb = bracos[melhor_braco]["serie_perda"] | |
| if t_global >= 10 and stats[melhor_braco][0] >= 3 and len(serie_mb) >= 3: | |
| janelas = serie_mb[-3:] | |
| margem = (janelas[0] - janelas[-1]) / max(abs(janelas[0]), 1e-9) | |
| if margem < 0.02: | |
| res["auditoria"].append({ | |
| "rodada": rodada, "tipo": "parada_marginal", "braco": melhor_braco, | |
| "acao": f"margem {100*margem:.2f}% < 2% em 3 avaliações — convergiu (A-2 meta)"}) | |
| parou_por_margem = True | |
| print(f" ⇒ PARADA MARGINAL: {melhor_braco} convergiu (margem {100*margem:.2f}% < 2%)", flush=True) | |
| break | |
| # ---------- consolidação ---------- | |
| medias = {a: (stats[a][1] / stats[a][0] if stats[a][0] else 0.0) for a in stats} | |
| vencedor = max(medias, key=medias.get) | |
| base = bracos["mlp-pequena"]["serie_perda"] | |
| final_moe = bracos[vencedor]["serie_perda"] | |
| gap = (sum(base[-3:]) / max(len(base[-3:]), 1) - sum(final_moe[-3:]) / max(len(final_moe[-3:]), 1)) \ | |
| / max(sum(base[-3:]) / max(len(base[-3:]), 1), 1e-9) | |
| # arrependimento empírico do UCB1 (auditoria — A-4) | |
| melhor_media = max(medias.values()) | |
| arrepend = sum(stats[a][0] * (melhor_media - medias[a]) for a in stats) / max(t_global, 1) | |
| print(f"\n=== CONSOLIDAÇÃO ===") | |
| print(f" vencedor UCB1: {vencedor} (recompensa média {medias[vencedor]:.4f})") | |
| print(f" gap final vs baseline (mlp-pequena): {100*gap:.1f}% de perda preditiva menor") | |
| print(f" arrependimento médio UCB1: {arrepend:.4f} (sublinear — Auer)") | |
| print(f" rodadas executadas: {len(res['rodadas'])} | parada marginal: {parou_por_margem}") | |
| # ---------- autoajuste A-1..A-5 (demo auditada — r-21 §3.5 fechado) ---------- | |
| print("\n=== Autoajuste A-1..A-5 (τ_c, B, N, g, UCB1) ===", flush=True) | |
| auto = nr.demo_autoajuste(semente=42) | |
| print(f" τ_c={auto['tau_c_final']:.3f} | N={auto['n_tentativas_final']} | " | |
| f"B={auto['orcamento_b_final']} | g={auto['gate_final']:.1f} | " | |
| f"ajustes auditados: {auto['ajustes_auditados']}") | |
| print(f" UCB1: melhor braço {auto['ucb1_melhor_braco']} | arrependimento médio {auto['ucb1_arrependimento_medio']:.4f}") | |
| # ---------- gates ---------- | |
| gates = { | |
| "gate_ucb1_encontrou_moe": vencedor.startswith("moe"), | |
| "gate_gap_positivo_vs_baseline": gap > 0.0, | |
| "gate_arrependimento_pequeno": arrepend < 0.15, | |
| "gate_regulacao_ocorreu": any(a["tipo"] == "regulacao" for a in res["auditoria"]), | |
| "gate_parada_ou_teto": parou_por_margem or len(res["rodadas"]) == teto_rodadas, | |
| "gate_melhor_aprendeu": melhor_melhora > 0.0, | |
| } | |
| gates.update({g[5:]: v for g, v in auto["gates"].items()}) | |
| res["consolidacao"] = { | |
| "vencedor": vencedor, "medias_recompensa": medias, "gap_vs_baseline_pct": 100 * gap, | |
| "arrependimento_medio": arrepend, "parou_por_margem": parou_por_margem, | |
| "melhor_melhora_rel": melhor_melhora, | |
| } | |
| res["autoajuste_demo"] = auto | |
| res["gates"] = gates | |
| ok = sum(1 for v in gates.values() if v) | |
| res["resumo"] = {"gates_totais": len(gates), "gates_ok": ok, "todos_verdes": ok == len(gates)} | |
| print(f"\n=== RESUMO: {ok}/{len(gates)} gates verdes ===") | |
| for g, v in gates.items(): | |
| print(f" {'OK ' if v else 'FALHOU'} {g}") | |
| os.makedirs(os.path.join(pasta, "resultados"), exist_ok=True) | |
| with open(os.path.join(pasta, "resultados", "autoajuste_iterativo_v060.json"), "w") as f: | |
| json.dump(res, f, indent=1, ensure_ascii=False) | |
| print("salvo: resultados/autoajuste_iterativo_v060.json") | |
| if not res["resumo"]["todos_verdes"]: | |
| raise SystemExit(1) | |
| if __name__ == "__main__": | |
| main() | |