#!/usr/bin/env python # paper_experiments/run.py — reproduce every table in the report with one command each. # # python paper_experiments/run.py table1 # Table 1: main progression (score AND cost) # python paper_experiments/run.py table2 # Table 2: ensemble ablation (calls per frame) # python paper_experiments/run.py table3 # Table 3: dedup ablation # python paper_experiments/run.py gate # Table 4: perception-gate cost/score frontier # python paper_experiments/run.py sensitivity # FS example-sensitivity check (Sec. 4.x) # python paper_experiments/run.py all # # Self-contained and DETERMINISTIC: replays the shipped per-frame VLM decision caches # (caches/web_*.jsonl, holdout videos only) through the system's merge + entailment dedup, and scores # with the official-protocol harness using the shipped LLM-judge cache (caches/judge_cache.json) — # so a rerun makes ZERO API calls and reproduces the reported numbers exactly. The perception gate is # replayed from the shipped per-frame novelty cache (caches/novelty_holdout.json). Requires only # `pip install -e .`; the draft embeddings ship too (caches/emb_cache.*), so no model download — # the run is fully offline. import argparse import json import os import shutil import sys from concurrent.futures import ThreadPoolExecutor HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) os.chdir(ROOT) # humomni modules use repo-root-relative cache paths sys.path.insert(0, os.path.join(ROOT, "src")) # The scoring harness reads its judge cache from cache/judge_cache.json (repo-root relative). # Ship-in the reproduction copy if the local one is absent so reruns are exact and free. os.makedirs("cache", exist_ok=True) if not os.path.exists("cache/judge_cache.json"): shutil.copyfile(os.path.join(HERE, "caches", "judge_cache.json"), "cache/judge_cache.json") import humomni.core.emb_cache as emb_cache from humomni.phase1.dedup import EntailmentGuard from humomni.tuning.cache import VLMCache from humomni.tuning.faithful_eval import faithful_score THETA = 0.6 # NEW_ANSWER confidence threshold (config.json "theta") GOLD = json.load(open(os.path.join(HERE, "data", "gold_holdout.json"), encoding="utf-8")) CACHES = {n: VLMCache(os.path.join(HERE, "caches", f"web_{n}.jsonl")) for n in ("base", "comp", "exh", "fs", "fs_alt")} NOVELTY = json.load(open(os.path.join(HERE, "caches", "novelty_holdout.json"), encoding="utf-8")) _TIMES = {} # qid -> ascending frame timestamps (from the cache itself) for c in CACHES.values(): for (qid, t) in c.mem: _TIMES.setdefault(qid, set()).add(t) _TIMES = {q: sorted(ts) for q, ts in _TIMES.items()} emb_cache.load_disk() emb_cache.load_disk(os.path.join(HERE, "caches", "emb_cache")) # shipped draft embeddings — # covers every draft below, so the MiniLM model (a ~90 MB first-run download) is never needed # for the replay; embed_many only encodes cache MISSES. emb_cache.embed_many([d["draft"].strip() for c in CACHES.values() for d in c.mem.values() if d.get("draft", "").strip()]) def replay(names, k, use_llm=True, sim_high=0.97, sim_low=0.55, gate=None): """Replay the cached decisions of ensemble `names` through the merge + dedup (identical to the live policy: prompts queried in order per frame, said=[]). `gate` = None (call the VLM on every frame) or a dict {novelty, max_silence_s, warmup_s}: the CPU perception gate is replayed from the per-frame novelty cache, and the VLM is consulted ONLY on frames where the gate fires. Returns (predictions, vlm_calls_per_video).""" cs = [CACHES[n] for n in names] def one(item): qid = item["question_id"] g = EntailmentGuard(sim_high=sim_high, sim_low=sim_low, recent_k=k, use_llm=use_llm) said, res, calls, last_fire = [], [], 0, -1e9 for t in _TIMES[qid]: if gate is not None: # Bit-identical replay of phase1.perception_gate.PerceptionGate.should_decide: the # cached novelty (tuning.cache.build_novelty, keys f"{t:.1f}") uses the SAME grayscale # 64x36 mean-abs-diff feature the live gate computes, and the live policy runs this # check BEFORE the VLM request (policy_api.step) — skip means no API call. nov = NOVELTY[qid][f"{t:.1f}"] fire = (t <= gate["warmup_s"]) or (nov >= gate["novelty"]) \ or (t - last_fire >= gate["max_silence_s"]) if not fire: continue last_fire = t calls += len(cs) for c in cs: d = c.get(qid, t) if d and d["decision"] == "NEW_ANSWER" and d["confidence"] >= THETA: dr = d["draft"].strip() if dr and not g.is_duplicate(dr, said): said.append(dr) res.append({"time": round(t, 3), "content": dr}) return qid, (res, calls) with ThreadPoolExecutor(max_workers=3) as ex: out = dict(ex.map(one, GOLD)) preds = {q: v[0] for q, v in out.items()} calls = sum(v[1] for v in out.values()) / len(out) return preds, calls def row(label, **kw): pred, calls = replay(**kw) r = faithful_score(GOLD, pred, workers=3) emit = sum(len(v) for v in pred.values()) / len(pred) print(f" {label:46} score={r['score']:.4f} PAUC={r['PAUC']:.4f} dup={r['duplicate']:.4f} " f"emits/video={emit:4.1f} VLM-calls/video={calls:5.1f}", flush=True) def table1(): print("Table 1 — main progression: score AND cost (holdout, official protocol)", flush=True) row("BASE (focused prompt), k=16", names=["base"], k=16) row("COMP (comprehensive), k=16", names=["comp"], k=16) row("EXH (exhaustive+OCR), k=16", names=["exh"], k=16) row("EXH+COMP ensemble, k=32", names=["exh", "comp"], k=32) row("EXH+COMP+FS ensemble, k=40 [max-score]", names=["exh", "comp", "fs"], k=40) def table2(): print("Table 2 — ensemble ablation (calls per frame)", flush=True) row("1x EXH alone, k=16", names=["exh"], k=16) row("1x FS alone (EXH rules + few-shot), k=16", names=["fs"], k=16) row("2x EXH+COMP, k=32", names=["exh", "comp"], k=32) row("3x EXH+COMP+FS, k=40", names=["exh", "comp", "fs"], k=40) def table3(): print("Table 3 — dedup ablation (on EXH+COMP, k=32)", flush=True) row("two-stage entailment (ours)", names=["exh", "comp"], k=32, use_llm=True) row("cosine-only", names=["exh", "comp"], k=32, use_llm=False) row("none (emit everything)", names=["exh", "comp"], k=32, use_llm=False, sim_high=2.0) # Gate operating points: "deployed" = the configuration our initial system shipped with. GATES = [("gate off (max coverage)", None), ("gate: nov>=8, silence 1.5s (deployed)", {"novelty": 8, "max_silence_s": 1.5, "warmup_s": 1.0}), ("gate: nov>=16, silence 3s (aggressive)", {"novelty": 16, "max_silence_s": 3.0, "warmup_s": 1.0})] def gate(): print("Table 4 — perception-gate cost/score frontier", flush=True) print(" on EXH alone (k=16):", flush=True) for lab, gcfg in GATES: row(" " + lab, names=["exh"], k=16, gate=gcfg) print(" on the full ensemble EXH+COMP+FS (k=40):", flush=True) for lab, gcfg in GATES: row(" " + lab, names=["exh", "comp", "fs"], k=40, gate=gcfg) def sensitivity(): print("FS example-sensitivity (same shape, different examples)", flush=True) row("EXH+COMP+FS (paper examples), k=40", names=["exh", "comp", "fs"], k=40) row("EXH+COMP+FS-alt (swapped examples), k=40", names=["exh", "comp", "fs_alt"], k=40) if __name__ == "__main__": ap = argparse.ArgumentParser(description="Reproduce the report's experiment tables.") ap.add_argument("what", choices=["table1", "table2", "table3", "gate", "sensitivity", "all"]) a = ap.parse_args() todo = [table1, table2, table3, gate, sensitivity] if a.what == "all" else [globals()[a.what]] for fn in todo: fn() print(flush=True) emb_cache.save_disk()