#!/usr/bin/env python3 # Published reference script. """Run the local model over 20 questions, Q01 through Q20. Examples: python3 run20.py --only q01 python3 run20.py --reps 1 The supplied historical Series 1 runner defaulted to two repetitions. The published scored corpus contains one r1 response per model and matter; the exact executed invocation is not included in the public record. Use --reps 1 to reproduce the published corpus shape. The runner sends no sampling parameters. The local server configuration governs sampling. Transport retries are recorded. A non-stop finish reason is preserved and is never silently repaired or replaced. """ from __future__ import annotations import argparse import hashlib import json import pathlib import re import sys import time from typing import Any import requests def classify_finish(finish_reason: str | None, transport_failed: bool, refusal: bool = False) -> dict[str, Any]: length_truncated = finish_reason == "length" delivery_complete = (not transport_failed) and finish_reason == "stop" and not refusal delivery_failure = transport_failed or refusal or finish_reason != "stop" return { "transport_failed": transport_failed, "finish_reason": finish_reason, "length_truncated": length_truncated, "delivery_complete": delivery_complete, "delivery_failure": delivery_failure, "refusal_detected": refusal, "scored_as_delivered": True, } def get_props(base: str) -> dict[str, Any]: url = base.rstrip("/").removesuffix("/v1") + "/props" response = requests.get(url, timeout=30) response.raise_for_status() return response.json() def auto_label(props: dict[str, Any]) -> str: name = pathlib.Path(props.get("model_path", "")).name name = re.sub(r"-\d{5}-of-\d{5}", "", name) name = re.sub(r"\.gguf$", "", name, flags=re.I) name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-.") return name or "local-model" def ask(url: str, prompt: str, max_tokens: int, timeout: int) -> tuple[dict[str, Any], list[dict[str, Any]]]: payload = {"messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "stream": False} history: list[dict[str, Any]] = [] for attempt in range(1, 4): try: response = requests.post(url, json=payload, timeout=timeout) if response.status_code in (429, 500, 502, 503, 504): raise requests.HTTPError(f"HTTP {response.status_code}: {response.text[:200]}") response.raise_for_status() data = response.json() choice = data["choices"][0] message = choice.get("message") or {} finish = choice.get("finish_reason") refusal = bool(message.get("refusal")) history.append({"attempt": attempt, "transport_failed": False, "finish_reason": finish, "superseded": False}) result = { "content": message.get("content") or "", "reasoning": message.get("reasoning_content") or message.get("reasoning") or "", "server_model": data.get("model"), "usage": data.get("usage") or {}, "attempt_number": attempt, "rerun_occurred": attempt > 1, "rerun_reason": "transport retry" if attempt > 1 else None, "scored_attempt": attempt, "earlier_attempt_superseded": attempt > 1, } result.update(classify_finish(finish, False, refusal)) return result, history except Exception as exc: history.append({"attempt": attempt, "transport_failed": True, "error": str(exc), "superseded": attempt < 3}) if attempt == 3: raise time.sleep(10 * attempt) raise RuntimeError("unreachable") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--base-url", default="http://127.0.0.1:8080/v1") parser.add_argument("--questions", default="questions/") parser.add_argument("--answers", default="answers/") parser.add_argument("--logs", default="logs/") parser.add_argument("--only") parser.add_argument("--reps", type=int, default=2) parser.add_argument("--max-tokens", type=int, default=64000) parser.add_argument("--timeout", type=int, default=3600) args = parser.parse_args() base = pathlib.Path(__file__).resolve().parent props = get_props(args.base_url) label = auto_label(props) qdir = base / args.questions questions = sorted(qdir.glob("q*.md")) if args.only: questions = [p for p in questions if p.stem == args.only] if not questions: raise SystemExit(f"No questions found in {qdir}") answer_dir = base / args.answers / label answer_dir.mkdir(parents=True, exist_ok=True) log_dir = base / args.logs log_dir.mkdir(parents=True, exist_ok=True) log_file = log_dir / "runs.jsonl" template = props.get("chat_template") or "" template_prefix = hashlib.sha256(template.encode()).hexdigest()[:16] if template else None model_filename = pathlib.Path(props.get("model_path", "")).name for question_path in questions: prompt = question_path.read_text(encoding="utf-8") for rep in range(1, args.reps + 1): answer_path = answer_dir / f"{label}__{question_path.stem}_r{rep}.md" if answer_path.exists(): continue result, attempts = ask(args.base_url.rstrip("/") + "/chat/completions", prompt, args.max_tokens, args.timeout) answer_path.write_text(result.pop("content"), encoding="utf-8") reasoning = result.pop("reasoning") if reasoning: answer_path.with_suffix(".reasoning.md").write_text(reasoning, encoding="utf-8") record = { **result, "attempt_history": attempts, "label": label, "question": question_path.stem, "rep": rep, "max_tokens": args.max_tokens, "base_url": args.base_url, "answer_file": str(answer_path.relative_to(base)), "answer_sha256": hashlib.sha256(answer_path.read_bytes()).hexdigest(), "model_filename": model_filename, "chat_template_sha256_prefix": template_prefix, "server_params": (props.get("default_generation_settings") or {}).get("params", {}), "_public_record": {"type": "prospective reproduction record", "policy": "../../PUBLIC_RECORD_POLICY.md"}, } with log_file.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") print(question_path.stem, label, record["finish_reason"], "delivery_complete=", record["delivery_complete"]) if __name__ == "__main__": main()