Python API

Reference for the agent_eval package, generated from the source docstrings and type hints. Most users drive the harness through the slash commands and eval.yaml — this page is for extending it (for example, writing a custom runner) or embedding it in your own tooling.

Autodoc

These entries are rendered by mkdocstrings. The documentation build runs pip install -e . so the package is importable; add more ::: directives to this page to surface additional modules.

Configuration

The entire eval.yaml surface is parsed into EvalConfig. See the eval.yaml reference for the YAML-level documentation of every field.

agent_eval.config.EvalConfig dataclass

EvalConfig(name='', description='', skill=None, permissions=dict(), hooks=HooksConfig(), execution=ExecutionConfig(), runner=RunnerConfig(), models=ModelsConfig(), mlflow=MlflowConfig(), dataset=DatasetConfig(), generation=GenerationConfig(), outputs=list(), inputs=InputsConfig(), traces=TracesConfig(), judges=list(), reward=None, thresholds=dict(), config_dir=None, config_path=None, model='', subagent_model='', run_id='', baseline='')

Complete evaluation suite configuration.

Structure is schema-driven: dataset and output structures are described in natural language. The harness interprets these descriptions via LLM (once, cached) to drive prepare, collect, and score steps.

project_root property

project_root

Project root directory (always CWD, not the eval.yaml location).

eval_name

eval_name()

Derive eval identifier with backward-compatible fallback chain.

Priority order (backward-compatible with existing skill evals): 1. skill field - preserves existing skill-based eval runs 2. name field - allows explicit naming for prompt-mode evals 3. directory/filename - pure path-based derivation 4. "eval" - final fallback

This ensures existing skill evals continue to work while enabling prompt mode to use either explicit names or path-based identifiers.

Source code in agent_eval/config.py

def eval_name(self) -> str:
    """Derive eval identifier with backward-compatible fallback chain.

    Priority order (backward-compatible with existing skill evals):
    1. skill field - preserves existing skill-based eval runs
    2. name field - allows explicit naming for prompt-mode evals
    3. directory/filename - pure path-based derivation
    4. "eval" - final fallback

    This ensures existing skill evals continue to work while enabling
    prompt mode to use either explicit names or path-based identifiers.
    """
    # Priority 1: skill field (backward compat with existing evals).
    # Resolve through resolve_skill() so execution.skill-only configs
    # still name the run after the skill under test.
    skill = self.resolve_skill()
    if skill:
        return skill

    # Priority 2: name field (explicit identifier, sanitized)
    # Skip if name == path.stem (auto-set default from from_yaml)
    if self.name and not (self.config_path and self.name == self.config_path.stem):
        # Sanitize: convert spaces to hyphens, keep only safe chars
        sanitized = self.name.lower().replace(" ", "-")
        sanitized = "".join(c for c in sanitized if c.isalnum() or c in "._-")
        if sanitized and _is_valid_eval_name(sanitized):
            return sanitized

    # Priority 3: derive from path (new behavior for prompt mode)
    if self.config_path:
        if self.config_path.name == "eval.yaml":
            # Nested: eval/user-guides/eval.yaml → "user-guides"
            # Check if grandparent directory is named "eval"
            if self.config_path.parent.parent.name == "eval":
                return self.config_path.parent.name
            # Root: eval.yaml at project root → "eval"
            else:
                return "eval"
        # Flat: eval/user-guides.yaml → "user-guides"
        else:
            return self.config_path.stem

    # Final fallback
    return "eval"

from_yaml classmethod

from_yaml(path)

Load config from a YAML file.

Source code in agent_eval/config.py

@classmethod
def from_yaml(cls, path: str | Path) -> "EvalConfig":
    """Load config from a YAML file."""
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"Config not found: {path}")

    with open(path) as f:
        raw = yaml.safe_load(f) or {}

    # Deprecation: top-level `skill:` is auto-normalized into
    # execution.skill (below) but the canonical home is the execution
    # block, symmetric with execution.prompt. Warn once per load; only
    # for a non-empty value that isn't already mirrored in execution.
    exec_raw = raw.get("execution", {})
    if raw.get("skill") and not (exec_raw.get("skill") or "").strip():
        import warnings
        warnings.warn(
            f"Top-level 'skill:' in {path} is deprecated; move it under "
            "execution.skill (it is auto-normalized for now and will be "
            "removed in a future release).",
            DeprecationWarning,
            stacklevel=2,
        )

    # Dataset
    dataset = raw.get("dataset", {})

    # Execution config
    execution = ExecutionConfig(
        mode=exec_raw.get("mode", "case"),
        skill=exec_raw.get("skill", "") or raw.get("skill", ""),
        prompt=exec_raw.get("prompt", ""),
        arguments=exec_raw.get("arguments", ""),
        timeout=exec_raw.get("timeout"),
        max_budget_usd=exec_raw.get("max_budget_usd"),
        parallelism=exec_raw.get("parallelism"),
        env=exec_raw.get("env") or {},
    )

    # Runner config (block form)
    runner_raw = raw.get("runner") or {}
    command = runner_raw.get("command")
    if command is not None:
        valid_list = isinstance(command, list) and all(
            isinstance(x, str) for x in command
        )
        if not (isinstance(command, str) or valid_list):
            raise ValueError("runner.command must be a string or list of strings")
    # Validate workspace_mode (prevent typos that silently change behavior)
    workspace_mode = runner_raw.get("workspace_mode")
    if workspace_mode is not None and workspace_mode not in ("repo",):
        raise ValueError(
            f"runner.workspace_mode must be None or 'repo', got: {workspace_mode!r}")

    runner = RunnerConfig(
        type=runner_raw.get("type", "claude-code"),
        command=command,
        workspace_mode=workspace_mode,
        settings=runner_raw.get("settings", {}) or {},
        plugin_dirs=runner_raw.get("plugin_dirs", []) or [],
        env=runner_raw.get("env", {}) or {},
        system_prompt=runner_raw.get("system_prompt"),
        effort=runner_raw.get("effort"),
    )

    # Models block
    models_raw = raw.get("models", {}) or {}
    models = ModelsConfig(
        skill=models_raw.get("skill"),
        subagent=models_raw.get("subagent"),
        judge=models_raw.get("judge"),
        hook=models_raw.get("hook"),
    )

    # MLflow block. Experiment defaults to the eval's top-level
    # `name` only when an `mlflow:` block is present — so omitting
    # the block entirely leaves MLflow off (no accidental experiment
    # creation on shared tracking servers).
    has_mlflow_block = "mlflow" in raw and raw["mlflow"] is not None
    mlflow_raw = raw.get("mlflow") or {}
    if has_mlflow_block:
        experiment = mlflow_raw.get("experiment") or raw.get("name", "")
    else:
        experiment = ""
    mlflow = MlflowConfig(
        experiment=experiment,
        tracking_uri=mlflow_raw.get("tracking_uri"),
        tags=mlflow_raw.get("tags", {}) or {},
    )

    # Dataset — path, schema, and workspace file provisioning
    ws_raw = dataset.get("workspace", {}) or {}
    ws_files_raw = ws_raw.get("files", []) or []
    ws_files = []
    for i, f in enumerate(ws_files_raw):
        if not isinstance(f, str):
            raise ValueError(
                f"dataset.workspace.files[{i}] must be a string, got {type(f).__name__}"
            )
        ws_files.append(
            _validate_relative_path(f.rstrip("/"), "dataset.workspace.files")
        )
    dataset_config = DatasetConfig(
        path=_validate_relative_path(
            dataset.get("path", ""), "dataset.path", allow_absolute=True
        ),
        schema=dataset.get("schema", ""),
        workspace=WorkspaceConfig(files=ws_files),
    )
    # Generation — synthetic test-case generation (optional) with validation
    gen_raw = raw.get("generation") or {}
    seeds = []
    for i, s in enumerate(gen_raw.get("seeds") or []):
        category = s.get("category", "")
        count = s.get("count")
        if not category or not isinstance(category, str):
            raise ValueError(
                f"generation.seeds[{i}].category must be a non-empty string, got: {category!r}")
        # count is required — a silent default would swallow a mistyped field name
        if not isinstance(count, int) or count < 1:
            raise ValueError(
                f"generation.seeds[{i}].count must be an integer >= 1, got: {count!r}")

        # Exactly one prompt discriminator (mirrors judges: builtin/prompt_file/prompt)
        discriminators = [
            k for k in ("builtin", "prompt_file", "prompt") if s.get(k)
        ]
        if len(discriminators) != 1:
            raise ValueError(
                f"generation.seeds[{i}] ('{category}') must set exactly one of "
                f"builtin / prompt_file / prompt, got: {discriminators or 'none'}")

        seeds.append(GenerationSeed(
            category=category,
            count=count,
            builtin=s.get("builtin", ""),
            prompt_file=s.get("prompt_file", ""),
            prompt=s.get("prompt", ""),
            description=s.get("description", ""),
        ))

    # Provenance: absent normalizes to 'skill' (the default source).
    strategy = gen_raw.get("strategy") or "skill"
    if strategy not in GENERATION_STRATEGIES:
        raise ValueError(
            f"generation.strategy must be one of "
            f"{', '.join(GENERATION_STRATEGIES)}, got: {strategy!r}")
    if strategy == "synthetic" and not seeds:
        raise ValueError(
            "generation.strategy is 'synthetic' but generation.seeds is empty.")
    if seeds and strategy != "synthetic":
        raise ValueError(
            f"generation.seeds are only valid with strategy: synthetic "
            f"(got strategy: {strategy}).")

    generation_config = GenerationConfig(
        strategy=strategy,
        context=gen_raw.get("context", {}),
        seeds=seeds,
    )

    config = cls(
        name=raw.get("name", path.stem),
        description=raw.get("description", ""),
        skill=raw.get("skill") or None,  # Convert empty string to None
        permissions=raw.get("permissions", {}),
        execution=execution,
        runner=runner,
        models=models,
        mlflow=mlflow,
        config_dir=path.resolve().parent,
        config_path=path.resolve(),
        dataset=dataset_config,
        generation=generation_config,
    )

    # Outputs (path or tool)
    for i, o in enumerate(raw.get("outputs", [])):
        config.outputs.append(
            OutputConfig(
                path=_validate_relative_path(
                    o.get("path", ""), f"outputs[{i}].path", reject_root=True
                ),
                tool=o.get("tool", ""),
                schema=o.get("schema", ""),
                batch_pattern=o.get("batch_pattern", ""),
                types=o.get("types") or None,
            )
        )

    # Inputs (tool interception)
    inputs_raw = raw.get("inputs", {})
    for t in inputs_raw.get("tools") or []:
        config.inputs.tools.append(
            ToolInputConfig(
                match=t.get("match", ""),
                prompt=t.get("prompt", ""),
                prompt_file=t.get("prompt_file", ""),
            )
        )

    # Traces
    traces = raw.get("traces", {})
    if traces:
        config.traces = TracesConfig(
            stdout=traces.get("stdout", True),
            stderr=traces.get("stderr", True),
            events=traces.get("events", True),
            metrics=traces.get("metrics", True),
        )

    # Judges
    for j in raw.get("judges", []):
        builtin_val = j.get("builtin", "")
        if builtin_val is None:
            builtin_val = ""
        if not isinstance(builtin_val, str):
            raise ValueError(
                f"Judge '{j.get('name', '')}': 'builtin' must be a string"
            )
        args_val = j.get("arguments")
        if args_val is None:
            args_val = {}
        elif not isinstance(args_val, dict):
            raise ValueError(
                f"Judge '{j.get('name', '')}': 'arguments' must be a mapping"
            )
        score_range_val = j.get("score_range")
        if score_range_val is not None:
            jname = j.get("name", "")
            if (not isinstance(score_range_val, list)
                    or len(score_range_val) != 2):
                raise ValueError(
                    f"Judge '{jname}': 'score_range' must be a [min, max] list")
            try:
                lo, hi = float(score_range_val[0]), float(score_range_val[1])
            except (TypeError, ValueError) as exc:
                raise ValueError(
                    f"Judge '{jname}': 'score_range' values must be numeric") from exc
            if lo >= hi:
                raise ValueError(
                    f"Judge '{jname}': 'score_range' must be increasing [min, max]")
            score_range_val = [lo, hi]
        config.judges.append(
            JudgeConfig(
                name=j.get("name", ""),
                description=j.get("description", ""),
                condition=j.get("if", ""),
                check=j.get("check", ""),
                prompt=j.get("prompt", ""),
                prompt_file=j.get("prompt_file", ""),
                llm_rubric=j.get("llm_rubric", ""),
                context=j.get("context", []),
                feedback_type=j.get("feedback_type", ""),
                score_range=score_range_val,
                model=j.get("model", ""),
                module=j.get("module", ""),
                function=j.get("function", ""),
                builtin=builtin_val,
                arguments=args_val,
                samples=int(j.get("samples", 1)),
            )
        )

    # Reward composition
    if "reward" in raw:
        reward_raw = raw.get("reward")
        if not isinstance(reward_raw, dict):
            raise ValueError("reward must be a mapping when provided")
        sr = reward_raw.get("score_range", [1, 5])
        if not isinstance(sr, list) or len(sr) != 2:
            raise ValueError("reward.score_range must be a [min, max] list")
        try:
            score_min = float(sr[0])
            score_max = float(sr[1])
        except (TypeError, ValueError) as exc:
            raise ValueError(
                "reward.score_range values must be numeric") from exc
        if not score_min < score_max:
            raise ValueError(
                "reward.score_range must be increasing [min, max]")
        weights = reward_raw.get("weights", {}) or {}
        if not isinstance(weights, dict):
            raise ValueError("reward.weights must be a mapping")
        try:
            weights = {str(k): float(v) for k, v in weights.items()}
        except (TypeError, ValueError) as exc:
            raise ValueError(
                "reward.weights values must be numeric") from exc
        if any(v < 0 for v in weights.values()):
            raise ValueError("reward.weights values must be non-negative")
        raw_list = reward_raw.get("raw", []) or []
        if not isinstance(raw_list, list):
            raw_list = [raw_list]
        # Single-judge mode: one judge's value is the reward. Mutually
        # exclusive with the composition inputs.
        judge = reward_raw.get("judge")
        if judge is not None:
            if not isinstance(judge, str) or not judge.strip():
                raise ValueError(
                    "reward.judge must be a non-empty judge name")
            conflicting = [k for k in ("formula", "weights", "raw")
                           if k in reward_raw]
            if conflicting:
                raise ValueError(
                    "reward.judge cannot be combined with "
                    f"{'/'.join(conflicting)}")
            judge_names = {j.name for j in config.judges if j.name}
            if judge not in judge_names:
                raise ValueError(
                    f"reward.judge '{judge}' does not match any defined "
                    "judge")
        normalize = reward_raw.get("normalize", False)
        if not isinstance(normalize, bool):
            raise ValueError("reward.normalize must be a boolean")
        # gate defaults to False in judge mode, True for composition.
        gate = reward_raw.get("gate", judge is None)
        if not isinstance(gate, bool):
            raise ValueError("reward.gate must be a boolean")
        formula = str(reward_raw.get("formula", "weighted"))
        # Validate expression formulas now so a typo or unsafe construct
        # fails loudly here, not silently as reward 0.0 on every case at
        # run time. Bare references ("weighted") are resolved at compute
        # time, so skip the expression check for them. Skipped in judge
        # mode, where formula is unused.
        if judge is None and not re.fullmatch(
                r"[A-Za-z_][\w.\-]*", formula.strip()):
            from agent_eval.harbor.reward import validate_formula
            try:
                validate_formula(formula)
            except ValueError as exc:
                raise ValueError(
                    f"reward.formula is invalid: {exc}") from exc
        config.reward = RewardConfig(
            formula=formula,
            weights=weights,
            gate=gate,
            score_range=[score_min, score_max],
            raw=[str(r) for r in raw_list],
            judge=judge,
            normalize=normalize,
        )

    # Thresholds
    config.thresholds = raw.get("thresholds", {})

    # Hooks
    hooks_raw = raw.get("hooks", {}) or {}
    phases = ["before_all", "before_each", "after_each",
              "before_scoring", "after_all"]
    for phase in phases:
        entries = []
        for h in (hooks_raw.get(phase) or []):
            on_failure_val = h.get("on_failure", "fail")
            if on_failure_val not in ("fail", "continue"):
                raise ValueError(
                    f"hooks.{phase}: on_failure must be 'fail' or "
                    f"'continue', got '{on_failure_val}'")
            timeout_val = h.get("timeout", 120)
            if not isinstance(timeout_val, int) or timeout_val <= 0:
                raise ValueError(
                    f"hooks.{phase}: timeout must be a positive "
                    f"integer, got {timeout_val}")
            entries.append(HookEntry(
                command=h.get("command", ""),
                timeout=timeout_val,
                description=h.get("description", ""),
                on_failure=on_failure_val,
                condition=h.get("condition", ""),
            ))
        setattr(config.hooks, phase, entries)

    if config.execution.mode == "batch":
        per_case = []
        if config.hooks.before_each:
            per_case.append("before_each")
        if config.hooks.after_each:
            per_case.append("after_each")
        if per_case:
            import warnings
            warnings.warn(
                f"hooks.{', '.join(per_case)} ignored in batch mode "
                f"(per-case hooks only run in case/prompt mode)",
                stacklevel=2,
            )

    resolved_skill = config.resolve_skill()
    if resolved_skill:
        try:
            _validate_path_segment(resolved_skill, f"skill name in {path}")
        except ValueError as e:
            raise ValueError(str(e)) from e

    return config

is_prompt_mode

is_prompt_mode()

True when the eval runs a direct prompt (no skill wrapper).

Source code in agent_eval/config.py

def is_prompt_mode(self) -> bool:
    """True when the eval runs a direct prompt (no skill wrapper)."""
    return bool(self.execution.prompt and self.execution.prompt.strip())

resolve_path

resolve_path(relative)

Resolve a path relative to the config file's directory.

Absolute paths are returned as-is. Relative paths resolve against config_dir (falling back to cwd when config_dir is None).

Source code in agent_eval/config.py

def resolve_path(self, relative: Path | str) -> Path:
    """Resolve a path relative to the config file's directory.

    Absolute paths are returned as-is. Relative paths resolve against
    config_dir (falling back to cwd when config_dir is None).
    """
    p = Path(relative)
    if p.is_absolute():
        return p
    base = self.config_dir if self.config_dir is not None else Path.cwd()
    return base / p

resolve_skill

resolve_skill()

Canonical skill name for skill mode, or None for prompt mode.

Prefers execution.skill (the current location) and falls back to the deprecated top-level skill field. Returns None when neither is set — i.e. prompt mode or an unconfigured target. All execution substrates (local, Harbor, EvalHub) MUST resolve the target through this method so a config authored with only execution.skill runs the skill instead of silently degrading to prompt mode.

Source code in agent_eval/config.py

def resolve_skill(self) -> Optional[str]:
    """Canonical skill name for skill mode, or None for prompt mode.

    Prefers ``execution.skill`` (the current location) and falls back to
    the deprecated top-level ``skill`` field.  Returns None when neither
    is set — i.e. prompt mode or an unconfigured target.  All execution
    substrates (local, Harbor, EvalHub) MUST resolve the target through
    this method so a config authored with only ``execution.skill`` runs
    the skill instead of silently degrading to prompt mode.
    """
    return self.execution.skill or self.skill or None

Runners

A runner adapts a generic evaluation call to a specific agent runtime and returns a normalized RunResult. To support a new agent, subclass EvalRunner, implement the three abstract members below, and register it in the RUNNERS registry — see Runners.

agent_eval/agent/base.py

class EvalRunner(ABC):
    """Abstract runner -- one implementation per agent platform."""

    @classmethod
    @abstractmethod
    def from_config(cls, config, *, log_prefix=None, **overrides):
        """Construct a runner from an EvalConfig."""

    @property
    @abstractmethod
    def name(self) -> str:
        """Short identifier for this runner (e.g. 'claude-code')."""

    @abstractmethod
    def execute(self, target, args, workspace, model, ...) -> RunResult:
        """Run one invocation and return a normalized RunResult."""

Every runner returns the same normalized result, so scoring and reporting are runner-agnostic:

agent_eval.agent.base.RunResult dataclass

RunResult(exit_code, stdout, stderr, duration_s, token_usage=None, cost_usd=None, num_turns=None, resolved_model=None, models_used=None, per_model_usage=None, per_model_turns=None, permission_denials=None, raw_output=None)

Result of a single skill invocation.