> ## Documentation Index > Fetch the complete documentation index at: https://agent-compass.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # 开发者指南 > 扩展 benchmark、harness、environment、recipe 和 analyzer 的开发入口。 扩展 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` 的实现。 ```python theme={null} 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` 中导出： ```python theme={null} from .my_benchmark import MyBenchmark # noqa: F401 ``` 确认组件可以被发现，并跑一个最小样例： ```bash theme={null} uv run agentcompass list benchmark ``` ```bash theme={null} agentcompass run \ my_benchmark \ openai_chat \ your-model \ --env \ --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` 的实现。 ```python theme={null} 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` 中导出： ```python theme={null} from .my_harness import MyHarness # noqa: F401 ``` 然后确认发现和 smoke run： ```bash theme={null} uv run agentcompass list harness ``` ```bash theme={null} agentcompass run \ scicode \ my_harness \ your-model \ --env \ --benchmark-params '{"sample_ids":[""]}' \ --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` 实现。 ```python theme={null} 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： ```python theme={null} try: from .my_environment import MyEnvironment # noqa: F401 except ModuleNotFoundError as exc: if exc.name != "my_provider_sdk": raise ``` 然后编译 provider，并用一个最小任务验证命令和文件 primitive： ```bash theme={null} python -m compileall src/agentcompass/environments/my_environment.py ``` ```bash theme={null} agentcompass run \ terminal_bench_2 \ terminus2 \ your-model \ --env my_environment \ --benchmark-params '{"sample_ids":[""]}' \ --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//` 下新增模块。不同 provider 的行为差异较大时，推荐一个 provider 一个文件。 ```python theme={null} 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//common.py`。 | 当 `execution.enabled_recipes` 为空时，recipe 会自动匹配。用户也可以通过 `--enabled-recipes` 限定 recipe 选择。 ### 注册与验证 从 package initializer 导出 recipe： ```python theme={null} # 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： ```bash theme={null} agentcompass run \ my_benchmark \ my_harness \ your-model \ --env \ --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.` 下。跨任务聚合字段会渲染到 `analysis_summary.md` 和 `analysis_summary.json`。 当逻辑属于后置诊断或统计时，新增 analyzer。不要把可重新运行的诊断逻辑塞进 benchmark 或 harness。 ### Analyzer 文件 在 `src/agentcompass/analyzers/basic/` 下新增文件；如果是一组较大的 analyzer，也可以创建新的 sub-package。每个 analyzer 都要继承 `BaseAnalyzer`，使用 `@ANALYZERS.register()` 注册，并实现 `analysis()`。 ```python theme={null} 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.` 合并来的 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 能看到它： ```python theme={null} # 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 ``` 确认它已经出现在组件列表中： ```bash theme={null} uv run agentcompass list analyzer ``` 在 evaluation 中启用： ```bash theme={null} agentcompass run \ terminal_bench_2 \ terminus2 \ your-model \ --env \ --benchmark-params '{"sample_ids":[""]}' \ --model-base-url "$MODEL_BASE_URL" \ --model-api-key "$MODEL_API_KEY" \ --enable-analysis \ --analysis-params '{"analyzers":["MyAnalyzer"]}' ``` 或对已有结果重新运行： ```bash theme={null} agentcompass analysis --input results/// --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。