Make the vision judge discover cost-effective OpenRouter models
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.
This commit is contained in:
1 parent
28f2a48811
commit
659b0e35b2
15 files changed
+535
-65
No files matched your search
@@ -0,0 +1,127 @@
|
||||
"""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")
|
||||
Reference in new issue
Block a user