|
| 1 | +"""Data loader abstractions""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from collections.abc import Sequence |
| 6 | +from dataclasses import dataclass |
| 7 | +from typing import Callable |
| 8 | +from typing_extensions import Protocol |
| 9 | +from abc import ABC, abstractmethod |
| 10 | + |
| 11 | +from pydantic import BaseModel, Field, field_validator |
| 12 | + |
| 13 | +from eval_protocol.models import EvaluationRow |
| 14 | + |
| 15 | + |
| 16 | +class DataLoaderResult(BaseModel): |
| 17 | + """Rows and metadata returned by a loader variant.""" |
| 18 | + |
| 19 | + rows: list[EvaluationRow] = Field( |
| 20 | + description="List of evaluation rows loaded from the data source. These are the " |
| 21 | + "processed and ready-to-use evaluation data that will be fed into the evaluation pipeline." |
| 22 | + ) |
| 23 | + |
| 24 | + type: str = Field( |
| 25 | + ..., |
| 26 | + description="Type of the data loader that produced this result. Used for identification " |
| 27 | + "and debugging purposes (e.g., 'InlineDataLoader', 'DynamicDataLoader').", |
| 28 | + ) |
| 29 | + |
| 30 | + variant_id: str = Field( |
| 31 | + ..., |
| 32 | + description="Unique identifier for the data loader variant that produced this result. " |
| 33 | + "Used for tracking and organizing evaluation results from different data sources.", |
| 34 | + ) |
| 35 | + |
| 36 | + variant_description: str | None = Field( |
| 37 | + default=None, |
| 38 | + description="Human-readable description of the data loader variant that produced this result. " |
| 39 | + "Provides context about what this variant represents, its purpose, or any special characteristics that distinguish " |
| 40 | + "it from other variants.", |
| 41 | + ) |
| 42 | + |
| 43 | + preprocessed: bool = Field( |
| 44 | + default=False, |
| 45 | + description="Whether the data has been preprocessed. This flag indicates if any " |
| 46 | + "preprocessing functions have been applied to the data, helping to avoid duplicate " |
| 47 | + "processing and track data transformation state.", |
| 48 | + ) |
| 49 | + |
| 50 | + @field_validator("type") |
| 51 | + @classmethod |
| 52 | + def validate_type(cls, v: str) -> str: |
| 53 | + if not v or not v.strip(): |
| 54 | + raise ValueError("type must be non-empty") |
| 55 | + return v |
| 56 | + |
| 57 | + @field_validator("variant_id") |
| 58 | + @classmethod |
| 59 | + def validate_variant_id(cls, v: str) -> str: |
| 60 | + if not v or not v.strip(): |
| 61 | + raise ValueError("variant_id must be non-empty") |
| 62 | + return v |
| 63 | + |
| 64 | + |
| 65 | +class DataLoaderVariant(Protocol): |
| 66 | + """Single parameterizable variant from a data loader.""" |
| 67 | + |
| 68 | + def __call__(self) -> DataLoaderResult: |
| 69 | + """Load a dataset for this variant using the provided context.""" |
| 70 | + ... |
| 71 | + |
| 72 | + |
| 73 | +@dataclass(kw_only=True) |
| 74 | +class EvaluationDataLoader(ABC): |
| 75 | + """Abstract base class for data loaders that can be consumed by ``evaluation_test``.""" |
| 76 | + |
| 77 | + preprocess_fn: Callable[[list[EvaluationRow]], list[EvaluationRow]] | None = None |
| 78 | + """Optional preprocessing function for evaluation rows. This function is applied |
| 79 | + to the loaded data before it's returned, allowing for data cleaning, transformation, |
| 80 | + filtering, or other modifications. The function receives a list of EvaluationRow objects |
| 81 | + and should return a modified list of EvaluationRow objects.""" |
| 82 | + |
| 83 | + @abstractmethod |
| 84 | + def variants(self) -> Sequence[DataLoaderVariant]: |
| 85 | + """Return parameterizable variants emitted by this loader.""" |
| 86 | + ... |
| 87 | + |
| 88 | + def load(self) -> list[DataLoaderResult]: |
| 89 | + """Loads all variants of this data loader and return a list of DataLoaderResult.""" |
| 90 | + results = [] |
| 91 | + for variant in self.variants(): |
| 92 | + result = variant() |
| 93 | + result = self._process_variant(result) |
| 94 | + results.append(result) |
| 95 | + return results |
| 96 | + |
| 97 | + def _process_variant(self, result: DataLoaderResult) -> DataLoaderResult: |
| 98 | + """Process a single variant: preprocess data and apply metadata.""" |
| 99 | + # Preprocess data |
| 100 | + original_count = len(result.rows) |
| 101 | + if self.preprocess_fn: |
| 102 | + result.rows = self.preprocess_fn(result.rows) |
| 103 | + result.preprocessed = True |
| 104 | + processed_count = len(result.rows) |
| 105 | + else: |
| 106 | + processed_count = original_count |
| 107 | + |
| 108 | + # Apply metadata to rows |
| 109 | + self._apply_metadata(result, original_count, processed_count) |
| 110 | + return result |
| 111 | + |
| 112 | + def _apply_metadata(self, result: DataLoaderResult, original_count: int, processed_count: int) -> None: |
| 113 | + """Apply metadata to all rows in the result.""" |
| 114 | + for row in result.rows: |
| 115 | + if row.input_metadata.dataset_info is None: |
| 116 | + row.input_metadata.dataset_info = {} |
| 117 | + |
| 118 | + # Apply result attributes as metadata |
| 119 | + for attr_name, attr_value in vars(result).items(): |
| 120 | + """ |
| 121 | + Exclude rows and private attributes from metadata. |
| 122 | + """ |
| 123 | + if attr_name != "rows" and not attr_name.startswith("_"): |
| 124 | + row.input_metadata.dataset_info[f"data_loader_{attr_name}"] = attr_value |
| 125 | + |
| 126 | + # Apply row counts |
| 127 | + row.input_metadata.dataset_info["data_loader_num_rows"] = original_count |
| 128 | + row.input_metadata.dataset_info["data_loader_num_rows_after_preprocessing"] = processed_count |
0 commit comments