commit a54bd57fbdb7ba540ef92204685b28447176a5c3 Author: Codex Bot Date: Tue Sep 15 19:15:05 2026 +0800 Implement provider-neutral Lean reasoning and evaluation foundation diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2c255f8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.egg-info/ +.venv/ +.env +results/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..6d6f081 --- /dev/null +++ b/README.md @@ -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)。 diff --git a/lean-toolchain b/lean-toolchain new file mode 100644 index 0000000..a8afa7d --- /dev/null +++ b/lean-toolchain @@ -0,0 +1 @@ +leanprover/lean4:v4.33.1 diff --git a/lean_agent/__init__.py b/lean_agent/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lean_agent/cli.py b/lean_agent/cli.py new file mode 100644 index 0000000..1aa6d3b --- /dev/null +++ b/lean_agent/cli.py @@ -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() diff --git a/lean_agent/core.py b/lean_agent/core.py new file mode 100644 index 0000000..dc045ea --- /dev/null +++ b/lean_agent/core.py @@ -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} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d0cff30 --- /dev/null +++ b/pyproject.toml @@ -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*"] diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..c80bced --- /dev/null +++ b/tests/test_core.py @@ -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()