| """Benchmark cheap OpenRouter models against AskMyDocs' real answer contract.
|
|
|
| The app does not just need valid JSON — the grounding pipeline splits the answer
|
| into claims and verifies each against its cited evidence, so the model MUST emit
|
| inline ``[C1]``/``[C2]`` markers inside the answer text. A model that returns a
|
| ``citations`` array but no inline markers silently breaks grounding, which a
|
| casual test will miss.
|
|
|
| Scores each model on:
|
| * json_ok — response parses as the expected JSON object
|
| * inline — number of [Cn] markers inside `answer` (must be > 0)
|
| * claims — extracted claims array length
|
| * latency — seconds
|
| * cost — USD for the run's token usage
|
|
|
| Usage:
|
| python scripts/model_benchmark.py # default shortlist
|
| python scripts/model_benchmark.py --top 12 # shortlist by price
|
| python scripts/model_benchmark.py --trials 2
|
| """
|
|
|
| from __future__ import annotations
|
|
|
| import argparse
|
| import json
|
| import os
|
| import time
|
|
|
| import httpx
|
| from dotenv import load_dotenv
|
|
|
| OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
|
|
|
| SYSTEM_PROMPT = (
|
| "You answer strictly from the provided evidence. "
|
| 'Respond with a single JSON object: {"answer": string, "citations": [string], '
|
| '"claims": [{"claim": string, "citation_ids": [string]}]}. '
|
| "Every factual sentence in `answer` must end with an inline marker like [C1] "
|
| "referring to the evidence item it came from. "
|
| "If the evidence is insufficient, refuse instead of guessing."
|
| )
|
|
|
|
|
| CASES = [
|
| {
|
| "question": "What dominates ingestion latency, and how is chunk identity preserved?",
|
| "evidence": (
|
| "[C1] The ingestion pipeline runs parsing, cleaning, chunking, "
|
| "embedding, vector upsert and keyword indexing.\n"
|
| "[C2] On CPU, embedding dominates: 46,977 ms of a 47.1 s run for a "
|
| "4-chunk document.\n"
|
| "[C3] Chunk IDs are deterministic hashes derived from the document id "
|
| "and chunk index, so they match across Postgres, Qdrant and the "
|
| "tsvector postings."
|
| ),
|
| },
|
| {
|
| "question": "How are API keys secured and what happens when they expire?",
|
| "evidence": (
|
| "[C1] API keys are prefixed with `amsk_` and are shown only once at "
|
| "creation time.\n"
|
| "[C2] Keys expire after 90 days and can be revoked at any time from "
|
| "the Admin Console.\n"
|
| "[C3] A key can be scoped to a set of permissions using the "
|
| "permissions claim; scoping is strongly recommended."
|
| ),
|
| },
|
| ]
|
|
|
|
|
| def fetch_models(api_key: str) -> list[dict]:
|
| r = httpx.get("https://openrouter.ai/api/v1/models", timeout=60)
|
| r.raise_for_status()
|
| return r.json()["data"]
|
|
|
|
|
| def price_of(model: dict) -> tuple[float, float] | None:
|
| p = model.get("pricing") or {}
|
| try:
|
| pin = float(p.get("prompt", "0")) * 1_000_000
|
| pout = float(p.get("completion", "0")) * 1_000_000
|
| except (TypeError, ValueError):
|
| return None
|
| if pin < 0 or pout < 0:
|
| return None
|
| return pin, pout
|
|
|
|
|
| def run_case(api_key: str, model_id: str, case: dict) -> dict:
|
| payload = {
|
| "model": model_id,
|
| "messages": [
|
| {"role": "system", "content": SYSTEM_PROMPT},
|
| {
|
| "role": "user",
|
| "content": f"Question: {case['question']}\n\nEvidence:\n{case['evidence']}",
|
| },
|
| ],
|
| "max_tokens": 900,
|
| "temperature": 0.2,
|
| "response_format": {"type": "json_object"},
|
| }
|
| start = time.time()
|
| try:
|
| r = httpx.post(
|
| OPENROUTER_URL,
|
| headers={
|
| "Authorization": f"Bearer {api_key}",
|
| "Content-Type": "application/json",
|
| },
|
| json=payload,
|
| timeout=180,
|
| )
|
| except Exception as exc:
|
| return {"ok": False, "error": type(exc).__name__, "secs": time.time() - start}
|
|
|
| secs = time.time() - start
|
| if r.status_code != 200:
|
| return {"ok": False, "error": f"HTTP {r.status_code}", "secs": secs}
|
|
|
| data = r.json()
|
| choices = data.get("choices") or []
|
| if not choices:
|
| return {"ok": False, "error": "no choices", "secs": secs}
|
|
|
| content = (choices[0].get("message") or {}).get("content") or ""
|
| usage = data.get("usage") or {}
|
| result = {
|
| "ok": True,
|
| "secs": secs,
|
| "finish": choices[0].get("finish_reason"),
|
| "prompt_tokens": usage.get("prompt_tokens", 0),
|
| "completion_tokens": usage.get("completion_tokens", 0),
|
| "empty": not content.strip(),
|
| }
|
| try:
|
| obj = json.loads(content)
|
| answer = obj.get("answer") or ""
|
| markers = sum(answer.count(m) for m in ("[C1]", "[C2]", "[C3]"))
|
| result.update(
|
| {
|
| "json": True,
|
| "inline": markers,
|
| "claims": len(obj.get("claims") or []),
|
| "answer_len": len(answer),
|
| }
|
| )
|
| except Exception:
|
| result.update({"json": False, "inline": 0, "claims": 0, "answer_len": 0})
|
| return result
|
|
|
|
|
| def main() -> int:
|
| load_dotenv()
|
| parser = argparse.ArgumentParser(description="Benchmark cheap OpenRouter models.")
|
| parser.add_argument("--top", type=int, default=10, help="How many cheap paid models to test.")
|
| parser.add_argument("--trials", type=int, default=1, help="Runs per model per case.")
|
| parser.add_argument(
|
| "--max-out-price",
|
| type=float,
|
| default=0.45,
|
| help="Max completion price (USD per 1M) to consider 'dead cheap'.",
|
| )
|
| parser.add_argument("--only", nargs="*", help="Explicit model ids to test.")
|
| parser.add_argument(
|
| "--include-free",
|
| action="store_true",
|
| help="Also test free (:free) variants.",
|
| )
|
| args = parser.parse_args()
|
|
|
| api_key = os.getenv("OPENROUTER_API_KEY")
|
| if not api_key:
|
| print("OPENROUTER_API_KEY not set")
|
| return 1
|
|
|
| models = fetch_models(api_key)
|
|
|
| if args.only:
|
| candidates = [m for m in models if m["id"] in args.only]
|
| else:
|
| priced = []
|
| for m in models:
|
| pr = price_of(m)
|
| if pr is None:
|
| continue
|
| pin, pout = pr
|
| if pin == 0 and pout == 0 and not args.include_free:
|
| continue
|
| if pout > args.max_out_price:
|
| continue
|
| if (m.get("context_length") or 0) < 32_000:
|
| continue
|
| priced.append((pout, pin, m["id"], m))
|
| priced.sort(key=lambda t: (t[0], t[1], t[2]))
|
| candidates = [t[3] for t in priced[: args.top]]
|
|
|
| print(f"Testing {len(candidates)} model(s), {len(CASES)} case(s), {args.trials} trial(s)\n")
|
|
|
| rows = []
|
| for model in candidates:
|
| mid = model["id"]
|
| pr = price_of(model) or (0.0, 0.0)
|
| runs = []
|
| for _ in range(args.trials):
|
| for case in CASES:
|
| runs.append(run_case(api_key, mid, case))
|
| good = [r for r in runs if r.get("ok")]
|
| total_cost = sum(
|
| (r.get("prompt_tokens", 0) / 1e6) * pr[0]
|
| + (r.get("completion_tokens", 0) / 1e6) * pr[1]
|
| for r in good
|
| )
|
| rows.append(
|
| {
|
| "model": mid,
|
| "in": pr[0],
|
| "out": pr[1],
|
| "runs": len(runs),
|
| "ok": len(good),
|
| "json": sum(1 for r in good if r.get("json")),
|
| "cited": sum(1 for r in good if (r.get("inline") or 0) > 0),
|
| "empty": sum(1 for r in good if r.get("empty")),
|
| "claims": sum(r.get("claims", 0) for r in good),
|
| "avg_s": round(sum(r["secs"] for r in good) / len(good), 1) if good else None,
|
| "cost": total_cost,
|
| }
|
| )
|
|
|
| hdr = (
|
| f"{'model':44s} {'$/1M in/out':>14s} {'json':>5s} {'cited':>6s} "
|
| f"{'empty':>6s} {'claims':>7s} {'avg s':>6s} {'$/query':>9s}"
|
| )
|
| print(hdr)
|
| print("-" * len(hdr))
|
| for r in rows:
|
| print(
|
| f"{r['model']:44s} {r['in']:6.3f}/{r['out']:<7.3f} "
|
| f"{r['json']:>3d}/{r['runs']:<2d} {r['cited']:>4d}/{r['runs']:<2d} "
|
| f"{r['empty']:>4d}/{r['runs']:<2d} {r['claims']:>7d} "
|
| f"{str(r['avg_s']):>6s} {r['cost']/max(1, r['runs']):>9.6f}"
|
| )
|
| return 0
|
|
|
|
|
| if __name__ == "__main__":
|
| raise SystemExit(main())
|
|
|
