Implement provider-neutral Lean reasoning and evaluation foundation

2026-09-15 19:15:05 +08:00
commit 99638be637
9 changed files with 117 additions and 0 deletions
@@ -0,0 +1,50 @@
+"""JSONL benchmark runner; failed tasks stay in the denominator."""
+import argparse
+import hashlib
+import json
+import subprocess
+from pathlib import Path
+from lean_agent.cli import options, components
+from lean_agent.core import solve
+
+def load_tasks(path):
+    tasks = [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+    seen = set()
+    for task in tasks:
+        if not isinstance(task.get('id'), str) or not isinstance(task.get('statement'), str):
+            raise ValueError('Every task needs string id and statement')
+        if task['id'] in seen: raise ValueError('Duplicate task id: ' + task['id'])
+        seen.add(task['id'])
+    if not tasks: raise ValueError('Empty dataset')
+    return tasks
+
+def main():
+    parser = argparse.ArgumentParser(description='Evaluate Lean agent on trusted JSONL tasks')
+    parser.add_argument('dataset', type=Path)
+    parser.add_argument('--output', type=Path, required=True, help='New directory for this run')
+    options(parser)
+    args = parser.parse_args()
+    if args.attempts < 1: parser.error('--attempts must be positive')
+    tasks = load_tasks(args.dataset)
+    provider, verifier = components(args)
+    args.output.mkdir(parents=True, exist_ok=False)
+    version = subprocess.run([*verifier.command, '--version'], cwd=verifier.cwd,
+                             capture_output=True, text=True, timeout=30).stdout.strip()
+    passed = errors = total_attempts = 0
+    with (args.output / 'results.jsonl').open('w') as stream:
+        for task in tasks:
+            result = solve(provider, verifier, task, args.attempts)
+            passed += result['success']
+            errors += 'error' in result
+            total_attempts += len(result['attempts'])
+            stream.write(json.dumps(result, ensure_ascii=False) + '\n')
+            stream.flush()
+    summary = {'tasks': len(tasks), 'solved': passed, 'solve_rate': passed / len(tasks),
+               'errors': errors, 'attempts': total_attempts, 'max_attempts': args.attempts,
+               'model': args.model, 'provider': 'command' if args.provider_command else 'http',
+               'lean_version': version,
+               'dataset_sha256': hashlib.sha256(args.dataset.read_bytes()).hexdigest()}
+    (args.output / 'summary.json').write_text(json.dumps(summary, indent=2) + '\n')
+    print(json.dumps(summary, indent=2))
+
+if __name__ == '__main__': main()