开发者指南 - AgentCompass Documentation

扩展 AgentCompass 时，优先遵守现有 runtime 边界：benchmark 负责任务和评分，harness 负责 agent 执行，environment 负责执行 primitive，recipe 负责 provider-specific 兼容性，analyzer 负责后置诊断。

添加 Benchmark

Benchmark 负责数据集协议、任务构造、任务准备、评分逻辑和指标聚合。它不应该负责模型调用、agent loop、sandbox 生命周期，也不应该硬编码某个 provider 的镜像选择。当你要引入新的任务族、新的评测协议或新的 scoring 逻辑时，添加 benchmark。如果只是要把现有 benchmark 适配到某个远程 sandbox 的镜像、workspace 或资源参数，应该添加 recipe。

Benchmark 文件

在 src/agentcompass/benchmarks/ 下新增模块。通常包含一个 RuntimeBenchmarkConfig、一个可选 BenchmarkPlan，以及一个继承 BaseBenchmark 的实现。

from dataclasses import dataclass

from agentcompass.benchmarks.config import RuntimeBenchmarkConfig
from agentcompass.runtime.base import BaseBenchmark, EnvironmentSession
from agentcompass.runtime.models import (
    BenchmarkPlan,
    EnvironmentSpec,
    ExecutionPlan,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskInput,
    TaskOutput,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.registry import BENCHMARKS


@dataclass(slots=True)
class MyBenchmarkConfig(RuntimeBenchmarkConfig):
    dataset_path: str = "data/my_benchmark/tasks.jsonl"


@dataclass(slots=True)
class MyBenchmarkPlan(BenchmarkPlan):
    workspace_root: str = "workspace/"


@BENCHMARKS.register()
class MyBenchmark(BaseBenchmark):
    id = "my_benchmark"
    description = "My Benchmark: one-sentence task and scoring description (paper or project URL)."
    config_class = MyBenchmarkConfig
    evaluation_environment_mode = "none"

    def load_tasks(self, req: RunRequest) -> list[TaskSpec]:
        config = self.build_config(req)
        return [
            TaskSpec(
                task_id="sample-1",
                question="...",
                category="default",
                ground_truth="...",
                metadata={"source": config.dataset_path},
            )
        ]

    def build_plan(self, task: TaskSpec, req: RunRequest, environment: EnvironmentSpec) -> MyBenchmarkPlan:
        return MyBenchmarkPlan()

    async def prepare_task(
        self,
        task: TaskSpec,
        env: EnvironmentSession,
        req: RunRequest,
        plan: MyBenchmarkPlan,
    ) -> PreparedTask:
        return PreparedTask(
            task_id=task.task_id,
            category=task.category,
            ground_truth=task.ground_truth,
            input=TaskInput(prompt=task.question, workspace=plan.workspace_root),
            output=TaskOutput(),
            metadata=dict(task.metadata),
        )

    async def evaluate(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        result: RunResult,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        result.correct = result.final_answer == prepared.ground_truth
        result.score = 1.0 if result.correct else 0.0
        result.status = TaskStatus.COMPLETED if not result.error else TaskStatus.EVAL_ERROR
        return result

Benchmark 职责

Hook	作用
`id`	稳定 CLI/config id，也会被 recipe、analyzer、结果路径和文档引用。
`description`	`agentcompass list dump` 导出的用户可见描述。Benchmark 和 harness 必须提供非空 description。
`config_class`	解析 `--benchmark-params` 和默认配置。数据路径、版本、split、采样参数应放在这里。
`load_tasks()`	加载原始任务并返回 `TaskSpec`。应保持确定性，并在数据缺失时尽早失败。
`select_tasks()`	可选自定义采样逻辑。基类已经支持 `sample_ids`。
`build_plan()`	为单个 task 生成 benchmark-side runtime state，后续 recipe 可以改写它。
`prepare_task()`	将 `TaskSpec` 转成 benchmark-to-harness 的 `PreparedTask` 契约。
`evaluate()`	对 harness 的 `RunResult` 评分，并返回标准化后的最终结果。
`aggregate_metrics()`	可选聚合逻辑。默认路径会聚合 binary correctness。

TaskSpec.metadata 适合保存原始记录、镜像名、expected files、benchmark-specific id 等信息；PreparedTask.input 和 PreparedTask.output 才是 harness 应消费的正式契约。

注册与验证

在 src/agentcompass/benchmarks/__init__.py 中导出：

from .my_benchmark import MyBenchmark  # noqa: F401

确认组件可以被发现，并跑一个最小样例：

uv run agentcompass list benchmark

agentcompass run \
  my_benchmark \
  openai_chat \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

添加 Harness

Harness 负责 agent 或 model 的执行 loop。它消费 PreparedTask、RunRequest.model 和 EnvironmentSession，并返回标准化的 RunResult。它不应该加载数据集、判断 benchmark correctness，或硬编码某个 provider 的 sandbox 镜像。当你要接入新的 agent framework、coding assistant、GUI agent、direct model-call path 或 tool-use 策略时，添加 harness。

Harness 文件

在 src/agentcompass/harnesses/ 下新增模块。通常包含一个用户可配置的 RuntimeHarnessConfig、一个 HarnessPlan，以及一个继承 BaseHarness 的实现。

from dataclasses import dataclass
from typing import Any

from agentcompass.runtime.base import BaseHarness, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeHarnessConfig
from agentcompass.runtime.models import (
    EnvironmentSpec,
    HarnessPlan,
    ModelSpec,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskStatus,
)
from agentcompass.runtime.registry import HARNESSES


@dataclass(slots=True)
class MyHarnessConfig(RuntimeHarnessConfig):
    max_steps: int = 30


@dataclass(slots=True)
class MyHarnessPlan(HarnessPlan):
    max_steps: int = 30


@HARNESSES.register()
class MyHarness(BaseHarness):
    id = "my_harness"
    description = "Runs MyAgent against text, file, or workspace tasks (official website: https://...)."
    config_class = MyHarnessConfig
    plan_class = MyHarnessPlan

    def supports(self, environment: EnvironmentSpec, model: ModelSpec) -> bool:
        return environment.id in {"host_process", "docker", "daytona", "modal"}

    async def start_session(self, env: EnvironmentSession, req: RunRequest, plan: MyHarnessPlan) -> dict[str, Any]:
        return {"env": env, "model": req.model.id, "max_steps": plan.max_steps}

    async def run_task(
        self,
        session: dict[str, Any],
        prepared: PreparedTask,
        req: RunRequest,
        plan: MyHarnessPlan,
    ) -> RunResult:
        return RunResult(
            task_id=prepared.task_id,
            category=prepared.category,
            status=TaskStatus.COMPLETED,
            final_answer="...",
            ground_truth=prepared.ground_truth,
            artifacts={"workspace": prepared.input.workspace},
        )

    async def close_session(self, session: dict[str, Any]) -> None:
        return None

Harness 职责

Hook	作用
`id` / `description`	稳定 CLI id 和导出描述。已有 official website 时应写入 description。
`supports()`	在执行前校验 environment 和 model protocol 是否兼容。协议不支持时应给出清晰错误。
`config_class`	解析 `--harness-params` 和默认配置。公开参数不要散落在 metadata 里。
`plan_class`	由 config 派生出的 runtime state。最终 plan 可被 recipe 查看或改写。
`start_session()`	初始化 client、agent config、后台进程或每次 run 的状态。
`run_task()`	执行一个 `PreparedTask`，返回 `RunResult`，尽量保留 trajectory、artifact、usage 和 error。
`close_session()`	释放外部进程、client、临时 session 或远程框架状态。

Harness 应消费 prepared.input.workspace、files、media、tools、messages 等正式字段，而不是反向依赖某个 benchmark 的私有 metadata。这样同一个 harness 才能复用到多个 benchmark。

注册与验证

在 src/agentcompass/harnesses/__init__.py 中导出：

from .my_harness import MyHarness  # noqa: F401

然后确认发现和 smoke run：

uv run agentcompass list harness

agentcompass run \
  scicode \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

添加 Environment

Environment provider 负责执行 primitive：命令执行、文件传输、文本读写、目录同步、endpoint discovery 和 teardown。它不应该知道 benchmark scoring，也不应该处理 agent 行为。当 AgentCompass 需要在新的本地进程、容器运行时、云 sandbox 或远程执行 provider 上运行同一套 benchmark/harness 矩阵时，添加 environment。

Environment 文件

在 src/agentcompass/environments/ 下新增模块。每个 provider 通常包含 session class、config dataclass 和 BaseEnvironment 实现。

from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agentcompass.runtime.base import BaseEnvironment, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeEnvironmentConfig
from agentcompass.runtime.models import ExecResult, ExecutionPlan, RunRequest
from agentcompass.runtime.registry import ENVIRONMENTS


class MyEnvironmentSession(EnvironmentSession):
    default_workspace_root = "/workspace/"

    async def exec(
        self,
        command: list[str] | str,
        *,
        shell: bool = False,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        timeout: float | None = None,
        detach: bool = False,
        flags: dict[str, Any] | None = None,
    ) -> ExecResult:
        command = self._validate_exec_command(command, shell=shell)
        return ExecResult(returncode=0, stdout="", stderr="")

    async def upload(self, src: str, dst: str) -> None: ...
    async def download(self, src: str, dst: str) -> None: ...
    async def write_text(self, path: str, content: str) -> None: ...
    async def read_text(self, path: str) -> str: ...
    async def upload_dir(self, src: Path | str, dst: str) -> None: ...
    async def download_dir(self, src: str, dst: Path | str) -> None: ...
    async def endpoint(self) -> str | None: ...


@dataclass(slots=True)
class MyEnvironmentConfig(RuntimeEnvironmentConfig):
    image: str = "python:3.12-slim"
    default_workspace_root: str = "/workspace/"


@ENVIRONMENTS.register()
class MyEnvironment(BaseEnvironment):
    id = "my_environment"
    config_class = MyEnvironmentConfig
    default_workspace_root = "/workspace/"

    async def open(self, req: RunRequest, plan: ExecutionPlan) -> MyEnvironmentSession:
        config = self.build_config(req, plan)
        return MyEnvironmentSession()

    async def close(self, env: EnvironmentSession) -> None:
        return None

Environment 职责

Surface	作用
`exec()`	实现 subprocess-like 语义，处理 `shell`、`cwd`、`env`、`timeout`、`detach`，并返回 `ExecResult`。
File primitives	实现 `upload`、`download`、`write_text`、`read_text`、`upload_dir`、`download_dir`。
`endpoint()`	provider 暴露可访问 URL 时返回 URL，否则返回 `None`。
`RuntimeEnvironmentConfig`	文档化 image、resources、region、credentials、workspace root 等公开参数。
`open()`	从 recipe-adjusted `ExecutionPlan` 构建 provider session。
`close()`	稳定释放远程资源，并容忍部分初始化失败的 session。

通用 provider 默认值可以放进 config/defaults.yaml。Benchmark-specific image、snapshot、workspace 或 resource 映射应放在 recipe 中。

注册与验证

在 src/agentcompass/environments/__init__.py 中导出。可选 SDK 依赖需要用 import guard：

try:
    from .my_environment import MyEnvironment  # noqa: F401
except ModuleNotFoundError as exc:
    if exc.name != "my_provider_sdk":
        raise

然后编译 provider，并用一个最小任务验证命令和文件 primitive：

python -m compileall src/agentcompass/environments/my_environment.py

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env my_environment \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

添加 Recipe

Recipe 是 benchmark family 和 environment provider 之间的兼容层。它在 sandbox 启动前改写单个 task 的 ExecutionPlan，通常用于选择镜像、workspace、resource request、snapshot、环境变量或 evaluation layout。当 task metadata 已经包含 provider 需要的信息，并且用户不应该在 CLI 手动传入这些参数时，添加 recipe。

Recipe 文件

在 src/agentcompass/recipes/<benchmark_family>/ 下新增模块。不同 provider 的行为差异较大时，推荐一个 provider 一个文件。

from copy import deepcopy

from agentcompass.runtime.base import BaseRecipe
from agentcompass.runtime.models import ExecutionPlan, RunRequest, TaskSpec
from agentcompass.runtime.registry import RECIPES


@RECIPES.register()
class MyBenchmarkModalRecipe(BaseRecipe):
    id = "my_benchmark_modal"
    enabled_by_default = True

    def matches(self, req: RunRequest, task: TaskSpec, plan: ExecutionPlan) -> bool:
        return req.benchmark.id == "my_benchmark" and req.environment.id == "modal"

    def apply(self, plan: ExecutionPlan, req: RunRequest, task: TaskSpec) -> ExecutionPlan:
        _ = req
        updated = deepcopy(plan)
        params = dict(updated.environment.params)

        user_image = str(params.get("image") or "").strip()
        task_image = str(task.metadata.get("docker_image") or "").strip()
        if not user_image and not task_image:
            raise ValueError(f"{self.id} requires image or task.metadata.docker_image")

        params.setdefault("image", task_image)
        params.setdefault("default_workspace_root", "/workspace")
        updated.environment.params = params
        return updated

Recipe 职责

规则	原因
Match narrowly	检查 benchmark id、environment id 和必要 task metadata，避免误触发。
Preserve user overrides	用户显式传入 image、resources、workspace、credentials 时，优先保留或做兼容性检查。
Fail before sandbox startup	重型 benchmark 缺少必要 task image 时，在 `apply()` 中给出清晰 `ValueError`。
Rewrite plans, not configs	改写复制后的 `ExecutionPlan`，不要修改原始 `RunRequest`。
Keep execution out	Recipe 不应运行命令、创建 session、评分结果或调用模型。
Share helpers by benchmark family	多个 provider 共享 workspace layout 时，放到 `recipes/<benchmark_family>/common.py`。

当 execution.enabled_recipes 为空时，recipe 会自动匹配。用户也可以通过 --enabled-recipes 限定 recipe 选择。

注册与验证

从 package initializer 导出 recipe：

# src/agentcompass/recipes/my_benchmark/__init__.py
from .modal import MyBenchmarkModalRecipe  # noqa: F401

# src/agentcompass/recipes/__init__.py
from .my_benchmark import MyBenchmarkModalRecipe  # noqa: F401

运行单个样例，并确认日志中出现 Recipe matched 和新 recipe id：

agentcompass run \
  my_benchmark \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

添加 Analyzer

Analyzer 在任务生成 RunResult 之后运行。它检查 trajectory、metrics、error、latency、模型输出或 tool calls，并把 AnalysisResult 写到每个 details JSON 的 analysis_result.<AnalyzerId> 下。跨任务聚合字段会渲染到 analysis_summary.md 和 analysis_summary.json。当逻辑属于后置诊断或统计时，新增 analyzer。不要把可重新运行的诊断逻辑塞进 benchmark 或 harness。

Analyzer 文件

在 src/agentcompass/analyzers/basic/ 下新增文件；如果是一组较大的 analyzer，也可以创建新的 sub-package。每个 analyzer 都要继承 BaseAnalyzer，使用 @ANALYZERS.register() 注册，并实现 analysis()。

from agentcompass.runtime.base import BaseAnalyzer
from agentcompass.runtime.models import AnalysisResult, AnalyzerCategory, RunResult
from agentcompass.runtime.registry import ANALYZERS


@ANALYZERS.register()
class MyAnalyzer(BaseAnalyzer):
    id = "MyAnalyzer"
    description = "Detects a custom trajectory pattern."
    category = AnalyzerCategory.BEHAVIOR
    datasets = []
    data_requirements = []
    conf = {"only_incorrect": False, "threshold": 0.0}
    distribution_fields = {
        "total_steps": "numeric_stats",
        "pattern_types": "value_counts",
    }

    async def analysis(self, task, prepared, result: RunResult, req, plan) -> AnalysisResult:
        if result is None or result.trajectory is None:
            return AnalysisResult(
                task_id=task.task_id,
                is_badcase=None,
                error="no trajectory available",
            )

        total_steps = len(result.trajectory.steps or [])

        return AnalysisResult(
            task_id=task.task_id,
            is_badcase=False,
            score=None,
            details={
                "total_steps": total_steps,
                "pattern_types": [],
            },
        )

Analyzer 属性

Attribute	作用
`id`	稳定 analyzer id，用于 registry、CLI selection、details JSON 和 summary。
`description`	用户可见描述，会被 `agentcompass list dump` 导出。
`category`	summary 分组。可使用 `AnalyzerCategory.ERROR`、`EFFICIENCY`、`BEHAVIOR`、`ABILITY`、`BASIC_BADCASE`、`ENV_FRAMEWORK_ERROR`。
`datasets`	支持的 benchmark id 列表。空列表表示所有 benchmark。
`data_requirements`	analysis 前检查的 JSONPath requirements；缺失字段时跳过该样本。
`conf`	从 `execution.analysis_params.<AnalyzerId>` 合并来的 analyzer 配置。常见字段包括 `only_incorrect` 和 `threshold`。
`distribution_fields`	需要跨任务聚合的 details 字段。值为 `numeric_stats` 或 `value_counts`。
`base_analyzer` / `priority`	benchmark-specific analyzer 覆盖 generic analyzer 的机制。同一 family 中 priority 更高者生效。

AnalysisResult.is_badcase 应设置为：True 表示 badcase，False 表示明确通过，None 表示 statistics-only analyzer，不参与 badcase ratio。

聚合字段

需要进入跨任务 summary 的字段必须声明在 distribution_fields 中：

Method	适用字段	Summary 输出
`numeric_stats`	数值或数值列表。	count、min、mean、percentiles 和 max。
`value_counts`	字符串或字符串列表。	value frequency 和 ratio。

只有 AnalysisResult.details 中返回的 key 才能被聚合。

注册与验证

把 analyzer 从 package initializer 导出，确保 registry import side effects 能看到它：

# src/agentcompass/analyzers/basic/__init__.py
from agentcompass.analyzers.basic.my_analyzer import MyAnalyzer  # noqa: F401

# src/agentcompass/analyzers/__init__.py
from agentcompass.analyzers.basic import MyAnalyzer  # noqa: F401

确认它已经出现在组件列表中：

uv run agentcompass list analyzer

在 evaluation 中启用：

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY" \
  --enable-analysis \
  --analysis-params '{"analyzers":["MyAnalyzer"]}'

或对已有结果重新运行：

agentcompass analysis --input results/<benchmark>/<model>/<run_id> --analysis-params '{"analyzers":["MyAnalyzer"]}' --task_concurrency 8

开发者说明

Analyzer 应只读取和诊断已有结果，不应修改 benchmark 执行行为。
Benchmark-specific analyzer 应优先使用 datasets 限定适用范围。
当 benchmark-specific analyzer 需要覆盖 generic analyzer 时，使用 base_analyzer 和 priority。
如果缺少可选 trajectory 字段，返回带 error 的 AnalysisResult，不要直接抛异常中断整批 analysis。

​添加 Benchmark

​Benchmark 文件

​Benchmark 职责

​注册与验证

​添加 Harness

​Harness 文件

​Harness 职责

​注册与验证

​添加 Environment

​Environment 文件

​Environment 职责

​注册与验证

​添加 Recipe

​Recipe 文件

​Recipe 职责

​注册与验证

​添加 Analyzer

​Analyzer 文件

​Analyzer 属性

​聚合字段

​注册与验证

​开发者说明

添加 Benchmark

Benchmark 文件

Benchmark 职责

注册与验证

添加 Harness

Harness 文件

Harness 职责

注册与验证

添加 Environment

Environment 文件

Environment 职责

注册与验证

添加 Recipe

Recipe 文件

Recipe 职责

注册与验证

添加 Analyzer

Analyzer 文件

Analyzer 属性

聚合字段

注册与验证

开发者说明