Public API¶

This page lists the public API. Everything not listed here is implementation detail.

Anything under clean_evals._internal may change without notice. The public surface follows SemVer; breaking changes ratchet majors. Pre-1.0, minor versions may break the public API with a CHANGELOG.md entry.

Stability tags¶

— won't change without a major bump.
— may change at minor bumps with deprecation notice.

Data models¶

clean_evals.Case ¶

Bases: _FrozenStrictModel

A single eval input + optional expected output.

Attributes:

Name	Type	Description
`id`	`str`	Stable identifier within the dataset. Allowed characters: alphanumerics, underscore, dash, dot, colon. No spaces.
`input`	`dict[str, Any]`	Free-form structured input passed to the prompt template.
`expected`	`dict[str, Any] \| None`	Expected output for the case. `None` while the case is being built in the Dataset Builder; required once the dataset is locked.
`tags`	`list[str]`	Free-form tags used for filtering and per-tag accuracy breakdowns.
`metadata`	`dict[str, Any]`	Free-form metadata. Not used by the runner; passed through to reporters.
`locked`	`bool`	When `True`, `expected` is immutable. The Dataset Builder sets this on save; editing a locked case must bump the dataset version.

Example

Case(id="case_001", input={"text": "I love it"}, expected={"label": "positive"}) Case(id='case_001', ...)

Source code in src/clean_evals/models.py

class Case(_FrozenStrictModel):
    """A single eval input + optional expected output.

    Attributes:
        id: Stable identifier within the dataset. Allowed characters: alphanumerics,
            underscore, dash, dot, colon. No spaces.
        input: Free-form structured input passed to the prompt template.
        expected: Expected output for the case. ``None`` while the case is being
            built in the Dataset Builder; required once the dataset is locked.
        tags: Free-form tags used for filtering and per-tag accuracy breakdowns.
        metadata: Free-form metadata. Not used by the runner; passed through to
            reporters.
        locked: When ``True``, ``expected`` is immutable. The Dataset Builder
            sets this on save; editing a locked case must bump the dataset
            version.

    Example:
        >>> Case(id="case_001", input={"text": "I love it"}, expected={"label": "positive"})
        Case(id='case_001', ...)
    """

    id: str
    input: dict[str, Any]
    expected: dict[str, Any] | None = None
    tags: list[str] = Field(default_factory=list)
    metadata: dict[str, Any] = Field(default_factory=dict)
    locked: bool = False

    @field_validator("id")
    @classmethod
    def _validate_id(cls, v: str) -> str:
        if not v:
            raise ValueError("Case.id may not be empty")
        if not _CASE_ID_RE.match(v):
            raise ValueError(
                f"Case.id {v!r} contains illegal characters; allowed: A-Z, a-z, 0-9, _ - . :"
            )
        return v

clean_evals.Dataset ¶

Bases: _StrictModel

A versioned collection of Case rows plus its scoring configuration.

Datasets are immutable once tagged: once locked_at is set on the backing storage row, the version is sealed and any edit must produce a new version. Historical runs against v1 remain comparable after v2 ships.

The scorer field names a scorer registered under the clean_evals.scorers entry-point group. scorer_config is passed verbatim to Scorer.from_config.

Loading from YAML

Dataset.from_yaml("examples/sentiment/dataset.yml") # doctest: +SKIP

Source code in src/clean_evals/models.py

class Dataset(_StrictModel):
    """A versioned collection of ``Case`` rows plus its scoring configuration.

    Datasets are immutable once tagged: once ``locked_at`` is set on the
    backing storage row, the version is sealed and any edit must produce a new
    version. Historical runs against ``v1`` remain comparable after ``v2`` ships.

    The ``scorer`` field names a scorer registered under the
    ``clean_evals.scorers`` entry-point group. ``scorer_config`` is passed
    verbatim to ``Scorer.from_config``.

    Loading from YAML:
        >>> Dataset.from_yaml("examples/sentiment/dataset.yml")  # doctest: +SKIP
    """

    name: str
    version: str
    description: str | None = None
    cases: list[Case]
    scorer: str
    scorer_config: dict[str, Any] = Field(default_factory=dict)
    # Prompt spec (docs/docs/flow.md, stage 1). "raw" sends each case
    # verbatim; "templated" assembles system prompt + context + variables.
    request_shape: Literal["raw", "templated"] = "raw"
    system_prompt: str | None = None
    shared_context: str | None = None
    user_template: str | None = None

    @field_validator("version")
    @classmethod
    def _validate_version(cls, v: str) -> str:
        if not _DATASET_VERSION_RE.match(v):
            raise ValueError(
                f"Dataset.version {v!r} must match SemVer-like pattern (e.g. 'v1', '1.2', '0.3.0')"
            )
        return v

    @field_validator("name")
    @classmethod
    def _validate_name(cls, v: str) -> str:
        if not v:
            raise ValueError("Dataset.name may not be empty")
        return v

    @model_validator(mode="after")
    def _validate_unique_case_ids(self) -> Dataset:
        seen: set[str] = set()
        for case in self.cases:
            if case.id in seen:
                raise ValueError(f"Duplicate case id within dataset: {case.id!r}")
            seen.add(case.id)
        return self

    @classmethod
    def from_yaml(
        cls,
        path: str | Path,
        scrubber: Scrubber | None = None,
    ) -> Dataset:
        """Load a dataset from a YAML file.

        The YAML may inline cases, or reference a sidecar JSONL via
        ``cases_jsonl: ./path/to/cases.jsonl`` relative to the YAML.

        Args:
            path: Path to the YAML file.
            scrubber: Optional ``Scrubber`` plugin run on every case after
                load. Useful for redacting PII before the case crosses any
                boundary.

        Returns:
            A validated ``Dataset``.
        """
        p = Path(path)
        with p.open("r", encoding="utf-8") as fh:
            raw = yaml.safe_load(fh)
        if not isinstance(raw, dict):
            raise ValueError(f"{p}: top-level must be a mapping")

        cases_jsonl = raw.pop("cases_jsonl", None)
        if cases_jsonl is not None:
            import json as _json

            jsonl_path = (p.parent / cases_jsonl).resolve()
            cases: list[dict[str, Any]] = []
            with jsonl_path.open("r", encoding="utf-8") as fh:
                for lineno, raw_line in enumerate(fh, start=1):
                    line = raw_line.strip()
                    if not line:
                        continue
                    try:
                        cases.append(_json.loads(line))
                    except _json.JSONDecodeError as e:
                        raise ValueError(f"{jsonl_path}:{lineno}: invalid JSON ({e.msg})") from e
            raw["cases"] = cases

        ds = cls.model_validate(raw)
        if scrubber is not None:
            ds = ds.model_copy(update={"cases": [scrubber.scrub(c) for c in ds.cases]})
        return ds

    def to_yaml(self, path: str | Path) -> None:
        """Write the dataset to a YAML file.

        Cases are inlined. For large datasets, write JSONL separately and
        reference it from the YAML — but that's a workflow you do by hand.
        """
        p = Path(path)
        p.parent.mkdir(parents=True, exist_ok=True)
        data = self.model_dump(mode="json")
        with p.open("w", encoding="utf-8") as fh:
            yaml.safe_dump(data, fh, sort_keys=False, allow_unicode=True)

from_yaml `classmethod` ¶

from_yaml(path: str | Path, scrubber: Scrubber | None = None) -> Dataset

Load a dataset from a YAML file.

The YAML may inline cases, or reference a sidecar JSONL via cases_jsonl: ./path/to/cases.jsonl relative to the YAML.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to the YAML file.	required
`scrubber`	`Scrubber \| None`	Optional `Scrubber` plugin run on every case after load. Useful for redacting PII before the case crosses any boundary.	`None`

Returns:

Type	Description
`Dataset`	A validated `Dataset`.

Source code in src/clean_evals/models.py

@classmethod
def from_yaml(
    cls,
    path: str | Path,
    scrubber: Scrubber | None = None,
) -> Dataset:
    """Load a dataset from a YAML file.

    The YAML may inline cases, or reference a sidecar JSONL via
    ``cases_jsonl: ./path/to/cases.jsonl`` relative to the YAML.

    Args:
        path: Path to the YAML file.
        scrubber: Optional ``Scrubber`` plugin run on every case after
            load. Useful for redacting PII before the case crosses any
            boundary.

    Returns:
        A validated ``Dataset``.
    """
    p = Path(path)
    with p.open("r", encoding="utf-8") as fh:
        raw = yaml.safe_load(fh)
    if not isinstance(raw, dict):
        raise ValueError(f"{p}: top-level must be a mapping")

    cases_jsonl = raw.pop("cases_jsonl", None)
    if cases_jsonl is not None:
        import json as _json

        jsonl_path = (p.parent / cases_jsonl).resolve()
        cases: list[dict[str, Any]] = []
        with jsonl_path.open("r", encoding="utf-8") as fh:
            for lineno, raw_line in enumerate(fh, start=1):
                line = raw_line.strip()
                if not line:
                    continue
                try:
                    cases.append(_json.loads(line))
                except _json.JSONDecodeError as e:
                    raise ValueError(f"{jsonl_path}:{lineno}: invalid JSON ({e.msg})") from e
        raw["cases"] = cases

    ds = cls.model_validate(raw)
    if scrubber is not None:
        ds = ds.model_copy(update={"cases": [scrubber.scrub(c) for c in ds.cases]})
    return ds

to_yaml ¶

to_yaml(path: str | Path) -> None

Write the dataset to a YAML file.

Cases are inlined. For large datasets, write JSONL separately and reference it from the YAML — but that's a workflow you do by hand.

Source code in src/clean_evals/models.py

def to_yaml(self, path: str | Path) -> None:
    """Write the dataset to a YAML file.

    Cases are inlined. For large datasets, write JSONL separately and
    reference it from the YAML — but that's a workflow you do by hand.
    """
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    data = self.model_dump(mode="json")
    with p.open("w", encoding="utf-8") as fh:
        yaml.safe_dump(data, fh, sort_keys=False, allow_unicode=True)

clean_evals.ModelResponse ¶

Bases: _FrozenStrictModel

A single model invocation's output and accounting.

Attributes:

Name	Type	Description
`content`	`str`	Raw response text.
`parsed`	`dict[str, Any] \| None`	`content` parsed as JSON when `response_format="json"` was requested. `None` otherwise, or when parsing failed (the runner then records `status="schema_invalid"` on the case result).
`tokens_in`	`int`	Input tokens reported by the provider. `-1` when the provider does not return a count.
`tokens_out`	`int`	Output tokens reported by the provider. `-1` when the provider does not return a count.
`latency_ms`	`int`	Wall-clock latency including network. Adapter-measured.
`cost_usd`	`float`	Computed via `pricing.py` at the time of the run.
`raw`	`dict[str, Any]`	Provider response as a dict, for debugging and per-case diff.

Source code in src/clean_evals/models.py

class ModelResponse(_FrozenStrictModel):
    """A single model invocation's output and accounting.

    Attributes:
        content: Raw response text.
        parsed: ``content`` parsed as JSON when ``response_format="json"`` was
            requested. ``None`` otherwise, or when parsing failed (the runner
            then records ``status="schema_invalid"`` on the case result).
        tokens_in: Input tokens reported by the provider. ``-1`` when the
            provider does not return a count.
        tokens_out: Output tokens reported by the provider. ``-1`` when the
            provider does not return a count.
        latency_ms: Wall-clock latency including network. Adapter-measured.
        cost_usd: Computed via ``pricing.py`` at the time of the run.
        raw: Provider response as a dict, for debugging and per-case diff.
    """

    content: str
    parsed: dict[str, Any] | None = None
    tokens_in: int = -1
    tokens_out: int = -1
    latency_ms: int
    cost_usd: float
    raw: dict[str, Any] = Field(default_factory=dict)

clean_evals.ScoreResult ¶

Bases: _FrozenStrictModel

Output of a single scorer invocation.

Attributes:

Name	Type	Description
`score`	`float`	Normalised 0.0–1.0 score. Required.
`passed`	`bool`	Whether this counts as a "pass" for leaderboard purposes. Scorers decide their own threshold; `passed` is reported alongside `score` so the runner can compute pass-rate without re-thresholding.
`breakdown`	`dict[str, float]`	Optional sub-component scores for richer reporting (e.g. `{"name_match": 1.0, "type_match": 0.5}`).
`notes`	`str \| None`	Optional free-form text — handy for LLM-judge rationales.

Source code in src/clean_evals/models.py

class ScoreResult(_FrozenStrictModel):
    """Output of a single scorer invocation.

    Attributes:
        score: Normalised 0.0–1.0 score. Required.
        passed: Whether this counts as a "pass" for leaderboard purposes.
            Scorers decide their own threshold; ``passed`` is reported alongside
            ``score`` so the runner can compute pass-rate without re-thresholding.
        breakdown: Optional sub-component scores for richer reporting
            (e.g. ``{"name_match": 1.0, "type_match": 0.5}``).
        notes: Optional free-form text — handy for LLM-judge rationales.
    """

    score: float = Field(ge=0.0, le=1.0)
    passed: bool
    breakdown: dict[str, float] = Field(default_factory=dict)
    notes: str | None = None

clean_evals.CaseResult ¶

Bases: _FrozenStrictModel

One (case, model) outcome.

A model erroring (timeout, 5xx, content filter, schema-invalid output) produces a CaseResult with the appropriate status rather than crashing the run. Reports distinguish wrong answers from infrastructure errors so a model that 500s on 30% of cases is not "70% accurate."

Attributes:

Name	Type	Description
`case_id`	`str`	Foreign key to `Case.id`.
`model`	`str`	Snapshot model id (e.g. `"gpt-4o-2024-11-20"`).
`status`	`_RUN_STATUS`	`"ok"` for a scored response; `"error"`, `"timeout"`, `"schema_invalid"`, or `"aborted_cost"` for failure modes.
`response`	`ModelResponse \| None`	The model's response, or `None` if the call did not complete.
`score`	`ScoreResult \| None`	Scorer output, or `None` if scoring was skipped (e.g. for an errored case).
`error`	`str \| None`	Captured error payload when `status != "ok"`.
`started_at,`	`finished_at`	UTC timestamps with microsecond precision.

Source code in src/clean_evals/models.py

class CaseResult(_FrozenStrictModel):
    """One ``(case, model)`` outcome.

    A model erroring (timeout, 5xx, content filter, schema-invalid output)
    produces a ``CaseResult`` with the appropriate ``status`` rather than
    crashing the run. Reports distinguish wrong answers from infrastructure
    errors so a model that 500s on 30% of cases is not "70% accurate."

    Attributes:
        case_id: Foreign key to ``Case.id``.
        model: Snapshot model id (e.g. ``"gpt-4o-2024-11-20"``).
        status: ``"ok"`` for a scored response; ``"error"``, ``"timeout"``,
            ``"schema_invalid"``, or ``"aborted_cost"`` for failure modes.
        response: The model's response, or ``None`` if the call did not
            complete.
        score: Scorer output, or ``None`` if scoring was skipped (e.g. for an
            errored case).
        error: Captured error payload when ``status != "ok"``.
        started_at, finished_at: UTC timestamps with microsecond precision.
    """

    case_id: str
    model: str
    status: _RUN_STATUS
    response: ModelResponse | None = None
    score: ScoreResult | None = None
    error: str | None = None
    started_at: datetime
    finished_at: datetime

clean_evals.RunConfig ¶

Bases: _StrictModel

Configuration for a single eval run.

Attributes:

Name	Type	Description
`models`	`list[str]`	Snapshot model ids only — floating aliases ending in `-latest` are rejected to keep historical runs comparable.
`model_params`	`dict[str, ModelParams]`	Per-model parameters keyed by model id (see :class:`ModelParams`). Missing keys fall back to the run-level settings.
`concurrency`	`dict[str, int]`	Optional per-provider concurrency caps. Empty = uncapped (the runner backs off on 429 anyway).
`timeout_s`	`float`	Per-call timeout, in seconds. Adapter-enforced.
`retries`	`int`	Retries on transient failures (429, 5xx, network errors).
`seed`	`int \| None`	Provider seed for determinism. `None` opts out.
`temperature`	`float`	Sampling temperature. `> 0` makes runs non-deterministic; the report carries a header warning when this is the case.
`max_cost_usd`	`float`	Best-effort ceiling for the run. Spend is checked as results arrive; cases that have not started when the ceiling trips are aborted with `status="aborted_cost"`, but calls already in flight complete, so actual spend can overshoot the ceiling (bounded by concurrency). Verify real spend in the provider's billing console.

Source code in src/clean_evals/models.py

class RunConfig(_StrictModel):
    """Configuration for a single eval run.

    Attributes:
        models: Snapshot model ids only — floating aliases ending in
            ``-latest`` are rejected to keep historical runs comparable.
        model_params: Per-model parameters keyed by model id (see
            :class:`ModelParams`). Missing keys fall back to the run-level
            settings.
        concurrency: Optional per-provider concurrency caps. Empty = uncapped
            (the runner backs off on 429 anyway).
        timeout_s: Per-call timeout, in seconds. Adapter-enforced.
        retries: Retries on transient failures (429, 5xx, network errors).
        seed: Provider seed for determinism. ``None`` opts out.
        temperature: Sampling temperature. ``> 0`` makes runs non-deterministic;
            the report carries a header warning when this is the case.
        max_cost_usd: Best-effort ceiling for the run. Spend is checked as
            results arrive; cases that have not started when the ceiling
            trips are aborted with ``status="aborted_cost"``, but calls
            already in flight complete, so actual spend can overshoot the
            ceiling (bounded by concurrency). Verify real spend in the
            provider's billing console.
    """

    models: list[str]
    model_params: dict[str, ModelParams] = Field(default_factory=dict)
    concurrency: dict[str, int] = Field(default_factory=dict)
    timeout_s: float = 120.0
    retries: int = 2
    seed: int | None = 42
    temperature: float = 0.0
    max_cost_usd: float = 5.0

    @field_validator("models")
    @classmethod
    def _validate_models(cls, v: list[str]) -> list[str]:
        if not v:
            raise ValueError("RunConfig.models must contain at least one model")
        for m in v:
            # local/ models are pinned by the file on disk; the dated-snapshot
            # rule exists for hosted providers whose aliases move.
            if m.startswith("local/"):
                continue
            if m.endswith("-latest") or m == "latest":
                raise ValueError(
                    f"Floating alias {m!r} rejected. clean-evals requires dated snapshot ids."
                )
        return v

    @field_validator("temperature")
    @classmethod
    def _validate_temperature(cls, v: float) -> float:
        if v < 0.0 or v > 2.0:
            raise ValueError("RunConfig.temperature must be in [0.0, 2.0]")
        return v

    @field_validator("max_cost_usd")
    @classmethod
    def _validate_max_cost(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("RunConfig.max_cost_usd must be > 0")
        return v

clean_evals.ModelSummary ¶

Bases: _FrozenStrictModel

Per-model leaderboard row in a RunResult.

Attributes:

Name	Type	Description
`cost_per_correct_usd`	`float \| None`	Cost per passed case. `None` when zero cases passed (avoids division-by-zero presented as "infinity good").
`pricing_version`	`str`	The frozen pricing table version used for cost attribution. Rerunning under newer pricing creates a new run; old summaries still report the spend that was actually charged.

Source code in src/clean_evals/models.py

class ModelSummary(_FrozenStrictModel):
    """Per-model leaderboard row in a ``RunResult``.

    Attributes:
        cost_per_correct_usd: Cost per passed case. ``None`` when zero cases
            passed (avoids division-by-zero presented as "infinity good").
        pricing_version: The frozen pricing table version used for cost
            attribution. Rerunning under newer pricing creates a new run; old
            summaries still report the spend that was actually charged.
    """

    model: str
    cases_run: int
    cases_passed: int
    score_mean: float
    score_p50: float
    latency_p95_ms: int
    error_rate: float
    total_cost_usd: float
    cost_per_correct_usd: float | None
    pricing_version: str

clean_evals.RunResult ¶

Bases: _StrictModel

Top-level result for a single Runner.run invocation.

Reporters consume this. Storage persists this. The Decision UI renders this. It is the unit of comparison.

Source code in src/clean_evals/models.py

class RunResult(_StrictModel):
    """Top-level result for a single ``Runner.run`` invocation.

    Reporters consume this. Storage persists this. The Decision UI renders
    this. It is the unit of comparison.
    """

    run_id: str
    dataset: str
    dataset_version: str
    config: RunConfig
    cases: list[CaseResult]
    summary: dict[str, ModelSummary]
    started_at: datetime
    finished_at: datetime
    pricing_version: str
    deterministic: bool
    notes: list[str] = Field(default_factory=list)

Runner¶

clean_evals.Runner ¶

Run a Dataset against a list of models.

Parameters:

Name	Type	Description	Default
`adapters`	`dict[str, ModelAdapter] \| None`	Mapping from provider id to adapter instance. `None` (the default) builds a fresh adapter per provider on first use, instantiated from the entry-point registry.	`None`
`scorer_registry_obj`	`Any`	Custom registry. Most callers leave this `None` and use the module-level singleton.	`None`
`prompt_template`	`str \| None`	How to render a case's input as a single prompt string. `None` (the default) JSON-serialises the input. Otherwise, must contain `{prompt}` exactly once. The literal string `"{prompt}"` is replaced with `json.dumps(case.input)`.	`None`
`event_sink`	`EventSink \| None`	Optional progress sink. Default no-op for headless runs.	`None`
`daily_cost_so_far_usd`	`float \| None`	How much was spent today before this run started. `None` treats prior spend as $0 — callers with storage access (the CLI, web, and queue paths) pass the persisted total for today. Only persisted runs leave a cost trail, so the daily limit is best-effort.	`None`

Example

import asyncio from clean_evals import Dataset, Runner, RunConfig ds = Dataset( # doctest: +SKIP ... name="hello", version="v1", scorer="exact_match", cases=[], ... ) runner = Runner() # doctest: +SKIP result = asyncio.run(runner.run( # doctest: +SKIP ... ds, RunConfig(models=["gpt-4o-mini-2024-07-18"]), ... ))

Source code in src/clean_evals/runner.py

class Runner:
    """Run a ``Dataset`` against a list of models.

    Args:
        adapters: Mapping from provider id to adapter instance. ``None`` (the
            default) builds a fresh adapter per provider on first use,
            instantiated from the entry-point registry.
        scorer_registry_obj: Custom registry. Most callers leave this ``None``
            and use the module-level singleton.
        prompt_template: How to render a case's input as a single prompt
            string. ``None`` (the default) JSON-serialises the input.
            Otherwise, must contain ``{prompt}`` exactly once. The literal
            string ``"{prompt}"`` is replaced with ``json.dumps(case.input)``.
        event_sink: Optional progress sink. Default no-op for headless runs.
        daily_cost_so_far_usd: How much was spent today before this run
            started. ``None`` treats prior spend as $0 — callers with
            storage access (the CLI, web, and queue paths) pass the
            persisted total for today. Only persisted runs leave a cost
            trail, so the daily limit is best-effort.

    Example:
        >>> import asyncio
        >>> from clean_evals import Dataset, Runner, RunConfig
        >>> ds = Dataset(  # doctest: +SKIP
        ...     name="hello", version="v1", scorer="exact_match", cases=[],
        ... )
        >>> runner = Runner()  # doctest: +SKIP
        >>> result = asyncio.run(runner.run(  # doctest: +SKIP
        ...     ds, RunConfig(models=["gpt-4o-mini-2024-07-18"]),
        ... ))
    """

    def __init__(
        self,
        *,
        adapters: dict[str, ModelAdapter] | None = None,
        scorer_registry_obj: Any = None,
        prompt_template: str | None = None,
        event_sink: EventSink | None = None,
        daily_cost_so_far_usd: float | None = None,
    ) -> None:
        self._adapters: dict[str, ModelAdapter] = adapters or {}
        self._owned_providers: set[str] = set()
        self._scorer_registry = scorer_registry_obj or scorer_registry
        self._prompt_template = prompt_template
        if prompt_template is not None and prompt_template.count(_PROMPT_PLACEHOLDER) != 1:
            raise ValueError(
                f"prompt_template must contain exactly one {_PROMPT_PLACEHOLDER!r} marker"
            )
        self._event_sink = event_sink or noop_sink
        self._daily_cost_so_far = daily_cost_so_far_usd

    # ------------------------------------------------------------------
    # Sync facade
    # ------------------------------------------------------------------

    def run_sync(self, dataset: Dataset, config: RunConfig) -> RunResult:
        """Synchronous wrapper. Spins up a fresh event loop and blocks.

        Adapters the runner instantiated itself are closed before the loop
        is torn down (their HTTP clients are bound to it). Explicitly
        injected adapters stay open — the caller owns their lifecycle.
        """

        async def _run_and_close() -> RunResult:
            try:
                return await self.run(dataset, config)
            finally:
                await self.aclose()

        return asyncio.run(_run_and_close())

    async def aclose(self) -> None:
        """Close HTTP clients of adapters this runner created lazily."""
        for provider in sorted(self._owned_providers):
            adapter = self._adapters.pop(provider, None)
            closer = getattr(adapter, "aclose", None)
            if closer is None:
                continue
            try:
                await closer()
            except Exception as exc:
                _log.warning("closing %s adapter raised %r; ignoring", provider, exc)
        self._owned_providers.clear()

    # ------------------------------------------------------------------
    # Async core
    # ------------------------------------------------------------------

    async def run(self, dataset: Dataset, config: RunConfig) -> RunResult:
        """Run ``dataset`` against ``config.models``."""
        run_id = _new_run_id()
        started_at = now()
        notes: list[str] = []

        self._check_daily_cost_limit()

        scorer = self._scorer_registry.build(dataset.scorer, dataset.scorer_config)

        deterministic = config.temperature == 0.0
        if not deterministic:
            notes.append(f"temperature={config.temperature} > 0; results are not deterministic.")

        # Group concurrency by provider. Empty cap = uncapped.
        per_provider_semaphores: dict[str, asyncio.Semaphore | None] = {}
        for provider, cap in config.concurrency.items():
            per_provider_semaphores[provider] = asyncio.Semaphore(max(1, cap))

        cost_state = _CostState(max_cost=config.max_cost_usd)

        self._emit(
            RunEvent(
                type="run.started",
                run_id=run_id,
                at=now(),
                payload={
                    "dataset": dataset.name,
                    "dataset_version": dataset.version,
                    "models": list(config.models),
                    "case_count": len(dataset.cases),
                },
            )
        )

        # Schedule (case, model) tasks.
        case_results: list[CaseResult] = []
        sem_lookup: dict[str, asyncio.Semaphore | None] = dict(per_provider_semaphores)

        async def run_one(case: Case, model: str) -> CaseResult:
            try:
                provider = self._resolve_provider(model)
            except ValueError as exc:
                # Failure is data: an unrecognised model id fails its own
                # cases instead of crashing the whole run.
                return _failed_case(case.id, model, status="error", reason=str(exc))
            sem = sem_lookup.get(provider)
            if sem is not None:
                async with sem:
                    return await self._run_single(
                        run_id=run_id,
                        case=case,
                        model=model,
                        provider=provider,
                        config=config,
                        scorer=scorer,
                        dataset=dataset,
                        cost_state=cost_state,
                    )
            return await self._run_single(
                run_id=run_id,
                case=case,
                model=model,
                provider=provider,
                config=config,
                scorer=scorer,
                dataset=dataset,
                cost_state=cost_state,
            )

        tasks = [
            asyncio.create_task(run_one(case, model))
            for case in dataset.cases
            for model in config.models
        ]
        try:
            for finished in asyncio.as_completed(tasks):
                case_results.append(await finished)
        finally:
            # Cancel anything that's left if we were cancelled externally.
            for t in tasks:
                if not t.done():
                    t.cancel()

        if cost_state.aborted:
            notes.append(f"Run aborted: cost ceiling ${config.max_cost_usd:.2f} exceeded.")

        finished_at = now()

        summary = _summarise(case_results, config.models)

        result = RunResult(
            run_id=run_id,
            dataset=dataset.name,
            dataset_version=dataset.version,
            config=config,
            cases=sorted(case_results, key=lambda r: (r.case_id, r.model)),
            summary=summary,
            started_at=started_at,
            finished_at=finished_at,
            pricing_version=effective_version(),
            deterministic=deterministic,
            notes=notes,
        )

        self._emit(
            RunEvent(
                type="run.finished",
                run_id=run_id,
                at=finished_at,
                payload={
                    "summary": {m: s.model_dump(mode="json") for m, s in summary.items()},
                    "aborted": cost_state.aborted,
                },
            )
        )
        return result

    # ------------------------------------------------------------------
    # Single-call execution path
    # ------------------------------------------------------------------

    async def _run_single(
        self,
        *,
        run_id: str,
        case: Case,
        model: str,
        provider: str,
        config: RunConfig,
        scorer: Scorer,
        dataset: Dataset,
        cost_state: _CostState,
    ) -> CaseResult:
        if cost_state.aborted:
            return _failed_case(
                case.id,
                model,
                status="aborted_cost",
                reason="cost ceiling reached before scheduling",
            )

        self._emit(
            RunEvent(
                type="run.case_started",
                run_id=run_id,
                at=now(),
                payload={"case_id": case.id, "model": model},
            )
        )

        try:
            adapter = await self._get_adapter(provider)
        except UnknownPlugin as exc:
            # Failure is data: a provider with no registered adapter fails
            # its own cases instead of crashing the run.
            return _failed_case(case.id, model, status="error", reason=str(exc))
        request = self._assemble_request(case, dataset)
        started_at = now()

        response: ModelResponse | None = None
        status: str = "ok"
        error: str | None = None

        for attempt in range(config.retries + 1):
            try:
                params = config.model_params.get(model)
                response = await adapter.complete(
                    prompt=request.user,
                    model=model,
                    temperature=(
                        params.temperature
                        if params is not None and params.temperature is not None
                        else config.temperature
                    ),
                    seed=config.seed,
                    timeout_s=config.timeout_s,
                    response_format=_response_format_for(dataset),
                    system=request.system,
                    reasoning_effort=params.reasoning_effort if params else None,
                    max_output_tokens=params.max_output_tokens if params else None,
                )
                break
            except RateLimited as exc:
                if attempt >= config.retries:
                    status, error = "error", f"rate-limited: {exc}"
                    break
                wait = exc.retry_after_s if exc.retry_after_s is not None else _backoff(attempt)
                # Honour Retry-After, but never let a hostile or garbled
                # header stall a worker indefinitely.
                wait = min(wait, _MAX_RETRY_WAIT_S)
                _log.info(
                    "rate-limited %s on %s/%s; sleeping %.1fs (attempt %d/%d)",
                    provider,
                    case.id,
                    model,
                    wait,
                    attempt + 1,
                    config.retries + 1,
                )
                await asyncio.sleep(wait)
            except ProviderTimeout as exc:
                if attempt >= config.retries:
                    status, error = "timeout", f"timeout: {exc}"
                    break
                await asyncio.sleep(_backoff(attempt))
            except SchemaInvalidResponse as exc:
                status, error = "schema_invalid", str(exc)
                break
            except AdapterError as exc:
                if attempt >= config.retries:
                    status, error = "error", f"adapter error: {exc}"
                    break
                await asyncio.sleep(_backoff(attempt))
            except Exception as exc:
                status, error = "error", f"unexpected: {exc!r}"
                break

        finished_at = now()

        score = None
        if status == "ok" and response is not None:
            cost_state.add(response.cost_usd)
            try:
                score = scorer.score(case, response)
            except Exception as exc:
                status, error = "error", f"scorer raised: {exc!r}"
                score = None

        if cost_state.exceeded() and not cost_state.aborted:
            cost_state.aborted = True
            self._emit(
                RunEvent(
                    type="run.cost_warning",
                    run_id=run_id,
                    at=now(),
                    payload={"spend_usd": cost_state.spend, "limit_usd": cost_state.max_cost},
                )
            )

        result = CaseResult(
            case_id=case.id,
            model=model,
            status=status,  # type: ignore[arg-type]
            response=response if status == "ok" else None,
            score=score,
            error=error,
            started_at=started_at,
            finished_at=finished_at,
        )

        self._emit(
            RunEvent(
                type="run.case_finished",
                run_id=run_id,
                at=finished_at,
                payload={
                    "case_id": case.id,
                    "model": model,
                    "status": status,
                    "score": score.score if score is not None else None,
                    "cost_usd": response.cost_usd if response is not None else 0.0,
                },
            )
        )
        return result

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    async def _get_adapter(self, provider: str) -> ModelAdapter:
        if provider in self._adapters:
            return self._adapters[provider]
        cls = adapter_registry.get(provider)
        adapter = cls()
        self._adapters[provider] = adapter
        self._owned_providers.add(provider)
        return adapter

    def _resolve_provider(self, model: str) -> str:
        """Infer the provider from the model id's prefix.

        Explicitly injected adapters are honoured downstream: when the
        inferred provider matches a key in ``self._adapters``,
        ``_get_adapter`` returns that instance instead of building one.

        Raises:
            ValueError: When no known prefix matches.
        """
        return infer_provider(model)

    def _assemble_request(self, case: Case, dataset: Dataset) -> AssembledRequest:
        if dataset.request_shape == "templated":
            return assemble(
                request_shape=dataset.request_shape,
                system_prompt=dataset.system_prompt,
                shared_context=dataset.shared_context,
                user_template=dataset.user_template,
                case_input=case.input,
            )

        # Raw shape — legacy path: single field (or JSON) plus the optional
        # {prompt} wrapper template from the constructor or scorer_config.
        import json as _json

        body = case.input.get("prompt") if isinstance(case.input.get("prompt"), str) else None
        if body is None:
            body = _json.dumps(case.input, ensure_ascii=False, indent=2)
        if self._prompt_template is None:
            tpl = dataset.scorer_config.get("prompt_template")
            if isinstance(tpl, str) and _PROMPT_PLACEHOLDER in tpl:
                return AssembledRequest(system=None, user=tpl.replace(_PROMPT_PLACEHOLDER, body))
            return AssembledRequest(system=None, user=body)
        return AssembledRequest(
            system=None, user=self._prompt_template.replace(_PROMPT_PLACEHOLDER, body)
        )

    def _check_daily_cost_limit(self) -> None:
        env_val = os.environ.get("CLEAN_EVALS_DAILY_COST_LIMIT_USD")
        if not env_val:
            return
        try:
            limit = float(env_val)
        except ValueError as exc:
            raise CleanEvalsError(
                f"CLEAN_EVALS_DAILY_COST_LIMIT_USD={env_val!r} is not a number"
            ) from exc
        spent_today = self._daily_cost_so_far if self._daily_cost_so_far is not None else 0.0
        if spent_today >= limit:
            raise DailyCostLimitExceeded(
                f"Today's spend ${spent_today:.2f} >= daily limit ${limit:.2f}"
            )

    def _emit(self, event: RunEvent) -> None:
        try:
            self._event_sink(event)
        except Exception as exc:
            _log.warning("event sink raised %r; ignoring", exc)

aclose `async` ¶

aclose() -> None

Close HTTP clients of adapters this runner created lazily.

Source code in src/clean_evals/runner.py

async def aclose(self) -> None:
    """Close HTTP clients of adapters this runner created lazily."""
    for provider in sorted(self._owned_providers):
        adapter = self._adapters.pop(provider, None)
        closer = getattr(adapter, "aclose", None)
        if closer is None:
            continue
        try:
            await closer()
        except Exception as exc:
            _log.warning("closing %s adapter raised %r; ignoring", provider, exc)
    self._owned_providers.clear()

run `async` ¶

run(dataset: Dataset, config: RunConfig) -> RunResult

Run dataset against config.models.

Source code in src/clean_evals/runner.py

async def run(self, dataset: Dataset, config: RunConfig) -> RunResult:
    """Run ``dataset`` against ``config.models``."""
    run_id = _new_run_id()
    started_at = now()
    notes: list[str] = []

    self._check_daily_cost_limit()

    scorer = self._scorer_registry.build(dataset.scorer, dataset.scorer_config)

    deterministic = config.temperature == 0.0
    if not deterministic:
        notes.append(f"temperature={config.temperature} > 0; results are not deterministic.")

    # Group concurrency by provider. Empty cap = uncapped.
    per_provider_semaphores: dict[str, asyncio.Semaphore | None] = {}
    for provider, cap in config.concurrency.items():
        per_provider_semaphores[provider] = asyncio.Semaphore(max(1, cap))

    cost_state = _CostState(max_cost=config.max_cost_usd)

    self._emit(
        RunEvent(
            type="run.started",
            run_id=run_id,
            at=now(),
            payload={
                "dataset": dataset.name,
                "dataset_version": dataset.version,
                "models": list(config.models),
                "case_count": len(dataset.cases),
            },
        )
    )

    # Schedule (case, model) tasks.
    case_results: list[CaseResult] = []
    sem_lookup: dict[str, asyncio.Semaphore | None] = dict(per_provider_semaphores)

    async def run_one(case: Case, model: str) -> CaseResult:
        try:
            provider = self._resolve_provider(model)
        except ValueError as exc:
            # Failure is data: an unrecognised model id fails its own
            # cases instead of crashing the whole run.
            return _failed_case(case.id, model, status="error", reason=str(exc))
        sem = sem_lookup.get(provider)
        if sem is not None:
            async with sem:
                return await self._run_single(
                    run_id=run_id,
                    case=case,
                    model=model,
                    provider=provider,
                    config=config,
                    scorer=scorer,
                    dataset=dataset,
                    cost_state=cost_state,
                )
        return await self._run_single(
            run_id=run_id,
            case=case,
            model=model,
            provider=provider,
            config=config,
            scorer=scorer,
            dataset=dataset,
            cost_state=cost_state,
        )

    tasks = [
        asyncio.create_task(run_one(case, model))
        for case in dataset.cases
        for model in config.models
    ]
    try:
        for finished in asyncio.as_completed(tasks):
            case_results.append(await finished)
    finally:
        # Cancel anything that's left if we were cancelled externally.
        for t in tasks:
            if not t.done():
                t.cancel()

    if cost_state.aborted:
        notes.append(f"Run aborted: cost ceiling ${config.max_cost_usd:.2f} exceeded.")

    finished_at = now()

    summary = _summarise(case_results, config.models)

    result = RunResult(
        run_id=run_id,
        dataset=dataset.name,
        dataset_version=dataset.version,
        config=config,
        cases=sorted(case_results, key=lambda r: (r.case_id, r.model)),
        summary=summary,
        started_at=started_at,
        finished_at=finished_at,
        pricing_version=effective_version(),
        deterministic=deterministic,
        notes=notes,
    )

    self._emit(
        RunEvent(
            type="run.finished",
            run_id=run_id,
            at=finished_at,
            payload={
                "summary": {m: s.model_dump(mode="json") for m, s in summary.items()},
                "aborted": cost_state.aborted,
            },
        )
    )
    return result

run_sync ¶

run_sync(dataset: Dataset, config: RunConfig) -> RunResult

Synchronous wrapper. Spins up a fresh event loop and blocks.

Adapters the runner instantiated itself are closed before the loop is torn down (their HTTP clients are bound to it). Explicitly injected adapters stay open — the caller owns their lifecycle.

Source code in src/clean_evals/runner.py

def run_sync(self, dataset: Dataset, config: RunConfig) -> RunResult:
    """Synchronous wrapper. Spins up a fresh event loop and blocks.

    Adapters the runner instantiated itself are closed before the loop
    is torn down (their HTTP clients are bound to it). Explicitly
    injected adapters stay open — the caller owns their lifecycle.
    """

    async def _run_and_close() -> RunResult:
        try:
            return await self.run(dataset, config)
        finally:
            await self.aclose()

    return asyncio.run(_run_and_close())

Plugin protocols¶

clean_evals.Scorer ¶

Bases: Protocol

Computes a 0.0–1.0 quality score for a model's output.

Implementations should be pure: same (case, response) always yields the same ScoreResult. Anything stochastic (LLM judges) should be seeded so the run is reproducible.

Class attributes

name: The registry key under which this scorer is discoverable. Datasets reference it via Dataset.scorer.

Source code in src/clean_evals/protocols.py

@runtime_checkable
class Scorer(Protocol):
    """Computes a 0.0–1.0 quality score for a model's output.

    Implementations should be pure: same ``(case, response)`` always yields
    the same ``ScoreResult``. Anything stochastic (LLM judges) should be
    seeded so the run is reproducible.

    Class attributes:
        name: The registry key under which this scorer is discoverable.
            Datasets reference it via ``Dataset.scorer``.
    """

    name: ClassVar[str]

    def score(self, case: Case, response: ModelResponse) -> ScoreResult: ...

    @classmethod
    def from_config(cls, config: dict[str, Any]) -> Scorer: ...

clean_evals.ModelAdapter ¶

Bases: Protocol

Talks to a single model provider.

Adapters MUST:

Be async-native (use httpx.AsyncClient, never requests).
Validate that the requested model is a dated snapshot. Floating aliases like -latest are rejected at RunConfig validation, but adapters should also defend in depth.
Return ModelResponse.cost_usd populated via clean_evals.pricing using the prompt's actual token counts.
On HTTP 429, raise RateLimited with the Retry-After value if present so the runner can back off.
On other transient failures, raise the standard exception types from :mod:clean_evals.errors so the runner can retry consistently.

Source code in src/clean_evals/protocols.py

@runtime_checkable
class ModelAdapter(Protocol):
    """Talks to a single model provider.

    Adapters MUST:

    - Be ``async``-native (use ``httpx.AsyncClient``, never ``requests``).
    - Validate that the requested ``model`` is a dated snapshot. Floating
      aliases like ``-latest`` are rejected at ``RunConfig`` validation, but
      adapters should also defend in depth.
    - Return ``ModelResponse.cost_usd`` populated via ``clean_evals.pricing``
      using the prompt's actual token counts.
    - On HTTP 429, raise ``RateLimited`` with the ``Retry-After`` value if
      present so the runner can back off.
    - On other transient failures, raise the standard exception types from
      :mod:`clean_evals.errors` so the runner can retry consistently.
    """

    provider: ClassVar[str]

    async def complete(
        self,
        prompt: str,
        model: str,
        *,
        temperature: float,
        seed: int | None,
        timeout_s: float,
        response_format: Literal["text", "json"] = "text",
        system: str | None = None,
        reasoning_effort: str | None = None,
        max_output_tokens: int | None = None,
    ) -> ModelResponse: ...

clean_evals.Reporter ¶

Bases: Protocol

Writes a RunResult to a destination.

Reporters are invoked synchronously after the run completes. Each reporter MAY write multiple files; write returns the path of the primary artifact (the one humans/CI typically open first).

Source code in src/clean_evals/protocols.py

@runtime_checkable
class Reporter(Protocol):
    """Writes a ``RunResult`` to a destination.

    Reporters are invoked synchronously after the run completes. Each
    reporter MAY write multiple files; ``write`` returns the path of the
    primary artifact (the one humans/CI typically open first).
    """

    name: ClassVar[str]

    def write(self, result: RunResult, output_dir: Path) -> Path: ...

clean_evals.Scrubber ¶

Bases: Protocol

Optional plugin for cleaning PII / sensitive data during dataset load.

Called by Dataset.from_yaml(..., scrubber=...) once per case after parsing. Implementations must be pure — same input, same output.

Source code in src/clean_evals/protocols.py

@runtime_checkable
class Scrubber(Protocol):
    """Optional plugin for cleaning PII / sensitive data during dataset load.

    Called by ``Dataset.from_yaml(..., scrubber=...)`` once per case after
    parsing. Implementations must be pure — same input, same output.
    """

    def scrub(self, case: Case) -> Case: ...

Public API¶

Stability tags¶

Data models¶

clean_evals.Case ¶

clean_evals.Dataset ¶

from_yaml classmethod ¶

to_yaml ¶

clean_evals.ModelResponse ¶

clean_evals.ScoreResult ¶

clean_evals.CaseResult ¶

clean_evals.RunConfig ¶

clean_evals.ModelSummary ¶

clean_evals.RunResult ¶

Runner¶

clean_evals.Runner ¶

aclose async ¶

run async ¶

run_sync ¶

Plugin protocols¶

clean_evals.Scorer ¶

clean_evals.ModelAdapter ¶

clean_evals.Reporter ¶

clean_evals.Scrubber ¶

from_yaml `classmethod` ¶

aclose `async` ¶

run `async` ¶