Execution Layer

The execute layer runs tasks against a dataset and captures traces. It is the first of three independent layers in the ragpill pipeline:

1. Execute — this module.
2. Evaluate — ragpill.evaluation.
3. Upload — ragpill.upload.

execute_dataset

ragpill.execution.execute_dataset async

execute_dataset(testset, task=None, task_factory=None, settings=None, mlflow_tracking_uri=None, capture_traces=True)

Run every case in a dataset and return the captured outputs + traces.

The function is pure with respect to the dataset: evaluators attached to cases are not invoked. That is the Phase 2 evaluator's job. What happens here is task execution and trace capture.

Tracing backends (selected by mlflow_tracking_uri):

• None — use a private temp SQLite database. The database is removed after the call, but captured Trace objects are copied into the returned :class:DatasetRunOutput before cleanup.
• A URI string — trace directly to that server (dual-backend model).

Parameters:

Name	Type	Description	Default
testset	Dataset[Any, Any, CaseMetadataT]	The dataset to execute.	required
task	TaskType | None	The task callable. Mutually exclusive with task_factory.	None
task_factory	Callable[[], TaskType] | None	A zero-arg callable that returns a fresh task instance per run (use for stateful tasks). Mutually exclusive with task.	None
settings	MLFlowSettings | None	MLflow settings; falls back to environment variables.	None
mlflow_tracking_uri	str | None	Override the tracking URI. When None, a temp SQLite backend is spun up and torn down.	None
capture_traces	bool	When False, tasks are run without MLflow spans; all Trace fields in the result will be None and run_span_id will be empty. Use for fast non-traced runs.	True

Returns:

Name	Type	Description
A	DatasetRunOutput	class:DatasetRunOutput with one :class:CaseRunOutput per case
	DatasetRunOutput	and (when capture_traces=True) attached Trace objects.

Raises:

Type	Description
ValueError	If both or neither of task and task_factory are provided.

Example

from ragpill import Case, Dataset, TestCaseMetadata
from ragpill.execution import execute_dataset

ds = Dataset(cases=[Case(inputs="hi", metadata=TestCaseMetadata())])

async def my_task(q: str) -> str:
    return f"answer: {q}"

run_output = await execute_dataset(ds, task=my_task)
assert run_output.cases[0].task_runs[0].output == "answer: hi"

See Also

DatasetRunOutput: the return type. ragpill.evaluation.evaluate_results: Phase 2 — run evaluators against a DatasetRunOutput.

Source code in src/ragpill/execution.py

async def execute_dataset(
    testset: Dataset[Any, Any, CaseMetadataT],
    task: TaskType | None = None,
    task_factory: Callable[[], TaskType] | None = None,
    settings: MLFlowSettings | None = None,
    mlflow_tracking_uri: str | None = None,
    capture_traces: bool = True,
) -> DatasetRunOutput:
    """Run every case in a dataset and return the captured outputs + traces.

    The function is pure with respect to the dataset: evaluators attached to
    cases are not invoked. That is the Phase 2 evaluator's job. What happens
    here is task execution and trace capture.

    Tracing backends (selected by ``mlflow_tracking_uri``):

    - ``None`` — use a private temp SQLite database. The database is removed
      after the call, but captured ``Trace`` objects are copied into the
      returned :class:`DatasetRunOutput` before cleanup.
    - A URI string — trace directly to that server (dual-backend model).

    Args:
        testset: The dataset to execute.
        task: The task callable. Mutually exclusive with ``task_factory``.
        task_factory: A zero-arg callable that returns a fresh task instance
            per run (use for stateful tasks). Mutually exclusive with ``task``.
        settings: MLflow settings; falls back to environment variables.
        mlflow_tracking_uri: Override the tracking URI. When ``None``, a
            temp SQLite backend is spun up and torn down.
        capture_traces: When ``False``, tasks are run without MLflow spans;
            all ``Trace`` fields in the result will be ``None`` and
            ``run_span_id`` will be empty. Use for fast non-traced runs.

    Returns:
        A :class:`DatasetRunOutput` with one :class:`CaseRunOutput` per case
        and (when ``capture_traces=True``) attached ``Trace`` objects.

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

    Example:
        ```python
        from ragpill import Case, Dataset, TestCaseMetadata
        from ragpill.execution import execute_dataset

        ds = Dataset(cases=[Case(inputs="hi", metadata=TestCaseMetadata())])

        async def my_task(q: str) -> str:
            return f"answer: {q}"

        run_output = await execute_dataset(ds, task=my_task)
        assert run_output.cases[0].task_runs[0].output == "answer: hi"
        ```

    See Also:
        [`DatasetRunOutput`][ragpill.execution.DatasetRunOutput]: the return type.
        [`ragpill.evaluation.evaluate_results`][ragpill.evaluation.evaluate_results]:
            Phase 2 — run evaluators against a ``DatasetRunOutput``.
    """
    if task is not None and task_factory is not None:
        raise ValueError("Provide either 'task' or 'task_factory', not both.")
    if task is None and task_factory is None:
        raise ValueError("Provide either 'task' or 'task_factory'.")

    _factory: Callable[[], TaskType]
    if task is not None:
        _task = task
        _factory = lambda: _task  # noqa: E731
    else:
        assert task_factory is not None
        _factory = task_factory

    _settings = settings or MLFlowSettings()  # pyright: ignore[reportCallIssue]
    _fix_evaluator_global_flag(testset)

    tracing: _TracingContext | None = None
    try:
        if capture_traces:
            if mlflow_tracking_uri:
                tracing = _setup_server_tracing(mlflow_tracking_uri, _settings)
            else:
                tracing = _setup_local_tracing(_settings.ragpill_experiment_name)

        case_outputs: list[CaseRunOutput] = []
        for case in testset.cases:
            case_metadata: TestCaseMetadata | None = (
                case.metadata if isinstance(case.metadata, TestCaseMetadata) else None
            )
            repeat, _ = resolve_repeat(case_metadata, _settings)
            case_output = await _execute_case_runs(
                case,
                _factory,
                default_input_to_key,
                repeat,
                capture_traces=capture_traces,
                tracing=tracing,
            )
            case_outputs.append(case_output)

        return DatasetRunOutput(
            cases=case_outputs,
            tracking_uri=tracing.tracking_uri if (tracing and tracing.temp_dir is None) else "",
            mlflow_run_id=tracing.run_id if (tracing and tracing.temp_dir is None) else "",
            mlflow_experiment_id=tracing.experiment_id if (tracing and tracing.temp_dir is None) else "",
        )
    finally:
        _teardown_tracing(tracing)

DatasetRunOutput

ragpill.execution.DatasetRunOutput dataclass

DatasetRunOutput(cases=list(), tracking_uri='', mlflow_run_id='', mlflow_experiment_id='')

Top-level output of :func:execute_dataset.

Attributes:

Name	Type	Description
cases	list[CaseRunOutput]	One :class:CaseRunOutput per case in the dataset.
tracking_uri	str	The MLflow tracking URI that was active during execution. Empty string when the local temp backend was used (the temp DB is cleaned up after execution).
mlflow_run_id	str	MLflow run ID under which traces were captured. Empty when tracing was disabled or the temp backend was used.
mlflow_experiment_id	str	MLflow experiment ID under which the run was created. Empty when tracing was disabled or the temp backend was used.

from_json classmethod

from_json(s)

Deserialize a :class:DatasetRunOutput produced by :meth:to_json.

Parameters:

Name	Type	Description	Default
s	str	JSON string produced by :meth:to_json.	required

Returns:

Name	Type	Description
A	DatasetRunOutput	class:DatasetRunOutput equivalent to the one that was
	DatasetRunOutput	serialized (trace data survives the round trip).

Example

with open("run.json") as f:
    run_output = DatasetRunOutput.from_json(f.read())

Source code in src/ragpill/execution.py

@classmethod
def from_json(cls, s: str) -> DatasetRunOutput:
    """Deserialize a :class:`DatasetRunOutput` produced by :meth:`to_json`.

    Args:
        s: JSON string produced by :meth:`to_json`.

    Returns:
        A :class:`DatasetRunOutput` equivalent to the one that was
        serialized (trace data survives the round trip).

    Example:
        ```python
        with open("run.json") as f:
            run_output = DatasetRunOutput.from_json(f.read())
        ```
    """
    return _dataset_run_from_dict(json.loads(s))

to_json

to_json()

Serialize this output to a JSON string.

Traces are serialized via Trace.to_json(). Other fields are preserved as-is.

Returns:

Name	Type	Description
	str	A JSON string that from_json can round-trip back into an
equivalent	str	class:DatasetRunOutput.

Example

run_output = await execute_dataset(dataset, task=my_task)
with open("run.json", "w") as f:
    f.write(run_output.to_json())

Source code in src/ragpill/execution.py

def to_json(self) -> str:
    """Serialize this output to a JSON string.

    Traces are serialized via ``Trace.to_json()``. Other fields are
    preserved as-is.

    Returns:
        A JSON string that ``from_json`` can round-trip back into an
        equivalent :class:`DatasetRunOutput`.

    Example:
        ```python
        run_output = await execute_dataset(dataset, task=my_task)
        with open("run.json", "w") as f:
            f.write(run_output.to_json())
        ```
    """
    return json.dumps(_dataset_run_to_dict(self))

to_llm_text

to_llm_text(*, max_chars=16000, include_spans=True, redact=True, redact_patterns=None)

Render an exploration-focused markdown view of this run.

See :func:ragpill.report.exploration.render_dataset_run_as_exploration.

Source code in src/ragpill/execution.py

def to_llm_text(
    self,
    *,
    max_chars: int = 16_000,
    include_spans: bool = True,
    redact: bool = True,
    redact_patterns: list[str] | None = None,
) -> str:
    """Render an exploration-focused markdown view of this run.

    See :func:`ragpill.report.exploration.render_dataset_run_as_exploration`.
    """
    from ragpill.report.exploration import render_dataset_run_as_exploration

    return render_dataset_run_as_exploration(
        self,
        max_chars=max_chars,
        include_spans=include_spans,
        redact=redact,
        redact_patterns=redact_patterns,
    )

CaseRunOutput

ragpill.execution.CaseRunOutput dataclass

CaseRunOutput(case_name, inputs, expected_output, metadata, base_input_key, trace, trace_id, task_runs)

All runs for a single case, plus the case-level trace.

Attributes:

Name	Type	Description
case_name	str	Display name (falls back to str(inputs)).
inputs	Any	The task inputs for this case.
expected_output	Any	The case's expected output, if any.
metadata	dict[str, Any]	Case metadata as a plain dict (for JSON round-trip).
base_input_key	str	Hash of the inputs (no run-index suffix).
trace	Trace | None	Full case-level MLflow Trace (including all run subtrees). None when tracing is disabled.
trace_id	str	MLflow trace id string.
task_runs	list[TaskRunOutput]	One :class:TaskRunOutput per repeat.

TaskRunOutput

ragpill.execution.TaskRunOutput dataclass

TaskRunOutput(run_index, input_key, output, duration, trace=None, run_span_id='', error=None)

Output of a single task execution (one run of one case).

Attributes:

Name	Type	Description
run_index	int	Zero-based index of this run within the case's repeat sequence.
input_key	str	Unique key for this run, formatted as {base_hash}_{run_index}.
output	Any	Return value of the task, or None if the task raised.
duration	float	Wall-clock seconds the task took to run.
trace	Trace | None	MLflow Trace scoped to this run (filtered to the run's subtree). None when tracing is disabled.
run_span_id	str	Span ID of the per-run parent span inside the case trace. Empty string when tracing is disabled.
error	str | None	String representation of any exception the task raised. None on success. We store the string (not the exception) so the dataclass stays JSON-serializable.

See Also

• Evaluation Layer — run evaluators against a DatasetRunOutput.
• Upload Layer — persist results to MLflow.
• Layered Architecture Guide — the three-layer model and use cases.