# /// script # requires-python = ">=3.10" # dependencies = ["httpx", "duckdb==1.5.6"] # /// """hotdogbench: ask vision models "is this a hot dog?" through OpenRouter and score the answers. Input is a table of photos with columns id, image (JPEG bytes), is_hot_dog, category: (default) the public dataset, read straight from S3 --input dist/hotdogbench.parquet a local copy (or any parquet with those columns) --input local --per-group 25 a balanced sample of downloaded photos (building the dataset) Every model gets the same photo bytes and the same question, one photo per request, and answers as JSON {"hot_dog": true|false}. Requests come off one queue with an adaptive in-flight limit: it starts at --start-concurrency, doubles until OpenRouter first pushes back (429/5xx), then cuts to 70% and climbs by one per round trip, so each model runs just under its rate cap. Throttled photos go back on the queue. Results go to ./out// (in the folder you run it from): .parquet one row per photo: answer, latency, tokens, cost (from OpenRouter usage), retries, error summary.json per model: accuracy (overall and by category), unanswered, wall time, latency, cost Usage: export OPENROUTER_API_KEY=... (or put it in ./.env) uv run https://hotdogbench.com/hotdogbench.py --models cheap # Jev, Luna, Haiku, Sol: about $2 uv run https://hotdogbench.com/hotdogbench.py --models all # all seven: about $21 """ import argparse import asyncio import base64 import hashlib import json import os import random import statistics import time from dataclasses import asdict, dataclass, field from pathlib import Path import duckdb import httpx HERE = Path(__file__).resolve().parent DATA_URL = "https://md-ecosystem-public.s3.amazonaws.com/hotdogbench.parquet" # the public dataset URL = "https://openrouter.ai/api/v1/chat/completions" # Jev answers typed questions through OpenRouter's Decisions API; images go in `state` as image_url parts. DECISIONS_URL = "https://openrouter.ai/api/alpha/decisions" QUESTION = "Is the food in this photo a hot dog? Answer with JSON: {\"hot_dog\": true} or {\"hot_dog\": false}." SCHEMA = { "type": "json_schema", "json_schema": { "name": "verdict", "strict": True, "schema": {"type": "object", "properties": {"hot_dog": {"type": "boolean"}}, "required": ["hot_dog"], "additionalProperties": False}, }, } # OpenRouter id -> extra request params. Lowest reasoning each model allows; temperature 0 where accepted. MODELS = { "openai/gpt-6.1-sol": {"reasoning": {"effort": "minimal"}}, "openai/gpt-6-luna": {"reasoning": {"effort": "minimal"}}, "openai/gpt-6-astra": {"reasoning": {"effort": "minimal"}}, "typesafe/jev-1.14": {"decisions": True}, # Decisions API. Pinned to Cloudflare's own endpoint: DeepInfra's rejects image_url parts. # (microsoft/microsoft-decision-1 rejects image_url parts on every provider, so it can't take this test.) "cloudflare/clef": {"decisions": True, "provider": {"only": ["Cloudflare"], "allow_fallbacks": False}}, "openai/gpt-6-luna-decisions": {"decisions": True}, "perplexity/pplx-decider-v1.1-27b": {"decisions": True}, "anthropic/claude-opus-5.5": {"temperature": 0}, "anthropic/claude-sonnet-5.5": {"temperature": 0}, "anthropic/claude-haiku-5.5": {"temperature": 0}, } RETRY = {408, 409, 425, 429, 500, 502, 503, 504, 520, 522, 524, 529} @dataclass class Result: id: int open_images_id: str | None category: str is_hot_dog: bool scored: bool = True pred: bool | None = None raw: str | None = None latency_s: float | None = None prompt_tokens: int | None = None completion_tokens: int | None = None reasoning_tokens: int | None = None cost_usd: float | None = None provider: str | None = None retries: list[str] = field(default_factory=list) error: str | None = None def load_photos(args) -> list[dict]: con = duckdb.connect() if args.input == "local": # usable downloads so far: saved, and Flickr page still CC BY 2.0 rows = con.sql(f""" WITH s AS (SELECT *, row_number() OVER () AS rn FROM read_csv('{HERE / "fetch_status.csv"}', all_varchar=true)), last AS (SELECT * FROM s QUALIFY row_number() OVER (PARTITION BY ImageID ORDER BY rn DESC) = 1) SELECT ImageID, grp FROM last WHERE status = 'ok' AND page_license = 'by/2.0' QUALIFY row_number() OVER (PARTITION BY grp ORDER BY hash(ImageID || 'sample')) <= {args.per_group} """).fetchall() photos = [{"id": i, "image": (HERE / "images" / f"{image_id}.jpg").read_bytes(), "category": grp, "is_hot_dog": grp == "hot_dog", "scored": True, "open_images_id": image_id} # stable order (Python's hash() is randomized per process, which scrambled ids between runs) for i, (image_id, grp) in enumerate(sorted(rows, key=lambda r: hashlib.md5(r[0].encode()).hexdigest()), 1)] bonus = HERE / "extras" / "magritte_hot_dog.jpg" if bonus.exists(): # the unscored bonus row, as in the built dataset photos.append({"id": 1001, "image": bonus.read_bytes(), "category": "magritte", "is_hot_dog": True, "scored": False}) return photos rel = con.sql(f"SELECT * FROM read_parquet('{args.input}', union_by_name = true) ORDER BY id") return [dict(zip(rel.columns, r)) for r in rel.fetchall()] def body_for(model: str, image: bytes) -> dict: url = "data:image/jpeg;base64," + base64.b64encode(image).decode() if MODELS[model].get("decisions"): extra = {k: v for k, v in MODELS[model].items() if k != "decisions"} # e.g. provider routing return {"model": model, "state": ["Photo:", {"type": "image_url", "image_url": {"url": url}}], "questions": {"hot_dog": {"type": "noul", "instructions": "Is the food in the photo a hot dog?"}}, **extra} body = { "model": model, "messages": [{"role": "user", "content": [ {"type": "image_url", "image_url": {"url": url}}, {"type": "text", "text": QUESTION}, ]}], "max_tokens": 1024, "response_format": SCHEMA, "usage": {"include": True}, **MODELS[model], } return body def parse(text: str) -> bool | None: try: v = json.loads(text.strip().removeprefix("```json").removesuffix("```").strip())["hot_dog"] return v if isinstance(v, bool) else None except (ValueError, KeyError, TypeError, AttributeError): low = (text or "").lower() if '"hot_dog": true' in low or '"hot_dog":true' in low: return True if '"hot_dog": false' in low or '"hot_dog":false' in low: return False return None class Throttle(Exception): """OpenRouter pushed back (429, 5xx, or an upstream rate-limit error); the photo goes back on the queue.""" def __init__(self, tag: str, retry_after: float | None = None, cut: bool = True): super().__init__(tag) self.tag, self.retry_after, self.cut = tag, retry_after, cut # cut=False: a dropped connection, not a rate signal class Limiter: """AIMD concurrency limit. Slow start (double per window) until the first throttle, then +1 per window of `limit` successes; on a throttle cut to 70%, at most once per cooldown so a burst of 429s counts once.""" def __init__(self, start: int, cap: int): self.limit, self.cap, self.in_flight = float(start), cap, 0 self.slow_start, self.successes, self.throttles, self.transport_errors = True, 0, 0, 0 self.last_cut, self.cond = 0.0, asyncio.Condition() self.trace: list[tuple[float, int]] = [] # (seconds since start, limit) on every change self.t0 = time.perf_counter() def _mark(self): self.trace.append((round(time.perf_counter() - self.t0, 2), int(self.limit))) async def acquire(self): async with self.cond: await self.cond.wait_for(lambda: self.in_flight < int(self.limit)) self.in_flight += 1 async def release(self, ok: bool | None, latency: float | None = None): """ok=True success, False throttle (cut), None neutral (dropped connection: retried, limit unchanged).""" async with self.cond: self.in_flight -= 1 now = time.perf_counter() if ok is None: self.transport_errors += 1 elif ok: self.successes += 1 if self.successes >= int(self.limit): # one window of successes self.successes = 0 self.limit = min(self.cap, self.limit * 2 if self.slow_start else self.limit + 1) self._mark() else: self.throttles += 1 if now - self.last_cut > max(1.0, latency or 1.0): # one cut per round trip self.slow_start, self.successes, self.last_cut = False, 0, now self.limit = max(1.0, self.limit * 0.7) self._mark() self.cond.notify_all() async def ask_once(client, key, model, body, res: Result) -> None: """One request. Fills `res` on an answer or a final error; raises Throttle when it should be retried.""" t0 = time.perf_counter() try: r = await client.post(DECISIONS_URL if "questions" in body else URL, json=body, headers={"Authorization": f"Bearer {key}"}) except (httpx.HTTPError, OSError) as e: # stale keep-alive connections show up as SSLError/ReadError raise Throttle(type(e).__name__, retry_after=0.0, cut=False) res.latency_s = time.perf_counter() - t0 if r.status_code in RETRY: ra = r.headers.get("retry-after") raise Throttle(str(r.status_code), float(ra) if ra and ra.replace(".", "").isdigit() else None) try: data = r.json() except ValueError: raise Throttle(f"{r.status_code}/badjson") if r.status_code != 200 or "error" in data: err = data.get("error") or {} code = err.get("code") if isinstance(err, dict) else None reason = ((err.get("metadata") or {}).get("reason") if isinstance(err, dict) else None) or "" if r.status_code == 402 and "in_flight" in reason: # credit reservations for in-flight requests: back off raise Throttle("402/in_flight", retry_after=2.0) if code in RETRY: raise Throttle(f"200/err{code}") res.error = f"{r.status_code}: {json.dumps(err)[:300]}" return u = data.get("usage") or {} if "answers" in data: # Decisions API: noul probability that it's a hot dog p = (data["answers"].get("hot_dog") or {}).get("noul") res.raw, res.pred = str(p), (None if p is None else p >= 0.5) res.prompt_tokens, res.completion_tokens = u.get("input_tokens"), u.get("output_tokens") res.cost_usd, res.provider = u.get("cost"), data.get("provider") return msg = data["choices"][0]["message"] res.raw = (msg.get("content") or "")[:500] res.pred = parse(res.raw) res.prompt_tokens, res.completion_tokens = u.get("prompt_tokens"), u.get("completion_tokens") res.reasoning_tokens = (u.get("completion_tokens_details") or {}).get("reasoning_tokens") res.cost_usd, res.provider = u.get("cost"), data.get("provider") MAX_ATTEMPTS = 8 def summarize(model: str, all_results: list[Result], wall: float) -> dict: bonus = {str(r.id): r.pred for r in all_results if not r.scored} # e.g. the Magritte painting results = [r for r in all_results if r.scored] answered = [r for r in results if r.pred is not None] def acc(rs): rs = [r for r in rs if r.pred is not None] return round(sum(r.pred == r.is_hot_dog for r in rs) / len(rs), 4) if rs else None lat = sorted(r.latency_s for r in results if r.latency_s is not None and r.error is None) cost = sum(r.cost_usd or 0 for r in results) pos = [r for r in answered if r.is_hot_dog] neg = [r for r in answered if not r.is_hot_dog] tpr = sum(r.pred for r in pos) / len(pos) if pos else None tnr = sum(not r.pred for r in neg) / len(neg) if neg else None return { "model": model, "photos": len(results), "answered": len(answered), "unanswered": len(results) - len(answered), "errors": sum(r.error is not None for r in results), "accuracy": acc(results), "balanced_accuracy": round((tpr + tnr) / 2, 4) if tpr is not None and tnr is not None else None, "accuracy_by_category": {c: acc([r for r in results if r.category == c]) for c in sorted({r.category for r in results})}, "false_alarms": sum(1 for r in neg if r.pred), "missed_hot_dogs": sum(1 for r in pos if not r.pred), "wall_s": round(wall, 2), "photos_per_s": round(len(all_results) / wall, 2) if wall else None, "latency_p50_s": round(statistics.median(lat), 3) if lat else None, "latency_p95_s": round(lat[int(0.95 * (len(lat) - 1))], 3) if lat else None, "cost_usd": round(cost, 5), "cost_per_1000_photos_usd": round(cost / len(results) * 1000, 3) if results else None, "prompt_tokens_per_photo": round(sum(r.prompt_tokens or 0 for r in results) / max(1, len(answered)), 1), "reasoning_tokens": sum(r.reasoning_tokens or 0 for r in results), "requests_retried": sum(1 for r in results if r.retries), "providers": sorted({r.provider for r in results if r.provider}), "unscored_answers": bonus, # id -> said hot dog? } async def run_model(model, photos, key, start, cap, out_dir) -> dict: """One queue of photos, `cap` workers, and an AIMD limiter deciding how many requests are in flight.""" lim = Limiter(start, cap) queue: asyncio.Queue = asyncio.Queue() results: dict[int, Result] = {} for p in photos: res = Result(id=int(p["id"]), open_images_id=p.get("open_images_id"), category=p["category"], is_hot_dog=bool(p["is_hot_dog"]), scored=p.get("scored") is not False) results[res.id] = res queue.put_nowait((0.0, p, body_for(model, p["image"]), res)) pending = len(photos) done_evt = asyncio.Event() peak_in_flight = 0 async def worker(client): nonlocal pending, peak_in_flight while not done_evt.is_set(): try: not_before, p, body, res = await asyncio.wait_for(queue.get(), timeout=0.5) except asyncio.TimeoutError: continue wait = not_before - time.perf_counter() if wait > 0: await asyncio.sleep(wait) await lim.acquire() peak_in_flight = max(peak_in_flight, lim.in_flight) try: await ask_once(client, key, model, body, res) await lim.release(True) except Throttle as t: await lim.release(False if t.cut else None, res.latency_s) res.retries.append(t.tag) if len(res.retries) >= MAX_ATTEMPTS: res.error = "retries exhausted: " + ",".join(res.retries[-3:]) else: # back on the queue; honour Retry-After, else a short jittered pause delay = t.retry_after if t.retry_after is not None else min(20, 0.5 * 2 ** len(res.retries)) * random.random() queue.put_nowait((time.perf_counter() + delay, p, body, res)) continue pending -= 1 if pending == 0: done_evt.set() async def progress(t0): while not done_evt.is_set(): await asyncio.sleep(5) n = len(photos) - pending print(f" {model}: {n}/{len(photos)} done, limit {int(lim.limit)}, in flight {lim.in_flight}, " f"throttles {lim.throttles}, dropped conns {lim.transport_errors}, {n / (time.perf_counter() - t0):.1f}/s", flush=True) limits = httpx.Limits(max_connections=cap, max_keepalive_connections=cap) async with httpx.AsyncClient(timeout=httpx.Timeout(120), limits=limits) as client: t0 = time.perf_counter() lim.t0 = t0 tasks = [asyncio.create_task(worker(client)) for _ in range(cap)] + [asyncio.create_task(progress(t0))] await done_evt.wait() wall = time.perf_counter() - t0 for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptions=True) ordered = [results[int(p["id"])] for p in photos] con = duckdb.connect() rows = [asdict(r) for r in ordered] for r in rows: r["retries"] = ",".join(r["retries"]) con.execute("CREATE TABLE t AS SELECT * FROM (SELECT unnest(?, recursive := true))", [rows]) con.execute(f"COPY t TO '{out_dir / (model.replace('/', '__') + '.parquet')}'") s = summarize(model, ordered, wall) s["concurrency"] = {"start": start, "cap": cap, "final_limit": int(lim.limit), "peak_in_flight": peak_in_flight, "throttles": lim.throttles, "transport_errors": lim.transport_errors, "limit_trace": lim.trace} return s def openrouter_key() -> str: key = os.environ.get("OPENROUTER_API_KEY") or os.environ.get("OPENROUTER_TOKEN") for env in (Path.cwd() / ".env", HERE.parent / ".env"): if key or not env.exists(): continue for line in env.read_text().splitlines(): k, _, v = line.partition("=") if k.strip() in ("OPENROUTER_API_KEY", "OPENROUTER_TOKEN"): key = v.strip().strip('"').strip("'") if not key: raise SystemExit("Set OPENROUTER_API_KEY") return key def main() -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) local = HERE / "dist" / "hotdogbench.parquet" ap.add_argument("--input", default=str(local) if local.exists() else DATA_URL) ap.add_argument("--per-group", type=int, default=25, help="with --input local: photos per category") ap.add_argument("--models", default="all", help="'all', 'cheap' (Jev, Luna, Haiku, Sol) or comma-separated OpenRouter ids") ap.add_argument("--start-concurrency", type=int, default=8, help="in-flight requests to start from") ap.add_argument("--max-concurrency", type=int, default=256, help="ceiling for the adaptive limit") ap.add_argument("--run", default=time.strftime("run_%Y%m%d_%H%M%S")) ap.add_argument("--out", default=str(Path.cwd() / "out"), help="results folder (default ./out)") args = ap.parse_args() cheap = ["typesafe/jev-1.14", "openai/gpt-6-luna", "anthropic/claude-haiku-5.5", "openai/gpt-6.1-sol"] models = {"all": list(MODELS), "cheap": cheap}.get(args.models) or args.models.split(",") photos = load_photos(args) out_dir = Path(args.out) / args.run out_dir.mkdir(parents=True, exist_ok=True) key = openrouter_key() print(f"{len(photos)} photos x {len(models)} models -> {out_dir}") summaries = [] for m in models: # models run one after another so their timings don't interfere s = asyncio.run(run_model(m, photos, key, args.start_concurrency, args.max_concurrency, out_dir)) summaries.append(s) line = {k: s[k] for k in ("model", "accuracy", "unanswered", "errors", "wall_s", "photos_per_s", "latency_p50_s", "cost_per_1000_photos_usd")} line.update({k: s["concurrency"][k] for k in ("final_limit", "peak_in_flight", "throttles", "transport_errors")}) print(json.dumps(line), flush=True) (out_dir / "summary.json").write_text(json.dumps(summaries, indent=2)) if __name__ == "__main__": main()