russell@unturf.com
8fec3a5d56
Empirically motivated by the 2026-05-02 emergent log:
Q: 'How can a swallowtail butterfly, gracefully fluttering amidst
the rockiest terrain, remain undeterred by the upbraiding
winds...'
A: 'The Macleay's Swallowtail butterfly is found in Eastern
Australia including the ACT, New South Wales, Queensland...'
The model traded the metaphor for literal Macleay's-Swallowtail
taxonomic facts. Warrant passed (the literal anchor IS in cited
spans), DEFLECTION_DETECTED didn't fire (the last content token
'flight' did echo somewhere), the bench landed HYBRID 3/3 — but
the user's metaphorical question was never engaged.
Honest gap: catching this structurally requires NLI-grade
semantics, which is the verifier-semantic-gap design proposal.
Until that lands, ship a SMELL SIDECAR — purely lexical, sidecar
only, never enters the binary verifier output.
Detection rule:
1. Extract metaphor cues from the question:
- -ly adverbs (gracefully, defiantly), excluding common
-ly nouns (butterfly, italy, july) via blocklist
- -ing present participles >=6 chars (upbraiding,
fluttering, brooding), excluding common verb -ing forms
- -est superlatives >=6 chars (rockiest, harshest)
- prepositional cues (amidst, despite, against, beneath)
2. Count overlap with answer's content tokens.
3. Fire metaphor_deflection when:
cue_count >= 3 AND answer_overlap_count == 0
The threshold is conservative; the smell only triggers on
STRONGLY poetic questions with PURELY literal answers.
Wire-up:
- aborist/qa/inspect.py:diagnose_metaphor_deflection
- bench/qa_sweep.py: rows gain metaphor_deflection_kind +
metaphor_cue_count + metaphor_overlap_count
- scripts/bench_emergent.py: same fields on emergent log rows
5 new tests in tests/test_inspect.py:
- swallowtail canary case fires metaphor_deflection
- literal questions (mona lisa) return no_signal
- questions whose answer engages cues return no_signal
- common -ly nouns (butterfly, italy, july, family) filtered
- sub-threshold cue counts return no_signal
756/34 tests pass (5 new + 751 prior).
672 lines
28 KiB
Python
672 lines
28 KiB
Python
"""QA-quality benchmark sweep.
|
||
|
||
Runs a fixed question set through ``aborist.qa.query.query`` under each
|
||
answer mode and tabulates STRICT / HYBRID / UNGROUNDED counts, the
|
||
``n_verified / n_quotes`` ratio, latency, and lazy-anchor signals.
|
||
|
||
Why this exists: G0 / claim-lattice-pointer mode landed in 2337b77, and
|
||
four hardenings stacked on top (smell sidecar, positive-form prompts,
|
||
source-role boost, title-purity rerank, CITATION_MISMATCH check). Every
|
||
new heuristic flies blind until we can show "this commit moved STRICT
|
||
from N to M". This is that scaffolding.
|
||
|
||
Outputs:
|
||
|
||
bench/qa_results/<utc-iso>.jsonl one row per (mode, question)
|
||
bench/qa_results/<utc-iso>.md markdown summary table
|
||
|
||
The JSONL is the durable artifact. The markdown is the thing fox reads.
|
||
|
||
Wire via ``make bench-qa``. Honors ``BURN=1`` to force fresh inference
|
||
under the current ``governance_policy_hash`` — useful after a prompt
|
||
change that doesn't bump any version field.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as _dt
|
||
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
from collections import defaultdict
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from pathlib import Path
|
||
from threading import Lock
|
||
|
||
# Defer aborist imports until argparse runs so `--help` works without
|
||
# the package installed.
|
||
|
||
ANSWER_MODES = ("quote", "claim_lattice_pointer", "claim_lattice")
|
||
|
||
|
||
def _read_questions(path: Path) -> list[str]:
|
||
out: list[str] = []
|
||
for raw in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
out.append(line)
|
||
return out
|
||
|
||
|
||
def _utc_iso_compact() -> str:
|
||
# YYYY-MM-DDTHH-MM-SSZ — colon-free so it's filesystem-safe.
|
||
return _dt.datetime.now(tz=_dt.timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ")
|
||
|
||
|
||
def _ratio(verdict: dict) -> float:
|
||
nq = verdict.get("n_quotes") or 0
|
||
nv = verdict.get("n_verified") or 0
|
||
return (nv / nq) if nq else 0.0
|
||
|
||
|
||
def _run_one(
|
||
*,
|
||
question: str,
|
||
answer_mode: str,
|
||
shards_dir: Path | None,
|
||
qa_db: Path | None,
|
||
top_k: int,
|
||
burn: bool,
|
||
endpoint: str,
|
||
model: str,
|
||
) -> dict:
|
||
"""Run one (question, mode) — returns a flat record for the JSONL."""
|
||
from aborist.qa.client import OpenAICompatibleClient
|
||
from aborist.qa.query import DEFAULT_QUERY_POLICY, query
|
||
|
||
policy = dict(DEFAULT_QUERY_POLICY)
|
||
policy["answer_mode"] = answer_mode
|
||
|
||
api_key = os.environ.get("ABORIST_LLM_API_KEY")
|
||
client = OpenAICompatibleClient(base_url=endpoint, api_key=api_key)
|
||
|
||
if qa_db is None:
|
||
qa_db = (shards_dir / "qa.db") if shards_dir else (Path.home() / ".aborist" / "qa.db")
|
||
|
||
t0 = time.monotonic()
|
||
err: str | None = None
|
||
result: dict = {}
|
||
try:
|
||
result = query(
|
||
question=question,
|
||
qa_db=qa_db,
|
||
chat_client=client,
|
||
model_id=model,
|
||
shards_dir=shards_dir,
|
||
top_k=top_k,
|
||
policy=policy,
|
||
burn_existing=burn,
|
||
)
|
||
except Exception as e: # noqa: BLE001 — bench captures all
|
||
err = f"{type(e).__name__}: {e}"
|
||
elapsed_s = round(time.monotonic() - t0, 2)
|
||
|
||
# Deflection sidecar — sidecar-only signal, never feeds the
|
||
# verifier. Surfaces the Mars-BDFL pattern (STRICT but answer
|
||
# never mentions the question's subject) at bench-aggregate
|
||
# scale so a creeping "model deflects rather than refuses"
|
||
# regression is legible across runs.
|
||
from aborist.qa.inspect import diagnose_deflection, diagnose_metaphor_deflection
|
||
deflection = diagnose_deflection(question, result.get("answer_text") or "")
|
||
metaphor = diagnose_metaphor_deflection(
|
||
question, result.get("answer_text") or ""
|
||
)
|
||
|
||
audit_mode = result.get("audit_mode")
|
||
return {
|
||
"question": question,
|
||
"answer_mode": answer_mode,
|
||
"status": result.get("status") or ("error" if err else "missing"),
|
||
"audit_mode": audit_mode,
|
||
"n_quotes": result.get("n_quotes"),
|
||
"n_verified": result.get("n_verified"),
|
||
"ratio": round(_ratio(result), 3),
|
||
"verifier_method": result.get("verifier_method"),
|
||
"lookup_path": result.get("lookup_path"),
|
||
"failure_stage": result.get("failure_stage"),
|
||
"lazy_anchor_ratio": result.get("lazy_anchor_ratio"),
|
||
"pointer_id_distribution": result.get("pointer_id_distribution"),
|
||
"cache_key": (result.get("cache_key") or "")[:12],
|
||
"n_sources": len(result.get("sources") or []),
|
||
"elapsed_s": elapsed_s,
|
||
"deflection_kind": deflection["kind"],
|
||
"subject_anchor": deflection["subject_anchor"],
|
||
"subject_in_answer": deflection["subject_in_answer"],
|
||
"metaphor_deflection_kind": metaphor["kind"],
|
||
"metaphor_cue_count": metaphor["cue_count"],
|
||
"metaphor_overlap_count": metaphor["answer_overlap_count"],
|
||
# Capacity metrics — char-level proxy for prompt-token budget.
|
||
# Surfaces "did STRICT come from a tight 5KB prompt or a 50KB
|
||
# context-stuffed one?" at aggregate scale. Lets the bench
|
||
# bucket strict-rate by input-size band.
|
||
"prompt_chars_total": (result.get("prompt_chars") or {}).get("messages_total", 0),
|
||
"prompt_chars_evidence": (result.get("prompt_chars") or {}).get("evidence_or_context", 0),
|
||
"prompt_chars_system": (result.get("prompt_chars") or {}).get("system_prompt", 0),
|
||
"prompt_chars_question": (result.get("prompt_chars") or {}).get("user_question", 0),
|
||
"answer_chars": result.get("answer_chars", 0),
|
||
# Seven-point program directive compliance — see
|
||
# docs/seven-point-program.md. Per-row booleans where the
|
||
# directive is observable from the result; aggregate
|
||
# coverage shows up in the markdown summary's
|
||
# "directive coverage" section. Directives that are global
|
||
# properties of the substrate (D1, D5, D8) don't appear
|
||
# per-row.
|
||
"directive_compliance": _directive_compliance(answer_mode, result, err),
|
||
"error": err,
|
||
}
|
||
|
||
|
||
def _directive_compliance(
|
||
answer_mode: str, result: dict, err: str | None
|
||
) -> dict:
|
||
"""Compute per-row pass/fail for the seven-point directives whose
|
||
pinning is observable from a single bench row.
|
||
|
||
Returns a dict mapping directive id -> bool (True = pass, False =
|
||
fail/pending). Some directives are system-global (D1 verifier
|
||
never calls LLM, D5 verifier_method enum, D8 discipline) and don't
|
||
appear per-row — track those once in the summary header.
|
||
"""
|
||
if err or result.get("status") == "error":
|
||
return {}
|
||
audit = result.get("audit_mode")
|
||
method = result.get("verifier_method") or ""
|
||
is_lattice = answer_mode in ("claim_lattice_pointer", "claim_lattice")
|
||
return {
|
||
# D2: lattice modes emit pointer clauses (claim_lattice_pointer
|
||
# or claim_lattice JSON variant). Quote-mode rows count as
|
||
# n/a for D2 — they predate the directive.
|
||
"D2_pointer_clauses": is_lattice,
|
||
# D3: build CTI internally — proxy is "phrase route had a
|
||
# chance to fire" (lattice mode + run_dag_root populated).
|
||
# Module L (the answer-side multi-frame compilation) lands
|
||
# via ticket #000002; until then this is a structural
|
||
# readiness check, not full coverage.
|
||
"D3_cti_substrate_ready": (
|
||
is_lattice and bool(result.get("run_dag_root"))
|
||
),
|
||
# D4: evidence_map_root + run_dag_root present in the
|
||
# 9-stage CTI run-DAG. Retrieval-plan-hash binding pending
|
||
# via ticket #000001.
|
||
"D4_evidence_map_bound": bool(result.get("run_dag_root")),
|
||
# D6: warrant ran (relation/date anchor classes — today).
|
||
# Generalization to entity-list / count / why-cause shapes
|
||
# pending via ticket #000003. Mark True only when the
|
||
# verifier verdict shows the warrant was checked (today,
|
||
# any lattice-mode row gets the warrant pass; future
|
||
# per-shape gating sharpens this).
|
||
"D6_warrant_fired": is_lattice,
|
||
# D7: schema audit_mode is in the canonical enum. Renderer
|
||
# then maps lattice-mode STRICT/HYBRID → EVIDENCE-LINKED at
|
||
# display time (pinned by tests/test_cli_render.py). Per-row
|
||
# signal here just confirms the row's audit token is valid —
|
||
# the renderer side is a deterministic transformation tested
|
||
# separately, not something each bench row can re-verify.
|
||
"D7_honest_label": audit in ("STRICT", "HYBRID", "UNGROUNDED"),
|
||
}
|
||
|
||
|
||
def _summarize(rows: list[dict]) -> dict:
|
||
"""Group rows by mode → STRICT/HYBRID/UNGROUNDED totals + means."""
|
||
by_mode: dict[str, dict] = {}
|
||
for r in rows:
|
||
m = r["answer_mode"]
|
||
b = by_mode.setdefault(m, {
|
||
"n": 0,
|
||
"STRICT": 0,
|
||
"HYBRID": 0,
|
||
"UNGROUNDED": 0,
|
||
"errors": 0,
|
||
"ratio_sum": 0.0,
|
||
"latency_sum": 0.0,
|
||
"deflections": 0,
|
||
# Per-directive pass counts (seven-point program). Init
|
||
# all known directive ids so absent rows report 0/N
|
||
# rather than missing-key.
|
||
"directive_pass": {
|
||
"D2_pointer_clauses": 0,
|
||
"D3_cti_substrate_ready": 0,
|
||
"D4_evidence_map_bound": 0,
|
||
"D6_warrant_fired": 0,
|
||
"D7_honest_label": 0,
|
||
},
|
||
})
|
||
b["n"] += 1
|
||
if r["error"] or r["status"] == "error":
|
||
b["errors"] += 1
|
||
elif r["audit_mode"] in ("STRICT", "HYBRID", "UNGROUNDED"):
|
||
b[r["audit_mode"]] += 1
|
||
b["ratio_sum"] += r["ratio"] or 0.0
|
||
b["latency_sum"] += r["elapsed_s"] or 0.0
|
||
# Deflection rate: STRICT/HYBRID with subject-anchor missing
|
||
# from answer. UNGROUNDED + deflection isn't interesting (no
|
||
# answer to deflect with), and no_question_tokens is vacuous.
|
||
if r.get("deflection_kind") == "deflection" and r["audit_mode"] in (
|
||
"STRICT", "HYBRID"
|
||
):
|
||
b["deflections"] += 1
|
||
# Directive compliance — sum the per-row booleans into
|
||
# per-mode pass counts.
|
||
for did, ok in (r.get("directive_compliance") or {}).items():
|
||
if did in b["directive_pass"] and ok:
|
||
b["directive_pass"][did] += 1
|
||
return by_mode
|
||
|
||
|
||
def _per_question_summary(rows: list[dict]) -> dict:
|
||
"""Group rows by (question, mode) → per-cell vote counts + medians."""
|
||
cells: dict[tuple[str, str], dict] = {}
|
||
for r in rows:
|
||
key = (r["question"], r["answer_mode"])
|
||
c = cells.setdefault(key, {
|
||
"STRICT": 0, "HYBRID": 0, "UNGROUNDED": 0, "errors": 0,
|
||
"ratios": [], "latencies": [],
|
||
})
|
||
if r["error"] or r["status"] == "error":
|
||
c["errors"] += 1
|
||
elif r["audit_mode"] in ("STRICT", "HYBRID", "UNGROUNDED"):
|
||
c[r["audit_mode"]] += 1
|
||
c["ratios"].append(r["ratio"] or 0.0)
|
||
c["latencies"].append(r["elapsed_s"] or 0.0)
|
||
return cells
|
||
|
||
|
||
def _median(xs: list[float]) -> float:
|
||
if not xs:
|
||
return 0.0
|
||
s = sorted(xs)
|
||
mid = len(s) // 2
|
||
return s[mid] if len(s) % 2 else (s[mid - 1] + s[mid]) / 2
|
||
|
||
|
||
def _render_markdown(
|
||
rows: list[dict],
|
||
summary: dict,
|
||
started_utc: str,
|
||
modes: list[str],
|
||
questions: list[str],
|
||
n_samples: int,
|
||
) -> str:
|
||
lines: list[str] = []
|
||
n_runs = len(rows)
|
||
lines.append(f"# aborist QA-quality benchmark — {started_utc}")
|
||
lines.append("")
|
||
lines.append(
|
||
f"questions: {len(questions)} · modes: {len(modes)} · samples per cell: "
|
||
f"{n_samples} · runs: {n_runs}"
|
||
)
|
||
lines.append("")
|
||
lines.append("## summary")
|
||
lines.append("")
|
||
lines.append("| mode | runs | STRICT | HYBRID | UNGROUNDED | err | strict-rate | mean ratio | mean latency | deflections |")
|
||
lines.append("|------|------|--------|--------|------------|-----|-------------|-----------|--------------|-------------|")
|
||
for mode in modes:
|
||
b = summary.get(mode)
|
||
if not b:
|
||
continue
|
||
n = b["n"] or 1
|
||
mean_ratio = b["ratio_sum"] / n
|
||
mean_lat = b["latency_sum"] / n
|
||
strict_rate = b["STRICT"] / n
|
||
deflections = b.get("deflections", 0)
|
||
lines.append(
|
||
f"| {mode} | {b['n']} | {b['STRICT']} | {b['HYBRID']} | "
|
||
f"{b['UNGROUNDED']} | {b['errors']} | "
|
||
f"{strict_rate:.2f} | {mean_ratio:.3f} | {mean_lat:.1f}s | "
|
||
f"{deflections}/{b['n']} |"
|
||
)
|
||
lines.append("")
|
||
lines.append("## directive coverage (seven-point program)")
|
||
lines.append("")
|
||
lines.append(
|
||
"Per-mode pass-rate for the directives whose pinning is observable "
|
||
"from a single bench row. See `docs/seven-point-program.md` for "
|
||
"the full directive list. D1 (no LLM in verifier), D5 (verifier_method "
|
||
"enum), and D8 (test-pinning discipline) are global properties of "
|
||
"the substrate and don't appear per-row. ½ in the doc means some "
|
||
"scope is implemented; bench numbers reflect the implemented portion."
|
||
)
|
||
lines.append("")
|
||
lines.append("| mode | D2 pointer | D3 cti-ready | D4 ev-map bound | D6 warrant | D7 honest label |")
|
||
lines.append("|------|-----------|--------------|-----------------|-----------|----------------|")
|
||
for mode in modes:
|
||
b = summary.get(mode)
|
||
if not b:
|
||
continue
|
||
n = b["n"] or 1
|
||
dp = b["directive_pass"]
|
||
cells = []
|
||
for did in (
|
||
"D2_pointer_clauses",
|
||
"D3_cti_substrate_ready",
|
||
"D4_evidence_map_bound",
|
||
"D6_warrant_fired",
|
||
"D7_honest_label",
|
||
):
|
||
count = dp.get(did, 0)
|
||
cells.append(f"{count}/{n} ({count / n:.0%})")
|
||
lines.append(f"| {mode} | " + " | ".join(cells) + " |")
|
||
lines.append("")
|
||
lines.append("## strict-rate by prompt size")
|
||
lines.append("")
|
||
lines.append(
|
||
"Buckets capacity in `messages_total` chars (system + reminder + "
|
||
"evidence/context + question). ~4 chars/token English prose; "
|
||
"~2.5 chars/token JSON-evidence-heavy. Tells you whether the "
|
||
"model strict-rate degrades with input size. Buckets log-scale "
|
||
"from 8 KB to >= 1 MB so the same harness covers Hermes-3-8B "
|
||
"(82K context) through 1M-context models without code change."
|
||
)
|
||
lines.append("")
|
||
# Log-scale buckets — covers the entire model spectrum from
|
||
# 8B-class models (Hermes 82K context, max useful prompt ~32-64KB)
|
||
# through giant-context models (Gemini 1.5 / Claude 1M, useful
|
||
# prompt potentially 100KB-800KB). Models simply don't populate
|
||
# buckets beyond their context window; the substrate stays
|
||
# universal. Adding new boundaries doesn't require a code change
|
||
# for existing data — empty buckets get skipped at render.
|
||
buckets = [
|
||
("<8KB", 0, 8000),
|
||
("8-16KB", 8000, 16000),
|
||
("16-32KB", 16000, 32000),
|
||
("32-64KB", 32000, 64000),
|
||
("64-128KB", 64000, 128000),
|
||
("128-256KB", 128000, 256000),
|
||
("256-512KB", 256000, 512000),
|
||
("512K-1M", 512000, 1048576),
|
||
(">=1M", 1048576, 10**12),
|
||
]
|
||
lines.append("| mode | bucket | runs | STRICT | strict-rate | mean evidence chars | mean answer chars |")
|
||
lines.append("|------|--------|------|--------|-------------|---------------------|-------------------|")
|
||
# Track per-mode (label, strict-rate, n) tuples for the
|
||
# "recommended context budget" recommendation below.
|
||
per_mode_buckets: dict[str, list[tuple[str, float, int]]] = {}
|
||
for mode in modes:
|
||
per_mode_buckets[mode] = []
|
||
for label, lo, hi in buckets:
|
||
sub = [
|
||
r for r in rows
|
||
if r["answer_mode"] == mode
|
||
and lo <= (r.get("prompt_chars_total") or 0) < hi
|
||
and not r["error"] and r["status"] != "error"
|
||
]
|
||
if not sub:
|
||
continue
|
||
n_b = len(sub)
|
||
n_strict = sum(1 for r in sub if r["audit_mode"] == "STRICT")
|
||
mean_ev = sum((r.get("prompt_chars_evidence") or 0) for r in sub) / n_b
|
||
mean_ans = sum((r.get("answer_chars") or 0) for r in sub) / n_b
|
||
strict_rate = n_strict / n_b
|
||
per_mode_buckets[mode].append((label, strict_rate, n_b))
|
||
lines.append(
|
||
f"| {mode} | {label} | {n_b} | {n_strict} | "
|
||
f"{strict_rate:.2f} | {int(mean_ev)} | {int(mean_ans)} |"
|
||
)
|
||
lines.append("")
|
||
lines.append("## recommended context budget (learned from this bench)")
|
||
lines.append("")
|
||
lines.append(
|
||
"Per-mode peak strict-rate bucket from the table above. Bench "
|
||
"observations feed back into a recommended `max_context_chars` "
|
||
"default. Operator-driven landing per the five-step algorithm "
|
||
"step 5 — surfaced here, not auto-applied. Minimum bucket size "
|
||
"of 5 runs to be considered (smaller samples are noise). The "
|
||
"log-scale buckets cover everything from 8B-class models "
|
||
"through 1M-context models; a model whose context window stops "
|
||
"at 82K simply never populates the giant buckets."
|
||
)
|
||
lines.append("")
|
||
lines.append("| mode | peak bucket | strict-rate | n |")
|
||
lines.append("|------|-------------|-------------|---|")
|
||
for mode in modes:
|
||
candidates = [t for t in per_mode_buckets.get(mode, []) if t[2] >= 5]
|
||
if not candidates:
|
||
lines.append(
|
||
f"| {mode} | (insufficient samples; ≥5 runs/bucket "
|
||
f"required) | n/a | n/a |"
|
||
)
|
||
continue
|
||
# Peak strict-rate; tie-break on smaller bucket (less context
|
||
# is cheaper at equivalent strict-rate).
|
||
peak_label, peak_rate, peak_n = max(
|
||
candidates, key=lambda t: (t[1], -buckets.index(
|
||
next(b for b in buckets if b[0] == t[0])
|
||
))
|
||
)
|
||
lines.append(
|
||
f"| {mode} | {peak_label} | {peak_rate:.2f} | {peak_n} |"
|
||
)
|
||
lines.append("")
|
||
lines.append("## per question")
|
||
lines.append("")
|
||
lines.append(
|
||
"Cell shows verdict counts across N samples (S=STRICT, H=HYBRID, U=UNGROUNDED, e=err). "
|
||
"median ratio = median n_verified/n_quotes."
|
||
)
|
||
lines.append("")
|
||
cells = _per_question_summary(rows)
|
||
lines.append("| question | mode | verdicts (N=" + str(n_samples) + ") | median ratio | median latency |")
|
||
lines.append("|----------|------|---------------------|--------------|----------------|")
|
||
for q in questions:
|
||
for mode in modes:
|
||
c = cells.get((q, mode))
|
||
if not c:
|
||
continue
|
||
verdicts = (
|
||
f"S:{c['STRICT']} H:{c['HYBRID']} U:{c['UNGROUNDED']}"
|
||
+ (f" e:{c['errors']}" if c["errors"] else "")
|
||
)
|
||
mr = _median(c["ratios"])
|
||
ml = _median(c["latencies"])
|
||
qd = q if len(q) <= 50 else q[:47] + "..."
|
||
lines.append(f"| {qd} | {mode} | {verdicts} | {mr:.3f} | {ml:.1f}s |")
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
ap.add_argument("--questions", type=Path, default=Path("bench/qa_questions.txt"))
|
||
ap.add_argument("--shards-dir", type=Path, default=Path.home() / ".aborist" / "shards")
|
||
ap.add_argument("--qa-db", type=Path, default=None)
|
||
ap.add_argument("--out-dir", type=Path, default=Path("bench/qa_results"))
|
||
ap.add_argument("--top-k", type=int, default=8)
|
||
ap.add_argument("--modes", default=",".join(ANSWER_MODES),
|
||
help="comma-separated subset of ANSWER_MODES to sweep")
|
||
ap.add_argument("--n", dest="n_samples", type=int, default=3,
|
||
help="samples per (question, mode); each sample burns the cached "
|
||
"record so Hermes nondeterminism becomes the variance source")
|
||
ap.add_argument("--limit", type=int, default=0,
|
||
help="truncate question list to N; 0 = all")
|
||
ap.add_argument("--concurrency", type=int, default=1,
|
||
help="parallel sample-level tasks. Each (question, "
|
||
"mode, sample_idx) is one task; tasks are shuffled "
|
||
"via --seed and dispatched concurrently. Per-cell "
|
||
"Lock serializes burn+insert on the shared "
|
||
"cache_key so two samples of the same cell never "
|
||
"race. vLLM handles concurrent requests well; "
|
||
"4-8 is a safe starting point.")
|
||
ap.add_argument("--seed", type=int, default=0,
|
||
help="deterministic shuffle seed for sample-level "
|
||
"task ordering. Same seed = same task order = "
|
||
"reproducible bench. Random ordering is the "
|
||
"design — uncorrelated samples give true i.i.d. "
|
||
"variance estimation and feed vLLM's continuous "
|
||
"batcher a diverse request stream.")
|
||
ap.add_argument("--resume", type=Path, default=None,
|
||
help="resume an interrupted bench by appending to "
|
||
"an existing JSONL. Already-completed (question, "
|
||
"mode, sample_idx) tasks are skipped; remaining "
|
||
"tasks run in the original shuffled order (use "
|
||
"the same --seed). The same .md path is "
|
||
"re-rendered from the union of pre-existing + "
|
||
"new rows. Stop/start-able bench.")
|
||
ap.add_argument("--endpoint", default=os.environ.get(
|
||
"ABORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1"))
|
||
ap.add_argument("--model", default=os.environ.get(
|
||
"ABORIST_LLM_MODEL", "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic"))
|
||
args = ap.parse_args(argv)
|
||
|
||
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 2
|
||
|
||
modes = [m.strip() for m in args.modes.split(",") if m.strip()]
|
||
bad = [m for m in modes if m not in ANSWER_MODES]
|
||
if bad:
|
||
print(f"unknown mode(s): {bad}; expected subset of {list(ANSWER_MODES)}", file=sys.stderr)
|
||
return 2
|
||
|
||
args.out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Resume support: when --resume <existing.jsonl> is given, reuse
|
||
# that file's stamp and rebuild done_tasks set so already-completed
|
||
# samples skip on this run. Markdown rolls up the union of
|
||
# pre-existing + freshly-written rows.
|
||
existing_rows: list[dict] = []
|
||
done_tasks: set[tuple[str, str, int]] = set()
|
||
if args.resume:
|
||
resume_path = Path(args.resume)
|
||
if not resume_path.exists():
|
||
print(f"--resume target does not exist: {resume_path}", file=sys.stderr)
|
||
return 2
|
||
with resume_path.open(encoding="utf-8") as rf:
|
||
for line in rf:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
row = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue # skip a partial trailing line
|
||
key = (row["question"], row["answer_mode"], int(row.get("sample_idx", 0)))
|
||
done_tasks.add(key)
|
||
existing_rows.append(row)
|
||
stamp = resume_path.stem # reuse the original "YYYY-MM-DDTHH-MM-SSZ"
|
||
jsonl_path = resume_path
|
||
md_path = resume_path.with_suffix(".md")
|
||
else:
|
||
stamp = _utc_iso_compact()
|
||
jsonl_path = args.out_dir / f"{stamp}.jsonl"
|
||
md_path = args.out_dir / f"{stamp}.md"
|
||
|
||
if args.n_samples < 1:
|
||
print(f"--n must be >= 1, got {args.n_samples}", file=sys.stderr)
|
||
return 2
|
||
|
||
n_runs_total = len(questions) * len(modes) * args.n_samples
|
||
print(
|
||
f"[bench] {len(questions)} question(s) × {len(modes)} mode(s) × "
|
||
f"{args.n_samples} sample(s) = {n_runs_total} run(s)"
|
||
)
|
||
print(f"[bench] shards_dir={args.shards_dir} top_k={args.top_k} "
|
||
f"burn=always (per-sample, forces fresh inference)")
|
||
print(f"[bench] concurrency={args.concurrency} seed={args.seed} "
|
||
f"(sample-level tasks; per-cell Lock serializes cache_key writes)")
|
||
if args.resume:
|
||
print(f"[bench] resuming from {jsonl_path}: "
|
||
f"{len(existing_rows)} task(s) already done; skipping")
|
||
else:
|
||
print(f"[bench] writing → {jsonl_path}")
|
||
print()
|
||
|
||
# Build sample-level task list. Each task is one (question, mode,
|
||
# sample_idx); shuffle so samples for the same cell get spread
|
||
# across time. Two effects:
|
||
# - True i.i.d. n-sample variance: consecutive samples of the
|
||
# same cell don't share vLLM batch composition or KV-cache
|
||
# locality, so what looks like model nondeterminism actually
|
||
# IS model nondeterminism.
|
||
# - vLLM's continuous batcher gets a diverse request stream;
|
||
# under the prior cell-grouped scheduling, batches were
|
||
# correlated and the engine couldn't fill efficiently.
|
||
tasks = [
|
||
(q, mode, sample_idx)
|
||
for q in questions
|
||
for mode in modes
|
||
for sample_idx in range(args.n_samples)
|
||
]
|
||
rng = random.Random(args.seed)
|
||
rng.shuffle(tasks)
|
||
total_tasks = len(tasks)
|
||
|
||
# Resume: filter out tasks already in the JSONL. Order preserved.
|
||
if done_tasks:
|
||
tasks = [t for t in tasks if t not in done_tasks]
|
||
print(f"[bench] {len(tasks)} task(s) remaining ({total_tasks} total)")
|
||
print()
|
||
|
||
# Per-cell lock — burn+insert against a shared cache_key must
|
||
# serialize. Two samples of the SAME (q, mode) ending up in
|
||
# adjacent worker slots could race; the lock makes that path
|
||
# safe without bottlenecking unrelated cells. Lock contention is
|
||
# rare under shuffle (samples spread out), so the throughput cost
|
||
# is near-zero while correctness is preserved.
|
||
cell_locks: dict[tuple[str, str], Lock] = defaultdict(Lock)
|
||
|
||
rows: list[dict] = list(existing_rows) # preserve resumed rows for the rollup
|
||
write_lock = Lock()
|
||
print_lock = Lock()
|
||
done = [len(existing_rows)] # countdown reflects total, not just-this-session
|
||
|
||
file_mode = "a" if args.resume else "w"
|
||
with jsonl_path.open(file_mode, encoding="utf-8") as f:
|
||
def _process(task: tuple[str, str, int]) -> None:
|
||
q, mode, sample_idx = task
|
||
with cell_locks[(q, mode)]:
|
||
row = _run_one(
|
||
question=q,
|
||
answer_mode=mode,
|
||
shards_dir=args.shards_dir,
|
||
qa_db=args.qa_db,
|
||
top_k=args.top_k,
|
||
burn=True,
|
||
endpoint=args.endpoint,
|
||
model=args.model,
|
||
)
|
||
row["sample_idx"] = sample_idx
|
||
with write_lock:
|
||
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||
rows.append(row)
|
||
f.flush()
|
||
with print_lock:
|
||
done[0] += 1
|
||
tag = (
|
||
f"err: {row['error']}" if row["error"]
|
||
else f"{row['audit_mode']} {row['n_verified']}/{row['n_quotes']} "
|
||
f"{row['elapsed_s']:.1f}s"
|
||
)
|
||
print(
|
||
f" [{done[0]:>3}/{total_tasks}] {mode:<22} "
|
||
f"#{sample_idx + 1}/{args.n_samples} "
|
||
f"{q[:42]:<42} → {tag}",
|
||
flush=True,
|
||
)
|
||
|
||
if args.concurrency <= 1:
|
||
for task in tasks:
|
||
_process(task)
|
||
else:
|
||
with ThreadPoolExecutor(max_workers=args.concurrency) as pool:
|
||
futures = [pool.submit(_process, t) for t in tasks]
|
||
for fut in as_completed(futures):
|
||
fut.result() # surfaces any exception
|
||
|
||
summary = _summarize(rows)
|
||
md = _render_markdown(rows, summary, stamp, modes, questions, args.n_samples)
|
||
md_path.write_text(md, encoding="utf-8")
|
||
|
||
print()
|
||
print(md)
|
||
print(f"\n[bench] results: {jsonl_path}\n[bench] summary: {md_path}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|