claude-bot
ca98703538
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FYqwNMyPUHeTbFWg83GQq6
111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
"""Run the pilot detectors over the pilot items. Resumable; one JSON line per item.
|
|
|
|
python experiments/pilot/run_pilot.py --items runs/p85-110/items.jsonl \
|
|
--out runs/pilot/results.jsonl --dry-run # no API calls: shows the plan
|
|
python experiments/pilot/run_pilot.py --items ... --out ... --max-calls 400
|
|
|
|
Detectors (all see the passage and/or the Lean code, never the human labels):
|
|
lint static patterns on the projected simplex (no LLM)
|
|
role source-side simplex (LLM sees only the passage) aligned to the Lean simplex
|
|
judge one LLM call that sees passage + Lean and says faithful / not
|
|
compile the stored Lean verdict (``verified``), read from the items file
|
|
|
|
Hard stops: --max-calls bounds API calls; a rerun skips items already in --out.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from leanbench import srcsimplex
|
|
from leanbench.align import align
|
|
from leanbench.config import Settings
|
|
from leanbench.lint import lint
|
|
from leanbench.llm import OpenAICompatibleClient
|
|
from leanbench.project import project
|
|
|
|
JUDGE_PROMPT_VERSION = "judge-v1"
|
|
JUDGE_SYSTEM = (
|
|
"You check a Lean 4 (Mathlib) formalization against a textbook passage. Decide whether the "
|
|
"Lean statement says what the passage says: same objects, same hypotheses, same conclusion, "
|
|
"same quantifiers. Ignore whether it compiles. Reply with JSON only: "
|
|
'{"faithful": true|false, "confidence": 0..1, "reason": "one sentence"}'
|
|
)
|
|
|
|
|
|
def judge(client, passage: str, lean: str) -> dict:
|
|
reply = client.complete_json([
|
|
{"role": "system", "content": JUDGE_SYSTEM},
|
|
{"role": "user", "content": f"Passage:\n{passage}\n\nLean:\n{lean}"},
|
|
])
|
|
if not isinstance(reply, dict) or "faithful" not in reply:
|
|
raise ValueError(f"unusable judge reply: {str(reply)[:200]}")
|
|
return {"faithful": bool(reply["faithful"]), "confidence": reply.get("confidence"),
|
|
"reason": str(reply.get("reason", ""))[:300]}
|
|
|
|
|
|
def run_item(client, item: dict) -> dict:
|
|
code = item.get("lean_code") or ""
|
|
row: dict = {"id": item["id"], "compiled": bool(item.get("verified")), "errors": {}}
|
|
proj = project(code)
|
|
row["projection"] = proj.as_dict()
|
|
row["lint"] = lint(proj)
|
|
try:
|
|
src = srcsimplex.extract_source_simplex(client, item.get("source_text", ""),
|
|
item.get("label", ""))
|
|
al = align(src, proj)
|
|
row["source"] = [r.__dict__ for r in src.roles]
|
|
row["role"] = al.as_dict()
|
|
except Exception as error: # a failed extraction is data, not a crash
|
|
row["errors"]["role"] = str(error)[:300]
|
|
try:
|
|
row["judge"] = judge(client, item.get("source_text", ""), code)
|
|
except Exception as error:
|
|
row["errors"]["judge"] = str(error)[:300]
|
|
return row
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--items", required=True, type=Path)
|
|
ap.add_argument("--out", required=True, type=Path)
|
|
ap.add_argument("--max-calls", type=int, default=400)
|
|
ap.add_argument("--dry-run", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
items = [json.loads(x) for x in args.items.read_text(encoding="utf-8").splitlines() if x.strip()]
|
|
done = set()
|
|
if args.out.exists():
|
|
done = {json.loads(x)["id"] for x in args.out.read_text(encoding="utf-8").splitlines() if x.strip()}
|
|
todo = [i for i in items if i["id"] not in done]
|
|
settings = Settings.from_env()
|
|
est = len(todo) * 3 # extract (<=2 tries) + judge
|
|
print(f"items={len(items)} done={len(done)} todo={len(todo)} model={settings.llm.model} "
|
|
f"est_calls<={est} (extract x2 + judge) cap={args.max_calls}")
|
|
print("prompts:", srcsimplex.PROMPT_VERSION, JUDGE_PROMPT_VERSION,
|
|
hashlib.sha256((srcsimplex.SYSTEM_PROMPT + JUDGE_SYSTEM).encode()).hexdigest()[:12])
|
|
if args.dry_run:
|
|
return 0
|
|
|
|
client = OpenAICompatibleClient.from_settings(settings.llm)
|
|
args.out.parent.mkdir(parents=True, exist_ok=True)
|
|
with args.out.open("a", encoding="utf-8") as fh:
|
|
for item in todo:
|
|
if client.usage.calls >= args.max_calls:
|
|
print("call cap reached; rerun to resume", file=sys.stderr)
|
|
break
|
|
fh.write(json.dumps(run_item(client, item), ensure_ascii=False) + "\n")
|
|
fh.flush()
|
|
meta = {"model": settings.llm.model, "usage": client.usage.as_dict(),
|
|
"prompts": [srcsimplex.PROMPT_VERSION, JUDGE_PROMPT_VERSION]}
|
|
(args.out.parent / "run_meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8")
|
|
print(json.dumps(meta))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|