run_eval.py

"""评测流水线骨架（benchmark / eval runner）

说明：
- 这个脚本负责准备每个测试用例的环境（通过 setup_script），
  触发 agent 去执行任务（此处留有调用占位），并运行 verify_script 校验结果。
- 因为不同用户可能用不同的方法运行 agent（CLI / API / 子进程 / RPC），
  本脚本把 agent 运行部分封装为 `run_agent_instruction`，示例中为占位实现，
  你可以把它改成用 `subprocess` 启动 `python -m mini_claude` 并通过 stdin 交互，
  或者直接通过已暴露的 HTTP 接口调用 agent。
"""

import os
import subprocess
import time
import sys
from pathlib import Path

EVAL_DATASET = [
    {
        "id": "task_01",
        "instruction": "当前目录下有一个 math_utils.py，里面有一个漏了右括号的 add 函数，请用工具帮我修复它。",
        "setup_script": "echo 'def add(a, b:\n    return a + b' > math_utils.py",
        "verify_script": "python -c \"from math_utils import add; assert add(1, 2) == 3\"",
    },
    {
        "id": "task_02",
        "instruction": "使用 search_code 工具找出处理数据库连接的逻辑，并把最大连接数从 10 改成 20。",
        "setup_script": "echo 'def db_connect():\n    max_conn = 10\n    return max_conn' > db_conn.py",
        "verify_script": "python -c \"from db_conn import db_connect; assert db_connect() == 20\"",
    },
]


def run_agent_instruction(instruction: str, timeout: int = 120) -> bool:
    """通过 subprocess 启动本地 agent，并把任务指令写入 stdin。"""
    workspace_root = Path(__file__).resolve().parent
    api_base = os.environ.get("MINI_CLAUDE_API_BASE", "http://127.0.0.1:11434/v1")
    api_key = os.environ.get("MINI_CLAUDE_API_KEY", "ollama")
    model = os.environ.get("MINI_CLAUDE_MODEL", "qwen2.5-coder:7b")
    memory_model = os.environ.get("MINI_CLAUDE_MEMORY_MODEL", model)

    command = [
        sys.executable,
        "-m",
        "mini_claude",
        "--api-base",
        api_base,
        "--api-key",
        api_key,
        "--model",
        model,
        "--memory-model",
        memory_model,
        "--verbose",
    ]

    print(f"[run_eval] 启动 agent: {' '.join(command)}")
    process = subprocess.Popen(
        command,
        cwd=workspace_root,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
    )

    try:
        stdout, _ = process.communicate(input=f"{instruction}\nexit\n", timeout=timeout)
        print(stdout)
        return process.returncode == 0
    except subprocess.TimeoutExpired:
        process.kill()
        stdout, _ = process.communicate()
        print(stdout)
        print(f"[run_eval] agent 超时，已终止 ({timeout}s)")
        return False


def run_eval(agent_version: str = "local") -> None:
    success_count = 0
    for task in EVAL_DATASET:
        print(f"正在评测任务: {task['id']}")

        # 1. 准备环境
        subprocess.run(task["setup_script"], shell=True)
        time.sleep(0.2)

        # 2. 触发 agent（占位）
        ok = run_agent_instruction(task["instruction"])  # 若实现为异步可等待

        # 3. 校验结果（如果 agent 已经在文件系统上修改，会由 verify_script 检查）
        result = subprocess.run(task["verify_script"], shell=True, capture_output=True)
        if result.returncode == 0 and ok:
            print("✅ 任务成功！")
            success_count += 1
        else:
            print("❌ 任务失败！")
            print(result.stdout.decode(errors='ignore'))
            print(result.stderr.decode(errors='ignore'))

    total = len(EVAL_DATASET)
    print(f"【{agent_version}】端到端完成率: {success_count / total * 100:.1f}% ({success_count}/{total})")


if __name__ == "__main__":
    print("run_eval: 运行评测骨架。请实现 run_agent_instruction 以使评测真实生效。")
    run_eval("V0_skeleton")