v0.5.0: suíte SOM-4D + suíte JEPA (12) + tecnologia Pensamento + telemetria Theil–Sen — 29/29 gates verdes (provas r-18..r-22)
e26f4cc verified | #!/usr/bin/env python3 | |
| """50_pensamento.py — Tecnologia Pensamento v0.5.0. | |
| Demonstra o ciclo completo (provas r-21): | |
| 1. (item a) treino cooperativo das N redes paralelas — perda cai; | |
| 2. (item c) consulta OOD (fora do padrão comum) — confiança baixa; | |
| 3. (item d) ciclo planejamento → distribuição de competências → | |
| (re-planejamento | resposta) com orçamento — termina (T-P4); | |
| 4. ganho de confiança da agregação vs 1 rede isolada (T-P2); | |
| 5. escalonamento com N: mais tentativas ⇒ mais confiança (T-P2 empírico). | |
| Também mede a neutralidade no nascimento (T-P5) via confiança da linha de | |
| base e o custo (tempo) por iteração. Saída: resultados/pensamento_v050.json | |
| """ | |
| import json, math, os, time | |
| def main(): | |
| import khtst_rust.nucleo_rust as nr | |
| pasta = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| res = {"versao": "0.5.0", "modulo": nr.versao(), "demos": {}} | |
| # -------------------------------------------- demo principal (N=5, ciclo d) | |
| print("=== Pensamento: demo completa (N=5 tentativas) ===", flush=True) | |
| t0 = time.time() | |
| d = nr.demo_pensamento( | |
| d_cond=8, d_saida=4, n_tentativas=5, | |
| passos_treino=220, limiar_conf=0.55, orcamento=6, semente=42, | |
| ) | |
| dt = time.time() - t0 | |
| print(f" treino cooperativo: perda {d['perda_treino_inicial']:.4f} -> {d['perda_treino_final']:.4f} " | |
| f"(aprendeu: {d['treino_aprendeu']})") | |
| print(f" ciclo d: {d['iteracoes_ciclo_d']} iteracoes | eventos: " | |
| f"{[e['tipo'] for e in d['eventos_ciclo']]}") | |
| print(f" confianca base (1 rede): {d['confianca_base_1_rede']:.4f}") | |
| print(f" confianca Pensamento (N=5): {d['confianca_pensamento']:.4f} " | |
| f"(ganho {d['ganho_confianca']:+.4f})") | |
| res["demos"]["principal"] = {**d, "tempo_s": dt, | |
| "gate_aprendeu": d["treino_aprendeu"], | |
| "gate_terminou": d["iteracoes_ciclo_d"] <= 6, | |
| "gate_ganhou": d["ganho_confianca"] > 0} | |
| # -------------------------------------------- escalonamento com N (T-P2) | |
| print("\n=== Escalonamento com N (T-P2: auto-consistência) ===", flush=True) | |
| escal = [] | |
| for n in (1, 2, 3, 5, 8): | |
| t0 = time.time() | |
| r = nr.demo_pensamento( | |
| d_cond=8, d_saida=4, n_tentativas=n, | |
| passos_treino=200, limiar_conf=0.95, orcamento=8, semente=100 + n, | |
| ) | |
| dt = time.time() - t0 | |
| escal.append({ | |
| "n": n, | |
| "conf_base": r["confianca_base_1_rede"], | |
| "conf_pensamento": r["confianca_pensamento"], | |
| "ganho": r["ganho_confianca"], | |
| "iteracoes": r["iteracoes_ciclo_d"], | |
| "tempo_s": dt, | |
| }) | |
| print(f" N={n}: base {r['confianca_base_1_rede']:.4f} -> pensamento " | |
| f"{r['confianca_pensamento']:.4f} (ganho {r['ganho_confianca']:+.4f}) | {dt:.1f}s") | |
| # T-P2 empírico: a média das confianças com N grande excede a com N pequeno | |
| confs = [e["conf_pensamento"] for e in escal] | |
| monotona = all(c2 >= c1 - 0.02 for c1, c2 in zip(confs, confs[1:])) | |
| n1, nn = escal[0]["conf_pensamento"], escal[-1]["conf_pensamento"] | |
| res["demos"]["escalonamento_n"] = { | |
| "serie": escal, | |
| "confianca_n1": n1, | |
| "confianca_n_max": nn, | |
| "melhora_com_n": nn > n1, | |
| "monotona_quase": monotona, | |
| "gate_tp2": nn >= n1 - 0.02, | |
| } | |
| print(f" N=1: {n1:.4f} | N=8: {nn:.4f} | melhora: {nn > n1}") | |
| # -------------------------------------------- T-P3: montador algébrico | |
| print("\n=== T-P3: montador algébrico (projeção ridge) ===", flush=True) | |
| rng_w = [[0.3 * ((i * 7 + j * 3) % 11) / 11.0 - 0.15 for j in range(4)] for i in range(8)] | |
| r_alvo = [[float((i + j) % 5) / 5.0 for j in range(3)] for i in range(8)] | |
| r = nr.projetar_ridge(rng_w, r_alvo, 1e-6) | |
| # resíduo mínimo: compara com perturbações | |
| def residuo(rr): | |
| return sum((sum(rng_w[i][k] * rr[k][j] for k in range(4)) - r_alvo[i][j]) ** 2 | |
| for i in range(8) for j in range(3)) | |
| f0 = residuo(r) | |
| pert_ok = True | |
| for seed in range(20): | |
| pr = [[r[i][j] + 0.1 * math.sin(seed + i + j) for j in range(3)] for i in range(4)] | |
| if residuo(pr) < f0 - 1e-9: | |
| pert_ok = False | |
| print(f" residuo da projecao: {f0:.6f} | nenhuma perturbacao melhora: {pert_ok}") | |
| res["demos"]["projeção_ridge"] = {"residuo": f0, "otimalidade_vs_perturbacoes": pert_ok, | |
| "gate_tp3": pert_ok} | |
| # ------------------------------------------------------------------ resumo | |
| gates = [] | |
| for nome, s in res["demos"].items(): | |
| gates += [(f"{nome}.{g[5:]}", v) for g, v in s.items() if g.startswith("gate_")] | |
| ok = sum(1 for _, v in gates 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 ===") | |
| os.makedirs(os.path.join(pasta, "resultados"), exist_ok=True) | |
| with open(os.path.join(pasta, "resultados", "pensamento_v050.json"), "w") as f: | |
| json.dump(res, f, indent=1, ensure_ascii=False) | |
| print("salvo: resultados/pensamento_v050.json") | |
| if not res["resumo"]["todos_verdes"]: | |
| raise SystemExit(1) | |
| if __name__ == "__main__": | |
| main() | |