zwitschi
659b0e35b2
Add openrouter_judge.py to rank live image+JSON models by advertised price, with ordered fallbacks, auth-error handling and auto selection. Wire both judges to --model auto, record the model used, and add fake-catalogue/mocked-call tests. Re-run the judges with the valid key: 13/13 screen designs match, 0 diff; the Forward+ A/B report records the selected model and ~$0.006 spend. Headless 231/231, Python 94/94.
128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
"""Pick affordable OpenRouter vision models for the screenshot judges.
|
|
|
|
The model catalogue is fetched from OpenRouter (no auth needed) and filtered to
|
|
models that accept image input and advertise `response_format`/structured output.
|
|
The ranking uses the advertised per-token prices, so it follows the account's
|
|
current catalogue instead of relying on one hard-coded model name. If the
|
|
catalogue cannot be fetched, a small known-good list keeps QA working offline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
MODELS_URL = "https://openrouter.ai/api/v1/models"
|
|
DEFAULT_MODEL = "auto"
|
|
|
|
# Used only when the catalogue is unavailable. Ordered by price/known quality.
|
|
FALLBACK_MODELS = [
|
|
"z-ai/glm-5.3-flash",
|
|
"qwen/qwen3.7-flash",
|
|
"google/gemma-3-4b-it",
|
|
"openai/gpt-4.1-nano",
|
|
"google/gemini-2.5-flash-lite",
|
|
]
|
|
|
|
|
|
def fetch_models(timeout: int = 30) -> list[dict]:
|
|
"""Return the OpenRouter model catalogue, or [] when it is unreachable."""
|
|
req = urllib.request.Request(MODELS_URL, headers={"Accept": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
payload = __import__("json").loads(resp.read().decode("utf-8"))
|
|
except (OSError, ValueError, urllib.error.HTTPError):
|
|
return []
|
|
return list(payload.get("data") or [])
|
|
|
|
|
|
def _price(model: dict) -> float:
|
|
pricing = model.get("pricing") or {}
|
|
try:
|
|
prompt = float(pricing.get("prompt") or 0.0)
|
|
completion = float(pricing.get("completion") or 0.0)
|
|
image = float(pricing.get("image") or 0.0)
|
|
except (TypeError, ValueError):
|
|
return float("inf")
|
|
# A judge request is image-heavy and emits a short JSON verdict.
|
|
return prompt + image + completion * 0.25
|
|
|
|
|
|
def vision_candidates(catalog: list[dict], limit: int = 6) -> list[str]:
|
|
"""Cheap, image-capable models that can return structured JSON."""
|
|
ranked: list[tuple[float, str]] = []
|
|
for model in catalog:
|
|
model_id = str(model.get("id") or "")
|
|
if not model_id or model_id.startswith("~") or ":batch" in model_id:
|
|
continue
|
|
architecture = model.get("architecture") or {}
|
|
inputs = architecture.get("input_modalities") or []
|
|
outputs = architecture.get("output_modalities") or []
|
|
if "image" not in inputs or "text" not in outputs:
|
|
continue
|
|
params = model.get("supported_parameters") or []
|
|
# Our request uses response_format={"type": "json_object"}; models that
|
|
# only advertise structured_outputs may reject that envelope.
|
|
if "response_format" not in params:
|
|
continue
|
|
price = _price(model)
|
|
if price <= 0.0 or price == float("inf"):
|
|
continue
|
|
ranked.append((price, model_id))
|
|
ranked.sort()
|
|
out: list[str] = []
|
|
for _price_value, model_id in ranked:
|
|
if model_id not in out:
|
|
out.append(model_id)
|
|
if len(out) >= max(1, limit):
|
|
break
|
|
return out
|
|
|
|
|
|
def resolve_models(requested: str = DEFAULT_MODEL, *, catalog: list[dict] | None = None,
|
|
limit: int = 6) -> list[str]:
|
|
"""Resolve ``auto``/an explicit model to a fallback-ordered model list.
|
|
|
|
`GIG_LINE_JUDGE_MODEL` overrides the automatic choice for reproducible runs;
|
|
an explicit CLI `--model` still wins.
|
|
"""
|
|
requested = str(requested or "").strip()
|
|
if requested and requested != DEFAULT_MODEL:
|
|
return [requested]
|
|
env_model = os.environ.get("GIG_LINE_JUDGE_MODEL", "").strip()
|
|
if env_model:
|
|
return [env_model]
|
|
models = vision_candidates(catalog if catalog is not None else fetch_models(), limit)
|
|
if not models:
|
|
models = FALLBACK_MODELS[:max(1, limit)]
|
|
return models
|
|
|
|
|
|
def is_auth_error(exc: BaseException) -> bool:
|
|
text = str(exc)
|
|
return "HTTP 401" in text or "HTTP 403" in text or "API key not valid" in text
|
|
|
|
|
|
def call_with_fallback(models: list[str], make_body, call, attempts: int = 2):
|
|
"""Try each model, retrying provider errors, and return (model, response).
|
|
|
|
Authentication errors abort immediately: switching models cannot repair a
|
|
bad key. Model/provider errors move to the next candidate.
|
|
"""
|
|
last: BaseException | None = None
|
|
for model in models:
|
|
body = make_body(model)
|
|
for attempt in range(max(1, attempts)):
|
|
try:
|
|
return model, call(body)
|
|
except (ValueError, OSError) as exc:
|
|
last = exc
|
|
if is_auth_error(exc):
|
|
raise
|
|
if attempt + 1 >= max(1, attempts):
|
|
break
|
|
if last is not None:
|
|
raise last
|
|
raise ValueError("no judge model was available")
|