Skip to content

MLflow Helper

High-level MLflow entry point that chains the three layers of the ragpill pipeline. For direct access to individual layers, see Execution, Evaluation, and Upload.

Recommendation

Create dedicated MLflow experiments for evaluations. Don't mix with production traces.

evaluate_testset_with_mlflow

ragpill.evaluate_testset_with_mlflow async 

evaluate_testset_with_mlflow(testset, task=None, task_factory=None, mlflow_settings=None, model_params=None)

Run the full evaluation pipeline against an MLflow server.

Chains the three layers of the refactored architecture:

  1. :func:~ragpill.execution.execute_dataset runs the task against every case and captures traces directly to the configured MLflow server.
  2. :func:~ragpill.evaluation.evaluate_results runs every evaluator against the captured outputs.
  3. :func:~ragpill.upload.upload_to_mlflow persists aggregated results (tables, metrics, assessments) to the MLflow run created by step 1.

Parameters:

Name Type Description Default
testset Dataset[Any, Any, CaseMetadataT]

The dataset to evaluate.

 required
task TaskType | None

The task callable. Mutually exclusive with task_factory.

 None
task_factory Callable[[], TaskType] | None

A zero-arg callable returning a fresh task instance per run. Mutually exclusive with task.

 None
mlflow_settings MLFlowSettings | None

MLflow configuration. Falls back to environment vars.

 None
model_params dict[str, str] | None

Optional model parameters to log for reproducibility.

 None

Returns:

Type Description
EvaluationOutput

class:EvaluationOutput with .runs, .cases, .summary
EvaluationOutput

DataFrames and .case_results.

Raises:

Type Description
ValueError

If both or neither of task and task_factory are provided.
Example 
from ragpill import evaluate_testset_with_mlflow

result = await evaluate_testset_with_mlflow(
    testset=my_dataset,
    task=my_task,
    mlflow_settings=my_settings,
)
print(result.summary)
Source code in src/ragpill/mlflow_helper.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def evaluate_testset_with_mlflow(
    testset: Dataset[Any, Any, CaseMetadataT],
    task: TaskType | None = None,
    task_factory: Callable[[], TaskType] | None = None,
    mlflow_settings: MLFlowSettings | None = None,
    model_params: dict[str, str] | None = None,
) -> EvaluationOutput:
    """Run the full evaluation pipeline against an MLflow server.

    Chains the three layers of the refactored architecture:

    1. :func:`~ragpill.execution.execute_dataset` runs the task against every
       case and captures traces directly to the configured MLflow server.
    2. :func:`~ragpill.evaluation.evaluate_results` runs every evaluator
       against the captured outputs.
    3. :func:`~ragpill.upload.upload_to_mlflow` persists aggregated results
       (tables, metrics, assessments) to the MLflow run created by step 1.

    Args:
        testset: The dataset to evaluate.
        task: The task callable. Mutually exclusive with ``task_factory``.
        task_factory: A zero-arg callable returning a fresh task instance per
            run. Mutually exclusive with ``task``.
        mlflow_settings: MLflow configuration. Falls back to environment vars.
        model_params: Optional model parameters to log for reproducibility.

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

    Raises:
        ValueError: If both or neither of ``task`` and ``task_factory`` are
            provided.

    Example:
        ```python
        from ragpill import evaluate_testset_with_mlflow

        result = await evaluate_testset_with_mlflow(
            testset=my_dataset,
            task=my_task,
            mlflow_settings=my_settings,
        )
        print(result.summary)
        ```
    """
    settings = mlflow_settings or MLFlowSettings()  # pyright: ignore[reportCallIssue]

    run_output = await execute_dataset(
        testset,
        task=task,
        task_factory=task_factory,
        settings=settings,
        mlflow_tracking_uri=settings.ragpill_tracking_uri,
        capture_traces=True,
    )
    eval_output = await evaluate_results(run_output, testset, settings=settings)
    upload_to_mlflow(eval_output, settings, model_params=model_params, upload_traces=False)
    return eval_output

See Also