File size: 14,019 Bytes
6b1bc4b | 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 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Verify and evaluate engine 3.0 through its REAL code path (raw post/user data -> features -> ONNX -> score).
1. feature parity : engine features == the training files, every validation and test row
2. score parity : engine model outputs == the offline ensemble predictions
3. replay (NDCG@10): validation / test periods (whole feeds and 25-post pages) + the two later P09 sessions, with
trained vectors (captions included, as in training) | serving vectors (title + body only, what the app stores) | no vectors
vs v2b alone, the warm model alone and random
4. brand-new users : same feeds with no history (interests only) and with no interests
Label everywhere: engaged OR active time >= 10 s (the app-style label the models were trained on)."""
import os, sys, json, time, collections
HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))); sys.path.insert(0, HERE); sys.path.insert(0, os.path.join(HERE, "engine_v2"))
import numpy as np, pandas as pd
from sklearn.metrics import roc_auc_score
from hybrid_engine import HybridRecommender
from engine_lib import CHILD, load_post_tags
ART = os.path.join(HERE, "hybrid_engine", "artifacts_v3"); OUT = os.path.join(HERE, "out"); DATA = os.path.join(HERE, "recbole_data")
eng = HybridRecommender(artifacts_dir=ART); fb = eng.fb; meta = eng.meta; K10 = 10
assert meta["version"] == "hybrid-engine-3.0-bge384" and fb.emb_dim == 384
rd = lambda name, sp: pd.read_csv(os.path.join(DATA, name, f"{name}.{sp}.inter"), sep="\t")
# ---------------- raw rows in training order ----------------
sl = pd.read_csv(os.path.join(OUT, "split_labels_bge.csv")); sl = pd.concat([sl[sl.split == s] for s in ("train", "valid", "test")], ignore_index=True)
it = pd.read_csv(os.path.join(OUT, "interactions_v2.csv"), low_memory=False).drop_duplicates(["user_id", "post_id", "ts"])
mm = sl[["user_id", "post_id", "ts", "split", "label"]].merge(it, on=["user_id", "post_id", "ts"], how="left"); assert len(mm) == len(sl) and mm.title.notna().sum() > 0
rr = np.load(os.path.join(OUT, "engine_v3", "real_rows_v3.npz"), allow_pickle=True); assert (rr["post"] == mm.post_id.values).all()
Qrows, split, label, user, ts = rr["Q"], rr["split"], rr["label"].astype(int), rr["user"], mm.ts.values
z = np.load(os.path.join(OUT, "post_emb_bge384.npz"), allow_pickle=True); EMB = {"fused": dict(zip(z["post_ids"], z["fused"])), "text": dict(zip(z["post_ids"], z["text"])), "none": None}
TAGS = {k: {CHILD[j]: 1.0 for j in np.where(v > 0)[0]} for k, v in load_post_tags(os.path.join(OUT, "post_tags.jsonl")).items()} # binary, like the app's tagger
s_ = lambda x: x if isinstance(x, str) else ""
def post_dict(i, vm="text"):
r = mm.iloc[i]
return dict(post_id=r.post_id, score=r.score, comments=r.comment_count, age_hours=None if pd.isna(r.age_hours) else float(r.age_hours), title=s_(r.title), body=s_(r.body),
tags=TAGS.get(r.post_id, {}), embedding=None if vm == "none" else EMB[vm][r.post_id], author=r.author_hash if isinstance(r.author_hash, str) else None, impressions=0)
def hist_item(i, vm="text"):
r = mm.iloc[i]
return dict(post_id=r.post_id, ts_ms=float(r.ts), visible_ms=float(r.visible_ms), focal_ms=float(r.active_ms), hover_ms=0.0, clicked=bool(r.clicked), engaged=bool(label[i]), # focal := active
community=None, author=r.author_hash if isinstance(r.author_hash, str) else None, embedding=None if vm == "none" else EMB[vm][r.post_id])
HIST = {vm: {u: [hist_item(i, vm) for i in np.where(user == u)[0]] for u in np.unique(user)} for vm in ("fused", "text", "none")}
picks = lambda i: [CHILD[j] for j in np.where(Qrows[i] > 0)[0]]
# ---------------- 1) feature parity ----------------
HP = {u: HybridRecommender._prep_history(h, fb) for u, h in HIST["fused"].items()}
worst = {"cold": 0.0, "warm": 0.0}; perf = {}; n_rows = 0
for sp in ("valid", "test"):
ce, we = rd("reco_engine_bge", sp), rd("reco_ctr_bge_noid_strict", sp); ix = np.where(split == sp)[0]
assert np.allclose(ts[ix] / 1000.0, ce["timestamp:float"].values, atol=1e-3) and np.allclose(ts[ix] / 1000.0, we["timestamp:float"].values, atol=1e-3), "row order"
for k, i in enumerate(ix):
post = post_dict(i, "fused"); X = post["embedding"] / np.linalg.norm(post["embedding"]); post = {**post, "embedding": X}
base = fb.post_floats(post); q, r_, ms = fb.interest_profile(picks(i)); tm = fb.tag_match(fb.tag_vector(post["tags"]), q, r_, ms)
zb = {a: fb.z(a, v) for a, v in base.items()}; zt = dict(zip(meta["stage1"]["features"], (tm - fb.s1_mu) / fb.s1_sd))
hf, _ = fb.history_floats(HP[user[i]], post, float(ts[i]), X); zh = {a: fb.z(a, v) for a, v in hf.items()}
for f in meta["cold"]["float_inputs"]: d = abs({**zb, **zt}[f] - ce.iloc[k][f"{f}:float"]); worst["cold"] = max(worst["cold"], d); perf["cold:" + f] = max(perf.get("cold:" + f, 0), d)
for f in meta["warm"]["float_inputs"]: d = abs({**zb, **zh}[f] - we.iloc[k][f"{f}:float"]); worst["warm"] = max(worst["warm"], d); perf["warm:" + f] = max(perf.get("warm:" + f, 0), d)
n_rows += 1
print("1) feature parity over", n_rows, "rows: max |engine - training file| =", {k: round(v, 5) for k, v in worst.items()}, "| worst:", sorted(((k, round(v, 4)) for k, v in perf.items()), key=lambda x: -x[1])[:4], flush=True)
# ---------------- 2) score parity ----------------
off = np.load(os.path.join(OUT, "engine_v3", "ship3_offline_scores.npz")); SC = {}; t0 = time.time()
for sp in ("valid", "test"):
for u in np.unique(user[split == sp]):
ix = np.where((user == u) & (split == sp))[0]; ix = ix[np.argsort(ts[ix], kind="stable")]
res = eng.score({"interests": picks(ix[0]), "history": HIST["fused"][u]}, [post_dict(i, "fused") for i in ix], [float(ts[i]) for i in ix])
for i, d in zip(ix, res): SC[i] = d
lat = (time.time() - t0) / len(SC) * 1000
ti = np.where(split == "test")[0]; a_, b_ = meta["cold"]["platt"]["coef"], meta["cold"]["platt"]["intercept"]
lgt = np.log(np.clip(off["cold_test"], 1e-6, 1 - 1e-6) / (1 - np.clip(off["cold_test"], 1e-6, 1 - 1e-6))); cold_off = 1 / (1 + np.exp(-(a_ * lgt + b_)))
cd = float(np.abs(np.array([SC[i]["cold_probability"] for i in ti]) - cold_off).max()); wd = float(np.abs(np.array([SC[i]["warm_probability"] for i in ti]) - off["warm_test"]).max())
print(f"2) score parity (test rows): cold max diff {cd:.1e} | warm max diff {wd:.1e} | {lat:.2f} ms per candidate incl. history features", flush=True)
# ---------------- 3) replay ----------------
def ndcg(order_ids, lab):
y = np.array([lab[p] for p in order_ids]); r = np.zeros(K10); r[:min(K10, len(y))] = y[:K10]; disc = 1 / np.log2(np.arange(2, K10 + 2))
return (r * disc).sum() / disc[:min(int(y.sum()), K10)].sum() if y.sum() else np.nan
def run(requests, vm_rank=None):
"""requests: (user, candidates, now, labels). Returns mean NDCG@10 of: shipped engine, v2b alone, warm alone (None if no history), and random."""
out = {"engine": [], "v2b only": [], "warm only": [], "random": []}; rs = np.random.default_rng(0)
for u, c, now, lab in requests:
if sum(lab.values()) == 0 or len(c) < K10 + 2: continue
sc = eng.score(u, c, now); ids = [d["post_id"] for d in sorted(eng.order([dict(d) for d in sc]), key=lambda d: d["position"])]
out["engine"].append(ndcg(ids, lab)); out["v2b only"].append(ndcg([d["post_id"] for d in sorted(sc, key=lambda d: -d["v2b"])], lab))
if sc[0]["warm_probability"] is not None: out["warm only"].append(ndcg([d["post_id"] for d in sorted(sc, key=lambda d: -d["warm_probability"])], lab))
allids = [p["post_id"] for p in c]; out["random"].append(np.mean([ndcg(list(rs.permutation(allids)), lab) for _ in range(50)]))
return {k: (round(float(np.nanmean(v)), 3) if v else None) for k, v in out.items()} | {"feeds": len(out["engine"])}
def period_requests(vm, history=True, interests=True):
res = {}
for sp in ("valid", "test"):
for size in (None, 25):
reqs = []
for u in np.unique(user[split == sp]):
ix = np.where((user == u) & (split == sp))[0]; ix = ix[np.argsort(ts[ix], kind="stable")]
for blk in ([ix] if size is None else [ix[k:k + size] for k in range(0, len(ix) - size // 2, size)]):
reqs.append(({"interests": picks(blk[0]) if interests else [], "history": HIST[vm][u] if history else []}, [post_dict(i, vm) for i in blk], float(ts[blk[0]]) - 1,
{post_dict(i, vm)["post_id"]: int(label[i]) for i in blk}))
res[f"{sp} period, {'whole feeds' if size is None else '25-post pages'}"] = reqs
return res
# -- later sessions
import importlib
lib = {"__name__": "lib", "__file__": os.path.join(HERE, "new_session2", "validate_session.py")}; exec(open(lib["__file__"], encoding="utf8").read(), lib)
from fastembed import TextEmbedding
_bge = TextEmbedding(model_name="BAAI/bge-small-en-v1.5", cache_dir=os.path.join(HERE, "bge_cache"), threads=4)
def load_ses(d):
S, _, T = lib["load_session"](os.path.join(HERE, d)); best = collections.defaultdict(float)
for l in open(os.path.join(HERE, d, "events.jsonl"), encoding="utf8"):
e = json.loads(l)
if e["type"] == "attention_update" and e["d"].get("target_type") == "post": best[e["d"]["post_id"]] = max(best[e["d"]["post_id"]], e["d"].get("active_ms") or 0)
S["active_ms"] = S.post_id.map(best).fillna(0.0); S["lab"] = ((S.engaged.astype(bool)) | (S.active_ms >= 10000)).astype(int)
cache = os.path.join(HERE, d, "new_post_emb_bge.npz")
if not os.path.exists(cache):
v = np.asarray(list(_bge.embed(["\n".join([s_(r.title), s_(r.body), "", ""])[:12000] for r in S.itertuples()], batch_size=32)), dtype=np.float64); v = v / np.linalg.norm(v, axis=1, keepdims=True)
np.savez_compressed(cache, post_ids=S.post_id.values, emb=v.astype("float32"))
c = np.load(cache, allow_pickle=True); return S, dict(zip(c["post_ids"], c["emb"])), {k: {t: 1.0 for t, w in v.items()} for k, v in T.items()}
def ses_hist(S, E, vm):
return [dict(post_id=r.post_id, ts_ms=float(r.ts), visible_ms=float(r.visible_ms), focal_ms=float(r.active_ms), hover_ms=0.0, clicked=bool(r.clicked), engaged=bool(r.lab), community=None,
author=r.author, embedding=None if vm == "none" else E[r.post_id]) for r in S.itertuples()]
def ses_cand(S, E, T, vm):
return [dict(post_id=r.post_id, score=r.score, comments=r.comments, age_hours=r.age_hours, title=s_(r.title), body=s_(r.body), tags=T.get(r.post_id, {}),
embedding=None if vm == "none" else E[r.post_id], author=r.author, impressions=0) for r in S.itertuples()]
S1, E1, T1 = load_ses("new_session"); S2, E2, T2 = load_ses("new_session2")
p09 = [i for i in range(len(user)) if user[i] == "P09" and split[i] == "test"][0]; P09_PICKS = picks(p09)
print(f" later sessions: session 1 {len(S1)} posts, {int(S1.lab.sum())} positives | session 2 {len(S2)} posts, {int(S2.lab.sum())} positives | P09 interests {P09_PICKS}", flush=True)
def session_requests(vm, history=True, interests=True):
base = HIST[vm]["P09"] if history else []; res = {}
pg = lambda n: [list(range(k, min(k + 25, n))) for k in range(0, n - 12, 25)]
for name, S, E, T, prior in (("later session 1", S1, E1, T1, []), ("later session 2", S2, E2, T2, ses_hist(S1, E1, vm) if history else [])):
C = ses_cand(S, E, T, vm); lab = dict(zip(S.post_id, S.lab)); hs = base + prior; pick = P09_PICKS if interests else []
res[f"{name}, one feed"] = [({"interests": pick, "history": hs}, C, float(S.ts.min()) - 1, lab)]
res[f"{name}, 25-post pages"] = [({"interests": pick, "history": hs + (ses_hist(S, E, vm) if history else [])}, [C[i] for i in p], float(S.ts.iloc[p[0]]) - 1, {C[i]["post_id"]: int(S.lab.iloc[i]) for i in p}) for p in pg(len(S))]
return res
results = {}
for vm, nm in (("fused", "trained vectors (captions in)"), ("text", "serving vectors (title + body only)"), ("none", "no content vectors")):
reqs = {**period_requests(vm), **session_requests(vm)}; results[nm] = {k: run(v) for k, v in reqs.items()}; print(f"3) done: {nm}", flush=True)
# ---------------- 4) new users (serving vectors) ----------------
results["new user: interests only, no history"] = {k: run(v) for k, v in {**period_requests("text", history=False), **session_requests("text", history=False)}.items()}
results["new user: no interests, no history"] = {k: run(v) for k, v in {**period_requests("text", history=False, interests=False), **session_requests("text", history=False, interests=False)}.items()}
feeds = list(results["serving vectors (title + body only)"].keys())
def mean(nm, key="engine"): return round(float(np.mean([results[nm][f][key] for f in feeds if results[nm][f][key] is not None])), 3)
summary = {nm: mean(nm) for nm in results}; summary["random (expected)"] = mean("serving vectors (title + body only)", "random")
summary["v2b only (serving vectors)"] = mean("serving vectors (title + body only)", "v2b only"); summary["warm only (serving vectors)"] = mean("serving vectors (title + body only)", "warm only")
print("\nNDCG@10 mean over", len(feeds), "feed settings:"); [print(f" {k:45s} {v}") for k, v in summary.items()]
print("\nper setting (engine):"); [print(f" {f:34s} " + " | ".join(f"{nm[:18]:18s} {results[nm][f]['engine']:.3f}" for nm in list(results)[:5]) + f" | random {results['serving vectors (title + body only)'][f]['random']:.3f}") for f in feeds]
json.dump(dict(engine=meta["version"], label="engaged OR active_ms >= 10 s", parity=dict(features_max_abs=worst, rows=n_rows, scores=dict(cold=cd, warm=wd), ms_per_candidate=lat), summary=summary, per_setting=results),
open(os.path.join(ART, "evaluation_report.json"), "w"), indent=1, default=float)
json.dump(dict(feature_parity_max_abs=worst, rows_checked=n_rows, score_parity=dict(cold=cd, warm=wd), ms_per_candidate=lat), open(os.path.join(ART, "verification_report.json"), "w"), indent=1)
print("VERIFY V3 DONE")
|