Datasets

dataset

Repeatable datasets and normalized evaluation-example ingestion.

DatasetSplit

Bases: str, Enum

Canonical purpose assigned to an evaluation example.

OPTIMIZE class-attribute instance-attribute

OPTIMIZE = 'optimize'

VALIDATION class-attribute instance-attribute

VALIDATION = 'validation'

TEST class-attribute instance-attribute

TEST = 'test'

SplitRatios dataclass

SplitRatios(optimize: float = 0.8, validation: float = 0.1, test: float = 0.1)

Deterministic split thresholds.

optimize class-attribute instance-attribute

optimize: float = 0.8

validation class-attribute instance-attribute

validation: float = 0.1

test class-attribute instance-attribute

test: float = 0.1

DatasetFieldMap dataclass

DatasetFieldMap(id: str = 'id', query: str = 'query', reference_answer: str = 'reference_answer', reference_contexts: str = 'reference_contexts', relevant_document_ids: str = 'relevant_document_ids', tags: str = 'tags', metadata: str = 'metadata', split: str = 'split')

Map source record fields to the canonical evaluation schema.

id class-attribute instance-attribute

id: str = 'id'

query class-attribute instance-attribute

query: str = 'query'

reference_answer class-attribute instance-attribute

reference_answer: str = 'reference_answer'

reference_contexts class-attribute instance-attribute

reference_contexts: str = 'reference_contexts'

relevant_document_ids class-attribute instance-attribute

relevant_document_ids: str = 'relevant_document_ids'

tags class-attribute instance-attribute

tags: str = 'tags'

metadata class-attribute instance-attribute

metadata: str = 'metadata'

split class-attribute instance-attribute

split: str = 'split'

EvaluationExample dataclass

EvaluationExample(id: str, query: str, reference_answer: str | None = None, reference_contexts: tuple[str, ...] = (), relevant_document_ids: tuple[str, ...] = (), tags: tuple[str, ...] = (), metadata: Mapping[str, Any] = dict(), split: DatasetSplit = OPTIMIZE, synthetic: bool = False)

One normalized, provenance-bearing evaluation example.

id instance-attribute

id: str

query instance-attribute

query: str

reference_answer class-attribute instance-attribute

reference_answer: str | None = None

reference_contexts class-attribute instance-attribute

reference_contexts: tuple[str, ...] = ()

relevant_document_ids class-attribute instance-attribute

relevant_document_ids: tuple[str, ...] = ()

tags class-attribute instance-attribute

tags: tuple[str, ...] = ()

metadata class-attribute instance-attribute

metadata: Mapping[str, Any] = field(default_factory=dict)

split class-attribute instance-attribute

split: DatasetSplit = DatasetSplit.OPTIMIZE

synthetic class-attribute instance-attribute

synthetic: bool = False

semantic_data

semantic_data() -> dict[str, Any]

Return normalized content used for version hashing.

Source code in src/tunarag/dataset.py

def semantic_data(self) -> dict[str, Any]:
    """Return normalized content used for version hashing."""

    return {
        "id": self.id,
        "query": self.query,
        "reference_answer": self.reference_answer,
        "reference_contexts": self.reference_contexts,
        "relevant_document_ids": self.relevant_document_ids,
        "tags": self.tags,
        "metadata": self.metadata,
        "split": self.split.value,
        "synthetic": self.synthetic,
    }

EvaluationDataset dataclass

EvaluationDataset(examples: tuple[EvaluationExample, ...])

A normalized, immutable dataset with a content-derived version.

examples instance-attribute

examples: tuple[EvaluationExample, ...]

version class-attribute instance-attribute

version: str = field(init=False)

from_records classmethod

from_records(records: Iterable[Mapping[str, Any]], *, field_map: DatasetFieldMap | None = None, source: str = 'records') -> EvaluationDataset

Normalize mapping records into an immutable dataset.

Source code in src/tunarag/dataset.py

@classmethod
def from_records(
    cls,
    records: Iterable[Mapping[str, Any]],
    *,
    field_map: DatasetFieldMap | None = None,
    source: str = "records",
) -> EvaluationDataset:
    """Normalize mapping records into an immutable dataset."""

    mapping = field_map or DatasetFieldMap()
    examples: list[EvaluationExample] = []
    for index, record in enumerate(records, start=1):
        if not isinstance(record, Mapping):
            raise DatasetError(
                "dataset record must be an object",
                details={"source": source, "record": index},
            )
        try:
            examples.append(_example_from_record(record, mapping))
        except (TypeError, ValueError, json.JSONDecodeError) as error:
            raise DatasetError(
                "dataset record is invalid",
                details={
                    "source": source,
                    "record": index,
                    "cause": type(error).__name__,
                },
            ) from error
    try:
        return cls(tuple(examples))
    except ValueError as error:
        raise DatasetError(
            "dataset validation failed",
            details={"source": source, "cause": type(error).__name__},
        ) from error

from_jsonl classmethod

from_jsonl(path: str | Path, *, field_map: DatasetFieldMap | None = None) -> EvaluationDataset

Load UTF-8 JSON Lines, ignoring blank lines.

Source code in src/tunarag/dataset.py

@classmethod
def from_jsonl(
    cls, path: str | Path, *, field_map: DatasetFieldMap | None = None
) -> EvaluationDataset:
    """Load UTF-8 JSON Lines, ignoring blank lines."""

    source = str(path)
    records: list[Mapping[str, Any]] = []
    try:
        with Path(path).open("r", encoding="utf-8-sig") as stream:
            for line_number, line in enumerate(stream, start=1):
                if not line.strip():
                    continue
                try:
                    value = json.loads(line)
                except json.JSONDecodeError as error:
                    raise DatasetError(
                        "could not load JSONL dataset",
                        details={
                            "source": source,
                            "record": line_number,
                            "cause": type(error).__name__,
                        },
                    ) from error
                if not isinstance(value, Mapping):
                    raise DatasetError(
                        "JSONL row must be an object",
                        details={"source": source, "record": line_number},
                    )
                records.append(value)
    except DatasetError:
        raise
    except (OSError, UnicodeError) as error:
        raise DatasetError(
            "could not load JSONL dataset",
            details={"source": source, "cause": type(error).__name__},
        ) from error
    return cls.from_records(records, field_map=field_map, source=source)

from_json classmethod

from_json(path: str | Path, *, field_map: DatasetFieldMap | None = None) -> EvaluationDataset

Load a UTF-8 JSON array of objects.

Source code in src/tunarag/dataset.py

@classmethod
def from_json(
    cls, path: str | Path, *, field_map: DatasetFieldMap | None = None
) -> EvaluationDataset:
    """Load a UTF-8 JSON array of objects."""

    source = str(path)
    try:
        with Path(path).open("r", encoding="utf-8-sig") as stream:
            value = json.load(stream)
    except (OSError, UnicodeError, json.JSONDecodeError) as error:
        raise DatasetError(
            "could not load JSON dataset",
            details={"source": source, "cause": type(error).__name__},
        ) from error
    if not isinstance(value, list):
        raise DatasetError("JSON dataset must be an array", details={"source": source})
    return cls.from_records(value, field_map=field_map, source=source)

from_csv classmethod

from_csv(path: str | Path, *, field_map: DatasetFieldMap | None = None) -> EvaluationDataset

Load a UTF-8 CSV with JSON-encoded collection fields.

Source code in src/tunarag/dataset.py

@classmethod
def from_csv(
    cls, path: str | Path, *, field_map: DatasetFieldMap | None = None
) -> EvaluationDataset:
    """Load a UTF-8 CSV with JSON-encoded collection fields."""

    source = str(path)
    try:
        with Path(path).open("r", encoding="utf-8-sig", newline="") as stream:
            records = list(csv.DictReader(stream))
    except (OSError, UnicodeError, csv.Error) as error:
        raise DatasetError(
            "could not load CSV dataset",
            details={"source": source, "cause": type(error).__name__},
        ) from error
    return cls.from_records(records, field_map=field_map, source=source)

with_splits

with_splits(*, seed: int = 0, ratios: SplitRatios | None = None, group_metadata_key: str | None = None) -> EvaluationDataset

Return a deterministically split copy, optionally keeping groups together.

Source code in src/tunarag/dataset.py

def with_splits(
    self,
    *,
    seed: int = 0,
    ratios: SplitRatios | None = None,
    group_metadata_key: str | None = None,
) -> EvaluationDataset:
    """Return a deterministically split copy, optionally keeping groups together."""

    if isinstance(seed, bool) or not isinstance(seed, int):
        raise TypeError("split seed must be an integer")
    if group_metadata_key is not None and not group_metadata_key.strip():
        raise ValueError("group metadata key must not be empty")
    thresholds = ratios or SplitRatios()
    examples = tuple(
        replace(
            example,
            split=_select_split(
                seed,
                _group_identity(example, group_metadata_key),
                thresholds,
            ),
        )
        for example in self.examples
    )
    return EvaluationDataset(examples)

select

select(split: DatasetSplit) -> EvaluationDataset

Return one nonempty canonical split.

Source code in src/tunarag/dataset.py

def select(self, split: DatasetSplit) -> EvaluationDataset:
    """Return one nonempty canonical split."""

    selected = tuple(example for example in self.examples if example.split is split)
    if not selected:
        raise DatasetError("selected dataset split is empty", details={"split": split.value})
    return EvaluationDataset(selected)

InMemoryDataset dataclass

InMemoryDataset(examples: tuple[Any, ...], version: str)

A versioned, repeatable dataset backed by an immutable tuple.

examples instance-attribute

examples: tuple[Any, ...]

version instance-attribute

version: str

from_iterable classmethod

from_iterable(examples: Iterable[Any], *, version: str) -> InMemoryDataset

Materialize a repeatable dataset from an iterable.

Source code in src/tunarag/dataset.py

@classmethod
def from_iterable(cls, examples: Iterable[Any], *, version: str) -> InMemoryDataset:
    """Materialize a repeatable dataset from an iterable."""

    return cls(tuple(examples), version)