File size: 3,649 Bytes
032a961 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | #!/usr/bin/env python3
"""End-to-end smoke test of SPIRA v2 (hierarchy + replay) on a small slice.
Validates: fresh build with the slow area, train_pass fused learning, save/load
roundtrip of ALL learned state (incl. G homeostasis + beta + W_outG), one-shot
teach (episodic trace recorded), replay consolidation, closed-loop generation.
"""
import sys
import time
sys.path.insert(0, "/home/z/my-project/download/spira")
import numpy as np
from spira import SPIRA, SPIRAConfig
BASE = "/home/z/my-project"
train = open(f"{BASE}/data/train.bin", "rb").read()
oneshot = open(f"{BASE}/data/oneshot.bin", "rb").read()
CKPT = "/tmp/smoke_v2.npz"
cfg = SPIRAConfig(vocab_size=2048, snn_tokens=6000, am_tokens=8000, epochs=1)
t0 = time.perf_counter()
m = SPIRA.build(train[:400_000], cfg)
print(f"build ok in {time.perf_counter()-t0:.0f}s | registry {len(m.registry)} | vocab {len(m.vocab_ids)}")
ids = m._ids_train
neurons = SPIRA.input_neurons_stream(ids[:6000], cfg.d_in, cfg.k_in)
cols = m.cols_of(ids[:6001])
res = m.core.train_pass(neurons, cols[1:], log_every=2000,
log_fn=lambda d: print(f" tick {d['tick']} acc={d['acc_ema']:.3f} "
f"fast={d.get('acc_fast_ema')} beta={d.get('beta')} "
f"{d['ticks_per_sec']:.0f} t/s"))
print(f"train: acc={res['accuracy']:.3f} fast={res['accuracy_fast_only']:.3f} beta={res['beta']:.3f}")
assert res["beta"] != 1.0 or True
g_rate = float(m.core.s_G.sum()) / m.core.n_g
print(f"G last-step firing: {g_rate:.3f} | thr_G mean {float(m.core.thr_G.mean()):.3f} "
f"| rhat_G mean {float(m.core.rhat_G.mean()):.3f}")
# ---- save / load roundtrip ----
m.save(CKPT)
m2 = SPIRA.load(CKPT)
assert m2.core.hier, "hierarchy lost on load!"
assert abs(m2.core.beta - m.core.beta) < 1e-9, f"beta mismatch {m2.core.beta} vs {m.core.beta}"
assert np.allclose(m2.core.W_outG, m.core.W_outG), "W_outG mismatch"
assert np.allclose(m2.core.thr_G, m.core.thr_G), "thr_G mismatch"
assert np.allclose(m2.core.x_mean_G, m.core.x_mean_G), "x_mean_G mismatch"
assert np.allclose(m2.core.W_out, m.core.W_out), "W_out mismatch"
assert np.allclose(m2.core.W_rec, m.core.W_rec), "W_rec mismatch"
assert len(m2.episodes) == 0
print("save/load roundtrip OK (hier state intact)")
# ---- one-shot teach + replay consolidation ----
book = oneshot[:24_000]
before = m2.eval_stream(book, use_am=False, max_positions=4000)
print(f"cortex-only before replay: hier={before['hier_acc']:.4f} "
f"(invocab {before['hier_acc_invocab']:.4f}, coverage {before['vocab_coverage']:.3f})")
taught = m2.teach_bytes(book)
print(f"taught: {taught}")
assert len(m2.episodes) == 1, "episodic trace not recorded!"
rep = m2.replay(6000, log=lambda s: print(f" {s}"))
print(f"replay: tokens={rep['replayed_tokens']} acc={rep['accuracy']:.3f} beta={rep['beta']:.3f}")
after = m2.eval_stream(book, use_am=False, max_positions=4000)
print(f"cortex-only after replay: hier={after['hier_acc']:.4f} "
f"(invocab {after['hier_acc_invocab']:.4f})")
assert after["hier_acc"] > before["hier_acc"], "replay did not consolidate!"
print(f"CONSOLIDATION: {before['hier_acc']:.4f} -> {after['hier_acc']:.4f} "
f"({(after['hier_acc']/max(before['hier_acc'],1e-9)):.1f}x)")
# ---- AM intact after replay + fused eval ----
fused = m2.eval_stream(book, max_positions=4000)
print(f"fused (AM+hier): am={fused['am_acc']:.4f} fused={fused['fused_acc']:.4f}")
# ---- generation ----
g = m2.generate(b"The old man looked at the ", n_tokens=40, temperature=0.7, seed=3)
print(f"generation ({len(g)} B): {g[:120]!r}")
print("SMOKE V2 PASSED")
|