File size: 5,668 Bytes
b319956 | 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 | # run_inference.py — drive every sample through a policy -> one JSONL line per sample.
# Works for any policy (SilentPolicy, APIPolicy, RLPolicy later). EVERY question_id must
# appear, even with an empty model_response_list. Robust for paid runs:
# - per-sample retry (transient Windows Errno 22 / API blips) with empty-record fallback,
# so ONE bad sample never aborts the whole run;
# - incremental flushed append, so a crash never loses (or re-pays for) finished samples;
# - --resume skips question_ids already present in the output.
import glob
import json
import os
import time
import argparse
import threading
from concurrent.futures import ThreadPoolExecutor
from humomni.core.streaming_driver import run_sample
def make_policy_factory(name, config):
"""Return a zero-arg factory that builds a FRESH policy per sample (no state leak)."""
if name == "silent":
from humomni.phase1.policies import SilentPolicy
return lambda: SilentPolicy()
if name == "api":
from humomni.phase1.policy_api import APIPolicy # imported lazily (needs deps + API key)
return lambda: APIPolicy(config=config)
raise ValueError(f"unknown policy {name!r}")
def _qid(d):
with open(os.path.join(d, "question.json"), encoding="utf-8") as f:
return json.load(f)["question_id"]
def _done_qids(out):
done = set()
if os.path.exists(out):
with open(out, encoding="utf-8") as f:
for ln in f:
ln = ln.strip()
if ln:
try:
done.add(json.loads(ln)["question_id"])
except Exception:
pass
return done
def main(data_dir, out, policy_factory, limit=0, workers=1, resume=False):
dirs = sorted(d for d in glob.glob(os.path.join(data_dir, "*")) if os.path.isdir(d))
if limit:
dirs = dirs[:limit] # used by experiment.py for fast partial Phase-2 evals
done = _done_qids(out) if resume else set()
if not resume:
open(out, "w", encoding="utf-8").close() # truncate (fresh run)
todo = [d for d in dirs if _qid(d) not in done]
print(f"{len(dirs)} samples | {len(done)} already done | {len(todo)} to run", flush=True)
lock = threading.Lock()
fails = []
wfile = open(out, "a", encoding="utf-8")
def process(d):
for attempt in range(3):
try:
return run_sample(d, policy_factory()) # fresh policy state per sample
except Exception as e:
if attempt == 2:
fails.append((_qid(d), str(e)[:120]))
return {"question_id": _qid(d), "model_response_list": []} # keep id present
time.sleep(0.5)
def handle(d):
rec = process(d)
with lock:
wfile.write(json.dumps(rec, ensure_ascii=False) + "\n")
wfile.flush()
return rec
try:
if workers > 1:
with ThreadPoolExecutor(max_workers=workers) as ex:
for n, _ in enumerate(ex.map(handle, todo), 1):
if n % 50 == 0:
print(f" ...{n}/{len(todo)}", flush=True)
else:
for n, d in enumerate(todo, 1):
handle(d)
if n % 50 == 0:
print(f" ...{n}/{len(todo)}", flush=True)
finally:
wfile.close()
print(f"wrote -> {out} (failed samples: {len(fails)})")
for q, e in fails[:10]:
print(" FAIL", q, e)
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--data_dir", default="data/phase1/data")
ap.add_argument("--out", default="submission.jsonl")
ap.add_argument("--policy", default="api", choices=["silent", "api"])
ap.add_argument("--config", default="config.json", help="orchestrator config (gate/theta/dedup)")
ap.add_argument("--limit", type=int, default=0, help="process only the first N samples (0=all; experiment.py uses it for fast Phase-2 evals)")
ap.add_argument("--workers", type=int, default=1, help="parallel samples (independent, still causal)")
ap.add_argument("--resume", action="store_true", help="skip question_ids already in --out")
a = ap.parse_args()
cfg = None
if a.policy == "api" and os.path.exists(a.config):
cfg = json.load(open(a.config, encoding="utf-8"))
if a.policy == "api" and a.workers > 1:
# Warm BOTH API clients + their lazy pydantic imports in the MAIN thread first;
# otherwise concurrent first-use across worker threads races and throws
# "No module named 'pydantic._migration'".
import humomni.phase1.vlm_client as vlm_client
try:
vlm_client._client_().chat.completions.create(
model=vlm_client.MODEL, max_tokens=1,
messages=[{"role": "user", "content": "ping"}])
except Exception:
pass
if (cfg or {}).get("dedup", {}).get("use_llm"):
import humomni.core.judges as judges
try:
judges.gemini("ping")
except Exception:
pass
try:
import humomni.core.emb_cache as emb_cache # warm MiniLM (cosine dedup prefilter) single-threaded
emb_cache.embed("ping")
except Exception:
pass
main(a.data_dir, a.out, make_policy_factory(a.policy, cfg),
limit=a.limit, workers=a.workers, resume=a.resume)
|