| """Run the app's classifier prompt against an Ollama model and score it. |
| |
| Mirrors app/agent/orchestrator._classify: same system prompt (dump_prompt.py), |
| same strict schema, temperature 0, thinking off unless --think. The one |
| post-processing step that changes routing is applied too: wanting a picture |
| forces a search. |
| |
| python3 eval.py qwen3.5:2b [--think] [--tag name] |
| python3 eval.py openrouter/anthropic/claude-haiku-5.5 [--think] # key from ../.env |
| """ |
| import argparse, json, os, statistics, sys, time, urllib.error, urllib.request |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| SYSTEM = open(os.path.join(HERE, "system_prompt.txt")).read() |
| SCHEMA = json.load(open(os.path.join(HERE, "schema.json"))) |
| HARD = {"day", "week", "month"} |
|
|
|
|
| def _key(): |
| with open(os.path.join(HERE, "..", ".env")) as f: |
| for line in f: |
| if line.startswith("OPENROUTER_API_KEY="): |
| return line.split("=", 1)[1].strip() |
| raise SystemExit("OPENROUTER_API_KEY missing in .env") |
|
|
|
|
| def call_openrouter(model, item, think): |
| """The app's own request: strict json_schema, temperature 0, and the |
| deployment's default 1024-token reasoning allowance when thinking.""" |
| messages = [{"role": "system", "content": SYSTEM}, *item["hist"], {"role": "user", "content": item["q"]}] |
| body = {"model": model, "messages": messages, "temperature": 0, "max_tokens": 500 + (1024 if think else 0), |
| "response_format": {"type": "json_schema", |
| "json_schema": {"name": "Classification", "strict": True, "schema": SCHEMA}}, |
| "reasoning": {"max_tokens": 1024} if think else {"enabled": False}, "usage": {"include": True}} |
| req = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", json.dumps(body).encode(), |
| {"Content-Type": "application/json", "Authorization": f"Bearer {_key()}"}) |
| t = time.time() |
| for attempt in range(4): |
| try: |
| with urllib.request.urlopen(req, timeout=120) as r: |
| out = json.load(r) |
| break |
| except urllib.error.HTTPError as e: |
| if e.code not in (429, 500, 502, 503) or attempt == 3: |
| raise SystemExit(f"OpenRouter {e.code}: {e.read()[:300]!r}") |
| time.sleep(2 ** attempt) |
| usage = out.get("usage") or {} |
| return ({"message": {"content": out["choices"][0]["message"].get("content") or ""}, |
| "eval_count": usage.get("completion_tokens"), "prompt_eval_count": usage.get("prompt_tokens"), |
| "cost": usage.get("cost")}, time.time() - t) |
|
|
|
|
| def call(model, item, think, num_gpu=None): |
| if model.startswith("openrouter/"): |
| return call_openrouter(model.removeprefix("openrouter/"), item, think) |
| messages = [{"role": "system", "content": SYSTEM}, *item["hist"], {"role": "user", "content": item["q"]}] |
| body = {"model": model, "messages": messages, "stream": False, "format": SCHEMA, "think": think, |
| "options": {"temperature": 0, "num_ctx": 8192, "num_predict": 500 + (1024 if think else 0)}, |
| "keep_alive": "10m"} |
| if num_gpu is not None: |
| |
| body["options"]["num_gpu"] = num_gpu |
| req = urllib.request.Request("http://localhost:11434/api/chat", json.dumps(body).encode(), |
| {"Content-Type": "application/json"}) |
| t = time.time() |
| with urllib.request.urlopen(req, timeout=300) as r: |
| out = json.load(r) |
| return out, time.time() - t |
|
|
|
|
| def score(item, pred): |
| s = {} |
| if pred is None: |
| return {"json": False} |
| s["json"] = True |
| search = bool(pred.get("needs_search", True)) or bool(pred.get("wants_image")) |
| s["search"] = search == item["search"] |
| if not item["search"]: |
| return s |
| if not search: |
| return s |
| cx = str(pred.get("complexity", "") or "").strip().lower() |
| cx = cx if cx in {"simple", "standard", "research"} else "standard" |
| s["complexity"] = cx in item["cx"] |
| s["shape"] = str(pred.get("answer_shape", "") or "").strip().lower() in item["shape"] |
| if item["img"] is not None: |
| s["image"] = bool(pred.get("wants_image")) == item["img"] |
| rec = str(pred.get("recency", "") or "").strip().lower() |
| s["recency"] = {"none": rec == "", "any": True, "set": rec != "", "hard": rec in HARD, |
| "nothard": rec not in HARD}[item["rec"]] |
| lenses = {str(x).strip().lower() for x in pred.get("lenses") or []} |
| if item["lens_in"] or item["lens_out"]: |
| s["lenses"] = set(item["lens_in"]) <= lenses and not (set(item["lens_out"]) & lenses) |
| queries = [str(x) for x in pred.get("search_queries") or [] if str(x).strip()] |
| text = " ".join(queries).lower() |
| s["queries_n"] = 1 <= len(queries) <= 3 |
| s["keys"] = all(any(k in text for k in group) for group in item["keys"]) |
| if item["sq"]: |
| sq = str(pred.get("standalone_question", "")).lower() |
| s["standalone"] = all(any(k in sq for k in group) for group in item["sq"]) |
| if item["place"]: |
| s["place"] = any(k in str(pred.get("place", "")).lower() for k in item["place"]) |
| s["route"] = s["search"] and s["complexity"] and s["shape"] |
| return s |
|
|
|
|
| def main(): |
| p = argparse.ArgumentParser() |
| p.add_argument("model") |
| p.add_argument("--think", action="store_true") |
| p.add_argument("--tag", default="") |
| p.add_argument("--num-gpu", type=int, default=None, help="layers on the GPU (99 = all)") |
| args = p.parse_args() |
| items = [json.loads(l) for l in open(os.path.join(HERE, "testset.jsonl"))] |
| tag = args.tag or (args.model.replace(":", "-").replace("/", "_") + ("-think" if args.think else "")) |
| out_path = os.path.join(HERE, f"results-{tag}.jsonl") |
| call(args.model, items[0], args.think, args.num_gpu) |
| with open(out_path, "w") as f: |
| for i, item in enumerate(items): |
| out, dt = call(args.model, item, args.think, args.num_gpu) |
| raw = out["message"]["content"] |
| try: |
| pred = json.loads(raw) |
| except json.JSONDecodeError: |
| pred = None |
| row = {"i": i, "q": item["q"], "lang": item["lang"], "tag": item["tag"], "pred": pred, "raw": raw, |
| "score": score(item, pred), "seconds": round(dt, 3), "out_tokens": out.get("eval_count"), |
| "prompt_tokens": out.get("prompt_eval_count"), "cost": out.get("cost")} |
| f.write(json.dumps(row, ensure_ascii=False) + "\n") |
| print(f"{i + 1}/{len(items)} {dt:.2f}s", file=sys.stderr, end="\r") |
| print(f"\n{out_path}", file=sys.stderr) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
