Evaluation Layer

The evaluate layer runs evaluators against a captured DatasetRunOutput and produces an EvaluationOutput with runs / cases DataFrames.

This layer has no MLflow server dependency — it is pure in the sense that it consumes dataclass inputs and returns dataclass outputs. This makes evaluation easy to unit-test without any external infrastructure.

evaluate_results

ragpill.evaluation.evaluate_results async

evaluate_results(dataset_run, testset, settings=None)

Run evaluators against a captured :class:DatasetRunOutput.

For every (case, task_run) pair, builds an :class:~ragpill.eval_types.EvaluatorContext with the captured trace and runs every case-level + dataset-level evaluator. Results are aggregated per case using threshold from settings (or per-case override).

Parameters:

Name	Type	Description	Default
dataset_run	DatasetRunOutput	Output of :func:ragpill.execution.execute_dataset.	required
testset	Dataset[Any, Any, CaseMetadataT]	The dataset whose cases align one-for-one with dataset_run.cases. Evaluators come from case.evaluators and testset.evaluators.	required
settings	MLFlowSettings | None	Global :class:MLFlowSettings. Only ragpill_repeat and ragpill_threshold are consulted — no MLflow connection is made.	None

Returns:

Type	Description
EvaluationOutput	class:EvaluationOutput with .runs, .cases, and
EvaluationOutput	.case_results.

Example

run_output = await execute_dataset(testset, task=my_task)
eval_output = await evaluate_results(run_output, testset)
print(eval_output.summary)

See Also

execute_dataset: Phase 1. upload_to_mlflow: Phase 3.

Source code in src/ragpill/evaluation.py

async def evaluate_results(
    dataset_run: DatasetRunOutput,
    testset: Dataset[Any, Any, CaseMetadataT],
    settings: MLFlowSettings | None = None,
) -> EvaluationOutput:
    """Run evaluators against a captured :class:`DatasetRunOutput`.

    For every ``(case, task_run)`` pair, builds an
    :class:`~ragpill.eval_types.EvaluatorContext` with the captured trace and
    runs every case-level + dataset-level evaluator. Results are aggregated
    per case using ``threshold`` from settings (or per-case override).

    Args:
        dataset_run: Output of :func:`ragpill.execution.execute_dataset`.
        testset: The dataset whose cases align one-for-one with
            ``dataset_run.cases``. Evaluators come from ``case.evaluators`` and
            ``testset.evaluators``.
        settings: Global :class:`MLFlowSettings`. Only ``ragpill_repeat`` and
            ``ragpill_threshold`` are consulted — no MLflow connection is made.

    Returns:
        :class:`EvaluationOutput` with ``.runs``, ``.cases``, and
        ``.case_results``.

    Example:
        ```python
        run_output = await execute_dataset(testset, task=my_task)
        eval_output = await evaluate_results(run_output, testset)
        print(eval_output.summary)
        ```

    See Also:
        [`execute_dataset`][ragpill.execution.execute_dataset]: Phase 1.
        [`upload_to_mlflow`][ragpill.upload.upload_to_mlflow]: Phase 3.
    """
    _settings = settings or MLFlowSettings()  # pyright: ignore[reportCallIssue]

    if len(dataset_run.cases) != len(testset.cases):
        raise ValueError(f"dataset_run has {len(dataset_run.cases)} cases but testset has {len(testset.cases)}")

    case_results: list[CaseResult] = []
    for case_run, case in zip(dataset_run.cases, testset.cases):
        # Resolve evaluators: case-level + dataset-level.
        evaluators: list[BaseEvaluator] = []
        for ev in case.evaluators:
            assert isinstance(ev, BaseEvaluator)
            evaluators.append(ev)
        for ev in testset.evaluators:
            assert isinstance(ev, BaseEvaluator)
            evaluators.append(ev)

        case_metadata: TestCaseMetadata | None = case.metadata if isinstance(case.metadata, TestCaseMetadata) else None
        _, threshold = resolve_repeat(case_metadata, _settings)

        run_results: list[RunResult] = []
        for task_run in case_run.task_runs:
            rr = await _evaluate_single_run(case, task_run, case_run, evaluators)
            run_results.append(rr)

        aggregated = _aggregate_runs(run_results, threshold)

        # CaseResult demands a TestCaseMetadata; fall back to an empty one.
        metadata_obj = case_metadata or TestCaseMetadata()

        case_results.append(
            CaseResult(
                case_name=case_run.case_name,
                inputs=case_run.inputs,
                metadata=metadata_obj,
                base_input_key=case_run.base_input_key,
                trace_id=case_run.trace_id,
                run_results=run_results,
                aggregated=aggregated,
            )
        )

    runs_df = _create_runs_dataframe(case_results)
    cases_df = _create_cases_dataframe(case_results)
    return EvaluationOutput(
        runs=runs_df,
        cases=cases_df,
        case_results=case_results,
        dataset_run=dataset_run,
    )

See Also

• Execution Layer — produce the DatasetRunOutput this layer consumes.
• Upload Layer — persist the EvaluationOutput produced here.
• Layered Architecture Guide.
• Result Types — EvaluationOutput, CaseResult, RunResult, etc.