Implement provider-neutral Lean reasoning and evaluation foundation

2026-09-15 19:15:05 +08:00
commit a54bd57fbd
8 changed files with 222 additions and 0 deletions
@@ -0,0 +1,5 @@
+__pycache__/
+*.egg-info/
+.venv/
+.env
+results/
@@ -0,0 +1,38 @@
+# Lean Agent
+
+独立 Python 3.11+ 项目：LLM 生成 tactic → Lean 验证 → 把错误反馈给模型修复。已在 Lean 4.33.1 验证，无 Python 运行时第三方依赖。
+
+## 使用
+
+```sh
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -e .
+printf '%s\n' '{"id":"example","statement":"∀ n : Nat, n = n"}' > /tmp/task.json
+export LLM_API_KEY='your-key'
+lean-agent /tmp/task.json --model YOUR_MODEL --base-url https://YOUR_HOST/v1
+```
+
+HTTP 接口使用 `/chat/completions`；本地服务可以不设密钥。任何非兼容服务可以使用适配器：
+
+```sh
+lean-agent /tmp/task.json --provider-command '["python3","/absolute/path/adapter.py"]'
+```
+
+适配命令 stdin 接收 `{"messages":[{"role":"user","content":"..."}]}`，stdout 返回 Lean tactic 正文；错误用非零退出码，日志用 stderr。也可实现 Python `Provider.complete(messages) -> str`。任何支持文本输出的 LLM 均可适配，但不保证证明能力。
+
+任务含 id、statement、可选 prelude（如 import Mathlib）。输出记录每轮证明、诊断、耗时。成功退出 0，失败退出 1。
+
+Mathlib 使用已有 Lake 项目：`--lean-project /absolute/project --lean-command '["lake","env","lean"]'`。不自动安装 Mathlib。可用 `--attempts`、`--lean-timeout`、`--llm-timeout` 控制预算。
+
+## 验证与限制
+
+拒绝 sorry/admit 和常见命令注入；要求编译成功、无 warning，并通过 #print axioms 确认仅依赖 propext、Classical.choice、Quot.sound 或无公理。[Lean 官方说明](https://lean-lang.org/doc/reference/latest/Tactic-Proofs/Tactic-Reference/)。
+
+文本过滤不是安全沙箱。数据集/prelude、导入模块、适配器必须可信；不可信模型代码应在无密钥、无网络、限制资源的容器或虚拟机中运行。当前仅提供进程超时和 Lean heartbeat 限制。不包含交互式 goal 搜索、token 计费。
+
+```sh
+python3 -m unittest discover -s tests -v
+```
+
+配套 [Lean Bench](https://gitea.entroia.com/codex-bot/lean-bench)。
@@ -0,0 +1 @@
+leanprover/lean4:v4.33.1
@@ -0,0 +1,31 @@
+import argparse
+import json
+from pathlib import Path
+from .core import HTTPProvider, CommandProvider, LeanVerifier, solve
+
+def options(parser):
+    parser.add_argument('--model', default='local-model')
+    parser.add_argument('--base-url', default='http://localhost:8000/v1')
+    parser.add_argument('--api-key-env', default='LLM_API_KEY')
+    parser.add_argument('--provider-command', help='JSON argv array; stdin JSON messages, stdout proof')
+    parser.add_argument('--lean-command', default='["lean"]', help='JSON argv array; e.g. ["lake","env","lean"]')
+    parser.add_argument('--lean-project', help='Lean/Lake project working directory')
+    parser.add_argument('--attempts', type=int, default=3)
+    parser.add_argument('--lean-timeout', type=float, default=30)
+    parser.add_argument('--llm-timeout', type=float, default=120)
+
+def components(args):
+    provider = (CommandProvider(json.loads(args.provider_command), args.llm_timeout)
+        if args.provider_command else HTTPProvider(args.model, args.base_url, args.api_key_env, args.llm_timeout))
+    return provider, LeanVerifier(json.loads(args.lean_command), args.lean_project, args.lean_timeout)
+
+def main():
+    parser = argparse.ArgumentParser(description='Prove one Lean theorem with any LLM')
+    parser.add_argument('task', type=Path, help='JSON with id, statement, optional prelude')
+    options(parser)
+    args = parser.parse_args()
+    result = solve(*components(args), json.loads(args.task.read_text()), args.attempts)
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    raise SystemExit(0 if result['success'] else 1)
+
+if __name__ == '__main__': main()
@@ -0,0 +1,111 @@
+"""Small provider-neutral proof repair loop. Dataset statements are trusted input."""
+import json
+import os
+import re
+import subprocess
+import tempfile
+import time
+import urllib.request
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Protocol
+
+class Provider(Protocol):
+    def complete(self, messages: list[dict[str, str]]) -> str: ...
+
+class HTTPProvider:
+    """OpenAI-compatible chat completions, including local serving endpoints."""
+    def __init__(self, model, base_url, api_key_env="LLM_API_KEY", timeout=120):
+        self.model, self.base_url = model, base_url.rstrip('/')
+        self.api_key_env, self.timeout = api_key_env, timeout
+
+    def complete(self, messages):
+        headers = {"Content-Type": "application/json"}
+        key = os.environ.get(self.api_key_env)
+        if key:
+            headers["Authorization"] = "Bearer " + key
+        request = urllib.request.Request(self.base_url + '/chat/completions',
+            data=json.dumps({"model": self.model, "messages": messages}).encode(), headers=headers)
+        with urllib.request.urlopen(request, timeout=self.timeout) as response:
+            content = json.load(response)["choices"][0]["message"]["content"]
+        if not isinstance(content, str):
+            raise ValueError("Provider returned non-text content")
+        return content
+
+class CommandProvider:
+    """Any model: executable receives {messages: [...]} on stdin, returns proof text."""
+    def __init__(self, argv, timeout=120):
+        if not argv: raise ValueError("Empty provider command")
+        self.argv, self.timeout = argv, timeout
+
+    def complete(self, messages):
+        proc = subprocess.run(self.argv, input=json.dumps({"messages": messages}),
+                              text=True, capture_output=True, timeout=self.timeout)
+        if proc.returncode: raise RuntimeError("Provider command failed")
+        return proc.stdout
+
+@dataclass
+class Check:
+    success: bool
+    diagnostics: str
+
+class LeanVerifier:
+    def __init__(self, command=None, cwd=None, timeout=30):
+        self.command = command or ["lean"]
+        self.cwd = str(Path(cwd).resolve()) if cwd else None
+        self.timeout = timeout
+
+    def check(self, statement, proof, prelude=""):
+        # Only a tactic body is accepted. Fail closed on holes and command injection.
+        if re.search(r'\b(sorry|admit|axiom|unsafe|implemented_by|run_tac|elab|macro|syntax)\b|--|/\*|/\-|\-/|#', proof):
+            return Check(False, "Forbidden proof token (holes, metaprogramming, comments or commands)")
+        if not proof.strip(): return Check(False, "Empty proof")
+        source = prelude + '\nset_option maxHeartbeats 200000\n'
+        source += 'theorem agent_target : ' + statement + ' := by\n'
+        source += '\n'.join('  ' + line for line in proof.splitlines())
+        source += '\n#print axioms agent_target\n'
+        with tempfile.TemporaryDirectory(prefix="lean-check-", dir=self.cwd) as directory:
+            path = Path(directory) / 'Candidate.lean'
+            path.write_text(source)
+            try:
+                result = subprocess.run([*self.command, str(path)], cwd=self.cwd,
+                    capture_output=True, text=True, timeout=self.timeout)
+            except subprocess.TimeoutExpired:
+                return Check(False, "Lean timeout")
+            diagnostics = result.stdout + result.stderr
+        # Reject warnings and require the axiom report; allow only Lean's standard axioms.
+        reports = re.findall(r"depends on axioms:\s*\[([^\]]*)\]", diagnostics)
+        clean = "does not depend on any axioms" in diagnostics
+        if reports:
+            clean = all(set(x.strip() for x in report.split(',') if x.strip()) <=
+                        {"propext", "Classical.choice", "Quot.sound"} for report in reports)
+        return Check(result.returncode == 0 and clean and 'warning:' not in diagnostics,
+                     diagnostics[-16000:])
+
+def extract(text):
+    match = re.search(r'```(?:lean4?|text)?\s*\n(.*?)```', text, re.S)
+    proof = (match.group(1) if match else text).strip()
+    if proof == 'by': return ''
+    if proof.startswith('by\n') or proof.startswith('by '): proof = proof[2:].strip()
+    return proof
+
+def solve(provider, verifier, task, max_attempts=3):
+    if max_attempts < 1: raise ValueError("max_attempts must be positive")
+    start = time.monotonic()
+    messages = [{"role": "system", "content": "Prove the Lean 4 theorem. Return ONLY the tactic body after by. No theorem declaration, comments, sorry, admit, metaprogramming or extra commands."},
+                {"role": "user", "content": task.get('prelude', '') + '\n⊢ ' + task['statement']}]
+    attempts = []
+    for _ in range(max_attempts):
+        try:
+            proof = extract(provider.complete(messages))
+            check = verifier.check(task['statement'], proof, task.get('prelude', ''))
+        except Exception as exc:
+            # Do not serialize exception bodies: providers may include credentials.
+            return {"id": task['id'], "success": False, "error": type(exc).__name__,
+                    "attempts": attempts, "seconds": time.monotonic() - start}
+        attempts.append({"proof": proof, **asdict(check)})
+        if check.success: break
+        messages.extend([{"role": "assistant", "content": proof},
+                         {"role": "user", "content": "Lean rejected this proof. Repair it:\n" + check.diagnostics}])
+    return {"id": task['id'], "success": attempts[-1]['success'],
+            "attempts": attempts, "seconds": time.monotonic() - start}
@@ -0,0 +1,16 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "lean-agent"
+version = "0.1.0"
+requires-python = ">=3.11"
+description = "Provider-independent Lean reasoning agent"
+dependencies = []
+
+[project.scripts]
+lean-agent = "lean_agent.cli:main"
+
+[tool.setuptools.packages.find]
+include = ["lean_agent*"]
@@ -0,0 +1,20 @@
+import unittest
+from lean_agent.core import LeanVerifier, solve, extract
+class Scripted:
+    def __init__(self): self.calls = 0
+    def complete(self, messages):
+        self.calls += 1
+        return 'exact False.elim' if self.calls == 1 else 'intro n; rfl'
+class Tests(unittest.TestCase):
+    def test_repair(self):
+        result = solve(Scripted(), LeanVerifier(), {'id':'x','statement':'∀ n : Nat, n = n'})
+        self.assertTrue(result['success'])
+        self.assertEqual(len(result['attempts']), 2)
+    def test_holes(self):
+        for proof in ['sorry', 'admit', 'exact (by sorry)', 'rfl\n#eval 1']:
+            self.assertFalse(LeanVerifier().check('1 = 1', proof).success)
+    def test_false(self):
+        self.assertFalse(LeanVerifier().check('False', 'trivial').success)
+    def test_extract(self):
+        self.assertEqual(extract('```lean\nby\n  rfl\n```'), 'rfl')
+if __name__ == '__main__': unittest.main()