"""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: # Ollama sometimes plans most layers onto the CPU with VRAM free. 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 # a missed search: the fields below never get used 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) # load the model; not timed 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()