Benchmark 驱动

如果上游评测要求使用专用的多角色交互、用户模拟器或领域状态机，可通过 HarnessFreeBenchmark 让 Benchmark 自己执行这套循环。

不要仅仅为了少写一个 Harness 而选择这条路径。如果一套 agent 循环可以服务多个 Benchmark，仍应将它实现为 Harness；只有协议、终止条件和轨迹语义都由当前 Benchmark 定义时，才应由 Benchmark 负责执行。

​

与普通 Benchmark 的区别

HarnessFreeBenchmark 继承 BaseBenchmark 的所有契约，并额外要求 run_task()：

prepare_task
  → Benchmark.run_task
  → collect_artifacts
  → evaluate

运行命令中的 Harness ID 使用 none：

uv run agentcompass run example_interactive none MODEL \
  --env docker

这里的 none 表示没有外部 Harness，不代表 evaluation_environment_mode 必须是 none。Benchmark 驱动的执行仍然可以选择 reuse 或 fresh 评测。

​

实现 run_task()

下面的片段展示与 BaseBenchmark 不同的部分。假设 prepare_task() 已在任务 Environment 中写入上游运行器的请求文件，并把公开路径放入 PreparedTask.metadata 的 official_runner 条目：

import json

from agentcompass.runtime import (
    EnvironmentSession,
    ExecutionPlan,
    HarnessFreeBenchmark,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.metrics import make_metric_contract


class ExampleInteractiveBenchmark(HarnessFreeBenchmark):
    evaluation_environment_mode = "reuse"
    metric_contract = make_metric_contract(
        primary="score",
        scalar=("score",),
        labels={"score": "Reward"},
    )

    async def run_task(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        _ = req, plan
        if env is None:
            raise RuntimeError(
                "example_interactive requires an EnvironmentSession"
            )

        runner = prepared.metadata["official_runner"]
        execution = await env.exec(
            [
                "python3",
                "-m",
                "example_interactive.runner",
                "run",
                "--request",
                runner["request_path"],
                "--output",
                runner["result_path"],
            ],
            timeout=runner["timeout_seconds"],
        )
        if execution.returncode != 0 or execution.timed_out:
            detail = (execution.stderr or execution.stdout).strip()
            return RunResult(
                task_id=task.task_id,
                category=task.category,
                status=TaskStatus.RUN_ERROR,
                error=detail or "official runner failed",
            )

        payload = json.loads(
            await env.read_text(runner["result_path"])
        )
        return RunResult(
            task_id=task.task_id,
            category=task.category,
            status=TaskStatus.COMPLETED,
            final_answer=payload.get("final_answer"),
            artifacts={"official_result": payload},
        )

这个片段不能独立运行，因为数据加载、任务准备和上游运行器都取决于具体协议。它只用于说明职责边界：runtime 管理 Environment 生命周期和错误流程，run_task() 负责执行 Benchmark 专属循环，并将输出转换为 RunResult。

​

保持执行与评分分离

即使上游运行器已经生成 reward 值，也应由 evaluate() 将其转换为最终判定，不要在 run_task() 中同时完成评分。这样 runtime 才能继续区分执行失败和评测失败：

from dataclasses import replace


async def evaluate(
    self,
    task: TaskSpec,
    prepared: PreparedTask,
    result: RunResult,
    req: RunRequest,
    plan: ExecutionPlan,
    env: EnvironmentSession | None = None,
) -> RunResult:
    _ = task, prepared, req, plan
    if result.status != TaskStatus.COMPLETED or result.error:
        return replace(result, metrics={})
    if env is None:
        raise RuntimeError("evaluation requires the task Environment")

    payload = result.artifacts["official_result"]
    reward = float(payload["reward"])
    return replace(
        result,
        metrics={"score": reward},
    )

如果需要运行另一个命令才能得到 reward 值，请在 evaluate() 中使用传入的 env 执行验证器，并将验证器崩溃标记为 EVAL_ERROR。Environment 的具体生命周期见评测模式与产物，状态组合见结果与聚合。

​

职责分工

• prepare_task() 负责把上游运行器、数据和请求材料放入 Environment；重试时重复执行也不能破坏任务状态。
• run_task() 负责 Benchmark 专属交互循环、Model 调用编排和轨迹转换，但不得自行创建或关闭 Environment。
• collect_artifacts() 只提取任务 Environment 中的提交物，不做评分。
• evaluate() 负责解释运行器输出或调用官方验证器，同时保留已有的执行错误。
• Model、模拟器 Model 或评审 Model 的凭证只能通过既有的 Model 配置边界传递，不得写入结果、日志或可公开的元数据。

完整的生产实现可参考 TauBenchBenchmark。它使用 none Harness，在 run_task() 中启动上游交互运行器，并通过 reuse 模式在同一个任务 Environment 中评测。