Prime Agent 开源：用递归模型与持续 Harness 构建可自我改进的编程 Agent

import json
import os
import subprocess
import urllib.request
from pathlib import Path

BASE_URL = os.environ.get("LLM_BASE_URL", "http://localhost:8000/v1")
MODEL = os.environ.get("LLM_MODEL", "your-model")
API_KEY = os.environ.get("LLM_API_KEY", "local")
WORK_FILE = Path("solution.py")
TEST_FILE = Path("test_solution.py")
MAX_ROUNDS = 5

SYSTEM_PROMPT = """You are a coding agent working on one Python file.
Return only the complete contents of solution.py, without Markdown fences.
Use the test output to repair the implementation. Do not modify the tests.
"""


def call_model(task: str, code: str, feedback: str) -> str:
    payload = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": f"Task:\n{task}\n\nCurrent code:\n{code}\n\nTest feedback:\n{feedback}",
            },
        ],
        "temperature": 0,
    }
    request = urllib.request.Request(
        f"{BASE_URL}/chat/completions",
        data=json.dumps(payload).encode(),
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
    )
    with urllib.request.urlopen(request, timeout=120) as response:
        body = json.load(response)
    return body["choices"][0]["message"]["content"].strip()


def run_tests() -> tuple[bool, str]:
    result = subprocess.run(
        ["python", "-m", "unittest", "-v", str(TEST_FILE)],
        text=True,
        capture_output=True,
        timeout=30,
    )
    output = (result.stdout + "\n" + result.stderr)[-6000:]
    return result.returncode == 0, output


def main() -> None:
    task = "Implement fibonacci(n). Reject negative n with ValueError."
    feedback = "No tests have been run yet."
    WORK_FILE.write_text("def fibonacci(n):\n    pass\n", encoding="utf-8")

    for round_no in range(1, MAX_ROUNDS + 1):
        current = WORK_FILE.read_text(encoding="utf-8")
        candidate = call_model(task, current, feedback)
        WORK_FILE.write_text(candidate + "\n", encoding="utf-8")

        passed, feedback = run_tests()
        print(f"round={round_no} passed={passed}")
        if passed:
            print("Accepted solution:\n" + candidate)
            return

    raise SystemExit("The agent exhausted its repair budget.\n" + feedback)


if __name__ == "__main__":
    main()

import unittest
from solution import fibonacci


class FibonacciTests(unittest.TestCase):
    def test_values(self):
        self.assertEqual([fibonacci(i) for i in range(8)], [0, 1, 1, 2, 3, 5, 8, 13])

    def test_negative_value(self):
        with self.assertRaises(ValueError):
            fibonacci(-1)


if __name__ == "__main__":
    unittest.main()

export LLM_BASE_URL="http://localhost:8000/v1"
export LLM_MODEL="your-model"
export LLM_API_KEY="your-api-key"
python continual_harness.py