Agent 基准测试套件：SWE-bench vs WebArena vs GAIA

# SWE-bench 任务示例
task = {
    "instance_id": "django__django-11099",
    "repo": "django/django",
    "base_commit": "a4b78c0e8e5d1b6...",
    "problem_statement": """
    When using ALLOWED_HOSTS with a domain that contains an underscore,
    ValidationError is raised. According to RFC 1034, underscores are
    technically not allowed in hostnames but are commonly used in
    certain DNS records. We should handle this gracefully.
    """,
    "test_patch": "diff --git a/tests/requests/tests.py ...",
    "FAIL_TO_PASS": [
        "tests.requests.tests.test_allowed_hosts_underscore"
    ],
    "PASS_TO_PASS": [
        "tests.requests.tests.test_basic"
    ],
    "environment_setup_commit": "a4b78c0e8e5d1b6..."
}

┌──────────────┐
│  读取 Issue   │
└──────┬───────┘
       v
┌──────────────┐
│  Clone 仓库   │
│  Checkout 基线 │
└──────┬───────┘
       v
┌──────────────┐
│  Agent 工作   │  <── 探索代码、定位问题、编写修复
│  (限时 30min) │
└──────┬───────┘
       v
┌──────────────┐
│  应用测试补丁  │
│  运行单元测试  │
└──────┬───────┘
       v
┌──────────────┐
│  评估结果     │  FAIL_TO_PASS 全部通过 + PASS_TO_PASS 不回归
└──────────────┘

import subprocess
import json
from pathlib import Path

class SWEBenchRunner:
    def __init__(self, agent_callable, benchmark_dir: str = "swe-bench"):
        self.agent = agent_callable
        self.bench_dir = Path(benchmark_dir)

    def setup_environment(self, instance_id: str):
        """为指定实例设置 Docker 环境"""
        subprocess.run([
            "python", "-m", "swebench.harness.run_evaluation",
            "--instance_id", instance_id,
            "--output_dir", str(self.bench_dir / "results"),
        ], check=True)

    def run_instance(self, instance: dict) -> dict:
        """运行单个 SWE-bench 实例"""
        # 1. 准备环境
        workdir = self.bench_dir / "workspace" / instance["instance_id"]
        workdir.mkdir(parents=True, exist_ok=True)

        # 2. 让 Agent 工作产生 patch
        agent_result = self.agent(
            problem=instance["problem_statement"],
            repo_path=str(workdir),
            base_commit=instance["base_commit"],
        )

        # 3. 应用 Agent 的 patch 并运行测试
        eval_result = subprocess.run([
            "python", "-m", "swebench.harness.run_evaluation",
            "--predictions_path", agent_result["patch_file"],
            "--swe_bench_tasks", "swe_bench_lite",
            "--log_level", "INFO",
            "--instance_ids", instance["instance_id"],
        ], capture_output=True, text=True)

        # 4. 解析结果
        report_path = self.bench_dir / "results" / instance["instance_id"] / "report.json"
        if report_path.exists():
            report = json.loads(report_path.read_text())
            return {
                "instance_id": instance["instance_id"],
                "resolved": report.get("resolved", False),
                "fail_to_pass": report.get("fail_to_pass", []),
                "pass_to_pass": report.get("pass_to_pass", []),
                "patch_size": len(agent_result.get("patch", "")),
            }
        return {"instance_id": instance["instance_id"], "resolved": False, "error": "no report"}

    def run_suite(self, instances: list[dict]) -> dict:
        results = [self.run_instance(inst) for inst in instances]
        resolved = sum(1 for r in results if r["resolved"])
        return {
            "total": len(results),
            "resolved": resolved,
            "resolve_rate": resolved / len(results) if results else 0,
            "details": results,
        }

┌─────────────────────────────────────────────────────┐
│                   WebArena 环境                      │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ Ecommerce │  │  Forum   │  │   CMS    │          │
│  │ (Magento) │  │(phpBB)   │  │(WordPress)│         │
│  └──────────┘  └──────────┘  └──────────┘          │
│       │              │              │                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ Git Host  │  │ Map      │  │ Analytics│          │
│  │(Gitea)    │  │(OSM)     │  │(Matomo)  │          │
│  └──────────┘  └──────────┘  └──────────┘          │
│                                                       │
│  ┌─────────────────────────────────────────┐        │
│  │          Reverse Proxy (Nginx)           │        │
│  └─────────────────────────────────────────┘        │
└─────────────────────────────────────────────────────┘

class WebArenaEvaluator:
    def __init__(self, config: dict):
        self.environments = config["environments"]
        self.verifiers = self._init_verifiers()

    def _init_verifiers(self):
        return {
            "url_match": self._verify_url,
            "element_exists": self._verify_element,
            "content_match": self._verify_content,
            "api_state": self._verify_api_state,
        }

    def evaluate_task(self, task: dict, agent_actions: list) -> dict:
        """
        评估 Agent 是否完成任务
        task 包含:
        - intent: 任务描述
        - evaluators: 验证条件列表
        - init_state: 初始状态
        """
        results = []
        for verifier_spec in task["evaluators"]:
            v_type = verifier_spec["type"]
            v_config = verifier_spec["config"]
            verifier = self.verifiers[v_type]
            passed = verifier(v_config, agent_actions)
            results.append({
                "verifier": v_type,
                "config": v_config,
                "passed": passed,
            })

        # 所有验证条件都通过才算成功
        success = all(r["passed"] for r in results)
        return {
            "task_id": task["task_id"],
            "intent": task["intent"],
            "success": success,
            "verifier_results": results,
            "num_steps": len(agent_actions),
        }

    def _verify_url(self, config, actions):
        """验证最终 URL 是否匹配"""
        import re
        pattern = config["url_pattern"]
        final_url = actions[-1]["url"] if actions else ""
        return bool(re.match(pattern, final_url))

    def _verify_element(self, config, actions):
        """验证页面上是否存在特定元素"""
        # 通过 API 查询数据库状态或通过浏览器检查 DOM
        selector = config["selector"]
        expected_text = config.get("expected_text", "")
        # 实际实现需要连接浏览器或查询后端
        return True  # 简化示例

    def _verify_content(self, config, actions):
        """验证页面内容是否包含期望文本"""
        expected = config["expected_content"]
        # 获取最终页面内容
        page_content = self._fetch_final_content(actions)
        return expected.lower() in page_content.lower()

    def _verify_api_state(self, config, actions):
        """通过 API 验证后端状态"""
        import requests
        api_url = config["api_url"]
        expected_state = config["expected_state"]
        resp = requests.get(api_url)
        actual_state = resp.json()
        return self._compare_states(actual_state, expected_state)

gaia_task = {
    "task_id": "gaia_001",
    "level": 2,
    "question": "附件 CSV 文件中列出的所有公司，哪家在 2023 年的营收增长率最高？请给出精确百分比。",
    "file_name": "companies.csv",
    "file_path": "/data/gaia/gaia_001/companies.csv",
    "answer": "Acme Corp, 47.3%",
    "annotator_metadata": {
        "required_steps": [
            "读取 CSV 文件",
            "搜索每家公司 2022 和 2023 年营收",
            "计算增长率",
            "比较并找到最高值"
        ],
        "tools_needed": ["file_reader", "web_search", "calculator"],
        "estimated_time_human": "5-10 minutes",
    }
}

class GAIAEvaluator:
    def __init__(self, data_dir: str):
        self.data_dir = Path(data_dir)
        self.tasks = self._load_tasks()

    def _load_tasks(self) -> list[dict]:
        tasks = []
        for json_file in (self.data_dir / "questions").glob("*.json"):
            tasks.append(json.loads(json_file.read_text(encoding="utf-8")))
        return tasks

    def evaluate_response(self, task: dict, agent_response: str) -> dict:
        """
        GAIA 使用精确匹配或包含匹配
        答案通常是简短的字符串（数字、名称、日期等）
        """
        expected = task["answer"].strip().lower()
        actual = agent_response.strip().lower()

        # 精确匹配
        exact_match = expected == actual

        # 包含匹配（允许额外文本）
        contains = expected in actual

        # 数字匹配（提取数字比较）
        number_match = self._compare_numbers(expected, actual)

        passed = exact_match or contains or number_match

        return {
            "task_id": task["task_id"],
            "level": task["level"],
            "passed": passed,
            "expected": task["answer"],
            "actual": agent_response,
            "match_type": "exact" if exact_match else ("contains" if contains else ("number" if number_match else "none")),
        }

    def _compare_numbers(self, expected: str, actual: str) -> bool:
        import re
        exp_nums = re.findall(r"[\d.]+", expected)
        act_nums = re.findall(r"[\d.]+", actual)
        if not exp_nums:
            return False
        try:
            exp_val = float(exp_nums[0])
            act_val = float(act_nums[0]) if act_nums else None
            if act_val is None:
                return False
            # 允许 1% 的误差
            return abs(exp_val - act_val) / max(abs(exp_val), 1e-10) < 0.01
        except ValueError:
            return False

    def benchmark_summary(self, results: list[dict]) -> dict:
        from collections import defaultdict
        by_level = defaultdict(list)
        for r in results:
            by_level[r["level"]].append(r["passed"])
        return {
            "overall_accuracy": sum(r["passed"] for r in results) / len(results),
            "by_level": {
                f"level_{lvl}": {
                    "total": len(passes),
                    "passed": sum(passes),
                    "accuracy": sum(passes) / len(passes),
                }
                for lvl, passes in sorted(by_level.items())
            },
        }

评估矩阵 = {
    "代码能力": "SWE-bench Lite (300题)",
    "Web交互": "WebArena (随机采样200题)",
    "通用推理": "GAIA Level 1+2 (200题)",
}