Harness 驱动

当 Harness 负责 agent 循环时，使用 BaseBenchmark 准备任务，并根据 Harness 返回的 RunResult 评分。

这条路径适用于问答、代码编辑、浏览和其他可以复用现有 Harness 的 Benchmark。Benchmark 不直接调用 Model，也不重新实现 Harness 的会话或工具循环。

​

实现一个完整的精确匹配示例

在 src/agentcompass/benchmarks/ 下创建 example_exact_match.py，内容如下：

from __future__ import annotations

from dataclasses import dataclass, replace

from agentcompass.benchmarks.config import RuntimeBenchmarkConfig
from agentcompass.runtime import (
    BENCHMARKS,
    BaseBenchmark,
    BenchmarkPlan,
    EnvironmentSession,
    EnvironmentSpec,
    ExecutionPlan,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskInput,
    TaskOutput,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.config import config_field, parse_bool
from agentcompass.runtime.metrics import make_metric_contract


@dataclass(slots=True)
class ExampleExactMatchConfig(RuntimeBenchmarkConfig):
    case_sensitive: bool = config_field(
        default=False,
        description="Compare answers with case sensitivity.",
    )

    def __post_init__(self) -> None:
        RuntimeBenchmarkConfig.__post_init__(self)
        self.case_sensitive = parse_bool(self.case_sensitive, "case_sensitive")


@dataclass(slots=True)
class ExampleExactMatchPlan(BenchmarkPlan):
    expected: str = ""
    case_sensitive: bool = False


@BENCHMARKS.register()
class ExampleExactMatchBenchmark(BaseBenchmark):
    id = "example_exact_match"
    description = "One-task exact-match Benchmark used by the developer tutorial."
    config_class = ExampleExactMatchConfig
    evaluation_environment_mode = "none"
    metric_contract = make_metric_contract(
        primary="correct",
        binary=("correct",),
        labels={"correct": "Accuracy"},
    )

    def load_tasks(self, req: RunRequest) -> list[TaskSpec]:
        _ = req
        return [
            TaskSpec(
                task_id="capital-france",
                question=(
                    "What is the capital of France? "
                    "Answer with only the city name."
                ),
                category="geography",
                ground_truth="Paris",
                metadata={"dataset_revision": "tutorial-v1"},
            )
        ]

    def build_plan(
        self,
        task: TaskSpec,
        req: RunRequest,
        environment: EnvironmentSpec,
    ) -> ExampleExactMatchPlan:
        _ = environment
        config = self.build_config(req)
        if not isinstance(config, ExampleExactMatchConfig):
            raise TypeError(
                "example_exact_match requires ExampleExactMatchConfig"
            )
        return ExampleExactMatchPlan(
            expected=str(task.ground_truth),
            case_sensitive=config.case_sensitive,
        )

    async def prepare_task(
        self,
        task: TaskSpec,
        env: EnvironmentSession,
        req: RunRequest,
        plan: BenchmarkPlan,
    ) -> PreparedTask:
        _ = env, req
        self._require_plan(plan)
        return PreparedTask(
            task_id=task.task_id,
            category=task.category,
            ground_truth=None,
            input=TaskInput(prompt=task.question),
            output=TaskOutput(answer="Return only the city name."),
            metadata={
                "dataset_revision": task.metadata["dataset_revision"],
            },
        )

    async def evaluate(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        result: RunResult,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        _ = req, env
        benchmark_plan = self._require_plan(plan.benchmark_plan)
        candidate = str(result.final_answer or "").strip()
        expected = benchmark_plan.expected.strip()
        if not benchmark_plan.case_sensitive:
            candidate = candidate.casefold()
            expected = expected.casefold()

        evaluated = (
            result.status == TaskStatus.COMPLETED and not result.error
        )
        correct = evaluated and candidate == expected
        return replace(
            result,
            task_id=prepared.task_id,
            category=prepared.category,
            metrics={"correct": correct},
        )

    @staticmethod
    def _require_plan(plan: BenchmarkPlan) -> ExampleExactMatchPlan:
        if not isinstance(plan, ExampleExactMatchPlan):
            raise TypeError(
                "example_exact_match requires ExampleExactMatchPlan"
            )
        return plan

这个实现完成了 BaseBenchmark 的三个抽象方法：load_tasks()、prepare_task() 和 evaluate()。其中，build_plan() 将答案保留在评测侧，避免把 ground_truth 交给 Harness；evaluate() 则通过 dataclasses.replace() 保留 Harness 写入的状态、错误、轨迹、产物和其他结果字段。

​

导出并检查注册

在 src/agentcompass/benchmarks/__init__.py 中添加导入：

from .example_exact_match import ExampleExactMatchBenchmark

检查注册和配置结构：

uv run agentcompass list benchmark
uv run agentcompass config docs benchmark example_exact_match

第一条命令应列出 example_exact_match；第二条命令应显示 case_sensitive 及 RuntimeBenchmarkConfig 的共享字段。如果组件未出现，先检查 __init__.py、重复 ID 和完整导入堆栈。

​

使用 Harness 运行

下面使用 Harness 实现教程中的 example_answer。该 Harness 直接返回配置的 final_answer，因此这次冒烟运行不需要可用的 Model 端点：

uv run agentcompass run example_exact_match example_answer unused-model \
  --env host_process \
  --benchmark-params '{"sample_ids":["capital-france"]}' \
  --harness-params '{"answer":"Paris"}' \
  --task-concurrency 1 \
  --no-enable-analysis \
  --results-dir results-dev \
  --run-name benchmark-smoke

在 run_info.json 中依次检查 request → benchmark → id、resolved_execution_plans，以及 details/*.json 中唯一的 attempt 记录。答案为 Paris 时，应得到 status: "completed" 和 metrics.correct: true；改为 Lyon 后，status 仍应是 completed，但 metrics.correct 应变为 false。这说明执行状态和 Benchmark 判定是两个独立维度。

​

接入真实数据时如何扩展

• 数据集读取、版本校验和稳定任务转换放在 load_tasks()，不要放在模块导入阶段。
• 逐任务评测器状态、超时或路径放在类型化 BenchmarkPlan，不要通过共享可变字典在不同尝试之间传递。
• 提示词、工作区和公开附件放在 PreparedTask；隐藏测试、答案和参考补丁不得进入 Harness 可见字段。
• 需要任务 Environment 或隔离验证器时，不要继续堆叠在这个 none 示例中；改用评测模式与产物中的 reuse 或 fresh 结构。
• 评分不是简单布尔值时，应把标量 score 声明为 Contract 主观测，并写入 RunResult.metrics；详见结果与聚合。

需要对照一个 Harness 驱动、进程内评分的生产实现时，可参考 browsecomp.py。