Root Package

The root package is intentionally small and benchmark-first.

Main imports:

• BenchmarkSpec
• SliceSpec
• DatasetQuerySpec
• PromptVariantSpec
• ParseSpec
• ScoreSpec
• ProjectSpec
• ModelSpec
• PluginRegistry
• Orchestrator
• BenchmarkResult
• generate_config_report

Curated public package surface for the current Themis runtime.

BenchmarkResult

Bases: ExperimentResult

Public result facade that speaks benchmark-native semantics.

Source code in themis/runtime/benchmark_result.py

class BenchmarkResult(ExperimentResult):
    """Public result facade that speaks benchmark-native semantics."""

    def __init__(
        self,
        *,
        projection_repo,
        trial_hashes: list[str],
        transform_hashes: list[str] | None = None,
        evaluation_hashes: list[str] | None = None,
        active_transform_hash: str | None = None,
        active_evaluation_hash: str | None = None,
        benchmark_id: str | None = None,
        slice_ids: list[str] | None = None,
        prompt_variant_ids: list[str] | None = None,
    ) -> None:
        super().__init__(
            projection_repo=projection_repo,
            trial_hashes=trial_hashes,
            transform_hashes=transform_hashes,
            evaluation_hashes=evaluation_hashes,
            active_transform_hash=active_transform_hash,
            active_evaluation_hash=active_evaluation_hash,
        )
        self.benchmark_id = benchmark_id
        self.slice_ids = list(slice_ids or [])
        self.prompt_variant_ids = list(prompt_variant_ids or [])

    def aggregate(
        self,
        *,
        group_by: list[str],
        metric_id: str | None = None,
    ) -> list[JSONDict]:
        """Aggregate score rows using benchmark-native summary fields."""

        summaries = {row.trial_hash: row for row in self.iter_trial_summaries()}
        self._validate_group_by_keys(summaries.values(), group_by)
        groups: dict[tuple[JSONValueType, ...], list[float]] = {}
        for row in self._iter_scores(metric_id=metric_id):
            summary = summaries.get(row.trial_hash)
            if summary is None:
                continue
            key_payload = self._group_payload(summary, row, group_by)
            groups.setdefault(tuple(key_payload.values()), []).append(row.score)

        results: list[JSONDict] = []
        for key, scores in sorted(
            groups.items(), key=lambda item: self._sort_group_key(item[0])
        ):
            payload = dict(zip(group_by, key, strict=True))
            payload["mean"] = sum(scores) / len(scores)
            payload["count"] = len(scores)
            results.append(payload)
        return results

    def paired_compare(
        self,
        *,
        metric_id: str,
        group_by: str = "slice_id",
        baseline_model_id: str | None = None,
        treatment_model_id: str | None = None,
        p_value_correction: PValueCorrection | str = PValueCorrection.NONE,
    ) -> list[JSONDict]:
        """Return paired comparisons by one benchmark grouping key."""

        trial_summaries = list(self.iter_trial_summaries())
        self._validate_group_by_keys(trial_summaries, [group_by])
        relevant_scores = list(self._iter_scores(metric_id=metric_id))
        scores_by_trial: dict[str, list[ScoreRow]] = {}
        for row in relevant_scores:
            scores_by_trial.setdefault(row.trial_hash, []).append(row)

        summaries_by_group: dict[JSONValueType, list[TrialSummaryRow]] = {}
        for summary in trial_summaries:
            group_value = self._resolve_group_value(
                summary,
                group_by,
                metric_id=metric_id,
            )
            summaries_by_group.setdefault(group_value, []).append(summary)

        comparison_rows: list[JSONDict] = []
        for group_value in sorted(summaries_by_group, key=self._group_value_sort_key):
            group_summaries = summaries_by_group[group_value]
            group_trial_hashes = {summary.trial_hash for summary in group_summaries}
            group_scores = [
                score
                for trial_hash in group_trial_hashes
                for score in scores_by_trial.get(trial_hash, [])
            ]
            table = build_comparison_table(
                group_summaries,
                group_scores,
                metric_id=metric_id,
                baseline_model_id=baseline_model_id,
                treatment_model_id=treatment_model_id,
                p_value_correction=p_value_correction,
            )
            for comparison_row in table.rows:
                payload = {
                    group_by: group_value,
                    "metric_id": comparison_row.metric_id,
                    "baseline_model_id": comparison_row.baseline_model_id,
                    "treatment_model_id": comparison_row.treatment_model_id,
                    "pair_count": comparison_row.pair_count,
                    "baseline_mean": comparison_row.baseline_mean,
                    "treatment_mean": comparison_row.treatment_mean,
                    "delta_mean": comparison_row.delta_mean,
                    "p_value": comparison_row.p_value,
                    "adjusted_p_value": comparison_row.adjusted_p_value,
                    "adjustment_method": comparison_row.adjustment_method,
                    "ci_lower": comparison_row.ci_lower,
                    "ci_upper": comparison_row.ci_upper,
                    "ci_level": comparison_row.ci_level,
                    "method": comparison_row.method,
                }
                comparison_rows.append(payload)
        return comparison_rows

    def for_transform(self, transform_hash: str) -> "BenchmarkResult":
        return BenchmarkResult(
            projection_repo=self.projection_repo,
            trial_hashes=self.trial_hashes,
            transform_hashes=self.transform_hashes,
            evaluation_hashes=self.evaluation_hashes,
            active_transform_hash=transform_hash,
            active_evaluation_hash=None,
            benchmark_id=self.benchmark_id,
            slice_ids=self.slice_ids,
            prompt_variant_ids=self.prompt_variant_ids,
        )

    def for_evaluation(self, evaluation_hash: str) -> "BenchmarkResult":
        return BenchmarkResult(
            projection_repo=self.projection_repo,
            trial_hashes=self.trial_hashes,
            transform_hashes=self.transform_hashes,
            evaluation_hashes=self.evaluation_hashes,
            active_transform_hash=None,
            active_evaluation_hash=evaluation_hash,
            benchmark_id=self.benchmark_id,
            slice_ids=self.slice_ids,
            prompt_variant_ids=self.prompt_variant_ids,
        )

    def persist_artifacts(
        self,
        *,
        storage_root: str | Path,
    ) -> ArtifactBundle:
        """Persist a small aggregate bundle for operator handoff."""

        root = Path(storage_root)
        root.mkdir(parents=True, exist_ok=True)
        aggregate_rows = self.aggregate(
            group_by=["model_id", "slice_id", "metric_id", "prompt_variant_id"]
        )
        scope = self._scope_metadata()
        scope_suffix = scope["overlay_key"].replace(":", "-")
        aggregate_path = root / f"benchmark-aggregate-{scope_suffix}.json"
        summary_path = root / f"benchmark-summary-{scope_suffix}.md"
        aggregate_path.write_text(
            json.dumps(
                {
                    "benchmark_id": self.benchmark_id,
                    "scope": scope,
                    "rows": aggregate_rows,
                },
                indent=2,
                sort_keys=True,
            )
        )
        summary_lines = ["# Benchmark Summary", ""]
        for row in aggregate_rows:
            mean_value = self._float_value(row, "mean")
            count_value = self._int_value(row, "count")
            summary_lines.append(
                "- "
                f"scope={scope['overlay_key']} "
                f"model={row.get('model_id')} "
                f"slice={row.get('slice_id')} "
                f"metric={row.get('metric_id')} "
                f"prompt={row.get('prompt_variant_id')} "
                f"mean={mean_value:.4f} "
                f"count={count_value}"
            )
        summary_path.write_text("\n".join(summary_lines) + "\n")
        return ArtifactBundle(
            aggregate_json_path=aggregate_path,
            summary_markdown_path=summary_path,
        )

    def _iter_scores(self, *, metric_id: str | None) -> Iterator[ScoreRow]:
        yield from self.projection_repo.iter_candidate_scores(
            trial_hashes=self.trial_hashes,
            metric_id=metric_id,
            evaluation_hash=self.active_evaluation_hash,
        )

    def _group_payload(
        self,
        summary: TrialSummaryRow,
        score_row: ScoreRow,
        group_by: list[str],
    ) -> JSONDict:
        payload: JSONDict = {}
        for key in group_by:
            payload[key] = self._resolve_group_value(
                summary,
                key,
                metric_id=score_row.metric_id,
            )
        return payload

    def _resolve_group_value(
        self,
        summary: TrialSummaryRow,
        key: str,
        *,
        metric_id: str | None = None,
    ) -> JSONValueType:
        if key == "metric_id":
            return metric_id
        if key in {
            "benchmark_id",
            "slice_id",
            "prompt_variant_id",
            "model_id",
            "item_id",
            "status",
        }:
            return getattr(summary, key)
        if key in summary.dimensions:
            return summary.dimensions[key]
        return None

    def _sort_group_key(
        self, values: tuple[JSONValueType, ...]
    ) -> tuple[tuple[int, str], ...]:
        return tuple(self._group_value_sort_key(value) for value in values)

    def _group_value_sort_key(self, value: JSONValueType) -> tuple[int, str]:
        if value is None:
            return (0, "")
        return (1, str(value))

    def _float_value(self, row: JSONDict, key: str) -> float:
        value = row.get(key)
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise TypeError(f"{key} must be numeric, got {value!r}")
        return float(value)

    def _int_value(self, row: JSONDict, key: str) -> int:
        value = row.get(key)
        if isinstance(value, bool) or not isinstance(value, int):
            raise TypeError(f"{key} must be an int, got {value!r}")
        return value

    def _scope_metadata(self) -> dict[str, str]:
        return OverlaySelection(
            transform_hash=self.active_transform_hash,
            evaluation_hash=self.active_evaluation_hash,
        ).metadata()

    def _validate_group_by_keys(
        self,
        summaries: Iterable[TrialSummaryRow],
        group_by: list[str],
    ) -> None:
        supported_keys = {
            "metric_id",
            "benchmark_id",
            "slice_id",
            "prompt_variant_id",
            "model_id",
            "item_id",
            "status",
        }
        dimension_keys = {key for summary in summaries for key in summary.dimensions}
        unknown_keys = sorted(set(group_by) - supported_keys - dimension_keys)
        if unknown_keys:
            raise ValueError(f"Unsupported group_by key: {', '.join(unknown_keys)}")

aggregate

aggregate(
    *, group_by: list[str], metric_id: str | None = None
) -> list[JSONDict]

Aggregate score rows using benchmark-native summary fields.

Source code in themis/runtime/benchmark_result.py

def aggregate(
    self,
    *,
    group_by: list[str],
    metric_id: str | None = None,
) -> list[JSONDict]:
    """Aggregate score rows using benchmark-native summary fields."""

    summaries = {row.trial_hash: row for row in self.iter_trial_summaries()}
    self._validate_group_by_keys(summaries.values(), group_by)
    groups: dict[tuple[JSONValueType, ...], list[float]] = {}
    for row in self._iter_scores(metric_id=metric_id):
        summary = summaries.get(row.trial_hash)
        if summary is None:
            continue
        key_payload = self._group_payload(summary, row, group_by)
        groups.setdefault(tuple(key_payload.values()), []).append(row.score)

    results: list[JSONDict] = []
    for key, scores in sorted(
        groups.items(), key=lambda item: self._sort_group_key(item[0])
    ):
        payload = dict(zip(group_by, key, strict=True))
        payload["mean"] = sum(scores) / len(scores)
        payload["count"] = len(scores)
        results.append(payload)
    return results

paired_compare

paired_compare(
    *,
    metric_id: str,
    group_by: str = "slice_id",
    baseline_model_id: str | None = None,
    treatment_model_id: str | None = None,
    p_value_correction: PValueCorrection | str = PValueCorrection.NONE,
) -> list[JSONDict]

Return paired comparisons by one benchmark grouping key.

Source code in themis/runtime/benchmark_result.py

def paired_compare(
    self,
    *,
    metric_id: str,
    group_by: str = "slice_id",
    baseline_model_id: str | None = None,
    treatment_model_id: str | None = None,
    p_value_correction: PValueCorrection | str = PValueCorrection.NONE,
) -> list[JSONDict]:
    """Return paired comparisons by one benchmark grouping key."""

    trial_summaries = list(self.iter_trial_summaries())
    self._validate_group_by_keys(trial_summaries, [group_by])
    relevant_scores = list(self._iter_scores(metric_id=metric_id))
    scores_by_trial: dict[str, list[ScoreRow]] = {}
    for row in relevant_scores:
        scores_by_trial.setdefault(row.trial_hash, []).append(row)

    summaries_by_group: dict[JSONValueType, list[TrialSummaryRow]] = {}
    for summary in trial_summaries:
        group_value = self._resolve_group_value(
            summary,
            group_by,
            metric_id=metric_id,
        )
        summaries_by_group.setdefault(group_value, []).append(summary)

    comparison_rows: list[JSONDict] = []
    for group_value in sorted(summaries_by_group, key=self._group_value_sort_key):
        group_summaries = summaries_by_group[group_value]
        group_trial_hashes = {summary.trial_hash for summary in group_summaries}
        group_scores = [
            score
            for trial_hash in group_trial_hashes
            for score in scores_by_trial.get(trial_hash, [])
        ]
        table = build_comparison_table(
            group_summaries,
            group_scores,
            metric_id=metric_id,
            baseline_model_id=baseline_model_id,
            treatment_model_id=treatment_model_id,
            p_value_correction=p_value_correction,
        )
        for comparison_row in table.rows:
            payload = {
                group_by: group_value,
                "metric_id": comparison_row.metric_id,
                "baseline_model_id": comparison_row.baseline_model_id,
                "treatment_model_id": comparison_row.treatment_model_id,
                "pair_count": comparison_row.pair_count,
                "baseline_mean": comparison_row.baseline_mean,
                "treatment_mean": comparison_row.treatment_mean,
                "delta_mean": comparison_row.delta_mean,
                "p_value": comparison_row.p_value,
                "adjusted_p_value": comparison_row.adjusted_p_value,
                "adjustment_method": comparison_row.adjustment_method,
                "ci_lower": comparison_row.ci_lower,
                "ci_upper": comparison_row.ci_upper,
                "ci_level": comparison_row.ci_level,
                "method": comparison_row.method,
            }
            comparison_rows.append(payload)
    return comparison_rows

persist_artifacts

persist_artifacts(*, storage_root: str | Path) -> ArtifactBundle

Persist a small aggregate bundle for operator handoff.

Source code in themis/runtime/benchmark_result.py

def persist_artifacts(
    self,
    *,
    storage_root: str | Path,
) -> ArtifactBundle:
    """Persist a small aggregate bundle for operator handoff."""

    root = Path(storage_root)
    root.mkdir(parents=True, exist_ok=True)
    aggregate_rows = self.aggregate(
        group_by=["model_id", "slice_id", "metric_id", "prompt_variant_id"]
    )
    scope = self._scope_metadata()
    scope_suffix = scope["overlay_key"].replace(":", "-")
    aggregate_path = root / f"benchmark-aggregate-{scope_suffix}.json"
    summary_path = root / f"benchmark-summary-{scope_suffix}.md"
    aggregate_path.write_text(
        json.dumps(
            {
                "benchmark_id": self.benchmark_id,
                "scope": scope,
                "rows": aggregate_rows,
            },
            indent=2,
            sort_keys=True,
        )
    )
    summary_lines = ["# Benchmark Summary", ""]
    for row in aggregate_rows:
        mean_value = self._float_value(row, "mean")
        count_value = self._int_value(row, "count")
        summary_lines.append(
            "- "
            f"scope={scope['overlay_key']} "
            f"model={row.get('model_id')} "
            f"slice={row.get('slice_id')} "
            f"metric={row.get('metric_id')} "
            f"prompt={row.get('prompt_variant_id')} "
            f"mean={mean_value:.4f} "
            f"count={count_value}"
        )
    summary_path.write_text("\n".join(summary_lines) + "\n")
    return ArtifactBundle(
        aggregate_json_path=aggregate_path,
        summary_markdown_path=summary_path,
    )

BenchmarkSpec

Bases: SpecBase

Top-level benchmark configuration compiled into an execution plan.

Source code in themis/benchmark/specs.py

class BenchmarkSpec(SpecBase):
    """Top-level benchmark configuration compiled into an execution plan."""

    benchmark_id: str = Field(..., min_length=1)
    models: list[ModelSpec] = Field(..., min_length=1)
    slices: list[SliceSpec] = Field(..., min_length=1)
    prompt_variants: list[PromptVariantSpec] = Field(..., min_length=1)
    inference_grid: InferenceGridSpec = Field(...)
    num_samples: int = Field(default=1, ge=1)

    @model_validator(mode="after")
    def _validate_semantic(self) -> "BenchmarkSpec":
        slice_ids = [slice_spec.slice_id for slice_spec in self.slices]
        if len(slice_ids) != len(set(slice_ids)):
            raise ValueError(
                f"BenchmarkSpec '{self.benchmark_id}' has duplicate slice_id."
            )
        prompt_variant_ids = [
            prompt_variant.id for prompt_variant in self.prompt_variants
        ]
        if len(prompt_variant_ids) != len(set(prompt_variant_ids)):
            raise ValueError(
                f"BenchmarkSpec '{self.benchmark_id}' has duplicate prompt variant id."
            )
        valid_prompt_variant_ids = set(prompt_variant_ids)
        for slice_spec in self.slices:
            missing_prompt_variant_ids = sorted(
                set(slice_spec.prompt_variant_ids) - valid_prompt_variant_ids
            )
            if missing_prompt_variant_ids:
                missing_joined = ", ".join(missing_prompt_variant_ids)
                raise ValueError(
                    f"SliceSpec '{slice_spec.slice_id}' references unknown prompt "
                    f"variant id(s): {missing_joined}."
                )
        return self

DatasetQuerySpec

Bases: SpecBase

Declarative slice query and sampling controls for dataset providers.

Source code in themis/benchmark/query.py

class DatasetQuerySpec(SpecBase):
    """Declarative slice query and sampling controls for dataset providers."""

    kind: SamplingKind = Field(default=SamplingKind.ALL)
    count: int | None = Field(default=None, gt=0)
    seed: int | None = Field(default=None)
    strata_field: str | None = Field(default=None)
    item_ids: list[str] = Field(default_factory=list)
    metadata_filters: dict[str, str] = Field(default_factory=dict)
    projected_fields: list[str] = Field(default_factory=list)

    @field_validator("kind", mode="before")
    @classmethod
    def _coerce_kind(cls, value: SamplingKind | str) -> SamplingKind:
        if isinstance(value, str):
            return SamplingKind(value)
        return value

    @classmethod
    def all(cls) -> "DatasetQuerySpec":
        return cls(kind=SamplingKind.ALL)

    @classmethod
    def subset(
        cls,
        count: int,
        *,
        seed: int | None = None,
    ) -> "DatasetQuerySpec":
        return cls(kind=SamplingKind.SUBSET, count=count, seed=seed)

    @classmethod
    def stratified(
        cls,
        count: int,
        *,
        strata_field: str,
        seed: int | None = None,
    ) -> "DatasetQuerySpec":
        return cls(
            kind=SamplingKind.STRATIFIED,
            count=count,
            seed=seed,
            strata_field=strata_field,
        )

    @model_validator(mode="after")
    def _validate_semantic(self) -> "DatasetQuerySpec":
        if (
            self.kind in {SamplingKind.SUBSET, SamplingKind.STRATIFIED}
            and self.count is None
        ):
            raise ValueError(
                f"DatasetQuerySpec kind='{self.kind.value}' requires a positive count."
            )
        if self.kind == SamplingKind.STRATIFIED and not self.strata_field:
            raise ValueError(
                "DatasetQuerySpec kind='stratified' requires strata_field."
            )
        if self.item_ids and self.kind in {
            SamplingKind.SUBSET,
            SamplingKind.STRATIFIED,
        }:
            raise ValueError(
                "DatasetQuerySpec item_ids is mutually exclusive with count-based "
                "sampling kinds."
            )
        return self

ExecutionPolicySpec

Bases: SpecBase

Retry, backoff, circuit-breaker, and concurrency controls for orchestration.

These settings live above provider SDK behavior. Engines are still responsible for classifying provider failures into stable retryable codes.

Source code in themis/specs/experiment.py

class ExecutionPolicySpec(SpecBase):
    """Retry, backoff, circuit-breaker, and concurrency controls for orchestration.

    These settings live above provider SDK behavior. Engines are still
    responsible for classifying provider failures into stable retryable codes.
    """

    max_retries: int = Field(default=3, ge=0)
    retry_backoff_factor: float = Field(default=1.5, gt=0.0)
    circuit_breaker_threshold: int = Field(default=5, ge=1)
    max_in_flight_work_items: int = Field(default=32, ge=1)
    retryable_error_codes: list[str] = Field(
        default_factory=list,
        description="Stable error-code values treated as retryable for persisted work items.",
    )

InferenceGridSpec

Bases: SpecBase

Typed inference sweep over base params and scalar override grids.

Use this for temperature, top-p, or provider-extra sweeps while keeping unchanged parameter combinations resumable across runs.

Source code in themis/specs/experiment.py

class InferenceGridSpec(SpecBase):
    """Typed inference sweep over base params and scalar override grids.

    Use this for temperature, top-p, or provider-extra sweeps while keeping
    unchanged parameter combinations resumable across runs.
    """

    params: list[InferenceParamsSpec] = Field(..., min_length=1)
    overrides: dict[str, list[str | int | float | bool]] = Field(default_factory=dict)

    def expand(self) -> list[InferenceParamsSpec]:
        """Expand the base inference params over all configured overrides."""
        if not self.overrides:
            return list(self.params)

        expanded: list[InferenceParamsSpec] = []
        override_keys = sorted(self.overrides)
        override_values = [self.overrides[key] for key in override_keys]

        for base in self.params:
            base_payload = base.model_dump()
            for combination in itertools.product(*override_values):
                payload = dict(base_payload)
                extras = dict(payload.get("extras", {}))
                for key, value in zip(override_keys, combination):
                    if key in InferenceParamsSpec.model_fields:
                        payload[key] = value
                    else:
                        extras[key] = value
                if extras:
                    payload["extras"] = extras
                expanded.append(InferenceParamsSpec.model_validate(payload))
        return expanded

expand

expand() -> list[InferenceParamsSpec]

Expand the base inference params over all configured overrides.

Source code in themis/specs/experiment.py

def expand(self) -> list[InferenceParamsSpec]:
    """Expand the base inference params over all configured overrides."""
    if not self.overrides:
        return list(self.params)

    expanded: list[InferenceParamsSpec] = []
    override_keys = sorted(self.overrides)
    override_values = [self.overrides[key] for key in override_keys]

    for base in self.params:
        base_payload = base.model_dump()
        for combination in itertools.product(*override_values):
            payload = dict(base_payload)
            extras = dict(payload.get("extras", {}))
            for key, value in zip(override_keys, combination):
                if key in InferenceParamsSpec.model_fields:
                    payload[key] = value
                else:
                    extras[key] = value
            if extras:
                payload["extras"] = extras
            expanded.append(InferenceParamsSpec.model_validate(payload))
    return expanded

InferenceParamsSpec

Bases: SpecBase

Sampling and response-shape settings forwarded to inference engines.

Source code in themis/specs/experiment.py

class InferenceParamsSpec(SpecBase):
    """Sampling and response-shape settings forwarded to inference engines."""

    temperature: float = Field(
        default=0.0, ge=0.0, description="Sampling randomness. 0.0 is deterministic."
    )
    top_p: float | None = Field(
        default=None, ge=0.0, le=1.0, description="Nucleus sampling threshold."
    )
    top_k: int | None = Field(default=None, ge=0, description="Top-k token threshold.")
    max_tokens: int = Field(
        default=1024, gt=0, description="Max string length generated."
    )
    stop_sequences: list[str] = Field(
        default_factory=list, description="Sequences that end generation."
    )
    logprobs: int | None = Field(
        default=None, ge=0, description="Request token logprobs if available."
    )
    response_format: ResponseFormat | None = Field(default=None)
    seed: int | None = Field(
        default=None, description="Optional deterministic PRNG seed."
    )
    extras: JSONDict = Field(
        default_factory=dict, description="Provider-specific sampling args."
    )

    @field_validator("response_format", mode="before")
    @classmethod
    def _coerce_response_format(
        cls, value: ResponseFormat | str | None
    ) -> ResponseFormat | str | None:
        if isinstance(value, str):
            return ResponseFormat(value)
        return value

ModelSpec

Bases: SpecBase

Configures one inference-engine target and its provider-specific extras.

Source code in themis/specs/foundational.py

class ModelSpec(SpecBase):
    """Configures one inference-engine target and its provider-specific extras."""

    model_id: str = Field(
        ..., description="The unique name/ID of the model (e.g., 'gpt-4')."
    )
    provider: str = Field(
        ..., description="The provider adapter to route to (e.g., 'openai')."
    )
    extras: JSONDict = Field(
        default_factory=dict, description="Provider-specific initialization arguments."
    )

Orchestrator

Main facade that plans, executes, projects, and returns experiment results.

Source code in themis/orchestration/orchestrator.py

class Orchestrator:
    """Main facade that plans, executes, projects, and returns experiment results."""

    @classmethod
    def from_project_spec(
        cls,
        project: ProjectSpec,
        *,
        registry: PluginRegistry | None = None,
        dataset_provider: DatasetProvider | None = None,
        dataset_loader: DatasetLoader | None = None,
        parallel_candidates: int = 5,
        telemetry_bus: TelemetryBus | None = None,
        storage_bundle: StorageBundle | None = None,
    ) -> Orchestrator:
        """Construct an orchestrator from a validated project specification."""
        return cls(
            registry or PluginRegistry(),
            storage_bundle or build_storage_bundle(project.storage),
            dataset_provider=dataset_provider,
            dataset_loader=dataset_loader,
            execution_policy=project.execution_policy,
            parallel_candidates=parallel_candidates,
            project_seed=project.global_seed,
            store_item_payloads=project.storage.store_item_payloads,
            telemetry_bus=telemetry_bus,
            project_spec=project,
            _allow_runtime_construction=True,
        )

    @classmethod
    def from_project_file(
        cls,
        path: str,
        *,
        registry: PluginRegistry | None = None,
        dataset_provider: DatasetProvider | None = None,
        dataset_loader: DatasetLoader | None = None,
        parallel_candidates: int = 5,
        telemetry_bus: TelemetryBus | None = None,
        storage_bundle: StorageBundle | None = None,
    ) -> Orchestrator:
        """Load a project file, validate it, and build an orchestrator."""
        project_path = Path(path)
        try:
            if project_path.suffix == ".toml":
                with project_path.open("rb") as fh:
                    project_data = tomllib.load(fh)
                project = ProjectSpec.model_validate(project_data)
            elif project_path.suffix == ".json":
                project = ProjectSpec.model_validate_json(project_path.read_text())
            else:
                raise ValueError("Project files must use .toml or .json.")
        except tomllib.TOMLDecodeError as exc:
            raise SpecValidationError(
                code=ErrorCode.SCHEMA_MISMATCH,
                message=f"Failed to parse project config {project_path.name}: {exc}",
            ) from exc
        except ValidationError as exc:
            raise SpecValidationError(
                code=ErrorCode.SCHEMA_MISMATCH,
                message=f"Failed to parse project config {project_path.name}: {format_validation_error(exc)}",
            ) from exc
        return cls.from_project_spec(
            project,
            registry=registry,
            dataset_provider=dataset_provider,
            dataset_loader=dataset_loader,
            parallel_candidates=parallel_candidates,
            telemetry_bus=telemetry_bus,
            storage_bundle=storage_bundle,
        )

    def __init__(
        self,
        registry: PluginRegistry | None = None,
        storage_bundle: StorageBundle | None = None,
        *,
        dataset_provider: DatasetProvider | None = None,
        dataset_loader: DatasetLoader | None = None,
        execution_policy: ExecutionPolicySpec | None = None,
        parallel_candidates: int = 5,
        project_seed: int | None = None,
        store_item_payloads: bool = True,
        telemetry_bus: TelemetryBus | None = None,
        project_spec: ProjectSpec | None = None,
        _allow_runtime_construction: bool = False,
    ) -> None:
        if not _allow_runtime_construction:
            raise TypeError(
                "Use Orchestrator.from_project_spec(...) or "
                "Orchestrator.from_project_file(...)."
            )
        if registry is None or storage_bundle is None or execution_policy is None:
            raise TypeError(
                "Internal orchestrator construction requires registry, "
                "storage_bundle, and execution_policy."
            )

        self.registry = registry
        self.dataset_provider = dataset_provider
        self.dataset_loader = dataset_loader
        self.project_spec = project_spec
        self.execution_policy = execution_policy
        self.telemetry_bus = telemetry_bus
        self._services: OrchestratorServices = build_orchestrator_services(
            registry=self.registry,
            storage_bundle=storage_bundle,
            dataset_provider=self.dataset_provider,
            dataset_loader=self.dataset_loader,
            execution_policy=self.execution_policy,
            parallel_candidates=parallel_candidates,
            project_seed=project_seed,
            store_item_payloads=store_item_payloads,
            telemetry_bus=self.telemetry_bus,
        )
        manifest_repo = RunManifestRepository(storage_bundle.manager)
        self._run_planning = RunPlanningService(
            planner=self._services.planner,
            event_repo=self._services.event_repo,
            projection_repo=self._services.projection_repo,
            projection_handler=self._services.projection_handler,
            manifest_repo=manifest_repo,
            project_spec=self.project_spec,
            registry=self.registry,
        )
        self._run_imports = RunImportService(
            event_repo=self._services.event_repo,
            projection_repo=self._services.projection_repo,
            projection_handler=self._services.projection_handler,
        )

    def run(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> ExperimentResult | BenchmarkResult:
        """Execute generation, transforms, and evaluations for one experiment."""
        normalized = self._normalize_source_spec(experiment)
        planned_trials = self._services.planner.plan_experiment(
            normalized.experiment_spec
        )
        progress_tracker = self._build_progress_tracker(
            normalized.experiment_spec,
            planned_trials,
            progress=progress,
            allowed_stages={
                RunStage.GENERATION,
                RunStage.TRANSFORM,
                RunStage.EVALUATION,
            },
            benchmark_spec=normalized.benchmark_spec,
        )
        pending_generation_trials = generation_trials(planned_trials)
        pending_transform_trials = transform_trials(planned_trials)
        pending_evaluation_trials = evaluation_trials(planned_trials)

        if progress_tracker is not None:
            progress_tracker.start_run()
        try:
            if pending_generation_trials:
                self._services.executor.execute_generation_trials(
                    pending_generation_trials,
                    runtime,
                    progress_tracker=progress_tracker,
                )
            if pending_transform_trials:
                self._services.executor.execute_transforms(
                    pending_transform_trials,
                    runtime,
                    progress_tracker=progress_tracker,
                )
            if pending_evaluation_trials:
                self._services.executor.execute_evaluations(
                    pending_evaluation_trials,
                    runtime,
                    progress_tracker=progress_tracker,
                )
        finally:
            if progress_tracker is not None:
                progress_tracker.finish_run()

        result = self._run_planning.build_result(
            planned_trials,
            transform_hashes=collect_transform_hashes(pending_transform_trials),
            evaluation_hashes=collect_evaluation_hashes(pending_evaluation_trials),
        )
        return self._wrap_source_result(normalized, result)

    def run_benchmark(
        self,
        benchmark: BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> BenchmarkResult:
        """Compile and execute one benchmark specification."""
        result = self.run(benchmark, runtime=runtime, progress=progress)
        if not isinstance(result, BenchmarkResult):
            raise TypeError(
                "Orchestrator.run_benchmark expected run() to return a "
                f"BenchmarkResult, got {type(result).__name__}."
            )
        return result

    def generate(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> ExperimentResult | BenchmarkResult:
        """Execute only generation-stage work for one experiment."""
        normalized = self._normalize_source_spec(experiment)
        planned_trials = generation_trials(
            self._services.planner.plan_experiment(
                normalized.experiment_spec,
                required_stages={RunStage.GENERATION},
            )
        )
        progress_tracker = self._build_progress_tracker(
            normalized.experiment_spec,
            planned_trials,
            progress=progress,
            allowed_stages={RunStage.GENERATION},
            benchmark_spec=normalized.benchmark_spec,
        )
        if planned_trials:
            if progress_tracker is not None:
                progress_tracker.start_run()
            try:
                self._services.executor.execute_generation_trials(
                    planned_trials,
                    runtime,
                    progress_tracker=progress_tracker,
                )
            finally:
                if progress_tracker is not None:
                    progress_tracker.finish_run()
        result = self._run_planning.build_result(planned_trials)
        return self._wrap_source_result(normalized, result)

    def transform(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> ExperimentResult | BenchmarkResult:
        """Execute output transforms against existing generation candidates."""
        normalized = self._normalize_source_spec(experiment)
        planned_trials = transform_trials(
            self._services.planner.plan_experiment(
                normalized.experiment_spec,
                required_stages={RunStage.TRANSFORM},
            )
        )
        transform_hashes = collect_transform_hashes(planned_trials)
        progress_tracker = self._build_progress_tracker(
            normalized.experiment_spec,
            planned_trials,
            progress=progress,
            allowed_stages={RunStage.TRANSFORM},
            benchmark_spec=normalized.benchmark_spec,
        )
        if planned_trials:
            if progress_tracker is not None:
                progress_tracker.start_run()
            try:
                self._services.executor.execute_transforms(
                    planned_trials,
                    runtime,
                    progress_tracker=progress_tracker,
                )
            finally:
                if progress_tracker is not None:
                    progress_tracker.finish_run()
        result = self._run_planning.build_result(
            planned_trials,
            transform_hashes=transform_hashes,
        )
        return self._wrap_source_result(normalized, result)

    def evaluate(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> ExperimentResult | BenchmarkResult:
        """Execute evaluation-stage work, reusing generation when possible."""
        normalized = self._normalize_source_spec(experiment)
        planned_trials = evaluation_trials(
            self._services.planner.plan_experiment(
                normalized.experiment_spec,
                required_stages={RunStage.TRANSFORM, RunStage.EVALUATION},
            )
        )
        progress_tracker = self._build_progress_tracker(
            normalized.experiment_spec,
            planned_trials,
            progress=progress,
            allowed_stages={RunStage.TRANSFORM, RunStage.EVALUATION},
            benchmark_spec=normalized.benchmark_spec,
        )
        if planned_trials:
            if progress_tracker is not None:
                progress_tracker.start_run()
            try:
                self._services.executor.execute_transforms(
                    planned_trials,
                    runtime,
                    resume=True,
                    progress_tracker=progress_tracker,
                )
                self._services.executor.execute_evaluations(
                    planned_trials,
                    runtime,
                    resume=True,
                    progress_tracker=progress_tracker,
                )
            finally:
                if progress_tracker is not None:
                    progress_tracker.finish_run()
        result = self._run_planning.build_result(
            planned_trials,
            transform_hashes=collect_transform_hashes(planned_trials),
            evaluation_hashes=collect_evaluation_hashes(planned_trials),
        )
        return self._wrap_source_result(normalized, result)

    def import_candidates(
        self,
        trial_records: list[TrialRecord],
    ) -> ExperimentResult:
        """Import prebuilt generation artifacts into the current store."""
        trial_hashes = self._run_imports.import_candidates(trial_records)
        return ExperimentResult(
            projection_repo=self._services.projection_repo,
            trial_hashes=list(trial_hashes),
        )

    def plan(self, experiment: ExperimentSpec | BenchmarkSpec) -> RunManifest:
        """Build and persist a deterministic run manifest for one experiment."""
        normalized = self._normalize_source_spec(experiment)
        return self._run_planning.plan(
            normalized.experiment_spec,
            benchmark_spec=normalized.benchmark_spec,
        )

    def diff_specs(
        self,
        baseline: ExperimentSpec | BenchmarkSpec,
        treatment: ExperimentSpec | BenchmarkSpec,
    ) -> RunDiff:
        """Return a high-level diff between two experiment specifications."""
        normalized_baseline = self._normalize_source_spec(baseline)
        normalized_treatment = self._normalize_source_spec(treatment)
        return self._run_planning.diff_specs(
            normalized_baseline.experiment_spec,
            normalized_treatment.experiment_spec,
            baseline_source_spec=normalized_baseline.source_spec,
            treatment_source_spec=normalized_treatment.source_spec,
        )

    def submit(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> RunHandle:
        """Persist one run manifest and start execution if the backend is local."""
        normalized = self._normalize_source_spec(experiment)
        return self._run_planning.submit(
            normalized.experiment_spec,
            benchmark_spec=normalized.benchmark_spec,
            runtime=runtime,
            execute_run=lambda spec, runtime_context: self.run(
                spec,
                runtime=runtime_context,
                progress=progress,
            ),
        )

    def resume(
        self,
        run_id: str,
        *,
        runtime: RuntimeContext | None = None,
        progress: ProgressConfig | None = None,
    ) -> RunHandle | ExperimentResult | BenchmarkResult:
        """Refresh a persisted run and continue it when possible."""
        resumed = self._run_planning.resume(
            run_id,
            runtime=runtime,
            execute_run=lambda spec, runtime_context: self.run(
                spec,
                runtime=runtime_context,
                progress=progress,
            ),
        )
        if isinstance(resumed, ExperimentResult):
            manifest = self._run_planning.manifest_repo.get_manifest(run_id)
            if manifest is not None:
                return self._wrap_manifest_result(manifest, resumed)
        return resumed

    def get_run_progress(self, run_id: str) -> RunProgressSnapshot | None:
        """Return the persisted progress snapshot for one known run."""
        return self._run_planning.get_progress_snapshot(run_id)

    def estimate(self, experiment: ExperimentSpec | BenchmarkSpec) -> CostEstimate:
        """Return a best-effort work-item and token estimate for an experiment."""
        normalized = self._normalize_source_spec(experiment)
        return self._run_planning.estimate(normalized.experiment_spec)

    def _build_progress_tracker(
        self,
        experiment: ExperimentSpec,
        planned_trials: list[PlannedTrial],
        *,
        progress: ProgressConfig | None,
        allowed_stages: set[RunStage],
        benchmark_spec: BenchmarkSpec | None = None,
    ) -> RunProgressTracker | None:
        if progress is None:
            return None
        manifest = self._run_planning.plan_from_trials(
            experiment,
            planned_trials,
            benchmark_spec=benchmark_spec,
        )
        return RunProgressTracker(
            manifest,
            self._run_planning.manifest_repo,
            progress,
            frozenset(allowed_stages),
        )

    def export_generation_bundle(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
    ) -> GenerationWorkBundle:
        """Export only pending generation items for an experiment."""
        normalized = self._normalize_source_spec(experiment)
        return self._run_planning.export_generation_bundle(
            normalized.experiment_spec,
            benchmark_spec=normalized.benchmark_spec,
        )

    def import_generation_results(
        self,
        bundle: GenerationWorkBundle,
        trial_records: list[TrialRecord],
    ) -> ExperimentResult | BenchmarkResult:
        """Import externally generated results for a previously exported bundle."""
        self._run_imports.import_generation_results(bundle, trial_records)
        manifest = self.plan(self._manifest_source_spec(bundle.manifest))
        result = self._run_planning.result_from_manifest(manifest)
        return self._wrap_manifest_result(manifest, result)

    def export_evaluation_bundle(
        self,
        experiment: ExperimentSpec | BenchmarkSpec,
    ) -> EvaluationWorkBundle:
        """Export only pending evaluation items for an experiment."""
        normalized = self._normalize_source_spec(experiment)
        return self._run_planning.export_evaluation_bundle(
            normalized.experiment_spec,
            benchmark_spec=normalized.benchmark_spec,
        )

    def import_evaluation_results(
        self,
        bundle: EvaluationWorkBundle,
        trial_records: list[TrialRecord],
    ) -> ExperimentResult | BenchmarkResult:
        """Import externally evaluated results for a previously exported bundle."""
        self._run_imports.import_evaluation_results(bundle, trial_records)
        manifest = self.plan(self._manifest_source_spec(bundle.manifest))
        result = self._run_planning.result_from_manifest(manifest)
        return self._wrap_manifest_result(manifest, result)

    @property
    def db_manager(self) -> StorageConnectionManager:
        """Return the active backend-specific database manager for this orchestrator."""
        return self._services.storage_bundle.manager

    def _benchmark_result_from_public_spec(
        self,
        benchmark: BenchmarkSpec,
        result: ExperimentResult,
    ) -> BenchmarkResult:
        prompt_variant_ids = [
            prompt.id for prompt in benchmark.prompt_variants if prompt.id is not None
        ]
        return BenchmarkResult(
            projection_repo=result.projection_repo,
            trial_hashes=result.trial_hashes,
            transform_hashes=result.transform_hashes,
            evaluation_hashes=result.evaluation_hashes,
            active_transform_hash=result.active_transform_hash,
            active_evaluation_hash=result.active_evaluation_hash,
            benchmark_id=benchmark.benchmark_id,
            slice_ids=[slice_spec.slice_id for slice_spec in benchmark.slices],
            prompt_variant_ids=prompt_variant_ids,
        )

    def _wrap_source_result(
        self,
        normalized: _NormalizedSourceSpec,
        result: ExperimentResult,
    ) -> ExperimentResult | BenchmarkResult:
        if normalized.benchmark_spec is not None:
            return self._benchmark_result_from_public_spec(
                normalized.benchmark_spec,
                result,
            )
        return result

    def _wrap_manifest_result(
        self,
        manifest: RunManifest,
        result: ExperimentResult,
    ) -> ExperimentResult | BenchmarkResult:
        if manifest.benchmark_spec is not None:
            return self._benchmark_result_from_public_spec(
                manifest.benchmark_spec,
                result,
            )
        experiment = manifest.experiment_spec
        benchmark_ids = {
            task.benchmark_id
            for task in experiment.tasks
            if task.benchmark_id is not None
        }
        if not benchmark_ids:
            return result
        slice_ids = [
            task.slice_id or task.task_id
            for task in experiment.tasks
            if (task.slice_id or task.task_id) is not None
        ]
        prompt_variant_ids = [
            prompt.id for prompt in experiment.prompt_templates if prompt.id is not None
        ]
        return BenchmarkResult(
            projection_repo=result.projection_repo,
            trial_hashes=result.trial_hashes,
            transform_hashes=result.transform_hashes,
            evaluation_hashes=result.evaluation_hashes,
            active_transform_hash=result.active_transform_hash,
            active_evaluation_hash=result.active_evaluation_hash,
            benchmark_id=sorted(benchmark_ids)[0],
            slice_ids=slice_ids,
            prompt_variant_ids=prompt_variant_ids,
        )

    def _manifest_source_spec(
        self, manifest: RunManifest
    ) -> ExperimentSpec | BenchmarkSpec:
        if manifest.source_kind == "benchmark" and manifest.benchmark_spec is not None:
            return manifest.benchmark_spec
        return manifest.experiment_spec

    def _normalize_source_spec(
        self,
        source_spec: ExperimentSpec | BenchmarkSpec,
    ) -> _NormalizedSourceSpec:
        if isinstance(source_spec, BenchmarkSpec):
            return _NormalizedSourceSpec(
                source_kind="benchmark",
                source_spec=source_spec,
                experiment_spec=compile_benchmark(source_spec),
                benchmark_spec=source_spec,
            )
        return _NormalizedSourceSpec(
            source_kind="experiment",
            source_spec=source_spec,
            experiment_spec=source_spec,
            benchmark_spec=None,
        )

db_manager property

db_manager: StorageConnectionManager

Return the active backend-specific database manager for this orchestrator.

diff_specs

diff_specs(
    baseline: ExperimentSpec | BenchmarkSpec,
    treatment: ExperimentSpec | BenchmarkSpec,
) -> RunDiff

Return a high-level diff between two experiment specifications.

Source code in themis/orchestration/orchestrator.py

def diff_specs(
    self,
    baseline: ExperimentSpec | BenchmarkSpec,
    treatment: ExperimentSpec | BenchmarkSpec,
) -> RunDiff:
    """Return a high-level diff between two experiment specifications."""
    normalized_baseline = self._normalize_source_spec(baseline)
    normalized_treatment = self._normalize_source_spec(treatment)
    return self._run_planning.diff_specs(
        normalized_baseline.experiment_spec,
        normalized_treatment.experiment_spec,
        baseline_source_spec=normalized_baseline.source_spec,
        treatment_source_spec=normalized_treatment.source_spec,
    )

estimate

estimate(experiment: ExperimentSpec | BenchmarkSpec) -> CostEstimate

Return a best-effort work-item and token estimate for an experiment.

Source code in themis/orchestration/orchestrator.py

def estimate(self, experiment: ExperimentSpec | BenchmarkSpec) -> CostEstimate:
    """Return a best-effort work-item and token estimate for an experiment."""
    normalized = self._normalize_source_spec(experiment)
    return self._run_planning.estimate(normalized.experiment_spec)

evaluate

evaluate(
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult

Execute evaluation-stage work, reusing generation when possible.

Source code in themis/orchestration/orchestrator.py

def evaluate(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult:
    """Execute evaluation-stage work, reusing generation when possible."""
    normalized = self._normalize_source_spec(experiment)
    planned_trials = evaluation_trials(
        self._services.planner.plan_experiment(
            normalized.experiment_spec,
            required_stages={RunStage.TRANSFORM, RunStage.EVALUATION},
        )
    )
    progress_tracker = self._build_progress_tracker(
        normalized.experiment_spec,
        planned_trials,
        progress=progress,
        allowed_stages={RunStage.TRANSFORM, RunStage.EVALUATION},
        benchmark_spec=normalized.benchmark_spec,
    )
    if planned_trials:
        if progress_tracker is not None:
            progress_tracker.start_run()
        try:
            self._services.executor.execute_transforms(
                planned_trials,
                runtime,
                resume=True,
                progress_tracker=progress_tracker,
            )
            self._services.executor.execute_evaluations(
                planned_trials,
                runtime,
                resume=True,
                progress_tracker=progress_tracker,
            )
        finally:
            if progress_tracker is not None:
                progress_tracker.finish_run()
    result = self._run_planning.build_result(
        planned_trials,
        transform_hashes=collect_transform_hashes(planned_trials),
        evaluation_hashes=collect_evaluation_hashes(planned_trials),
    )
    return self._wrap_source_result(normalized, result)

export_evaluation_bundle

export_evaluation_bundle(
    experiment: ExperimentSpec | BenchmarkSpec,
) -> EvaluationWorkBundle

Export only pending evaluation items for an experiment.

Source code in themis/orchestration/orchestrator.py

def export_evaluation_bundle(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
) -> EvaluationWorkBundle:
    """Export only pending evaluation items for an experiment."""
    normalized = self._normalize_source_spec(experiment)
    return self._run_planning.export_evaluation_bundle(
        normalized.experiment_spec,
        benchmark_spec=normalized.benchmark_spec,
    )

export_generation_bundle

export_generation_bundle(
    experiment: ExperimentSpec | BenchmarkSpec,
) -> GenerationWorkBundle

Export only pending generation items for an experiment.

Source code in themis/orchestration/orchestrator.py

def export_generation_bundle(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
) -> GenerationWorkBundle:
    """Export only pending generation items for an experiment."""
    normalized = self._normalize_source_spec(experiment)
    return self._run_planning.export_generation_bundle(
        normalized.experiment_spec,
        benchmark_spec=normalized.benchmark_spec,
    )

from_project_file classmethod

from_project_file(
    path: str,
    *,
    registry: PluginRegistry | None = None,
    dataset_provider: DatasetProvider | None = None,
    dataset_loader: DatasetLoader | None = None,
    parallel_candidates: int = 5,
    telemetry_bus: TelemetryBus | None = None,
    storage_bundle: StorageBundle | None = None,
) -> Orchestrator

Load a project file, validate it, and build an orchestrator.

Source code in themis/orchestration/orchestrator.py

@classmethod
def from_project_file(
    cls,
    path: str,
    *,
    registry: PluginRegistry | None = None,
    dataset_provider: DatasetProvider | None = None,
    dataset_loader: DatasetLoader | None = None,
    parallel_candidates: int = 5,
    telemetry_bus: TelemetryBus | None = None,
    storage_bundle: StorageBundle | None = None,
) -> Orchestrator:
    """Load a project file, validate it, and build an orchestrator."""
    project_path = Path(path)
    try:
        if project_path.suffix == ".toml":
            with project_path.open("rb") as fh:
                project_data = tomllib.load(fh)
            project = ProjectSpec.model_validate(project_data)
        elif project_path.suffix == ".json":
            project = ProjectSpec.model_validate_json(project_path.read_text())
        else:
            raise ValueError("Project files must use .toml or .json.")
    except tomllib.TOMLDecodeError as exc:
        raise SpecValidationError(
            code=ErrorCode.SCHEMA_MISMATCH,
            message=f"Failed to parse project config {project_path.name}: {exc}",
        ) from exc
    except ValidationError as exc:
        raise SpecValidationError(
            code=ErrorCode.SCHEMA_MISMATCH,
            message=f"Failed to parse project config {project_path.name}: {format_validation_error(exc)}",
        ) from exc
    return cls.from_project_spec(
        project,
        registry=registry,
        dataset_provider=dataset_provider,
        dataset_loader=dataset_loader,
        parallel_candidates=parallel_candidates,
        telemetry_bus=telemetry_bus,
        storage_bundle=storage_bundle,
    )

from_project_spec classmethod

from_project_spec(
    project: ProjectSpec,
    *,
    registry: PluginRegistry | None = None,
    dataset_provider: DatasetProvider | None = None,
    dataset_loader: DatasetLoader | None = None,
    parallel_candidates: int = 5,
    telemetry_bus: TelemetryBus | None = None,
    storage_bundle: StorageBundle | None = None,
) -> Orchestrator

Construct an orchestrator from a validated project specification.

Source code in themis/orchestration/orchestrator.py

@classmethod
def from_project_spec(
    cls,
    project: ProjectSpec,
    *,
    registry: PluginRegistry | None = None,
    dataset_provider: DatasetProvider | None = None,
    dataset_loader: DatasetLoader | None = None,
    parallel_candidates: int = 5,
    telemetry_bus: TelemetryBus | None = None,
    storage_bundle: StorageBundle | None = None,
) -> Orchestrator:
    """Construct an orchestrator from a validated project specification."""
    return cls(
        registry or PluginRegistry(),
        storage_bundle or build_storage_bundle(project.storage),
        dataset_provider=dataset_provider,
        dataset_loader=dataset_loader,
        execution_policy=project.execution_policy,
        parallel_candidates=parallel_candidates,
        project_seed=project.global_seed,
        store_item_payloads=project.storage.store_item_payloads,
        telemetry_bus=telemetry_bus,
        project_spec=project,
        _allow_runtime_construction=True,
    )

generate

generate(
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult

Execute only generation-stage work for one experiment.

Source code in themis/orchestration/orchestrator.py

def generate(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult:
    """Execute only generation-stage work for one experiment."""
    normalized = self._normalize_source_spec(experiment)
    planned_trials = generation_trials(
        self._services.planner.plan_experiment(
            normalized.experiment_spec,
            required_stages={RunStage.GENERATION},
        )
    )
    progress_tracker = self._build_progress_tracker(
        normalized.experiment_spec,
        planned_trials,
        progress=progress,
        allowed_stages={RunStage.GENERATION},
        benchmark_spec=normalized.benchmark_spec,
    )
    if planned_trials:
        if progress_tracker is not None:
            progress_tracker.start_run()
        try:
            self._services.executor.execute_generation_trials(
                planned_trials,
                runtime,
                progress_tracker=progress_tracker,
            )
        finally:
            if progress_tracker is not None:
                progress_tracker.finish_run()
    result = self._run_planning.build_result(planned_trials)
    return self._wrap_source_result(normalized, result)

get_run_progress

get_run_progress(run_id: str) -> RunProgressSnapshot | None

Return the persisted progress snapshot for one known run.

Source code in themis/orchestration/orchestrator.py

def get_run_progress(self, run_id: str) -> RunProgressSnapshot | None:
    """Return the persisted progress snapshot for one known run."""
    return self._run_planning.get_progress_snapshot(run_id)

import_candidates

import_candidates(trial_records: list[TrialRecord]) -> ExperimentResult

Import prebuilt generation artifacts into the current store.

Source code in themis/orchestration/orchestrator.py

def import_candidates(
    self,
    trial_records: list[TrialRecord],
) -> ExperimentResult:
    """Import prebuilt generation artifacts into the current store."""
    trial_hashes = self._run_imports.import_candidates(trial_records)
    return ExperimentResult(
        projection_repo=self._services.projection_repo,
        trial_hashes=list(trial_hashes),
    )

import_evaluation_results

import_evaluation_results(
    bundle: EvaluationWorkBundle, trial_records: list[TrialRecord]
) -> ExperimentResult | BenchmarkResult

Import externally evaluated results for a previously exported bundle.

Source code in themis/orchestration/orchestrator.py

def import_evaluation_results(
    self,
    bundle: EvaluationWorkBundle,
    trial_records: list[TrialRecord],
) -> ExperimentResult | BenchmarkResult:
    """Import externally evaluated results for a previously exported bundle."""
    self._run_imports.import_evaluation_results(bundle, trial_records)
    manifest = self.plan(self._manifest_source_spec(bundle.manifest))
    result = self._run_planning.result_from_manifest(manifest)
    return self._wrap_manifest_result(manifest, result)

import_generation_results

import_generation_results(
    bundle: GenerationWorkBundle, trial_records: list[TrialRecord]
) -> ExperimentResult | BenchmarkResult

Import externally generated results for a previously exported bundle.

Source code in themis/orchestration/orchestrator.py

def import_generation_results(
    self,
    bundle: GenerationWorkBundle,
    trial_records: list[TrialRecord],
) -> ExperimentResult | BenchmarkResult:
    """Import externally generated results for a previously exported bundle."""
    self._run_imports.import_generation_results(bundle, trial_records)
    manifest = self.plan(self._manifest_source_spec(bundle.manifest))
    result = self._run_planning.result_from_manifest(manifest)
    return self._wrap_manifest_result(manifest, result)

plan

plan(experiment: ExperimentSpec | BenchmarkSpec) -> RunManifest

Build and persist a deterministic run manifest for one experiment.

Source code in themis/orchestration/orchestrator.py

def plan(self, experiment: ExperimentSpec | BenchmarkSpec) -> RunManifest:
    """Build and persist a deterministic run manifest for one experiment."""
    normalized = self._normalize_source_spec(experiment)
    return self._run_planning.plan(
        normalized.experiment_spec,
        benchmark_spec=normalized.benchmark_spec,
    )

resume

resume(
    run_id: str,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> RunHandle | ExperimentResult | BenchmarkResult

Refresh a persisted run and continue it when possible.

Source code in themis/orchestration/orchestrator.py

def resume(
    self,
    run_id: str,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> RunHandle | ExperimentResult | BenchmarkResult:
    """Refresh a persisted run and continue it when possible."""
    resumed = self._run_planning.resume(
        run_id,
        runtime=runtime,
        execute_run=lambda spec, runtime_context: self.run(
            spec,
            runtime=runtime_context,
            progress=progress,
        ),
    )
    if isinstance(resumed, ExperimentResult):
        manifest = self._run_planning.manifest_repo.get_manifest(run_id)
        if manifest is not None:
            return self._wrap_manifest_result(manifest, resumed)
    return resumed

run

run(
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult

Execute generation, transforms, and evaluations for one experiment.

Source code in themis/orchestration/orchestrator.py

def run(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult:
    """Execute generation, transforms, and evaluations for one experiment."""
    normalized = self._normalize_source_spec(experiment)
    planned_trials = self._services.planner.plan_experiment(
        normalized.experiment_spec
    )
    progress_tracker = self._build_progress_tracker(
        normalized.experiment_spec,
        planned_trials,
        progress=progress,
        allowed_stages={
            RunStage.GENERATION,
            RunStage.TRANSFORM,
            RunStage.EVALUATION,
        },
        benchmark_spec=normalized.benchmark_spec,
    )
    pending_generation_trials = generation_trials(planned_trials)
    pending_transform_trials = transform_trials(planned_trials)
    pending_evaluation_trials = evaluation_trials(planned_trials)

    if progress_tracker is not None:
        progress_tracker.start_run()
    try:
        if pending_generation_trials:
            self._services.executor.execute_generation_trials(
                pending_generation_trials,
                runtime,
                progress_tracker=progress_tracker,
            )
        if pending_transform_trials:
            self._services.executor.execute_transforms(
                pending_transform_trials,
                runtime,
                progress_tracker=progress_tracker,
            )
        if pending_evaluation_trials:
            self._services.executor.execute_evaluations(
                pending_evaluation_trials,
                runtime,
                progress_tracker=progress_tracker,
            )
    finally:
        if progress_tracker is not None:
            progress_tracker.finish_run()

    result = self._run_planning.build_result(
        planned_trials,
        transform_hashes=collect_transform_hashes(pending_transform_trials),
        evaluation_hashes=collect_evaluation_hashes(pending_evaluation_trials),
    )
    return self._wrap_source_result(normalized, result)

run_benchmark

run_benchmark(
    benchmark: BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> BenchmarkResult

Compile and execute one benchmark specification.

Source code in themis/orchestration/orchestrator.py

def run_benchmark(
    self,
    benchmark: BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> BenchmarkResult:
    """Compile and execute one benchmark specification."""
    result = self.run(benchmark, runtime=runtime, progress=progress)
    if not isinstance(result, BenchmarkResult):
        raise TypeError(
            "Orchestrator.run_benchmark expected run() to return a "
            f"BenchmarkResult, got {type(result).__name__}."
        )
    return result

submit

submit(
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> RunHandle

Persist one run manifest and start execution if the backend is local.

Source code in themis/orchestration/orchestrator.py

def submit(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> RunHandle:
    """Persist one run manifest and start execution if the backend is local."""
    normalized = self._normalize_source_spec(experiment)
    return self._run_planning.submit(
        normalized.experiment_spec,
        benchmark_spec=normalized.benchmark_spec,
        runtime=runtime,
        execute_run=lambda spec, runtime_context: self.run(
            spec,
            runtime=runtime_context,
            progress=progress,
        ),
    )

transform

transform(
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult

Execute output transforms against existing generation candidates.

Source code in themis/orchestration/orchestrator.py

def transform(
    self,
    experiment: ExperimentSpec | BenchmarkSpec,
    *,
    runtime: RuntimeContext | None = None,
    progress: ProgressConfig | None = None,
) -> ExperimentResult | BenchmarkResult:
    """Execute output transforms against existing generation candidates."""
    normalized = self._normalize_source_spec(experiment)
    planned_trials = transform_trials(
        self._services.planner.plan_experiment(
            normalized.experiment_spec,
            required_stages={RunStage.TRANSFORM},
        )
    )
    transform_hashes = collect_transform_hashes(planned_trials)
    progress_tracker = self._build_progress_tracker(
        normalized.experiment_spec,
        planned_trials,
        progress=progress,
        allowed_stages={RunStage.TRANSFORM},
        benchmark_spec=normalized.benchmark_spec,
    )
    if planned_trials:
        if progress_tracker is not None:
            progress_tracker.start_run()
        try:
            self._services.executor.execute_transforms(
                planned_trials,
                runtime,
                progress_tracker=progress_tracker,
            )
        finally:
            if progress_tracker is not None:
                progress_tracker.finish_run()
    result = self._run_planning.build_result(
        planned_trials,
        transform_hashes=transform_hashes,
    )
    return self._wrap_source_result(normalized, result)

ParseSpec

Bases: SpecBase

Named parse pipeline backed by one extractor chain.

Source code in themis/benchmark/specs.py

class ParseSpec(SpecBase):
    """Named parse pipeline backed by one extractor chain."""

    name: str = Field(..., min_length=1)
    extractors: list[ExtractorRefSpec] = Field(default_factory=list)

    @field_validator("extractors", mode="before")
    @classmethod
    def _coerce_extractors(cls, value: object, info: ValidationInfo) -> object:
        del info
        if not isinstance(value, list):
            return value
        coerced: list[ExtractorRefSpec | object] = []
        for item in value:
            if isinstance(item, str):
                coerced.append(ExtractorRefSpec(id=item))
            else:
                coerced.append(item)
        return coerced

PluginRegistry

Instance-scoped registry for protocol implementations and plugin metadata.

Source code in themis/registry/plugin_registry.py

class PluginRegistry:
    """
    Instance-scoped registry for protocol implementations and plugin metadata.
    """

    def __init__(self) -> None:
        self._registration_order = 0
        self._inference_engines: dict[str, InferenceEngineRegistration] = {}
        self._extractors: dict[str, PluginRegistration[Extractor]] = {}
        self._metrics: dict[str, PluginRegistration[Metric]] = {}
        self._judges: dict[str, PluginRegistration[JudgeService]] = {}
        self._hooks: list[HookRegistration] = []
        self._register_builtin_extractors()

    def register_inference_engine(
        self,
        name: str,
        factory: Callable[[], InferenceEngine]
        | type[InferenceEngine]
        | InferenceEngine,
        *,
        version: str = "0.0.0",
        capabilities: EngineCapabilities | None = None,
        plugin_api: str = "1.0",
        prompt_token_estimator: PromptTokenEstimator | None = None,
    ) -> None:
        """Register an inference engine implementation under a provider name."""
        self._inference_engines[name] = InferenceEngineRegistration(
            name=name,
            factory=factory,
            version=version,
            plugin_api=plugin_api,
            registration_order=self._next_registration_order(),
            capabilities=capabilities or EngineCapabilities(),
            prompt_token_estimator=prompt_token_estimator,
        )

    def register_extractor(
        self,
        name: str,
        factory: Callable[[], Extractor] | type[Extractor] | Extractor,
        *,
        version: str = "0.0.0",
        plugin_api: str = "1.0",
    ) -> None:
        """Register an extractor implementation."""
        self._extractors[name] = PluginRegistration(
            name=name,
            factory=factory,
            version=version,
            plugin_api=plugin_api,
            registration_order=self._next_registration_order(),
        )

    def register_metric(
        self,
        name: str,
        factory: Callable[[], Metric] | type[Metric] | Metric,
        *,
        version: str = "0.0.0",
        plugin_api: str = "1.0",
    ) -> None:
        """Register a metric implementation."""
        self._metrics[name] = PluginRegistration(
            name=name,
            factory=factory,
            version=version,
            plugin_api=plugin_api,
            registration_order=self._next_registration_order(),
        )

    def register_judge(
        self,
        name: str,
        factory: Callable[[], JudgeService] | type[JudgeService] | JudgeService,
        *,
        version: str = "0.0.0",
        plugin_api: str = "1.0",
    ) -> None:
        """Register a judge service implementation."""
        self._judges[name] = PluginRegistration(
            name=name,
            factory=factory,
            version=version,
            plugin_api=plugin_api,
            registration_order=self._next_registration_order(),
        )

    def register_hook(
        self,
        name: str,
        hook: PipelineHook,
        *,
        priority: int = 100,
        idempotent: bool = True,
    ) -> None:
        """Register a pipeline hook and preserve deterministic ordering metadata."""
        if not isinstance(hook, PipelineHook):
            raise SpecValidationError(
                code=ErrorCode.PLUGIN_INCOMPATIBLE,
                message=(
                    "Hook registrations must implement the full PipelineHook contract."
                ),
            )
        self._hooks.append(
            HookRegistration(
                name=name,
                hook=hook,
                priority=priority,
                idempotent=idempotent,
                registration_order=self._next_registration_order(),
            )
        )

    def has_inference_engine(self, name: str) -> bool:
        """Return whether an inference engine exists for ``name``."""
        return name in self._inference_engines

    def has_extractor(self, name: str) -> bool:
        """Return whether an extractor exists for ``name``."""
        return name in self._extractors

    def has_metric(self, name: str) -> bool:
        """Return whether a metric exists for ``name``."""
        return name in self._metrics

    def has_judge(self, name: str) -> bool:
        """Return whether a judge service exists for ``name``."""
        return name in self._judges

    def get_inference_engine_registration(
        self, name: str
    ) -> InferenceEngineRegistration:
        """Fetch inference-engine registration metadata for ``name``."""
        if name not in self._inference_engines:
            raise SpecValidationError(
                code=ErrorCode.PLUGIN_INCOMPATIBLE,
                message=f"Provider {name} not found in registry.",
            )
        return self._inference_engines[name]

    def get_extractor_registration(self, name: str) -> PluginRegistration[Extractor]:
        """Fetch extractor registration metadata for ``name``."""
        if name not in self._extractors:
            raise SpecValidationError(
                code=ErrorCode.PLUGIN_INCOMPATIBLE,
                message=f"Extractor {name} not found in registry.",
            )
        return self._extractors[name]

    def get_metric_registration(self, name: str) -> PluginRegistration[Metric]:
        """Fetch metric registration metadata for ``name``."""
        if name not in self._metrics:
            raise SpecValidationError(
                code=ErrorCode.PLUGIN_INCOMPATIBLE,
                message=f"Metric {name} not found in registry.",
            )
        return self._metrics[name]

    def get_inference_engine(self, name: str) -> InferenceEngine:
        """Instantiate or return the registered inference engine for ``name``."""
        registration = self.get_inference_engine_registration(name)
        return self._instantiate(registration.factory, required_methods=("infer",))

    def get_extractor(self, name: str) -> Extractor:
        """Instantiate or return the registered extractor for ``name``."""
        registration = self.get_extractor_registration(name)
        extractor = self._instantiate(
            registration.factory,
            required_methods=("extract",),
        )
        self._validate_extractor_signature(name, extractor)
        return extractor

    def get_metric(self, name: str) -> Metric:
        """Instantiate or return the registered metric for ``name``."""
        registration = self.get_metric_registration(name)
        return self._instantiate(registration.factory, required_methods=("score",))

    def get_judge(self, name: str) -> JudgeService:
        """Instantiate or return the registered judge service for ``name``."""
        if name not in self._judges:
            raise SpecValidationError(
                code=ErrorCode.PLUGIN_INCOMPATIBLE,
                message=f"Judge service {name} not found in registry.",
            )
        return self._instantiate(
            self._judges[name].factory, required_methods=("judge",)
        )

    def iter_hook_registrations(self) -> list[HookRegistration]:
        """Return hook registrations ordered by priority then registration order."""
        return sorted(
            self._hooks,
            key=lambda registration: (
                registration.priority,
                registration.registration_order,
            ),
        )

    def iter_hooks(self) -> Iterator[PipelineHook]:
        """Yield hook instances in the same order used by pipeline execution."""
        for registration in self.iter_hook_registrations():
            yield registration.hook

    def _register_builtin_extractors(self) -> None:
        from themis.extractors.builtin import (
            BoxedTextExtractor,
            ChoiceLetterExtractor,
            FirstNumberExtractor,
            JsonSchemaExtractor,
            NormalizedTextExtractor,
            RegexExtractor,
        )

        self.register_extractor(
            "regex", RegexExtractor, version="1.0.0", plugin_api="1.0"
        )
        self.register_extractor(
            "json_schema", JsonSchemaExtractor, version="1.0.0", plugin_api="1.0"
        )
        self.register_extractor(
            "first_number", FirstNumberExtractor, version="1.0.0", plugin_api="1.0"
        )
        self.register_extractor(
            "choice_letter", ChoiceLetterExtractor, version="1.0.0", plugin_api="1.0"
        )
        self.register_extractor(
            "boxed_text", BoxedTextExtractor, version="1.0.0", plugin_api="1.0"
        )
        self.register_extractor(
            "normalized_text",
            NormalizedTextExtractor,
            version="1.0.0",
            plugin_api="1.0",
        )

    def _next_registration_order(self) -> int:
        self._registration_order += 1
        return self._registration_order

    def _instantiate(
        self,
        factory: Callable[[], _PluginT] | type[_PluginT] | _PluginT,
        *,
        required_methods: tuple[str, ...],
    ) -> _PluginT:
        if isinstance(factory, type):
            return cast(_PluginT, factory())
        if not callable(factory):
            return factory
        if all(hasattr(factory, method_name) for method_name in required_methods):
            return cast(_PluginT, factory)
        return cast(_PluginT, factory())

    def _validate_extractor_signature(
        self,
        name: str,
        extractor: Extractor,
    ) -> None:
        signature = inspect.signature(extractor.extract)
        parameters = tuple(signature.parameters.values())
        has_config_parameter = "config" in signature.parameters
        positional_count = sum(
            parameter.kind
            in (
                inspect.Parameter.POSITIONAL_ONLY,
                inspect.Parameter.POSITIONAL_OR_KEYWORD,
            )
            for parameter in parameters
        )
        has_varargs = any(
            parameter.kind is inspect.Parameter.VAR_POSITIONAL
            for parameter in parameters
        )
        if has_config_parameter or positional_count >= 3 or has_varargs:
            return
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=(
                f"Extractor '{name}' must accept (trial, candidate, config); "
                "legacy two-argument extractors are no longer supported."
            ),
        )

get_extractor

get_extractor(name: str) -> Extractor

Instantiate or return the registered extractor for name.

Source code in themis/registry/plugin_registry.py

def get_extractor(self, name: str) -> Extractor:
    """Instantiate or return the registered extractor for ``name``."""
    registration = self.get_extractor_registration(name)
    extractor = self._instantiate(
        registration.factory,
        required_methods=("extract",),
    )
    self._validate_extractor_signature(name, extractor)
    return extractor

get_extractor_registration

get_extractor_registration(name: str) -> PluginRegistration[Extractor]

Fetch extractor registration metadata for name.

Source code in themis/registry/plugin_registry.py

def get_extractor_registration(self, name: str) -> PluginRegistration[Extractor]:
    """Fetch extractor registration metadata for ``name``."""
    if name not in self._extractors:
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=f"Extractor {name} not found in registry.",
        )
    return self._extractors[name]

get_inference_engine

get_inference_engine(name: str) -> InferenceEngine

Instantiate or return the registered inference engine for name.

Source code in themis/registry/plugin_registry.py

def get_inference_engine(self, name: str) -> InferenceEngine:
    """Instantiate or return the registered inference engine for ``name``."""
    registration = self.get_inference_engine_registration(name)
    return self._instantiate(registration.factory, required_methods=("infer",))

get_inference_engine_registration

get_inference_engine_registration(name: str) -> InferenceEngineRegistration

Fetch inference-engine registration metadata for name.

Source code in themis/registry/plugin_registry.py

def get_inference_engine_registration(
    self, name: str
) -> InferenceEngineRegistration:
    """Fetch inference-engine registration metadata for ``name``."""
    if name not in self._inference_engines:
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=f"Provider {name} not found in registry.",
        )
    return self._inference_engines[name]

get_judge

get_judge(name: str) -> JudgeService

Instantiate or return the registered judge service for name.

Source code in themis/registry/plugin_registry.py

def get_judge(self, name: str) -> JudgeService:
    """Instantiate or return the registered judge service for ``name``."""
    if name not in self._judges:
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=f"Judge service {name} not found in registry.",
        )
    return self._instantiate(
        self._judges[name].factory, required_methods=("judge",)
    )

get_metric

get_metric(name: str) -> Metric

Instantiate or return the registered metric for name.

Source code in themis/registry/plugin_registry.py

def get_metric(self, name: str) -> Metric:
    """Instantiate or return the registered metric for ``name``."""
    registration = self.get_metric_registration(name)
    return self._instantiate(registration.factory, required_methods=("score",))

get_metric_registration

get_metric_registration(name: str) -> PluginRegistration[Metric]

Fetch metric registration metadata for name.

Source code in themis/registry/plugin_registry.py

def get_metric_registration(self, name: str) -> PluginRegistration[Metric]:
    """Fetch metric registration metadata for ``name``."""
    if name not in self._metrics:
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=f"Metric {name} not found in registry.",
        )
    return self._metrics[name]

has_extractor

has_extractor(name: str) -> bool

Return whether an extractor exists for name.

Source code in themis/registry/plugin_registry.py

def has_extractor(self, name: str) -> bool:
    """Return whether an extractor exists for ``name``."""
    return name in self._extractors

has_inference_engine

has_inference_engine(name: str) -> bool

Return whether an inference engine exists for name.

Source code in themis/registry/plugin_registry.py

def has_inference_engine(self, name: str) -> bool:
    """Return whether an inference engine exists for ``name``."""
    return name in self._inference_engines

has_judge

has_judge(name: str) -> bool

Return whether a judge service exists for name.

Source code in themis/registry/plugin_registry.py

def has_judge(self, name: str) -> bool:
    """Return whether a judge service exists for ``name``."""
    return name in self._judges

has_metric

has_metric(name: str) -> bool

Return whether a metric exists for name.

Source code in themis/registry/plugin_registry.py

def has_metric(self, name: str) -> bool:
    """Return whether a metric exists for ``name``."""
    return name in self._metrics

iter_hook_registrations

iter_hook_registrations() -> list[HookRegistration]

Return hook registrations ordered by priority then registration order.

Source code in themis/registry/plugin_registry.py

def iter_hook_registrations(self) -> list[HookRegistration]:
    """Return hook registrations ordered by priority then registration order."""
    return sorted(
        self._hooks,
        key=lambda registration: (
            registration.priority,
            registration.registration_order,
        ),
    )

iter_hooks

iter_hooks() -> Iterator[PipelineHook]

Yield hook instances in the same order used by pipeline execution.

Source code in themis/registry/plugin_registry.py

def iter_hooks(self) -> Iterator[PipelineHook]:
    """Yield hook instances in the same order used by pipeline execution."""
    for registration in self.iter_hook_registrations():
        yield registration.hook

register_extractor

register_extractor(
    name: str,
    factory: Callable[[], Extractor] | type[Extractor] | Extractor,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None

Register an extractor implementation.

Source code in themis/registry/plugin_registry.py

def register_extractor(
    self,
    name: str,
    factory: Callable[[], Extractor] | type[Extractor] | Extractor,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None:
    """Register an extractor implementation."""
    self._extractors[name] = PluginRegistration(
        name=name,
        factory=factory,
        version=version,
        plugin_api=plugin_api,
        registration_order=self._next_registration_order(),
    )

register_hook

register_hook(
    name: str,
    hook: PipelineHook,
    *,
    priority: int = 100,
    idempotent: bool = True,
) -> None

Register a pipeline hook and preserve deterministic ordering metadata.

Source code in themis/registry/plugin_registry.py

def register_hook(
    self,
    name: str,
    hook: PipelineHook,
    *,
    priority: int = 100,
    idempotent: bool = True,
) -> None:
    """Register a pipeline hook and preserve deterministic ordering metadata."""
    if not isinstance(hook, PipelineHook):
        raise SpecValidationError(
            code=ErrorCode.PLUGIN_INCOMPATIBLE,
            message=(
                "Hook registrations must implement the full PipelineHook contract."
            ),
        )
    self._hooks.append(
        HookRegistration(
            name=name,
            hook=hook,
            priority=priority,
            idempotent=idempotent,
            registration_order=self._next_registration_order(),
        )
    )

register_inference_engine

register_inference_engine(
    name: str,
    factory: Callable[[], InferenceEngine]
    | type[InferenceEngine]
    | InferenceEngine,
    *,
    version: str = "0.0.0",
    capabilities: EngineCapabilities | None = None,
    plugin_api: str = "1.0",
    prompt_token_estimator: PromptTokenEstimator | None = None,
) -> None

Register an inference engine implementation under a provider name.

Source code in themis/registry/plugin_registry.py

def register_inference_engine(
    self,
    name: str,
    factory: Callable[[], InferenceEngine]
    | type[InferenceEngine]
    | InferenceEngine,
    *,
    version: str = "0.0.0",
    capabilities: EngineCapabilities | None = None,
    plugin_api: str = "1.0",
    prompt_token_estimator: PromptTokenEstimator | None = None,
) -> None:
    """Register an inference engine implementation under a provider name."""
    self._inference_engines[name] = InferenceEngineRegistration(
        name=name,
        factory=factory,
        version=version,
        plugin_api=plugin_api,
        registration_order=self._next_registration_order(),
        capabilities=capabilities or EngineCapabilities(),
        prompt_token_estimator=prompt_token_estimator,
    )

register_judge

register_judge(
    name: str,
    factory: Callable[[], JudgeService] | type[JudgeService] | JudgeService,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None

Register a judge service implementation.

Source code in themis/registry/plugin_registry.py

def register_judge(
    self,
    name: str,
    factory: Callable[[], JudgeService] | type[JudgeService] | JudgeService,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None:
    """Register a judge service implementation."""
    self._judges[name] = PluginRegistration(
        name=name,
        factory=factory,
        version=version,
        plugin_api=plugin_api,
        registration_order=self._next_registration_order(),
    )

register_metric

register_metric(
    name: str,
    factory: Callable[[], Metric] | type[Metric] | Metric,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None

Register a metric implementation.

Source code in themis/registry/plugin_registry.py

def register_metric(
    self,
    name: str,
    factory: Callable[[], Metric] | type[Metric] | Metric,
    *,
    version: str = "0.0.0",
    plugin_api: str = "1.0",
) -> None:
    """Register a metric implementation."""
    self._metrics[name] = PluginRegistration(
        name=name,
        factory=factory,
        version=version,
        plugin_api=plugin_api,
        registration_order=self._next_registration_order(),
    )

PostgresBlobStorageSpec

Bases: _StorageSpecBase

Postgres event/projection store plus local filesystem blob persistence.

Source code in themis/specs/experiment.py

class PostgresBlobStorageSpec(_StorageSpecBase):
    """Postgres event/projection store plus local filesystem blob persistence."""

    backend: Literal[StorageBackend.POSTGRES_BLOB] = Field(
        default=StorageBackend.POSTGRES_BLOB
    )
    database_url: str = Field(
        ..., description="Postgres connection URL for events and projections."
    )
    blob_root_dir: str = Field(
        ..., description="Local blob root for content-addressed artifact storage."
    )

ProjectSpec

Bases: SpecBase

Shared project-level identity, storage defaults, and execution policy.

Keep this stable across related experiment runs so resume behavior and run manifests refer to the same storage and backend context.

Source code in themis/specs/experiment.py

class ProjectSpec(SpecBase):
    """Shared project-level identity, storage defaults, and execution policy.

    Keep this stable across related experiment runs so resume behavior and run
    manifests refer to the same storage and backend context.
    """

    project_name: str = Field(..., description="Human readable project name.")
    researcher_id: str = Field(
        ..., description="Stable owner identifier for experiment lineage."
    )
    global_seed: int = Field(
        ..., description="Default deterministic seed shared across experiments."
    )
    storage: StorageConfig = Field(..., description="Shared storage defaults.")
    execution_policy: ExecutionPolicySpec = Field(
        ..., description="Shared retry and circuit-breaker policy."
    )
    execution_backend: ExecutionBackendConfig = Field(
        default_factory=LocalExecutionBackendSpec,
        description="Execution backend used for local, worker-pool, or batch orchestration.",
    )
    metadata: dict[str, str] = Field(
        default_factory=dict, description="User-defined project metadata."
    )

PromptMessage

Bases: BaseModel

One structured chat message in a prompt template.

Source code in themis/specs/experiment.py

class PromptMessage(BaseModel):
    """One structured chat message in a prompt template."""

    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)

    role: PromptRole
    content: str

    @field_validator("role", mode="before")
    @classmethod
    def _coerce_role(cls, value: PromptRole | str) -> PromptRole | str:
        if isinstance(value, str):
            return PromptRole(value)
        return value

PromptVariantSpec

Bases: SpecBase

Structured prompt variant scoped to one family or benchmark workflow.

Source code in themis/benchmark/specs.py

class PromptVariantSpec(SpecBase):
    """Structured prompt variant scoped to one family or benchmark workflow."""

    id: str = Field(..., min_length=1)
    family: str | None = Field(default=None)
    messages: list[PromptMessage] = Field(..., min_length=1)
    variables: JSONDict = Field(
        default_factory=dict,
        description="Static prompt-scoped variables exposed to prompt rendering.",
    )

ScoreSpec

Bases: SpecBase

Named scoring pass over raw or parsed candidate outputs.

Source code in themis/benchmark/specs.py

class ScoreSpec(SpecBase):
    """Named scoring pass over raw or parsed candidate outputs."""

    name: str = Field(..., min_length=1)
    parse: str | None = Field(default=None)
    metrics: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def _validate_semantic(self) -> "ScoreSpec":
        if not self.metrics:
            raise ValueError("ScoreSpec must define at least one metric.")
        return self

SliceSpec

Bases: SpecBase

One benchmark slice with dataset identity, queries, prompts, and scoring.

Source code in themis/benchmark/specs.py

class SliceSpec(SpecBase):
    """One benchmark slice with dataset identity, queries, prompts, and scoring."""

    slice_id: str = Field(..., min_length=1)
    dataset: DatasetSpec = Field(...)
    dataset_query: DatasetQuerySpec = Field(default_factory=DatasetQuerySpec)
    dimensions: dict[str, str] = Field(default_factory=dict)
    prompt_variant_ids: list[str] = Field(default_factory=list)
    prompt_families: list[str] = Field(default_factory=list)
    generation: GenerationSpec | None = Field(default=None)
    parses: list[ParseSpec] = Field(default_factory=list)
    scores: list[ScoreSpec] = Field(default_factory=list)

    @model_validator(mode="after")
    def _validate_semantic(self) -> "SliceSpec":
        if self.generation is None and not self.parses and not self.scores:
            raise ValueError(
                f"SliceSpec '{self.slice_id}' must define at least one stage."
            )
        parse_names = [parse.name for parse in self.parses]
        parse_name_set = set(parse_names)
        if len(parse_names) != len(set(parse_names)):
            raise ValueError(f"SliceSpec '{self.slice_id}' has duplicate parse name.")
        score_names = [score.name for score in self.scores]
        if len(score_names) != len(set(score_names)):
            raise ValueError(f"SliceSpec '{self.slice_id}' has duplicate score name.")
        for score in self.scores:
            if score.parse is not None and score.parse not in parse_name_set:
                raise ValueError(
                    f"SliceSpec '{self.slice_id}' references unknown parse "
                    f"'{score.parse}' in score '{score.name}'."
                )
        return self

SqliteBlobStorageSpec

Bases: _StorageSpecBase

SQLite event/projection store plus local filesystem blob persistence.

Source code in themis/specs/experiment.py

class SqliteBlobStorageSpec(_StorageSpecBase):
    """SQLite event/projection store plus local filesystem blob persistence."""

    backend: Literal[StorageBackend.SQLITE_BLOB] = Field(
        default=StorageBackend.SQLITE_BLOB
    )
    root_dir: str = Field(
        ..., description="Storage root for event, projection, and blob data."
    )

generate_config_report

generate_config_report(
    config: object,
    format: ConfigReportFormat | str = "markdown",
    output: str | PathLike[str] | None = None,
    *,
    entrypoint: str | None = None,
    verbosity: ConfigReportVerbosity = "default",
) -> str

Collect and render one nested configuration report.

PARAMETER	DESCRIPTION
config	Root config object or config bundle to collect. TYPE: object
format	Output format name or enum accepted by the renderer registry. TYPE: ConfigReportFormat | str DEFAULT: 'markdown'
output	Optional filesystem path where the rendered report should also be written. TYPE: str | PathLike[str] | None DEFAULT: None
entrypoint	Optional source label to record in the report header. TYPE: str | None DEFAULT: None
verbosity	Visibility level used to filter the collected document before rendering. TYPE: ConfigReportVerbosity DEFAULT: 'default'

RETURNS	DESCRIPTION
str	The rendered config report as a string, even when output is supplied.

RAISES	DESCRIPTION
KeyError	If no renderer is registered for the requested format.
OSError	If output is supplied and the rendered report cannot be written to disk.

Source code in themis/config_report/api.py

def generate_config_report(
    config: object,
    format: ConfigReportFormat | str = "markdown",
    output: str | PathLike[str] | None = None,
    *,
    entrypoint: str | None = None,
    verbosity: ConfigReportVerbosity = "default",
) -> str:
    """Collect and render one nested configuration report.

    Args:
        config: Root config object or config bundle to collect.
        format: Output format name or enum accepted by the renderer registry.
        output: Optional filesystem path where the rendered report should also be
            written.
        entrypoint: Optional source label to record in the report header.
        verbosity: Visibility level used to filter the collected document before
            rendering.

    Returns:
        The rendered config report as a string, even when `output` is supplied.

    Raises:
        KeyError: If no renderer is registered for the requested format.
        OSError: If `output` is supplied and the rendered report cannot be
            written to disk.
    """

    document = build_config_report_document(config, entrypoint=entrypoint)
    rendered = render_config_report(document, format=format, verbosity=verbosity)
    if output is not None:
        Path(output).write_text(rendered, encoding="utf-8")
    return rendered