arborist/bench/cross_model_selfplay.py
russell@unturf.com 2b8c12303b
bench: add --teacher-model-judge flag — emit prompts for downstream correctness grading
STRICT audit ≠ factually correct. The verifier passes any answer whose
quotes match source text; a model can quote correctly and still draw a
wrong conclusion. Without a judge, the bench can't detect false-STRICTs.

This flag emits <ts>.judge_input.jsonl alongside the regular results —
one row per (question, model) with a model-agnostic teacher prompt.
Feed to any teacher (claude -p per row, remote API, Hermes self-judge)
to get CORRECT|WRONG|PARTIAL|UNCERTAIN verdicts. Skips rows that
errored or returned empty answers.

Grounded in observed reality from the 76-question 2010-wiki bench:
Hermes produced 2 confidently-wrong STRICTs (Q12 conflated Roman Empire
with Holy Roman Empire; Q48 answered the Niger River when asked about
the Nile). Qwen had 0 false STRICTs over the same fixture. Naming the
flag --teacher-model-judge (not --opus-judge) keeps the harness
provider-agnostic.
2026-05-31 11:40:33 -04:00

513 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Cross-model self-play bench — same question, multiple models, compare.
The pattern: the uncloseai-cli harness can call arborist twice per
question with different models to get two opinions. This script bakes
that pattern in as a benchmark. For each question in a fixture, run
arborist once per configured model, then tabulate:
* audit_mode per model (EVIDENCE-WARRANTED → ANCHOR-WARRANTED →
POINTER-LINKED → UNGROUNDED; HYBRID variants flatten to base)
* agreement on primary source URI
* grounding rate per model
* implied spend in fox's natural unit: $/1000 grounded answers
(Hermes $0.09, Qwen $0.16 by default — override via --baseline)
* latency per call
The COGS unit is **dollars per 1000 grounded answers**, an outcome
metric — failed/ungrounded calls are already amortized into the
baseline. A model's baseline × grounded_count / 1000 = implied spend
for those grounded answers.
Outputs:
bench/cross_model_results/<utc-iso>.jsonl one row per (question, model)
bench/cross_model_results/<utc-iso>.md markdown summary
Usage:
make bench-cross-model # smoke fixture, Hermes + Qwen
make bench-cross-model BENCH_CM_QUESTIONS=bench/qa_questions.txt
# custom baseline (dollars per 1000 grounded answers):
python3 bench/cross_model_selfplay.py --baseline 'hermes=0.09,qwen=0.16'
# custom model list (name=endpoint:model_id, comma-separated):
python3 bench/cross_model_selfplay.py \\
--models 'hermes=https://hermes.ai.unturf.com/v1:adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic,qwen=https://qwen.ai.unturf.com/v1:Qwen3.6-27B-UD-Q4_K_XL.gguf'
# emit teacher-judge prompts for downstream correctness grading:
python3 bench/cross_model_selfplay.py --teacher-model-judge
# → produces <ts>.judge_input.jsonl with one row per (question, model);
# feed to a teacher (claude -p, remote API, etc.) to detect false STRICTs
# — answers that audit STRICT but are factually wrong.
"""
from __future__ import annotations
import argparse
import datetime as _dt
import json
import os
import subprocess
import sys
import time
from collections import defaultdict
from pathlib import Path
REPO = Path(__file__).resolve().parent.parent
ARBORIST = REPO / ".venv" / "bin" / "arborist"
DEFAULT_MODELS = [
("hermes", "https://hermes.ai.unturf.com/v1",
"adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic"),
("qwen", "https://qwen.ai.unturf.com/v1",
"Qwen3.6-27B-UD-Q4_K_XL.gguf"),
]
# Per fox 2026-05-31 — baseline COGS in dollars per 1000 grounded
# answers. Hermes (8B) on 3090, Qwen (30B) on 4090. The baseline
# already amortizes ungrounded calls — it's an outcome metric, not a
# per-call price. Override via --baseline.
DEFAULT_DOLLARS_PER_1K_GROUNDED = {"hermes": 0.09, "qwen": 0.16}
# audit_mode → numeric rank for "stronger" comparison. Strips
# "-PARTIAL" suffix that HYBRID modes carry.
_AUDIT_RANK = {
"EVIDENCE-WARRANTED": 3,
"ANCHOR-WARRANTED": 2,
"POINTER-LINKED": 1,
"UNGROUNDED": 0,
"STRICT": 2, # legacy alias for ANCHOR-WARRANTED
"HYBRID": 1, # legacy alias for POINTER-LINKED
}
def _audit_rank(mode: str | None) -> int:
if not mode:
return 0
base = mode.replace("-PARTIAL", "")
return _AUDIT_RANK.get(base, 0)
def _is_grounded(mode: str | None) -> bool:
"""Anything not UNGROUNDED counts as grounded for cost accounting."""
if not mode:
return False
return not mode.startswith("UNGROUNDED")
def _read_questions(path: Path) -> list[str]:
out: list[str] = []
for line in path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
out.append(line)
return out
def _parse_models(spec: str) -> list[tuple[str, str, str]]:
"""Parse name=endpoint:model_id,... into a list of tuples."""
out = []
for chunk in spec.split(","):
chunk = chunk.strip()
if not chunk:
continue
if "=" not in chunk or ":" not in chunk.split("=", 1)[1]:
raise SystemExit(
f"--models entry must be name=endpoint:model_id, got: {chunk!r}"
)
name, rest = chunk.split("=", 1)
# split on the LAST colon — model_id may contain colons (rare)
# but endpoint always ends at the path boundary, so split on
# the colon BEFORE a non-slash. Simpler: assume model_id
# follows the first colon AFTER the endpoint scheme://host[:port]/path.
# We split on " :" sentinel — but easiest: rsplit once on `:`
# because vLLM model_ids don't contain colons in practice.
endpoint, model_id = rest.rsplit(":", 1)
out.append((name.strip(), endpoint.strip(), model_id.strip()))
return out
def _parse_baseline(spec: str | None) -> dict[str, float]:
"""Parse name=dollars_per_1k_grounded,... into a dict."""
if not spec:
return dict(DEFAULT_DOLLARS_PER_1K_GROUNDED)
out = dict(DEFAULT_DOLLARS_PER_1K_GROUNDED)
for chunk in spec.split(","):
if "=" not in chunk:
continue
name, dollars = chunk.split("=", 1)
try:
out[name.strip()] = float(dollars.strip())
except ValueError:
pass
return out
def _run_one(
question: str, *, shards_dir: Path | None, endpoint: str, model: str,
top_k: int, timeout_s: int, burn: bool,
) -> dict:
cmd = [str(ARBORIST)]
if shards_dir:
cmd += ["--shards-dir", str(shards_dir)]
cmd += [
"query", "--json",
"--top-k", str(top_k),
"--endpoint", endpoint,
"--model", model,
]
if burn:
cmd.append("--burn")
cmd.append(question)
t0 = time.time()
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout_s,
env={**os.environ, "ARBORIST_PROGRESS": "0"},
)
except subprocess.TimeoutExpired:
return {"_error": f"timeout after {timeout_s}s",
"_elapsed_s": float(timeout_s)}
elapsed = round(time.time() - t0, 2)
if proc.returncode not in (0, 1):
# Exit 1 is "ungrounded" — still has JSON. Other codes are real errors.
return {"_error": f"exit {proc.returncode}",
"_stderr": (proc.stderr or "")[:500],
"_elapsed_s": elapsed}
try:
data = json.loads(proc.stdout)
except json.JSONDecodeError as e:
return {"_error": f"parse: {e}",
"_stdout_head": proc.stdout[:200],
"_elapsed_s": elapsed}
data["_elapsed_s"] = elapsed
data["_exit_code"] = proc.returncode
return data
def _primary_uri(result: dict) -> str:
srcs = result.get("sources") or []
primary = next(
(s for s in srcs if s.get("source_role") == "primary_answer_source"),
srcs[0] if srcs else {},
)
return primary.get("document_uri") or ""
def _stronger_model(per_model: dict[str, dict]) -> str:
"""Return the name of the model with the higher audit_mode rank
(ties broken by alpha for stability). Empty string if all ungrounded."""
best = ""
best_rank = -1
for name, res in sorted(per_model.items()):
r = _audit_rank(res.get("audit_mode"))
if r > best_rank:
best, best_rank = name, r
return best if best_rank > 0 else ""
def _judge_prompt(question: str, model_name: str, audit_mode: str, answer: str) -> str:
"""Build the prompt a teacher model receives to grade one (question, answer)
pair. Asks for a single verdict token + a one-sentence rationale.
Designed to be model-agnostic — any teacher (Opus via `claude -p`, a remote
API, a Hermes self-judge run) can consume it. The teacher should reply with
the first non-empty line as: `CORRECT|WRONG|PARTIAL|UNCERTAIN` followed by
a colon and a one-sentence rationale.
"""
return (
"You are a hermetic judge grading the factual correctness of one answer "
"to one question. Use your own parametric knowledge as ground truth. "
"Ignore the answer's stylistic quality and whether it cites sources — "
"judge only whether the factual claim is true.\n\n"
f"Question: {question}\n\n"
f"Answer: {answer}\n\n"
"Reply on a single line in this exact format:\n"
" VERDICT: <one sentence rationale>\n\n"
"Where VERDICT is exactly one of:\n"
" CORRECT — the answer's main factual claim is true\n"
" WRONG — the answer's main factual claim is false\n"
" PARTIAL — main claim correct but missing required detail "
"(or correct on part, wrong on part)\n"
" UNCERTAIN — you don't know the domain well enough to judge\n"
)
def _emit_judge_inputs(
rows: list[dict], out_path: Path,
) -> int:
"""Write one JSONL row per (question, model) ready for a teacher to grade.
Skips rows that errored or returned no answer text. Returns count written."""
written = 0
with out_path.open("w") as f:
for r in rows:
res = r.get("result") or {}
if "_error" in res:
continue
answer = (res.get("answer_text") or "").strip()
if not answer:
continue
audit = res.get("audit_mode") or "UNKNOWN"
payload = {
"question": r["question"],
"model_name": r["model_name"],
"audit_mode": audit,
"answer_text": answer,
"judge_prompt": _judge_prompt(
r["question"], r["model_name"], audit, answer,
),
}
f.write(json.dumps(payload, ensure_ascii=False) + "\n")
written += 1
return written
def _summarize(
rows: list[dict], models: list[tuple[str, str, str]],
baseline_per_1k: dict[str, float],
) -> str:
"""Build the markdown summary. COGS unit: $ per 1k grounded answers."""
by_q: dict[str, dict[str, dict]] = defaultdict(dict)
for r in rows:
by_q[r["question"]][r["model_name"]] = r["result"]
n_q = len(by_q)
model_names = [m[0] for m in models]
out = []
out.append(f"# Cross-model self-play bench — {_dt.datetime.now(_dt.timezone.utc).isoformat()}")
out.append("")
out.append(f"Questions: {n_q}. Models: {', '.join(model_names)}.")
out.append("")
out.append("COGS unit: **dollars per 1000 grounded answers** (fox's natural unit). "
"Implied spend uses each model's baseline × grounded_count / 1000.")
out.append("")
# Per-model summary
out.append("## Per-model summary")
out.append("")
out.append("| model | grounded | rate | avg latency | baseline $/1k grounded | implied spend |")
out.append("|---|---:|---:|---:|---:|---:|")
for name in model_names:
grounded = sum(
1 for q in by_q if _is_grounded(by_q[q].get(name, {}).get("audit_mode"))
)
total = sum(1 for q in by_q if name in by_q[q] and "_error" not in by_q[q][name])
rate = (grounded / total) if total else 0.0
avg_lat = (
sum(by_q[q].get(name, {}).get("_elapsed_s", 0) for q in by_q) / max(total, 1)
)
base = baseline_per_1k.get(name, 0.0)
spend = grounded * base / 1000
out.append(
f"| {name} | {grounded}/{total} | {rate:.0%} "
f"| {avg_lat:.1f}s | ${base:.2f} | ${spend:.6f} |"
)
out.append("")
# Agreement
out.append("## Per-question results")
out.append("")
header = ["question"] + model_names + ["stronger", "primary agree"]
out.append("| " + " | ".join(header) + " |")
out.append("|" + "|".join(["---"] * len(header)) + "|")
for q in by_q:
cells = [q[:64] + ("" if len(q) > 64 else "")]
uris = set()
for name in model_names:
res = by_q[q].get(name, {})
audit = res.get("audit_mode") or ""
if "_error" in res:
audit = f"ERR ({res['_error'][:30]})"
cells.append(audit)
uri = _primary_uri(res)
if uri:
uris.add(uri)
cells.append(_stronger_model(by_q[q]) or "")
cells.append("" if len(uris) == 1 else ("" if len(uris) > 1 else ""))
out.append("| " + " | ".join(cells) + " |")
out.append("")
# Cost analysis
out.append("## Cost analysis (north star: $/1k grounded)")
out.append("")
total_spend = 0.0
total_grounded = 0
for name in model_names:
base = baseline_per_1k.get(name, 0.0)
n_g = sum(1 for q in by_q if _is_grounded(by_q[q].get(name, {}).get("audit_mode")))
spent = n_g * base / 1000
total_spend += spent
total_grounded += n_g
out.append(f"* **{name}** — {n_g} grounded × ${base:.2f}/1k = ${spent:.6f}")
out.append("")
out.append(f"**Total implied spend**: ${total_spend:.6f}. **Total grounded answers**: {total_grounded}.")
if total_grounded:
# Effective $/1k = total_spend × (1000 / total_grounded)
out.append(f"**Effective COGS**: ${total_spend * 1000 / total_grounded:.4f} per 1k grounded "
f"(weighted across models in this run).")
out.append("")
# Routing hint — cheap-first cascade
out.append("## Cheap-first cascade (try cheapest, escalate on UNGROUNDED)")
out.append("")
ordered = sorted(model_names, key=lambda n: baseline_per_1k.get(n, 0))
# Per-q: which model ultimately grounded? Count by source.
cascade_grounded_by_model: dict[str, int] = defaultdict(int)
cascade_total_grounded = 0
for q in by_q:
for name in ordered:
res = by_q[q].get(name, {})
if "_error" in res:
continue
if _is_grounded(res.get("audit_mode")):
cascade_grounded_by_model[name] += 1
cascade_total_grounded += 1
break
# else fall through to next model in cascade
cascade_spend = sum(
cnt * baseline_per_1k.get(name, 0) / 1000
for name, cnt in cascade_grounded_by_model.items()
)
if cascade_total_grounded:
breakdown = ", ".join(
f"{cnt} from {name}" for name, cnt in cascade_grounded_by_model.items()
)
cascade_cogs = cascade_spend * 1000 / cascade_total_grounded
out.append(
f"Cascade ({''.join(ordered)}): {cascade_total_grounded}/{n_q} grounded "
f"({breakdown}). Implied spend ${cascade_spend:.6f}"
f"**${cascade_cogs:.4f}/1k grounded**."
)
# Compare to always-most-expensive (defensive baseline)
most_expensive = ordered[-1]
max_base = baseline_per_1k.get(most_expensive, 0)
always_grounded = sum(
1 for q in by_q
if _is_grounded(by_q[q].get(most_expensive, {}).get("audit_mode"))
)
always_spend = always_grounded * max_base / 1000
if always_grounded:
always_cogs = always_spend * 1000 / always_grounded
delta_pct = (cascade_cogs / always_cogs - 1) * 100 if always_cogs else 0
sign = "+" if delta_pct >= 0 else ""
out.append("")
out.append(
f"vs always-{most_expensive}: {always_grounded}/{n_q} grounded, "
f"${always_spend:.6f} spend, ${always_cogs:.4f}/1k. "
f"Cascade delta: **{sign}{delta_pct:.1f}%**."
)
out.append("")
out.append("Caveat: cascade math credits each grounded answer to its source model's baseline. "
"Real-world failed-call overhead (an UNGROUNDED hermes call that escalates to qwen) "
"is not separately accounted — it's amortized into hermes's baseline by definition.")
out.append("")
return "\n".join(out)
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument(
"--questions", type=Path,
default=Path("bench/qa_questions_smoke.txt"),
)
ap.add_argument("--shards-dir", type=Path, default=None)
ap.add_argument(
"--models", default=None,
help="comma-separated name=endpoint:model_id; defaults to hermes + qwen",
)
ap.add_argument(
"--baseline", default=None,
help="comma-separated name=dollars per 1k grounded answers; "
"defaults to hermes=0.09,qwen=0.16 (fox's COGS unit)",
)
ap.add_argument("--top-k", type=int, default=8)
# 600s default — queue depth on shared endpoints (e.g. hermes
# under contention from another agent) can push a single call
# past 3 minutes even when the model is healthy. 180s was too
# aggressive against real-world infrastructure load.
ap.add_argument("--timeout", type=int, default=600)
ap.add_argument(
"--burn", action="store_true",
help="force fresh inference (skip cache) on every call",
)
ap.add_argument("--limit", type=int, default=0, help="cap questions (0 = no cap)")
ap.add_argument(
"--out-dir", type=Path, default=Path("bench/cross_model_results"),
)
ap.add_argument(
"--teacher-model-judge", action="store_true",
help="emit <ts>.judge_input.jsonl alongside results — one row per "
"(question, model) with a teacher-ready prompt for grading. "
"Feed to any teacher (claude -p, remote API, Hermes self-judge) "
"to get correctness verdicts. STRICT audit ≠ correct — this is "
"how you measure false-STRICTs.",
)
args = ap.parse_args()
questions = _read_questions(args.questions)
if args.limit:
questions = questions[: args.limit]
if not questions:
print(f"no questions in {args.questions}", file=sys.stderr)
return 1
if args.models:
models = _parse_models(args.models)
else:
models = list(DEFAULT_MODELS)
baseline = _parse_baseline(args.baseline)
if not ARBORIST.exists():
print(f"arborist binary not found: {ARBORIST}", file=sys.stderr)
return 2
args.out_dir.mkdir(parents=True, exist_ok=True)
ts = _dt.datetime.now(_dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
jsonl_path = args.out_dir / f"{ts}.jsonl"
md_path = args.out_dir / f"{ts}.md"
rows: list[dict] = []
print(f"# {len(questions)} questions × {len(models)} models = "
f"{len(questions) * len(models)} calls", file=sys.stderr)
with jsonl_path.open("w") as f:
for q in questions:
for name, endpoint, model_id in models:
print(f" [{name}] {q[:80]}", file=sys.stderr)
res = _run_one(
q, shards_dir=args.shards_dir,
endpoint=endpoint, model=model_id,
top_k=args.top_k, timeout_s=args.timeout, burn=args.burn,
)
row = {
"question": q,
"model_name": name,
"model_id": model_id,
"endpoint": endpoint,
"result": res,
}
rows.append(row)
f.write(json.dumps(row, ensure_ascii=False) + "\n")
f.flush()
audit = res.get("audit_mode") or res.get("_error") or "?"
elapsed = res.get("_elapsed_s", "?")
print(f"{audit} ({elapsed}s)", file=sys.stderr)
md_path.write_text(_summarize(rows, models, baseline))
print(f"\nJSONL: {jsonl_path}", file=sys.stderr)
print(f"Summary: {md_path}", file=sys.stderr)
if args.teacher_model_judge:
judge_input_path = args.out_dir / f"{ts}.judge_input.jsonl"
n = _emit_judge_inputs(rows, judge_input_path)
print(f"Teacher-judge inputs: {judge_input_path} ({n} rows)", file=sys.stderr)
print(f" Feed to your teacher of choice (e.g. `claude -p` per row) "
f"to grade. Verdicts are CORRECT|WRONG|PARTIAL|UNCERTAIN.",
file=sys.stderr)
print()
print(md_path.read_text())
return 0
if __name__ == "__main__":
sys.exit(main())