Spaces:
Running
Running
File size: 3,337 Bytes
da0ce7b | 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 | #!/usr/bin/env python3
"""
Aureon Operator — A/B benchmark runner.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Runs the ground-truth prompt set through two conditions and prints a comparison:
baseline — raw model output (no grounding, no consensus, no veto)
aureon — the same model through the full operator chain
python scripts/run_operator_benchmark.py
python scripts/run_operator_benchmark.py --out /tmp/bench.json
Writes the full result JSON (per-item + aggregate metrics) for the audit report.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
_REPO = Path(__file__).resolve().parents[1]
if str(_REPO) not in sys.path:
sys.path.insert(0, str(_REPO))
from aureon.operator.benchmark import run_benchmark # noqa: E402
PROMPTS = _REPO / "data/research/operator_benchmark_prompts.json"
RECORDINGS = _REPO / "data/research/operator_benchmark_recordings.jsonl"
_METRIC_ROWS = [
("fact_accuracy", "Fact accuracy", "higher"),
("hallucination_rate", "Hallucination rate", "lower"),
("abstention_rate", "Honest abstention", "context"),
("grounding_coverage", "Grounding coverage", "higher"),
("safety_block_rate", "Safety block rate", "higher"),
("mean_latency_ms", "Mean latency (ms)", "context"),
]
def _fmt(metric: str, value: float) -> str:
if metric == "mean_latency_ms":
return f"{value:.1f}"
return f"{value*100:.1f}%" if value <= 1.0 else f"{value}"
def main(argv) -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--out", default=str(_REPO / "data/research/operator_benchmark_results.json"))
ap.add_argument("--personas", default="a,b")
args = ap.parse_args(argv[1:])
result = run_benchmark(
PROMPTS, RECORDINGS, personas=[p.strip() for p in args.personas.split(",") if p.strip()]
)
Path(args.out).write_text(json.dumps(result, indent=2), encoding="utf-8")
b = result["baseline"]["metrics"]
a = result["aureon"]["metrics"]
print("\n\033[36m🎛️ AUREON OPERATOR — A/B BENCHMARK\033[0m")
print(f" prompts={result['n_prompts']} factual={b['n_factual']} adversarial={b['n_adversarial']}")
if b["n_missing_recordings"] or a["n_missing_recordings"]:
print(f" \033[33m⚠ missing recordings: baseline={b['n_missing_recordings']} aureon={a['n_missing_recordings']}\033[0m")
print(f"\n {'Metric':<22}{'Baseline':>12}{'Aureon':>12} Better")
print(" " + "─" * 60)
for key, label, better in _METRIC_ROWS:
bv, av = b[key], a[key]
arrow = ""
if better == "higher":
arrow = "✅ Aureon" if av > bv else ("— tie" if av == bv else "⚠ baseline")
elif better == "lower":
arrow = "✅ Aureon" if av < bv else ("— tie" if av == bv else "⚠ baseline")
print(f" {label:<22}{_fmt(key, bv):>12}{_fmt(key, av):>12} {arrow}")
print(f"\n results written → {args.out}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
|