Synthetic Data

synthetic

Provider-neutral synthetic QA dataset generation.

SourceDocument dataclass

SourceDocument(id: str, text: str, metadata: Mapping[str, Any] = dict())

One source document approved for synthetic generation.

id instance-attribute

id: str

text instance-attribute

text: str

metadata class-attribute instance-attribute

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

SourceSpan dataclass

SourceSpan(start: int, end: int)

Half-open source character range supporting a generated QA pair.

start instance-attribute

start: int

end instance-attribute

end: int

SyntheticQA dataclass

SyntheticQA(query: str, answer: str, source_spans: tuple[SourceSpan, ...] = (), confidence: float | None = None)

One provider-produced question, answer, and supporting provenance.

query instance-attribute

query: str

answer instance-attribute

answer: str

source_spans class-attribute instance-attribute

source_spans: tuple[SourceSpan, ...] = ()

confidence class-attribute instance-attribute

confidence: float | None = None

SyntheticGenerationRequest dataclass

SyntheticGenerationRequest(document: SourceDocument, count: int, seed: int)

Structured request sent to a synthetic QA provider.

document instance-attribute

document: SourceDocument

count instance-attribute

count: int

seed instance-attribute

seed: int

SyntheticGenerationResponse dataclass

SyntheticGenerationResponse(items: tuple[SyntheticQA, ...], usage: tuple[UsageRecord, ...] = ())

Provider output with exact or estimated usage records.

items instance-attribute

items: tuple[SyntheticQA, ...]

usage class-attribute instance-attribute

usage: tuple[UsageRecord, ...] = ()

SyntheticQAProvider

Bases: Protocol

Generate typed QA pairs for one source document.

generate async

generate(request: SyntheticGenerationRequest) -> SyntheticGenerationResponse

Source code in src/tunarag/synthetic.py

async def generate(
    self, request: SyntheticGenerationRequest
) -> SyntheticGenerationResponse: ...

SyntheticDatasetResult dataclass

SyntheticDatasetResult(dataset: EvaluationDataset, usage: tuple[UsageRecord, ...])

Generated evaluation dataset and provider usage.

dataset instance-attribute

dataset: EvaluationDataset

usage instance-attribute

usage: tuple[UsageRecord, ...]

SyntheticDatasetBuilder

SyntheticDatasetBuilder(provider: SyntheticQAProvider, *, samples_per_document: int = 3, concurrency: int = 4, seed: int = 0, generator_version: str = 'unspecified')

Generate a validated dataset with bounded document concurrency.

Source code in src/tunarag/synthetic.py

def __init__(
    self,
    provider: SyntheticQAProvider,
    *,
    samples_per_document: int = 3,
    concurrency: int = 4,
    seed: int = 0,
    generator_version: str = "unspecified",
) -> None:
    for name, value in (
        ("samples_per_document", samples_per_document),
        ("concurrency", concurrency),
    ):
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise ValueError(f"{name} must be a positive integer")
    if isinstance(seed, bool) or not isinstance(seed, int):
        raise TypeError("synthetic generation seed must be an integer")
    self._provider = provider
    self._samples_per_document = samples_per_document
    self._concurrency = concurrency
    self._seed = seed
    self._generator_version = _required_text(generator_version, "generator version")

generate async

generate(documents: Sequence[SourceDocument]) -> SyntheticDatasetResult

Generate and validate examples in source-document order.

Source code in src/tunarag/synthetic.py

async def generate(self, documents: Sequence[SourceDocument]) -> SyntheticDatasetResult:
    """Generate and validate examples in source-document order."""

    if not documents:
        raise DatasetError("synthetic generation requires source documents")
    if len({document.id for document in documents}) != len(documents):
        raise DatasetError("source document ids must be unique")
    semaphore = asyncio.Semaphore(self._concurrency)

    async def bounded(document: SourceDocument) -> SyntheticGenerationResponse:
        async with semaphore:
            request = SyntheticGenerationRequest(
                document,
                self._samples_per_document,
                _document_seed(self._seed, document.id),
            )
            try:
                response = await self._provider.generate(request)
            except Exception as error:
                raise DatasetError(
                    "synthetic QA provider failed",
                    details={"document_id": document.id, "cause": type(error).__name__},
                ) from error
            return _validate_response(response, request)

    responses = await asyncio.gather(*(bounded(document) for document in documents))
    examples: list[EvaluationExample] = []
    usage: list[UsageRecord] = []
    provider_identity = _provider_identity(self._provider)
    for document, response in zip(documents, responses, strict=True):
        request_seed = _document_seed(self._seed, document.id)
        usage.extend(response.usage)
        for index, item in enumerate(response.items):
            contexts = tuple(document.text[span.start : span.end] for span in item.source_spans)
            if not contexts:
                contexts = (document.text,)
            metadata = {
                "source_document_id": document.id,
                "source_document_metadata": document.metadata,
                "source_spans": [
                    {"start": span.start, "end": span.end} for span in item.source_spans
                ],
                "generator": provider_identity,
                "generator_version": self._generator_version,
                "seed": request_seed,
                "confidence": item.confidence,
                "review_status": "unreviewed",
            }
            example_id = content_hash(
                {
                    "document_id": document.id,
                    "index": index,
                    "qa": metadata,
                    "query": item.query,
                    "answer": item.answer,
                },
                namespace="tunarag:synthetic-example:v1",
            )
            examples.append(
                EvaluationExample(
                    id=example_id,
                    query=item.query,
                    reference_answer=item.answer,
                    reference_contexts=contexts,
                    relevant_document_ids=(document.id,),
                    tags=("synthetic",),
                    metadata=metadata,
                    synthetic=True,
                )
            )
    return SyntheticDatasetResult(EvaluationDataset(tuple(examples)), tuple(usage))