"""Measure pass@1 on a held-out task set, before and after a reinforcement-learning run. Purpose: the number the GRPO lab and the reality check are decided on. Sends every held-out problem to the model, grades the answer with the same verifier the reward used, and reports the share solved with its standard error and a breakdown by task family. Running it once on the starting model and once on the trained one is the whole measurement; running it on a family the model never trained on is what turns it into a claim about generalisation. Platform: all. The model is loaded locally through transformers and PEFT by default, so no server is needed; --base-url sends the same prompts to any OpenAI-compatible endpoint instead, which is the faster path when you already have one running. Minimum memory: 16 GB for a 1B to 4B model loaded locally; very little against a server. Assumes: Python 3.10 or newer; torch, transformers and peft for the local path; make-tasks.py has written the held-out files; rewards.py and runlog.py sit next to this file. Usage: python3 eval-pass-at-1.py --tasks tasks/heldout-same.jsonl --model Qwen/Qwen3-1.7B \ --label before --labbook labbook.md python3 eval-pass-at-1.py --tasks tasks/heldout-same.jsonl --model Qwen/Qwen3-1.7B \ --adapter runs/grpo-qwen3-1.7b --label after --labbook labbook.md python3 eval-pass-at-1.py --tasks tasks/heldout-different.jsonl \ --base-url http://127.0.0.1:8080/v1 --served-model local --attempts 4 --temperature 0.7 """ from __future__ import annotations import argparse import json import math import time import urllib.error import urllib.request from collections import defaultdict from pathlib import Path from typing import Optional import rewards as reward_lib import runlog def post_json(url: str, payload: dict, api_key: Optional[str], timeout: int) -> dict: body = json.dumps(payload).encode("utf-8") headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" request = urllib.request.Request(url, data=body, headers=headers, method="POST") try: with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:400] raise RuntimeError(f"{url} returned HTTP {exc.code}: {detail}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"cannot reach {url}: {exc.reason}") from exc class ServedModel: def __init__(self, base_url: str, model: str, api_key: Optional[str], timeout: int, temperature: float, max_tokens: int): self.endpoint = base_url.rstrip("/") + "/chat/completions" self.model, self.api_key, self.timeout = model, api_key, timeout self.temperature, self.max_tokens = temperature, max_tokens def answer(self, messages: list[dict], seed: int) -> str: payload = {"model": self.model, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, "seed": seed} body = post_json(self.endpoint, payload, self.api_key, self.timeout) return (body["choices"][0]["message"]["content"] or "").strip() class LocalModel: def __init__(self, model_id: str, adapter: Optional[str], temperature: float, max_tokens: int): import torch # noqa: PLC0415 from transformers import AutoModelForCausalLM, AutoTokenizer # noqa: PLC0415 self.torch = torch self.temperature, self.max_tokens = temperature, max_tokens if adapter: from peft import AutoPeftModelForCausalLM # noqa: PLC0415 self.model = AutoPeftModelForCausalLM.from_pretrained(adapter) self.tokenizer = AutoTokenizer.from_pretrained(adapter) else: self.model = AutoModelForCausalLM.from_pretrained(model_id) self.tokenizer = AutoTokenizer.from_pretrained(model_id) self.model.eval() print(f"evaluating {adapter or model_id} on {self.model.device}") def answer(self, messages: list[dict], seed: int) -> str: self.torch.manual_seed(seed) inputs = self.tokenizer.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, return_dict=True, return_tensors="pt" ).to(self.model.device) sampling = {"do_sample": False} if self.temperature <= 0 else { "do_sample": True, "temperature": self.temperature, "top_p": 0.95} with self.torch.no_grad(): out = self.model.generate(**inputs, max_new_tokens=self.max_tokens, **sampling) generated = out[0][inputs["input_ids"].shape[-1]:] return self.tokenizer.decode(generated, skip_special_tokens=True).strip() def standard_error(successes: float, trials: int) -> float: """The binomial standard error of a proportion, so a difference can be read honestly.""" if trials <= 0: return float("nan") p = successes / trials return math.sqrt(max(0.0, p * (1 - p)) / trials) def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--tasks", required=True, help="a JSONL file from make-tasks.py") parser.add_argument("--model", default="Qwen/Qwen3-1.7B") parser.add_argument("--adapter", default=None) parser.add_argument("--base-url", default=None, help="evaluate a served model instead") parser.add_argument("--served-model", default=None) parser.add_argument("--api-key", default=None) parser.add_argument("--attempts", type=int, default=1, help="samples per problem; 1 at temperature 0 is deterministic pass@1, " "more at a higher temperature estimates it with an error bar") parser.add_argument("--temperature", type=float, default=0.0) parser.add_argument("--max-tokens", type=int, default=512) parser.add_argument("--limit", type=int, default=None, help="evaluate only the first N problems") parser.add_argument("--tolerance", type=float, default=1e-6) parser.add_argument("--fallback-to-last-number", action="store_true") parser.add_argument("--label", default="run", help="a name for this side of the comparison") parser.add_argument("--out", default=None, help="write every answer here as JSON") parser.add_argument("--timeout", type=int, default=300) parser.add_argument("--seed", type=int, default=0) parser.add_argument("--labbook", default=None) parser.add_argument("--notes", default=None) args = parser.parse_args() if args.attempts > 1 and args.temperature <= 0: raise SystemExit("--attempts above 1 with temperature 0 samples the same answer every " "time; raise --temperature or drop back to one attempt") rows = [json.loads(line) for line in Path(args.tasks).read_text(encoding="utf-8").splitlines() if line.strip()] if args.limit: rows = rows[: args.limit] print(f"{len(rows)} problems from {args.tasks}, {args.attempts} attempt(s) each") if args.base_url: if not args.served_model: raise SystemExit("--base-url needs --served-model") model = ServedModel(args.base_url, args.served_model, args.api_key, args.timeout, args.temperature, args.max_tokens) else: model = LocalModel(args.model, args.adapter, args.temperature, args.max_tokens) grade = reward_lib.numeric_reward(args.tolerance, args.fallback_to_last_number) per_family: dict[str, list[float]] = defaultdict(list) answers = [] started = time.time() for i, row in enumerate(rows, start=1): scores = [] for attempt in range(args.attempts): text = model.answer(row["prompt"], args.seed + attempt) correct = grade(completions=[text], answer=[row["answer"]])[0] scores.append(correct) answers.append({"id": row["id"], "family": row["family"], "attempt": attempt, "answer": text, "reference": row["answer"], "correct": bool(correct)}) share = sum(scores) / len(scores) per_family[row["family"]].append(share) print(f" [{i}/{len(rows)}] {row['id']:14s} {share:.2f}") elapsed = time.time() - started overall = [s for values in per_family.values() for s in values] scores = { "label": args.label, "tasks": args.tasks, "problems": len(rows), "attempts_each": args.attempts, "pass_at_1": round(sum(overall) / len(overall), 4) if overall else None, "standard_error": round(standard_error(sum(overall), len(overall)), 4) if overall else None, "by_family": {family: round(sum(v) / len(v), 4) for family, v in sorted(per_family.items())}, "seconds": round(elapsed, 1), } print("\n" + json.dumps(scores, indent=2)) print("\nThe standard error is the one on this sample of problems. Two runs whose " "intervals overlap have not been shown to differ.") if args.out: Path(args.out).write_text(json.dumps({"scores": scores, "answers": answers}, indent=2), encoding="utf-8") print(f"answers written to {args.out}") if args.labbook: record = runlog.record( labbook=args.labbook, lab="part-14/eval-pass-at-1", model=args.adapter or args.served_model or args.model, dataset={"path": args.tasks, "sha256": runlog.file_sha256(args.tasks), "problems": len(rows)}, hyperparameters={"attempts": args.attempts, "temperature": args.temperature, "max_tokens": args.max_tokens, "tolerance": args.tolerance, "fallback_to_last_number": args.fallback_to_last_number, "label": args.label}, seed=args.seed, losses={}, scores=scores, config_path=__file__, notes=args.notes, ) print(f"recorded run {record['run_id']} in {args.labbook}") if __name__ == "__main__": main()