SPIRA / v3_scale /code /gen /gen_eval.py

v3_scale: code banc v4/gen/v3, rapports validés, corpus_v3 (sha256 vérifié), README

519e2ef verified 17 days ago

6.44 kB

#!/usr/bin/env python3
"""Checker — éval de génération v1. Usage : gen_eval.py CKPT CORPUS_NAME TAG   (PYTHONHASHSEED=0, OMP_NUM_THREADS=1)
Prompts : prompts_v1.json (7 prompts frais, 64-o overlap = 0 avec old/broad/v3 train). 200 octets générés par prompt.
Décodages (aucune variante nouvelle) :
  État : snapshot SNN (core.snapshot + W_rec) restauré avant CHAQUE génération (SPIRA.generate seul n'est pas déterministe d'un appel à l'autre : plasticité).
  fused : boucle = copie de SPIRA.generate du repo (top_k=cfg.top_k, T=cfg.temperature, seed=0) ; vérifiée == m.generate sur le 1er prompt.
  v2b   : référence m9g v2b — même boucle, mais si c_AM=1-exp(-am_w/m)<0.5 : scores = logsoftmax(fused) + 2·log p_JM(col|p2,p1)
          (NGram JM 0.2/0.5/0.3 copiée de m9g/eval_m9g_v2.py, comptée sur le train du modèle).
Métriques vs vraie continuation : byte_acc (octet à octet sur 200), valid_word_rate (lexique corpus_v3 freq≥3, FR+EN), rep_rate (8-grammes d'octets
déjà vus dans la génération), lang/domain (stopwords FR/EN ; code si ≥4 % de []{}()=;_#<>) et drift vs continuation, longest_copy (plus long extrait
de la génération présent dans le train du modèle). Sortie results_gen_v1.jsonl + samples/."""
import os, sys, json, re, time, resource
sys.path.insert(0, "/workspace/spira_shared/checker/scale")
import numpy as np
from bench_common import load_model, CORPORA, rd, fnv1a64
D = os.path.dirname(os.path.abspath(__file__)); ckpt, cname, tag = sys.argv[1:4]; NB = 200; GATE, LAMB = 0.5, 2.0; JM = (0.2, 0.5, 0.3)
OUT = os.environ.get("OUT", f"{D}/results_gen_v1.jsonl")
P = json.load(open(f"{D}/prompts_v1.json")); LEX = set(json.load(open(f"{D}/lexicon_v3.json")))
t0 = time.time(); train = rd(f"{CORPORA[cname]}/train.bin"); m, kind, saved, cal = load_model(ckpt, train)
V = m.cfg.vocab_size; K = V + 1
src = open("/workspace/spira_shared/checker/m9g/eval_m9g_v2.py").read(); g = {"np": np, "K": K, "JM": JM}
exec(src[src.index("class NGram"):src.index("ng = NGram")], g); ng = g["NGram"](m.cols_of(m._ids_train))
S0 = (m.core.snapshot(), m.core.W_rec.copy())
def pristine(): m.core.restore(S0[0]); m.core.W_rec = S0[1].copy()
def lsm(u): z = u - u.max(); return z - np.log(np.exp(z).sum())
def gen(prompt, n_tokens, v2b, seed=0):
    pristine(); cfg = m.cfg; rng = np.random.default_rng(seed); m.core.reset_dynamics(); m._win_count = 0
    ids = [fnv1a64(p) for p in m.ept.segment(prompt)]; out = bytearray(prompt); x = np.zeros(m.core.n_res, dtype=np.float32); ng_used = 0
    for pid in ids: x = m._feed(pid)
    for _ in range(n_tokens):
        fused, am_val, am_w, bypass_id = m.fuse_scores(x, ids)
        if bypass_id is not None: chosen_id = bypass_id
        else:
            scores = np.asarray(fused, dtype=np.float64).copy()
            if v2b and 1.0 - np.exp(-am_w / m.am.m) < GATE:
                cc = m.cols_of(np.array(ids[-2:], dtype=np.uint64)); p1 = int(cc[-1]); p2 = int(cc[0]) if len(cc) > 1 else None
                scores = lsm(scores[:K]) + LAMB * ng.logp(p2, p1); ng_used += 1
            scores[V] = -np.inf; k = max(1, min(cfg.top_k, V)); top = np.argpartition(scores, -k)[-k:]
            w = np.exp((scores[top] - scores[top].max()) / max(cfg.temperature, 1e-6)); w /= w.sum()
            chosen_id = int(m.vocab_ids[int(top[rng.choice(len(top), p=w)])])
        b = m.registry.bytes_of(chosen_id)
        if b is None: break
        out.extend(b); ids.append(chosen_id); x = m._feed(chosen_id)
        if len(out) - len(prompt) >= NB: break
    return bytes(out[len(prompt):len(prompt) + NB]), ng_used
FR = set("le la les de des du un une et est que qui dans pour pas sur au aux il elle ne se ce son sa ses avec par plus mais".split())
EN = set("the of and to a in is that it was for on with as his her he she they not but at by be this had".split())
WRE = re.compile(r"[a-zà-ÿœæç']+")
def words(t): return [w.strip("'") for w in WRE.findall(t.lower()) if len(w.strip("'")) >= 2]
def lang(t):
    ws = words(t); f = sum(w in FR for w in ws); e = sum(w in EN for w in ws); return "fr" if f > e else ("en" if e > f else "?")
def dom(t): return "code" if t and sum(c in "[]{}()=;_#<>" for c in t) / len(t) >= 0.04 else "prose"
def longest_copy(gb):
    L, best = 0, b""
    for i in range(len(gb)):
        while i + L < len(gb) and gb[i:i + L + 1] in train: L += 1; best = gb[i:i + L]
    return L, best
def rep(gb, n=8):
    seen, r, tot = set(), 0, 0
    for i in range(len(gb) - n + 1):
        s = gb[i:i + n]; tot += 1; r += s in seen; seen.add(s)
    return r / max(tot, 1)
os.makedirs(f"{D}/samples", exist_ok=True); tl = time.time() - t0
p0 = bytes.fromhex(P[0]["prompt_b64"]); pristine(); ref = m.generate(p0, 120, seed=0)[len(p0):][:NB]; mine, _ = gen(p0, 120, False)
mine2, _ = gen(p0, 120, False)
assert mine == mine2, "non déterministe"
assert ref[:len(mine)] == mine[:len(ref)], "boucle fused != SPIRA.generate"
print("CHECK determinism OK, loop==SPIRA.generate OK", flush=True)
for dec in ("fused", "v2b"):
    for p in P:
        pb, cb = bytes.fromhex(p["prompt_b64"]), bytes.fromhex(p["cont_hex"]); t1 = time.time(); gb, nu = gen(pb, 400, dec == "v2b")
        gt, ct = gb.decode("utf-8", "replace"), cb.decode("utf-8", "replace"); ws = words(gt); L, ex = longest_copy(gb)
        r = {"tag": tag, "corpus": cname, "kind": kind, "decoding": dec, "prompt": p["name"], "n_bytes": len(gb),
             "byte_acc": sum(a == b for a, b in zip(gb, cb)) / NB, "valid_word_rate": sum(w in LEX for w in ws) / max(len(ws), 1), "n_words": len(ws),
             "rep_rate8": rep(gb), "lang_true": lang(ct), "lang_gen": lang(gt), "dom_true": dom(ct), "dom_gen": dom(gt),
             "lang_drift": lang(gt) != lang(ct), "dom_drift": dom(gt) != dom(ct), "longest_copy": L, "copy_excerpt": ex.decode("utf-8", "replace"),
             "ngram_steps": nu, "gen_s": round(time.time() - t1, 1), "load_s": round(tl, 1), "fusion_cal": cal,
             "peak_rss_mb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024), "seed": 0}
        open(OUT, "a").write(json.dumps(r, ensure_ascii=False) + "\n")
        open(f"{D}/samples/{tag}__{dec}__{p['name']}.txt", "w").write(f"PROMPT:\n{p['prompt']}\n\nVRAI:\n{ct}\n\nGÉNÉRÉ:\n{gt}\n")
        print(dec, p["name"], f"acc={r['byte_acc']:.3f} vw={r['valid_word_rate']:.2f} rep={r['rep_rate8']:.2f} {r['lang_gen']}/{r['lang_true']} {r['dom_gen']}/{r['dom_true']} copy={L}", flush=True)

	#!/usr/bin/env python3
	"""Checker — éval de génération v1. Usage : gen_eval.py CKPT CORPUS_NAME TAG (PYTHONHASHSEED=0, OMP_NUM_THREADS=1)
	Prompts : prompts_v1.json (7 prompts frais, 64-o overlap = 0 avec old/broad/v3 train). 200 octets générés par prompt.
	Décodages (aucune variante nouvelle) :
	État : snapshot SNN (core.snapshot + W_rec) restauré avant CHAQUE génération (SPIRA.generate seul n'est pas déterministe d'un appel à l'autre : plasticité).
	fused : boucle = copie de SPIRA.generate du repo (top_k=cfg.top_k, T=cfg.temperature, seed=0) ; vérifiée == m.generate sur le 1er prompt.
	v2b : référence m9g v2b — même boucle, mais si c_AM=1-exp(-am_w/m)<0.5 : scores = logsoftmax(fused) + 2·log p_JM(col\|p2,p1)
	(NGram JM 0.2/0.5/0.3 copiée de m9g/eval_m9g_v2.py, comptée sur le train du modèle).
	Métriques vs vraie continuation : byte_acc (octet à octet sur 200), valid_word_rate (lexique corpus_v3 freq≥3, FR+EN), rep_rate (8-grammes d'octets
	déjà vus dans la génération), lang/domain (stopwords FR/EN ; code si ≥4 % de []{}()=;_#<>) et drift vs continuation, longest_copy (plus long extrait
	de la génération présent dans le train du modèle). Sortie results_gen_v1.jsonl + samples/."""
	import os, sys, json, re, time, resource
	sys.path.insert(0, "/workspace/spira_shared/checker/scale")
	import numpy as np
	from bench_common import load_model, CORPORA, rd, fnv1a64
	D = os.path.dirname(os.path.abspath(__file__)); ckpt, cname, tag = sys.argv[1:4]; NB = 200; GATE, LAMB = 0.5, 2.0; JM = (0.2, 0.5, 0.3)
	OUT = os.environ.get("OUT", f"{D}/results_gen_v1.jsonl")
	P = json.load(open(f"{D}/prompts_v1.json")); LEX = set(json.load(open(f"{D}/lexicon_v3.json")))
	t0 = time.time(); train = rd(f"{CORPORA[cname]}/train.bin"); m, kind, saved, cal = load_model(ckpt, train)
	V = m.cfg.vocab_size; K = V + 1
	src = open("/workspace/spira_shared/checker/m9g/eval_m9g_v2.py").read(); g = {"np": np, "K": K, "JM": JM}
	exec(src[src.index("class NGram"):src.index("ng = NGram")], g); ng = g["NGram"](m.cols_of(m._ids_train))
	S0 = (m.core.snapshot(), m.core.W_rec.copy())
	def pristine(): m.core.restore(S0[0]); m.core.W_rec = S0[1].copy()
	def lsm(u): z = u - u.max(); return z - np.log(np.exp(z).sum())
	def gen(prompt, n_tokens, v2b, seed=0):
	pristine(); cfg = m.cfg; rng = np.random.default_rng(seed); m.core.reset_dynamics(); m._win_count = 0
	ids = [fnv1a64(p) for p in m.ept.segment(prompt)]; out = bytearray(prompt); x = np.zeros(m.core.n_res, dtype=np.float32); ng_used = 0
	for pid in ids: x = m._feed(pid)
	for _ in range(n_tokens):
	fused, am_val, am_w, bypass_id = m.fuse_scores(x, ids)
	if bypass_id is not None: chosen_id = bypass_id
	else:
	scores = np.asarray(fused, dtype=np.float64).copy()
	if v2b and 1.0 - np.exp(-am_w / m.am.m) < GATE:
	cc = m.cols_of(np.array(ids[-2:], dtype=np.uint64)); p1 = int(cc[-1]); p2 = int(cc[0]) if len(cc) > 1 else None
	scores = lsm(scores[:K]) + LAMB * ng.logp(p2, p1); ng_used += 1
	scores[V] = -np.inf; k = max(1, min(cfg.top_k, V)); top = np.argpartition(scores, -k)[-k:]
	w = np.exp((scores[top] - scores[top].max()) / max(cfg.temperature, 1e-6)); w /= w.sum()
	chosen_id = int(m.vocab_ids[int(top[rng.choice(len(top), p=w)])])
	b = m.registry.bytes_of(chosen_id)
	if b is None: break
	out.extend(b); ids.append(chosen_id); x = m._feed(chosen_id)
	if len(out) - len(prompt) >= NB: break
	return bytes(out[len(prompt):len(prompt) + NB]), ng_used
	FR = set("le la les de des du un une et est que qui dans pour pas sur au aux il elle ne se ce son sa ses avec par plus mais".split())
	EN = set("the of and to a in is that it was for on with as his her he she they not but at by be this had".split())
	WRE = re.compile(r"[a-zà-ÿœæç']+")
	def words(t): return [w.strip("'") for w in WRE.findall(t.lower()) if len(w.strip("'")) >= 2]
	def lang(t):
	ws = words(t); f = sum(w in FR for w in ws); e = sum(w in EN for w in ws); return "fr" if f > e else ("en" if e > f else "?")
	def dom(t): return "code" if t and sum(c in "[]{}()=;_#<>" for c in t) / len(t) >= 0.04 else "prose"
	def longest_copy(gb):
	L, best = 0, b""
	for i in range(len(gb)):
	while i + L < len(gb) and gb[i:i + L + 1] in train: L += 1; best = gb[i:i + L]
	return L, best
	def rep(gb, n=8):
	seen, r, tot = set(), 0, 0
	for i in range(len(gb) - n + 1):
	s = gb[i:i + n]; tot += 1; r += s in seen; seen.add(s)
	return r / max(tot, 1)
	os.makedirs(f"{D}/samples", exist_ok=True); tl = time.time() - t0
	p0 = bytes.fromhex(P[0]["prompt_b64"]); pristine(); ref = m.generate(p0, 120, seed=0)[len(p0):][:NB]; mine, _ = gen(p0, 120, False)
	mine2, _ = gen(p0, 120, False)
	assert mine == mine2, "non déterministe"
	assert ref[:len(mine)] == mine[:len(ref)], "boucle fused != SPIRA.generate"
	print("CHECK determinism OK, loop==SPIRA.generate OK", flush=True)
	for dec in ("fused", "v2b"):
	for p in P:
	pb, cb = bytes.fromhex(p["prompt_b64"]), bytes.fromhex(p["cont_hex"]); t1 = time.time(); gb, nu = gen(pb, 400, dec == "v2b")
	gt, ct = gb.decode("utf-8", "replace"), cb.decode("utf-8", "replace"); ws = words(gt); L, ex = longest_copy(gb)
	r = {"tag": tag, "corpus": cname, "kind": kind, "decoding": dec, "prompt": p["name"], "n_bytes": len(gb),
	"byte_acc": sum(a == b for a, b in zip(gb, cb)) / NB, "valid_word_rate": sum(w in LEX for w in ws) / max(len(ws), 1), "n_words": len(ws),
	"rep_rate8": rep(gb), "lang_true": lang(ct), "lang_gen": lang(gt), "dom_true": dom(ct), "dom_gen": dom(gt),
	"lang_drift": lang(gt) != lang(ct), "dom_drift": dom(gt) != dom(ct), "longest_copy": L, "copy_excerpt": ex.decode("utf-8", "replace"),
	"ngram_steps": nu, "gen_s": round(time.time() - t1, 1), "load_s": round(tl, 1), "fusion_cal": cal,
	"peak_rss_mb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024), "seed": 0}
	open(OUT, "a").write(json.dumps(r, ensure_ascii=False) + "\n")
	open(f"{D}/samples/{tag}__{dec}__{p['name']}.txt", "w").write(f"PROMPT:\n{p['prompt']}\n\nVRAI:\n{ct}\n\nGÉNÉRÉ:\n{gt}\n")
	print(dec, p["name"], f"acc={r['byte_acc']:.3f} vw={r['valid_word_rate']:.2f} rep={r['rep_rate8']:.2f} {r['lang_gen']}/{r['lang_true']} {r['dom_gen']}/{r['dom_true']} copy={L}", flush=True)