Generator — Python API

The blessed programmatic surface. Import everything from the package top level:

from seed_data import (
    Generator, ModelConfig, Schema, InferredSchema,
    GeneratedDoc, BatchResult, StructuredResult,
)

The mental model: configure a Generator once (models, threshold, renderer, output directory), then call one of its verbs. Configuration lives on the Generator; per-call arguments describe only what to make. Every verb returns a typed result — no stringly-typed dicts.

Generation — make synthetic documents:

Verb	Makes	Returns
`gen.generate(...)`	one document	`GeneratedDoc`
`gen.generate_batch(...)`	N diverse documents from one brief	`BatchResult`
`gen.generate_packet(...)`	a coordinated multi-document packet	`PacketResult` (or `list`)

Inference — reverse-engineer a schema from real example documents, then feed it back into generation (see Schema from Documents):

Verb	Makes	Returns
`gen.infer_schema(...)`	a `Schema` from sample doc(s) of one type	`Schema`
`gen.infer_packet(...)`	a packet dir from one concatenated multi-doc PDF	`str` (output dir)
`gen.generate_from_samples(...)`	infer, then generate one document	`GeneratedDoc`
`gen.generate_batch_from_samples(...)`	infer, then generate a batch	`BatchResult`

Planning & structured data — turn any input into a schema, generate tabular data from it, or do both in one call:

Verb	Makes	Returns
`gen.plan(...)`	one schema from any mix of inputs	`InferredSchema`
`gen.generate_structured(...)`	tabular data (CSV/Parquet/Excel/JSON)	`StructuredResult`
`gen.plan_and_generate(...)`	plan, then generate — end to end	`GeneratedDoc` \| `BatchResult` \| `StructuredResult`

See also the task-oriented guides: Single Document · Batch Generation · Packets · Schema from Documents · Plan · Structured Data.

`Generator(...)` — configure once

Every argument is keyword-only and optional; the defaults shown below are the real defaults.

from seed_data import Generator, ModelConfig

gen = Generator(
    models=ModelConfig(doc="gpt-oss", critic="haiku"),  # per-agent model choice; per-field defaults
    threshold=7,             # critic acceptance threshold, 1-10
    renderer="xhtml2pdf",    # PDF backend: "xhtml2pdf" (default) | "weasyprint" | "reportlab"
    output_dir="./output",   # root directory for generated artifacts
    max_attempts=5,          # max critic-retry attempts per document
    timeout=3600,            # per-document timeout, seconds
    critic_samples=True,     # include reference sample PDFs in the doc critic
    augment=False,           # apply image augmentation (aging/scanning) by default
    session=None,            # optional boto3 Session (containers/Lambda/AgentCore)
)

# Discover what's bundled with the package:
Generator.available_schemas()      # -> ["invoice", ...]
Generator.available_packets()      # -> ["lending-package", ...]
Generator.available_input_types()  # -> ["free_text", "example_data", ...] (plan)

threshold=7 is the Python default; the CLI's --threshold defaults to 5.

Passing session lets the generator run in-process with explicit credentials (containers, Lambda, AgentCore). If omitted, credentials resolve from the environment (AWS_PROFILE), and model IDs in models may be raw Bedrock IDs for region portability (EU/GovCloud).

`seed_data.api.Generator`

Configure once, generate many. The main entry point for seed-data.

Parameters:

Name	Type	Description	Default
`models`	`ModelConfig \| None`	Which model each agent uses (`ModelConfig`). Defaults applied per-field, so `ModelConfig(doc="gpt-oss")` overrides only the doc model.	`None`
`threshold`	`int`	Critic acceptance threshold, 1-10.	`7`
`renderer`	`str`	PDF backend — "xhtml2pdf" (default, pure Python), "weasyprint", or "reportlab".	`'xhtml2pdf'`
`output_dir`	`str`	Root directory for generated artifacts.	`'./output'`
`max_attempts`	`int`	Max critic-retry attempts per document.	`5`
`timeout`	`int`	Per-document timeout (seconds).	`3600`
`critic_samples`	`bool`	Include reference sample PDFs in the doc critic.	`True`
`augment`	`bool`	Apply image augmentation (aging/scanning artifacts) by default.	`False`
`session`	`Any`	Optional boto3 Session for in-process use (containers, Lambda, AgentCore). If omitted, credentials resolve from the environment (`AWS_PROFILE`). Model IDs in `models` may be raw Bedrock IDs for region portability (EU/GovCloud).	`None`

Source code in seed_data/api.py

class Generator:
    """Configure once, generate many. The main entry point for seed-data.

    Args:
        models: Which model each agent uses (``ModelConfig``). Defaults applied
            per-field, so ``ModelConfig(doc="gpt-oss")`` overrides only the doc model.
        threshold: Critic acceptance threshold, 1-10.
        renderer: PDF backend — "xhtml2pdf" (default, pure Python), "weasyprint",
            or "reportlab".
        output_dir: Root directory for generated artifacts.
        max_attempts: Max critic-retry attempts per document.
        timeout: Per-document timeout (seconds).
        critic_samples: Include reference sample PDFs in the doc critic.
        augment: Apply image augmentation (aging/scanning artifacts) by default.
        session: Optional boto3 Session for in-process use (containers, Lambda,
            AgentCore). If omitted, credentials resolve from the environment
            (``AWS_PROFILE``). Model IDs in ``models`` may be raw Bedrock IDs for
            region portability (EU/GovCloud).
    """

    def __init__(
        self,
        *,
        models: ModelConfig | None = None,
        threshold: int = 7,
        renderer: str = "xhtml2pdf",
        output_dir: str = "./output",
        max_attempts: int = 5,
        timeout: int = 3600,
        critic_samples: bool = True,
        augment: bool = False,
        session: Any = None,
    ):
        self.models = models or ModelConfig()
        self.threshold = threshold
        self.renderer = renderer
        self.output_dir = output_dir
        self.max_attempts = max_attempts
        self.timeout = timeout
        self.critic_samples = critic_samples
        self.augment = augment
        self.session = session

    def __repr__(self) -> str:
        return (
            f"Generator(models={self.models!r}, threshold={self.threshold}, "
            f"renderer={self.renderer!r}, output_dir={self.output_dir!r})"
        )

    # -- verbs ---------------------------------------------------------------

    def generate(
        self,
        schema: "str | Schema | InferredSchema",
        *,
        scenario: str = "",
        augment: bool | None = None,
        entity: str | None = None,
        verbose: bool = True,
    ) -> GeneratedDoc:
        """Generate a single document. Returns a typed ``GeneratedDoc``.

        Args:
            schema: One of —
                - a bundled schema name (e.g. ``"invoice"``),
                - a path to a schema directory,
                - a :class:`Schema` object defined in code (JSON schema or a
                  pydantic model, plus generation guidance), or
                - an :class:`InferredSchema` (e.g. from ``plan``), adapted
                  internally to the pipeline's schema triple.
            scenario: Free-text describing what to generate this run (the vendor,
                industry, region, size, etc.) — steers the content.
            augment: Override the instance's augment setting for this call.
            entity: For a multi-entity ``InferredSchema``, which entity to render
                as the document type. Defaults to the first.
            verbose: Print stage progress.
        """
        common = dict(
            output_dir=self.output_dir,
            extra=scenario,
            models=self.models,
            threshold=self.threshold,
            max_attempts=self.max_attempts,
            timeout=self.timeout,
            renderer=self.renderer,
            critic_samples=self.critic_samples,
            augment=self.augment if augment is None else augment,
            verbose=verbose,
            session=self.session,
        )
        resolved = self._resolve_for_documents(schema, entity=entity)
        if resolved is not None:
            return _pipeline_generate(resolved=resolved, **common)
        return _pipeline_generate(schema_dir=self._resolve_schema(schema), **common)

    def generate_batch(
        self,
        schema: "str | Schema | InferredSchema",
        *,
        count: int,
        scenario: str,
        augment: bool | None = None,
        entity: str | None = None,
        seed: int | None = None,
        on_document: Callable[[int, int, GeneratedDoc], None] | None = None,
        verbose: bool = True,
    ) -> BatchResult:
        """Generate a diverse batch of documents from one high-level scenario.

        A planner turns ``scenario`` into ``count`` distinct, specific scenarios;
        each runs its own self-contained pipeline graph as a sibling node, and
        Strands executes them concurrently.

        ``schema`` accepts a bundled name, a directory path, a ``Schema``, or an
        ``InferredSchema``.

        Args:
            scenario: The high-level theme the planner diversifies into ``count``
                specific documents. A specific scenario yields far more varied
                output than a generic one.
            entity: For a multi-entity ``InferredSchema``, which entity to render.
            seed: optional seed for scenario planning (regression-stable sets).
            on_document: optional ``callback(index, total, GeneratedDoc)`` fired as
                each document's result is collected — for host-side progress UIs.
        """
        from seed_data.stages.batch import generate_batch as _batch

        common = dict(
            count=count, brief=scenario, output_dir=self.output_dir,
            models=self.models, threshold=self.threshold,
            max_attempts=self.max_attempts, timeout=self.timeout,
            renderer=self.renderer, critic_samples=self.critic_samples,
            augment=self.augment if augment is None else augment,
            verbose=verbose, session=self.session, seed=seed,
            on_document=on_document,
        )
        resolved = self._resolve_for_documents(schema, entity=entity)
        if resolved is not None:
            docs = _batch(resolved=resolved, **common)
        else:
            docs = _batch(schema_dir=self._resolve_schema(schema), **common)

        succeeded = sum(1 for d in docs if d.success)
        return BatchResult(
            count_requested=count,
            count_succeeded=succeeded,
            count_failed=len(docs) - succeeded,
            documents=docs,
        )

    def generate_packet(
        self,
        packet: str,
        *,
        count: int = 1,
        scenario: str = "",
        shuffle: bool = False,
        doc_workers: int = 1,
        augment: bool | None = None,
    ):
        """Generate a coordinated multi-document packet.

        Args:
            packet: Path to a packet directory (containing a packet config), or a
                bundled packet name.
            count: Number of packets to generate.
            scenario: Free-text scenario shared across the packet's documents
                (the same applicant, situation, etc.).
            shuffle: Randomize sub-document order in the merged PDF.
            doc_workers: Parallel workers for sub-documents within a packet.

        Returns a ``PacketResult`` (count==1) or a list of them (count>1).
        """
        from seed_data.packet import load_packet_config, generate_packet as _gen_packet

        config = load_packet_config(self._resolve_packet(packet))
        common = dict(
            output_dir=self.output_dir,
            extra=scenario,
            shuffle=shuffle,
            doc_workers=doc_workers,
            data_model=self.models.data,
            doc_model=self.models.doc,
            critic_model=self.models.critic,
            aug_model=self.models.aug,
            context_model=self.models.batch,
            threshold=self.threshold,
            max_attempts=self.max_attempts,
            timeout=self.timeout,
            augment=self.augment if augment is None else augment,
            critic_samples=self.critic_samples,
            session=self.session,
            renderer=self.renderer,
        )
        if count == 1:
            return _gen_packet(config=config, **common)
        return [_gen_packet(config=config, **common) for _ in range(count)]

    # -- schema inference (documents -> Schema) ------------------------------

    def infer_schema(
        self,
        inputs: "str | list[str]",
        *,
        name: str,
        model: str | None = None,
        max_docs: int = 5,
        output_dir: str | None = None,
        on_question: "Callable[[str], str] | None" = None,
        verbose: bool = True,
    ) -> "Schema":
        """Infer a document-type ``Schema`` from real sample documents.

        The inverse of ``generate``: reads one or more real examples (PDF, PNG, or
        JPEG; local paths/globs/dirs and/or ``s3://`` URIs) with a vision model and
        returns a ``Schema`` (JSON Schema + generation guidance) ready to feed back
        into ``generate`` / ``generate_batch``.

        Args:
            inputs: sample document location(s) — mixed local and S3 accepted.
            name: the document-type name (becomes the schema title / doctype).
            model: vision-capable model key; defaults to ``sonnet``.
            max_docs: cap on how many examples to feed the model.
            output_dir: if given, also write the inferred schema there as
                ``schema.json`` + ``generation_guidance.md`` for review/reuse.
            on_question: optional ``callback(question) -> answer`` enabling a
                clarifying dialogue — the model may ask about ambiguous details
                (required-vs-optional, value ranges) and the callback collects the
                answer (stdin, notebook, web). ``None`` runs non-interactively.
            verbose: print progress.

        Returns:
            The inferred ``Schema`` (always returned, even when also written).
        """
        from seed_data.infer import infer_schema as _infer, write_schema_dir, DEFAULT_INFER_MODEL
        schema = _infer(
            inputs, name=name, model=model or DEFAULT_INFER_MODEL,
            max_docs=max_docs, session=self.session,
            on_question=on_question, verbose=verbose,
        )
        if output_dir:
            write_schema_dir(schema, output_dir)
            if verbose:
                print(f"Wrote inferred schema to {output_dir}/ (schema.json + generation_guidance.md)")
        return schema

    def infer_packet(
        self,
        inputs: "str | list[str]",
        *,
        name: str,
        output_dir: str,
        model: str | None = None,
        boundaries: str | None = None,
        on_question: "Callable[[str], str] | None" = None,
        verbose: bool = True,
    ) -> str:
        """Infer a packet definition from ONE concatenated multi-document PDF.

        The inverse of ``generate_packet``: takes a single file containing several
        *different* document types concatenated (e.g. a lending package), detects
        the document boundaries + classes with a vision model (or a ``boundaries``
        page-range override), infers a schema per segment, and writes a
        ``packet.json`` + one schema dir per segment to ``output_dir`` — the shape
        ``generate_packet`` / ``seed-data packet`` consumes.

        Args:
            inputs: a single concatenated PDF (path or ``s3://`` URI).
            name: the packet name.
            output_dir: where to write packet.json + per-segment schema dirs.
            model: vision-capable model key; defaults to ``sonnet``.
            boundaries: optional ``"1-2,3,4-5"`` page-range override for splitting.
            verbose: print progress.

        Returns:
            The output directory path (containing packet.json + schema dirs).
        """
        from seed_data.packet_infer import infer_packet as _infer_packet
        from seed_data.infer import DEFAULT_INFER_MODEL
        return _infer_packet(
            inputs, name=name, output_dir=output_dir,
            model=model or DEFAULT_INFER_MODEL, boundaries=boundaries,
            session=self.session, on_question=on_question, verbose=verbose,
        )

    def generate_from_samples(
        self,
        inputs: "str | list[str]",
        *,
        name: str,
        scenario: str = "",
        infer_model: str | None = None,
        max_docs: int = 5,
        output_dir: str | None = None,
        on_question: "Callable[[str], str] | None" = None,
        augment: bool | None = None,
        verbose: bool = True,
    ) -> GeneratedDoc:
        """Infer a schema from sample documents, then generate ONE synthetic doc.

        A convenience one-shot: ``infer_schema(...)`` followed by ``generate(...)``
        against the inferred ``Schema``. If ``output_dir`` is given the inferred
        schema is also persisted there for review/reuse (recommended), so you keep
        an editable schema rather than a throwaway. ``on_question`` enables the
        clarifying dialogue during inference (see ``infer_schema``).
        """
        schema = self.infer_schema(
            inputs, name=name, model=infer_model, max_docs=max_docs,
            output_dir=output_dir, on_question=on_question, verbose=verbose,
        )
        return self.generate(schema, scenario=scenario, augment=augment, verbose=verbose)

    def generate_batch_from_samples(
        self,
        inputs: "str | list[str]",
        *,
        name: str,
        count: int,
        scenario: str,
        infer_model: str | None = None,
        max_docs: int = 5,
        output_dir: str | None = None,
        on_question: "Callable[[str], str] | None" = None,
        augment: bool | None = None,
        seed: int | None = None,
        on_document: Callable[[int, int, GeneratedDoc], None] | None = None,
        verbose: bool = True,
    ) -> BatchResult:
        """Infer a schema from sample documents, then generate a diverse BATCH.

        A convenience one-shot: ``infer_schema(...)`` followed by
        ``generate_batch(...)`` against the inferred ``Schema``. ``on_question``
        enables the clarifying dialogue during inference (see ``infer_schema``).
        """
        schema = self.infer_schema(
            inputs, name=name, model=infer_model, max_docs=max_docs,
            output_dir=output_dir, on_question=on_question, verbose=verbose,
        )
        return self.generate_batch(
            schema, count=count, scenario=scenario, augment=augment,
            seed=seed, on_document=on_document, verbose=verbose,
        )

    # -- structured data (tabular) ------------------------------------------

    def plan(self, *inputs: str, name: str = "dataset", verbose: bool = True) -> InferredSchema:
        """Plan a dataset: turn any inputs (text, CSV, PDF, JSON Schema, SQL DDL,
        ERD) into a unified :class:`InferredSchema`. Auto-detects each input's type.

        Named for what you get back — a *plan* for the data, to read and edit
        before anything is generated — rather than for the reading of the inputs.

        Documents/images/``s3://`` inputs are routed through the existing vision
        path (``infer_schema``); non-document inputs go through the schema
        extraction agent. Returns a typed ``InferredSchema`` — not a dict —
        consistent with the other verbs.

        Args:
            *inputs: paths, globs, ``s3://`` URIs, or bare free-text descriptions.
            name: logical dataset name (used when delegating documents).
            verbose: print progress.
        """
        from seed_data.ingest import run_ingest
        return run_ingest(
            *inputs, name=name, models=self.models,
            session=self.session, verbose=verbose,
        )

    def ingest(self, *inputs: str, name: str = "dataset", verbose: bool = True) -> InferredSchema:
        """Deprecated alias for :meth:`plan`.

        Never shipped in a release — both names have only ever existed on this
        branch — so this is a courtesy for in-flight callers, not a compatibility
        guarantee. Slated for removal.
        """
        warnings.warn(
            "Generator.ingest() is deprecated; use Generator.plan() instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.plan(*inputs, name=name, verbose=verbose)

    def generate_structured(
        self,
        schema: "str | InferredSchema",
        *,
        rows: int = 100,
        format: str = "csv",
        seed: int | None = None,
        verbose: bool = True,
    ) -> StructuredResult:
        """Generate structured data (CSV/Parquet/Excel/JSON) from a schema.

        ``schema`` accepts a bundled schema name, a path to an ``InferredSchema``
        JSON file, or an ``InferredSchema`` object. Returns a typed
        ``StructuredResult`` — consistent with ``GeneratedDoc`` / ``BatchResult`` /
        ``PacketResult``.

        Args:
            schema: bundled name, JSON path, or ``InferredSchema`` object.
            rows: target records per entity.
            format: ``csv`` / ``parquet`` / ``excel`` / ``json``.
            seed: RNG seed. Pins the programmatically generated columns (numeric,
                enum, date, ID, pattern) so a re-run with the same schema and seed
                reproduces them. Free-text fields come from an LLM and are not
                seedable, so reproducibility covers the structured columns only.
            verbose: print progress.
        """
        from seed_data.common.deps import require_structured

        # Checked before anything else: without the extra, the pipeline's own
        # pandas import fails deep inside run_graph_pipeline, where the broad
        # `except` in run_structured turns it into
        # StructuredResult(error="No module named 'pandas'") — a missing install
        # reported as a generation failure. Fail here, naming the extra.
        require_structured("Generator.generate_structured")

        from seed_data.structured import run_structured
        resolved = self._resolve_inferred(schema)
        return run_structured(
            resolved, target_count=rows, export_format=format,
            output_dir=self.output_dir, models=self.models,
            threshold=self.threshold, session=self.session,
            seed=seed, verbose=verbose,
        )

    # -- end-to-end ----------------------------------------------------------

    def plan_and_generate(
        self,
        *inputs: str,
        output: str = "structured",
        name: str = "dataset",
        rows: int = 100,
        format: str = "csv",
        count: int = 1,
        scenario: str = "",
        entity: str | None = None,
        augment: bool | None = None,
        seed: int | None = None,
        verbose: bool = True,
    ) -> "GeneratedDoc | BatchResult | StructuredResult":
        """End-to-end: plan a schema from the inputs, then generate — in one call.

        Chains :meth:`plan` into :meth:`generate_structured` (``output="structured"``)
        or :meth:`generate` / :meth:`generate_batch` (``output="documents"``).
        Configuration stays on the ``Generator``; per-call args describe only what
        to make.

        This skips the review step :meth:`plan` exists to enable — the inferred
        schema goes straight to generation unseen. Prefer ``plan`` then a generate
        verb when the schema's accuracy matters.

        Args:
            *inputs: anything ``plan`` accepts — free text, paths, globs, ``s3://``.
            output: ``"structured"`` (CSV/Parquet/Excel) or ``"documents"`` (PDFs).
            name: logical dataset name passed through to ``plan``.
            rows: structured only — target records per entity.
            format: structured only — ``csv`` / ``parquet`` / ``excel`` / ``json``.
            count: documents only — how many to generate (>1 dispatches to batch).
            scenario: documents only — free-text steering the content.
            entity: documents only — which entity of a multi-entity schema to render.
            augment: documents only — override the instance augment setting.
            seed: RNG seed. For ``output="structured"`` it pins the programmatic
                columns; for ``output="documents"`` with ``count > 1`` it pins
                batch scenario planning. The planning step is an LLM run and is
                never seedable, so the schema itself can still differ between runs.
            verbose: print progress.

        Returns:
            ``StructuredResult`` for structured output; ``GeneratedDoc``
            (``count == 1``) or ``BatchResult`` (``count > 1``) for documents.

        Raises:
            ValueError: if ``output`` is not ``"structured"`` or ``"documents"``.
        """
        if output not in ("structured", "documents"):
            raise ValueError(
                f"output must be 'structured' or 'documents', got {output!r}"
            )

        if output == "structured":
            # Validated up front rather than at the generate_structured call
            # below: planning is a multi-agent LLM run, and failing after it would
            # bill the user for the expensive half of the chain before reporting
            # a missing install that was knowable from the start.
            from seed_data.common.deps import require_structured

            require_structured("Generator.plan_and_generate(output='structured')")

        schema = self.plan(*inputs, name=name, verbose=verbose)

        if output == "structured":
            return self.generate_structured(
                schema, rows=rows, format=format, seed=seed, verbose=verbose,
            )

        if count == 1:
            # `generate` has no seedable randomness — a single document is one LLM
            # call — so `seed` is deliberately not forwarded here.
            return self.generate(
                schema, scenario=scenario, augment=augment,
                entity=entity, verbose=verbose,
            )
        return self.generate_batch(
            schema, count=count, scenario=scenario, augment=augment,
            entity=entity, seed=seed, verbose=verbose,
        )

    def run(self, *inputs: str, **kwargs) -> "GeneratedDoc | BatchResult | StructuredResult":
        """Deprecated alias for :meth:`plan_and_generate`.

        ``**kwargs`` rather than the full signature so the two cannot drift: a new
        keyword on ``plan_and_generate`` is forwarded without being restated here.
        """
        warnings.warn(
            "Generator.run() is deprecated; use Generator.plan_and_generate() instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        return self.plan_and_generate(*inputs, **kwargs)

    # -- discovery -----------------------------------------------------------

    @staticmethod
    def available_schemas() -> list[str]:
        """Names of schemas bundled with the package."""
        root = _bundled_dir("schemas")
        return _list_subdirs(root)

    @staticmethod
    def available_packets() -> list[str]:
        """Names of packets bundled with the package."""
        root = _bundled_dir("packets")
        return _list_subdirs(root)

    @staticmethod
    def available_input_types() -> list[str]:
        """Input types the ``plan`` verb can auto-detect."""
        from seed_data.ingest import InputType
        return [t.value for t in InputType]

    # -- internals -----------------------------------------------------------

    def _resolve_schema(self, schema: str) -> str:
        return _resolve(schema, "schemas")

    def _resolve_packet(self, packet: str) -> str:
        return _resolve(packet, "packets")

    def _resolve_for_documents(
        self, schema: "str | Schema | InferredSchema", *, entity: str | None = None,
    ) -> tuple[dict, str, list[str]] | None:
        """Resolve an in-code schema to the pipeline's ``(dict, guidance, samples)``.

        Returns ``None`` for a plain string, signalling the caller to fall back to
        the ``schema_dir=`` path (bundled name or directory) — which keeps the
        existing behaviour for every already-published call shape.
        """
        from seed_data.schema import Schema

        if isinstance(schema, Schema):
            return schema.resolve()
        if isinstance(schema, InferredSchema):
            from seed_data.schema.adapter import inferred_to_resolved
            return inferred_to_resolved(schema, entity_name=entity)
        return None

    def _resolve_inferred(self, schema: "str | InferredSchema") -> InferredSchema:
        """Resolve a schema spec into an :class:`InferredSchema`.

        Accepts an ``InferredSchema`` (returned as-is), a path to a JSON file
        (an ``InferredSchema`` dump or a JSON-Schema document), or a bundled
        schema name (resolved to its ``schema.json`` and converted).
        """
        if isinstance(schema, InferredSchema):
            return schema

        from seed_data.schema.io import from_json_schema

        if isinstance(schema, str):
            path = schema if os.path.isfile(schema) else None
            if path is None:
                # Try a bundled schema directory.
                from seed_data.schema.io import from_schema_dir
                resolved_dir = _resolve(schema, "schemas")
                return from_schema_dir(resolved_dir)

            with open(path) as f:
                data = json.load(f)
            # An InferredSchema dump has a top-level "entities" list.
            if isinstance(data, dict) and "entities" in data:
                return InferredSchema.model_validate(data)
            return from_json_schema(data)

        raise TypeError(f"Cannot resolve schema of type {type(schema).__name__}")

`generate(schema, *, scenario='', augment=None, entity=None, verbose=True)`

Generate a single document. Returns a typed GeneratedDoc.

Parameters:

Name	Type	Description	Default
`schema`	`'str \| Schema \| InferredSchema'`	One of — - a bundled schema name (e.g. `"invoice"`), - a path to a schema directory, - a :class:`Schema` object defined in code (JSON schema or a pydantic model, plus generation guidance), or - an :class:`InferredSchema` (e.g. from `plan`), adapted internally to the pipeline's schema triple.	required
`scenario`	`str`	Free-text describing what to generate this run (the vendor, industry, region, size, etc.) — steers the content.	`''`
`augment`	`bool \| None`	Override the instance's augment setting for this call.	`None`
`entity`	`str \| None`	For a multi-entity `InferredSchema`, which entity to render as the document type. Defaults to the first.	`None`
`verbose`	`bool`	Print stage progress.	`True`

Source code in seed_data/api.py

def generate(
    self,
    schema: "str | Schema | InferredSchema",
    *,
    scenario: str = "",
    augment: bool | None = None,
    entity: str | None = None,
    verbose: bool = True,
) -> GeneratedDoc:
    """Generate a single document. Returns a typed ``GeneratedDoc``.

    Args:
        schema: One of —
            - a bundled schema name (e.g. ``"invoice"``),
            - a path to a schema directory,
            - a :class:`Schema` object defined in code (JSON schema or a
              pydantic model, plus generation guidance), or
            - an :class:`InferredSchema` (e.g. from ``plan``), adapted
              internally to the pipeline's schema triple.
        scenario: Free-text describing what to generate this run (the vendor,
            industry, region, size, etc.) — steers the content.
        augment: Override the instance's augment setting for this call.
        entity: For a multi-entity ``InferredSchema``, which entity to render
            as the document type. Defaults to the first.
        verbose: Print stage progress.
    """
    common = dict(
        output_dir=self.output_dir,
        extra=scenario,
        models=self.models,
        threshold=self.threshold,
        max_attempts=self.max_attempts,
        timeout=self.timeout,
        renderer=self.renderer,
        critic_samples=self.critic_samples,
        augment=self.augment if augment is None else augment,
        verbose=verbose,
        session=self.session,
    )
    resolved = self._resolve_for_documents(schema, entity=entity)
    if resolved is not None:
        return _pipeline_generate(resolved=resolved, **common)
    return _pipeline_generate(schema_dir=self._resolve_schema(schema), **common)

`generate_batch(schema, *, count, scenario, augment=None, entity=None, seed=None, on_document=None, verbose=True)`

Generate a diverse batch of documents from one high-level scenario.

A planner turns scenario into count distinct, specific scenarios; each runs its own self-contained pipeline graph as a sibling node, and Strands executes them concurrently.

schema accepts a bundled name, a directory path, a Schema, or an InferredSchema.

Parameters:

Name	Type	Description	Default
`scenario`	`str`	The high-level theme the planner diversifies into `count` specific documents. A specific scenario yields far more varied output than a generic one.	required
`entity`	`str \| None`	For a multi-entity `InferredSchema`, which entity to render.	`None`
`seed`	`int \| None`	optional seed for scenario planning (regression-stable sets).	`None`
`on_document`	`Callable[[int, int, GeneratedDoc], None] \| None`	optional `callback(index, total, GeneratedDoc)` fired as each document's result is collected — for host-side progress UIs.	`None`

Source code in seed_data/api.py

def generate_batch(
    self,
    schema: "str | Schema | InferredSchema",
    *,
    count: int,
    scenario: str,
    augment: bool | None = None,
    entity: str | None = None,
    seed: int | None = None,
    on_document: Callable[[int, int, GeneratedDoc], None] | None = None,
    verbose: bool = True,
) -> BatchResult:
    """Generate a diverse batch of documents from one high-level scenario.

    A planner turns ``scenario`` into ``count`` distinct, specific scenarios;
    each runs its own self-contained pipeline graph as a sibling node, and
    Strands executes them concurrently.

    ``schema`` accepts a bundled name, a directory path, a ``Schema``, or an
    ``InferredSchema``.

    Args:
        scenario: The high-level theme the planner diversifies into ``count``
            specific documents. A specific scenario yields far more varied
            output than a generic one.
        entity: For a multi-entity ``InferredSchema``, which entity to render.
        seed: optional seed for scenario planning (regression-stable sets).
        on_document: optional ``callback(index, total, GeneratedDoc)`` fired as
            each document's result is collected — for host-side progress UIs.
    """
    from seed_data.stages.batch import generate_batch as _batch

    common = dict(
        count=count, brief=scenario, output_dir=self.output_dir,
        models=self.models, threshold=self.threshold,
        max_attempts=self.max_attempts, timeout=self.timeout,
        renderer=self.renderer, critic_samples=self.critic_samples,
        augment=self.augment if augment is None else augment,
        verbose=verbose, session=self.session, seed=seed,
        on_document=on_document,
    )
    resolved = self._resolve_for_documents(schema, entity=entity)
    if resolved is not None:
        docs = _batch(resolved=resolved, **common)
    else:
        docs = _batch(schema_dir=self._resolve_schema(schema), **common)

    succeeded = sum(1 for d in docs if d.success)
    return BatchResult(
        count_requested=count,
        count_succeeded=succeeded,
        count_failed=len(docs) - succeeded,
        documents=docs,
    )

`generate_packet(packet, *, count=1, scenario='', shuffle=False, doc_workers=1, augment=None)`

Generate a coordinated multi-document packet.

Parameters:

Name	Type	Description	Default
`packet`	`str`	Path to a packet directory (containing a packet config), or a bundled packet name.	required
`count`	`int`	Number of packets to generate.	`1`
`scenario`	`str`	Free-text scenario shared across the packet's documents (the same applicant, situation, etc.).	`''`
`shuffle`	`bool`	Randomize sub-document order in the merged PDF.	`False`
`doc_workers`	`int`	Parallel workers for sub-documents within a packet.	`1`

Returns a PacketResult (count==1) or a list of them (count>1).

Source code in seed_data/api.py

def generate_packet(
    self,
    packet: str,
    *,
    count: int = 1,
    scenario: str = "",
    shuffle: bool = False,
    doc_workers: int = 1,
    augment: bool | None = None,
):
    """Generate a coordinated multi-document packet.

    Args:
        packet: Path to a packet directory (containing a packet config), or a
            bundled packet name.
        count: Number of packets to generate.
        scenario: Free-text scenario shared across the packet's documents
            (the same applicant, situation, etc.).
        shuffle: Randomize sub-document order in the merged PDF.
        doc_workers: Parallel workers for sub-documents within a packet.

    Returns a ``PacketResult`` (count==1) or a list of them (count>1).
    """
    from seed_data.packet import load_packet_config, generate_packet as _gen_packet

    config = load_packet_config(self._resolve_packet(packet))
    common = dict(
        output_dir=self.output_dir,
        extra=scenario,
        shuffle=shuffle,
        doc_workers=doc_workers,
        data_model=self.models.data,
        doc_model=self.models.doc,
        critic_model=self.models.critic,
        aug_model=self.models.aug,
        context_model=self.models.batch,
        threshold=self.threshold,
        max_attempts=self.max_attempts,
        timeout=self.timeout,
        augment=self.augment if augment is None else augment,
        critic_samples=self.critic_samples,
        session=self.session,
        renderer=self.renderer,
    )
    if count == 1:
        return _gen_packet(config=config, **common)
    return [_gen_packet(config=config, **common) for _ in range(count)]

`infer_schema(inputs, *, name, model=None, max_docs=5, output_dir=None, on_question=None, verbose=True)`

Infer a document-type Schema from real sample documents.

The inverse of generate: reads one or more real examples (PDF, PNG, or JPEG; local paths/globs/dirs and/or s3:// URIs) with a vision model and returns a Schema (JSON Schema + generation guidance) ready to feed back into generate / generate_batch.

Parameters:

Name	Type	Description	Default
`inputs`	`'str \| list[str]'`	sample document location(s) — mixed local and S3 accepted.	required
`name`	`str`	the document-type name (becomes the schema title / doctype).	required
`model`	`str \| None`	vision-capable model key; defaults to `sonnet`.	`None`
`max_docs`	`int`	cap on how many examples to feed the model.	`5`
`output_dir`	`str \| None`	if given, also write the inferred schema there as `schema.json` + `generation_guidance.md` for review/reuse.	`None`
`on_question`	`'Callable[[str], str] \| None'`	optional `callback(question) -> answer` enabling a clarifying dialogue — the model may ask about ambiguous details (required-vs-optional, value ranges) and the callback collects the answer (stdin, notebook, web). `None` runs non-interactively.	`None`
`verbose`	`bool`	print progress.	`True`

Returns:

Type	Description
`'Schema'`	The inferred `Schema` (always returned, even when also written).

Source code in seed_data/api.py

def infer_schema(
    self,
    inputs: "str | list[str]",
    *,
    name: str,
    model: str | None = None,
    max_docs: int = 5,
    output_dir: str | None = None,
    on_question: "Callable[[str], str] | None" = None,
    verbose: bool = True,
) -> "Schema":
    """Infer a document-type ``Schema`` from real sample documents.

    The inverse of ``generate``: reads one or more real examples (PDF, PNG, or
    JPEG; local paths/globs/dirs and/or ``s3://`` URIs) with a vision model and
    returns a ``Schema`` (JSON Schema + generation guidance) ready to feed back
    into ``generate`` / ``generate_batch``.

    Args:
        inputs: sample document location(s) — mixed local and S3 accepted.
        name: the document-type name (becomes the schema title / doctype).
        model: vision-capable model key; defaults to ``sonnet``.
        max_docs: cap on how many examples to feed the model.
        output_dir: if given, also write the inferred schema there as
            ``schema.json`` + ``generation_guidance.md`` for review/reuse.
        on_question: optional ``callback(question) -> answer`` enabling a
            clarifying dialogue — the model may ask about ambiguous details
            (required-vs-optional, value ranges) and the callback collects the
            answer (stdin, notebook, web). ``None`` runs non-interactively.
        verbose: print progress.

    Returns:
        The inferred ``Schema`` (always returned, even when also written).
    """
    from seed_data.infer import infer_schema as _infer, write_schema_dir, DEFAULT_INFER_MODEL
    schema = _infer(
        inputs, name=name, model=model or DEFAULT_INFER_MODEL,
        max_docs=max_docs, session=self.session,
        on_question=on_question, verbose=verbose,
    )
    if output_dir:
        write_schema_dir(schema, output_dir)
        if verbose:
            print(f"Wrote inferred schema to {output_dir}/ (schema.json + generation_guidance.md)")
    return schema

`infer_packet(inputs, *, name, output_dir, model=None, boundaries=None, on_question=None, verbose=True)`

Infer a packet definition from ONE concatenated multi-document PDF.

The inverse of generate_packet: takes a single file containing several different document types concatenated (e.g. a lending package), detects the document boundaries + classes with a vision model (or a boundaries page-range override), infers a schema per segment, and writes a packet.json + one schema dir per segment to output_dir — the shape generate_packet / seed-data packet consumes.

Parameters:

Name	Type	Description	Default
`inputs`	`'str \| list[str]'`	a single concatenated PDF (path or `s3://` URI).	required
`name`	`str`	the packet name.	required
`output_dir`	`str`	where to write packet.json + per-segment schema dirs.	required
`model`	`str \| None`	vision-capable model key; defaults to `sonnet`.	`None`
`boundaries`	`str \| None`	optional `"1-2,3,4-5"` page-range override for splitting.	`None`
`verbose`	`bool`	print progress.	`True`

Returns:

Type	Description
`str`	The output directory path (containing packet.json + schema dirs).

Source code in seed_data/api.py

def infer_packet(
    self,
    inputs: "str | list[str]",
    *,
    name: str,
    output_dir: str,
    model: str | None = None,
    boundaries: str | None = None,
    on_question: "Callable[[str], str] | None" = None,
    verbose: bool = True,
) -> str:
    """Infer a packet definition from ONE concatenated multi-document PDF.

    The inverse of ``generate_packet``: takes a single file containing several
    *different* document types concatenated (e.g. a lending package), detects
    the document boundaries + classes with a vision model (or a ``boundaries``
    page-range override), infers a schema per segment, and writes a
    ``packet.json`` + one schema dir per segment to ``output_dir`` — the shape
    ``generate_packet`` / ``seed-data packet`` consumes.

    Args:
        inputs: a single concatenated PDF (path or ``s3://`` URI).
        name: the packet name.
        output_dir: where to write packet.json + per-segment schema dirs.
        model: vision-capable model key; defaults to ``sonnet``.
        boundaries: optional ``"1-2,3,4-5"`` page-range override for splitting.
        verbose: print progress.

    Returns:
        The output directory path (containing packet.json + schema dirs).
    """
    from seed_data.packet_infer import infer_packet as _infer_packet
    from seed_data.infer import DEFAULT_INFER_MODEL
    return _infer_packet(
        inputs, name=name, output_dir=output_dir,
        model=model or DEFAULT_INFER_MODEL, boundaries=boundaries,
        session=self.session, on_question=on_question, verbose=verbose,
    )

`generate_from_samples(inputs, *, name, scenario='', infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, verbose=True)`

Infer a schema from sample documents, then generate ONE synthetic doc.

A convenience one-shot: infer_schema(...) followed by generate(...) against the inferred Schema. If output_dir is given the inferred schema is also persisted there for review/reuse (recommended), so you keep an editable schema rather than a throwaway. on_question enables the clarifying dialogue during inference (see infer_schema).

Source code in seed_data/api.py

def generate_from_samples(
    self,
    inputs: "str | list[str]",
    *,
    name: str,
    scenario: str = "",
    infer_model: str | None = None,
    max_docs: int = 5,
    output_dir: str | None = None,
    on_question: "Callable[[str], str] | None" = None,
    augment: bool | None = None,
    verbose: bool = True,
) -> GeneratedDoc:
    """Infer a schema from sample documents, then generate ONE synthetic doc.

    A convenience one-shot: ``infer_schema(...)`` followed by ``generate(...)``
    against the inferred ``Schema``. If ``output_dir`` is given the inferred
    schema is also persisted there for review/reuse (recommended), so you keep
    an editable schema rather than a throwaway. ``on_question`` enables the
    clarifying dialogue during inference (see ``infer_schema``).
    """
    schema = self.infer_schema(
        inputs, name=name, model=infer_model, max_docs=max_docs,
        output_dir=output_dir, on_question=on_question, verbose=verbose,
    )
    return self.generate(schema, scenario=scenario, augment=augment, verbose=verbose)

`generate_batch_from_samples(inputs, *, name, count, scenario, infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, seed=None, on_document=None, verbose=True)`

Infer a schema from sample documents, then generate a diverse BATCH.

A convenience one-shot: infer_schema(...) followed by generate_batch(...) against the inferred Schema. on_question enables the clarifying dialogue during inference (see infer_schema).

Source code in seed_data/api.py

def generate_batch_from_samples(
    self,
    inputs: "str | list[str]",
    *,
    name: str,
    count: int,
    scenario: str,
    infer_model: str | None = None,
    max_docs: int = 5,
    output_dir: str | None = None,
    on_question: "Callable[[str], str] | None" = None,
    augment: bool | None = None,
    seed: int | None = None,
    on_document: Callable[[int, int, GeneratedDoc], None] | None = None,
    verbose: bool = True,
) -> BatchResult:
    """Infer a schema from sample documents, then generate a diverse BATCH.

    A convenience one-shot: ``infer_schema(...)`` followed by
    ``generate_batch(...)`` against the inferred ``Schema``. ``on_question``
    enables the clarifying dialogue during inference (see ``infer_schema``).
    """
    schema = self.infer_schema(
        inputs, name=name, model=infer_model, max_docs=max_docs,
        output_dir=output_dir, on_question=on_question, verbose=verbose,
    )
    return self.generate_batch(
        schema, count=count, scenario=scenario, augment=augment,
        seed=seed, on_document=on_document, verbose=verbose,
    )

`plan(*inputs, name='dataset', verbose=True)`

Plan a dataset: turn any inputs (text, CSV, PDF, JSON Schema, SQL DDL, ERD) into a unified :class:InferredSchema. Auto-detects each input's type.

Named for what you get back — a plan for the data, to read and edit before anything is generated — rather than for the reading of the inputs.

Documents/images/s3:// inputs are routed through the existing vision path (infer_schema); non-document inputs go through the schema extraction agent. Returns a typed InferredSchema — not a dict — consistent with the other verbs.

Parameters:

Name	Type	Description	Default
`*inputs`	`str`	paths, globs, `s3://` URIs, or bare free-text descriptions.	`()`
`name`	`str`	logical dataset name (used when delegating documents).	`'dataset'`
`verbose`	`bool`	print progress.	`True`

Source code in seed_data/api.py

def plan(self, *inputs: str, name: str = "dataset", verbose: bool = True) -> InferredSchema:
    """Plan a dataset: turn any inputs (text, CSV, PDF, JSON Schema, SQL DDL,
    ERD) into a unified :class:`InferredSchema`. Auto-detects each input's type.

    Named for what you get back — a *plan* for the data, to read and edit
    before anything is generated — rather than for the reading of the inputs.

    Documents/images/``s3://`` inputs are routed through the existing vision
    path (``infer_schema``); non-document inputs go through the schema
    extraction agent. Returns a typed ``InferredSchema`` — not a dict —
    consistent with the other verbs.

    Args:
        *inputs: paths, globs, ``s3://`` URIs, or bare free-text descriptions.
        name: logical dataset name (used when delegating documents).
        verbose: print progress.
    """
    from seed_data.ingest import run_ingest
    return run_ingest(
        *inputs, name=name, models=self.models,
        session=self.session, verbose=verbose,
    )

`ingest(*inputs, name='dataset', verbose=True)`

Deprecated alias for :meth:plan.

Never shipped in a release — both names have only ever existed on this branch — so this is a courtesy for in-flight callers, not a compatibility guarantee. Slated for removal.

Source code in seed_data/api.py

def ingest(self, *inputs: str, name: str = "dataset", verbose: bool = True) -> InferredSchema:
    """Deprecated alias for :meth:`plan`.

    Never shipped in a release — both names have only ever existed on this
    branch — so this is a courtesy for in-flight callers, not a compatibility
    guarantee. Slated for removal.
    """
    warnings.warn(
        "Generator.ingest() is deprecated; use Generator.plan() instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.plan(*inputs, name=name, verbose=verbose)

`generate_structured(schema, *, rows=100, format='csv', seed=None, verbose=True)`

Generate structured data (CSV/Parquet/Excel/JSON) from a schema.

schema accepts a bundled schema name, a path to an InferredSchema JSON file, or an InferredSchema object. Returns a typed StructuredResult — consistent with GeneratedDoc / BatchResult / PacketResult.

Parameters:

Name	Type	Description	Default
`schema`	`'str \| InferredSchema'`	bundled name, JSON path, or `InferredSchema` object.	required
`rows`	`int`	target records per entity.	`100`
`format`	`str`	`csv` / `parquet` / `excel` / `json`.	`'csv'`
`seed`	`int \| None`	RNG seed. Pins the programmatically generated columns (numeric, enum, date, ID, pattern) so a re-run with the same schema and seed reproduces them. Free-text fields come from an LLM and are not seedable, so reproducibility covers the structured columns only.	`None`
`verbose`	`bool`	print progress.	`True`

Source code in seed_data/api.py

def generate_structured(
    self,
    schema: "str | InferredSchema",
    *,
    rows: int = 100,
    format: str = "csv",
    seed: int | None = None,
    verbose: bool = True,
) -> StructuredResult:
    """Generate structured data (CSV/Parquet/Excel/JSON) from a schema.

    ``schema`` accepts a bundled schema name, a path to an ``InferredSchema``
    JSON file, or an ``InferredSchema`` object. Returns a typed
    ``StructuredResult`` — consistent with ``GeneratedDoc`` / ``BatchResult`` /
    ``PacketResult``.

    Args:
        schema: bundled name, JSON path, or ``InferredSchema`` object.
        rows: target records per entity.
        format: ``csv`` / ``parquet`` / ``excel`` / ``json``.
        seed: RNG seed. Pins the programmatically generated columns (numeric,
            enum, date, ID, pattern) so a re-run with the same schema and seed
            reproduces them. Free-text fields come from an LLM and are not
            seedable, so reproducibility covers the structured columns only.
        verbose: print progress.
    """
    from seed_data.common.deps import require_structured

    # Checked before anything else: without the extra, the pipeline's own
    # pandas import fails deep inside run_graph_pipeline, where the broad
    # `except` in run_structured turns it into
    # StructuredResult(error="No module named 'pandas'") — a missing install
    # reported as a generation failure. Fail here, naming the extra.
    require_structured("Generator.generate_structured")

    from seed_data.structured import run_structured
    resolved = self._resolve_inferred(schema)
    return run_structured(
        resolved, target_count=rows, export_format=format,
        output_dir=self.output_dir, models=self.models,
        threshold=self.threshold, session=self.session,
        seed=seed, verbose=verbose,
    )

`plan_and_generate(*inputs, output='structured', name='dataset', rows=100, format='csv', count=1, scenario='', entity=None, augment=None, seed=None, verbose=True)`

End-to-end: plan a schema from the inputs, then generate — in one call.

Chains :meth:plan into :meth:generate_structured (output="structured") or :meth:generate / :meth:generate_batch (output="documents"). Configuration stays on the Generator; per-call args describe only what to make.

This skips the review step :meth:plan exists to enable — the inferred schema goes straight to generation unseen. Prefer plan then a generate verb when the schema's accuracy matters.

Parameters:

Name	Type	Description	Default
`*inputs`	`str`	anything `plan` accepts — free text, paths, globs, `s3://`.	`()`
`output`	`str`	`"structured"` (CSV/Parquet/Excel) or `"documents"` (PDFs).	`'structured'`
`name`	`str`	logical dataset name passed through to `plan`.	`'dataset'`
`rows`	`int`	structured only — target records per entity.	`100`
`format`	`str`	structured only — `csv` / `parquet` / `excel` / `json`.	`'csv'`
`count`	`int`	documents only — how many to generate (>1 dispatches to batch).	`1`
`scenario`	`str`	documents only — free-text steering the content.	`''`
`entity`	`str \| None`	documents only — which entity of a multi-entity schema to render.	`None`
`augment`	`bool \| None`	documents only — override the instance augment setting.	`None`
`seed`	`int \| None`	RNG seed. For `output="structured"` it pins the programmatic columns; for `output="documents"` with `count > 1` it pins batch scenario planning. The planning step is an LLM run and is never seedable, so the schema itself can still differ between runs.	`None`
`verbose`	`bool`	print progress.	`True`

Returns:

Type	Description
`'GeneratedDoc \| BatchResult \| StructuredResult'`	`StructuredResult` for structured output; `GeneratedDoc`
`'GeneratedDoc \| BatchResult \| StructuredResult'`	(`count == 1`) or `BatchResult` (`count > 1`) for documents.

Raises:

Type	Description
`ValueError`	if `output` is not `"structured"` or `"documents"`.

Source code in seed_data/api.py

def plan_and_generate(
    self,
    *inputs: str,
    output: str = "structured",
    name: str = "dataset",
    rows: int = 100,
    format: str = "csv",
    count: int = 1,
    scenario: str = "",
    entity: str | None = None,
    augment: bool | None = None,
    seed: int | None = None,
    verbose: bool = True,
) -> "GeneratedDoc | BatchResult | StructuredResult":
    """End-to-end: plan a schema from the inputs, then generate — in one call.

    Chains :meth:`plan` into :meth:`generate_structured` (``output="structured"``)
    or :meth:`generate` / :meth:`generate_batch` (``output="documents"``).
    Configuration stays on the ``Generator``; per-call args describe only what
    to make.

    This skips the review step :meth:`plan` exists to enable — the inferred
    schema goes straight to generation unseen. Prefer ``plan`` then a generate
    verb when the schema's accuracy matters.

    Args:
        *inputs: anything ``plan`` accepts — free text, paths, globs, ``s3://``.
        output: ``"structured"`` (CSV/Parquet/Excel) or ``"documents"`` (PDFs).
        name: logical dataset name passed through to ``plan``.
        rows: structured only — target records per entity.
        format: structured only — ``csv`` / ``parquet`` / ``excel`` / ``json``.
        count: documents only — how many to generate (>1 dispatches to batch).
        scenario: documents only — free-text steering the content.
        entity: documents only — which entity of a multi-entity schema to render.
        augment: documents only — override the instance augment setting.
        seed: RNG seed. For ``output="structured"`` it pins the programmatic
            columns; for ``output="documents"`` with ``count > 1`` it pins
            batch scenario planning. The planning step is an LLM run and is
            never seedable, so the schema itself can still differ between runs.
        verbose: print progress.

    Returns:
        ``StructuredResult`` for structured output; ``GeneratedDoc``
        (``count == 1``) or ``BatchResult`` (``count > 1``) for documents.

    Raises:
        ValueError: if ``output`` is not ``"structured"`` or ``"documents"``.
    """
    if output not in ("structured", "documents"):
        raise ValueError(
            f"output must be 'structured' or 'documents', got {output!r}"
        )

    if output == "structured":
        # Validated up front rather than at the generate_structured call
        # below: planning is a multi-agent LLM run, and failing after it would
        # bill the user for the expensive half of the chain before reporting
        # a missing install that was knowable from the start.
        from seed_data.common.deps import require_structured

        require_structured("Generator.plan_and_generate(output='structured')")

    schema = self.plan(*inputs, name=name, verbose=verbose)

    if output == "structured":
        return self.generate_structured(
            schema, rows=rows, format=format, seed=seed, verbose=verbose,
        )

    if count == 1:
        # `generate` has no seedable randomness — a single document is one LLM
        # call — so `seed` is deliberately not forwarded here.
        return self.generate(
            schema, scenario=scenario, augment=augment,
            entity=entity, verbose=verbose,
        )
    return self.generate_batch(
        schema, count=count, scenario=scenario, augment=augment,
        entity=entity, seed=seed, verbose=verbose,
    )

`run(*inputs, **kwargs)`

Deprecated alias for :meth:plan_and_generate.

**kwargs rather than the full signature so the two cannot drift: a new keyword on plan_and_generate is forwarded without being restated here.

Source code in seed_data/api.py

def run(self, *inputs: str, **kwargs) -> "GeneratedDoc | BatchResult | StructuredResult":
    """Deprecated alias for :meth:`plan_and_generate`.

    ``**kwargs`` rather than the full signature so the two cannot drift: a new
    keyword on ``plan_and_generate`` is forwarded without being restated here.
    """
    warnings.warn(
        "Generator.run() is deprecated; use Generator.plan_and_generate() instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.plan_and_generate(*inputs, **kwargs)

`available_schemas()` `staticmethod`

Names of schemas bundled with the package.

Source code in seed_data/api.py

@staticmethod
def available_schemas() -> list[str]:
    """Names of schemas bundled with the package."""
    root = _bundled_dir("schemas")
    return _list_subdirs(root)

`available_packets()` `staticmethod`

Names of packets bundled with the package.

Source code in seed_data/api.py

@staticmethod
def available_packets() -> list[str]:
    """Names of packets bundled with the package."""
    root = _bundled_dir("packets")
    return _list_subdirs(root)

`available_input_types()` `staticmethod`

Input types the plan verb can auto-detect.

Source code in seed_data/api.py

@staticmethod
def available_input_types() -> list[str]:
    """Input types the ``plan`` verb can auto-detect."""
    from seed_data.ingest import InputType
    return [t.value for t in InputType]

`gen.generate(...)` — one document

Returns a GeneratedDoc. schema accepts a bundled schema name, a path to a schema directory, a Schema object, or an InferredSchema (e.g. straight from plan).

from seed_data import Generator, ModelConfig

gen = Generator(models=ModelConfig(doc="gpt-oss"), threshold=7)

doc = gen.generate(
    "invoice",                              # bundled name, dir path, Schema, or InferredSchema
    scenario="Midwest food distributor, net-30 terms",  # what to generate this run
    augment=None,                           # None -> use the Generator's default; True/False overrides
    entity=None,                            # multi-entity InferredSchema: which entity to render
    verbose=True,                           # print stage progress
)

if doc.success:
    print(doc.pdf_path)          # rendered PDF on disk
    print(doc.data_json_path)    # ground-truth JSON label on disk
    print(doc.data)              # the label loaded as a dict (lazy; None if missing)
    print(doc.verdict)           # "accepted" | "rejected" | "error" | "unknown"
    print(doc.score)             # critic score 0-10 (or None)
    print(doc.token_usage)       # {"inputTokens": ..., "outputTokens": ..., "totalTokens": ...}
else:
    print("failed:", doc.error)

`gen.generate_batch(...)` — N documents from one brief

A planner turns scenario into count distinct scenarios; each runs its own pipeline concurrently. Returns a BatchResult.

from seed_data import Generator, GeneratedDoc

gen = Generator(threshold=7)

def on_document(index: int, total: int, doc: GeneratedDoc) -> None:
    # Fired as each document's result is collected — for host-side progress UIs.
    print(f"[{index}/{total}] {doc.doctype}: {doc.verdict}")

batch = gen.generate_batch(
    "invoice",                       # bundled name, dir path, Schema, or InferredSchema
    count=10,
    scenario="Distributors across the US Midwest, varied totals and terms",
    augment=None,                    # None -> Generator default; True/False overrides
    entity=None,                     # multi-entity InferredSchema: which entity to render
    seed=42,                         # optional: stable scenario set for regressions
    on_document=on_document,
)

print(batch.count_requested, batch.count_succeeded, batch.count_failed)
print(batch.total_tokens)            # summed across all documents

for doc in batch.succeeded:          # only the successful GeneratedDocs
    print(doc.pdf_path, doc.data_json_path)

for doc in batch.documents:          # every attempt, success or not
    print(doc.doctype, doc.success, doc.verdict)

`gen.generate_packet(...)` — a coordinated packet

A packet is several inter-related document types that share context (e.g. a lending package: credit report + pay stubs + statement of intent, all for one applicant). Returns a PacketResult when count == 1, or a list[PacketResult] when count > 1.

from seed_data import Generator

gen = Generator(threshold=7)

result = gen.generate_packet(
    "lending-package",       # packet dir path or bundled packet name
    count=1,                 # >1 returns a list[PacketResult]
    scenario="First-time homebuyer, 720 credit score",  # shared across the packet
    shuffle=False,           # randomize sub-document order in the merged PDF
    doc_workers=1,           # parallel workers for sub-documents within the packet
    augment=None,            # None -> Generator default; True/False overrides
)

print(result.merged_pdf)     # path to the single merged PDF for the whole packet

for section in result.sections:
    print(section.document_class)     # e.g. "credit_report"
    print(section.page_indices)       # pages this section occupies in the merged PDF
    print(section.inference_result)   # the section's ground-truth JSON label (dict)

`gen.infer_schema(...)` — a Schema from example documents

Reverse-engineer a Schema from one or more real example documents of the same type. inputs accepts local paths/globs/directories and/or s3:// URIs (PDF, PNG, or JPEG). Returns the Schema (and, with output_dir, also writes schema.json + generation_guidance.md there for review/reuse). See Schema from Documents.

from seed_data import Generator

gen = Generator()

schema = gen.infer_schema(
    "./samples/*.pdf",          # path, glob, dir, or s3:// URI(s); PDF/PNG/JPEG
    name="invoice",             # the document-type name (schema title)
    model=None,                 # vision-capable model key; None -> "sonnet"
    max_docs=5,                 # cap on how many examples to feed the model
    output_dir="./schemas/invoice",  # optional: also write the schema for review
    on_question=None,           # optional callback -> clarifying dialogue (see below)
)

doc = gen.generate(schema, scenario="Midwest food distributor")

`gen.infer_packet(...)` — a packet from one concatenated PDF

The inverse of generate_packet: takes a single file that is several different document types concatenated, detects the boundaries + classes, infers a schema per segment, and writes a packet directory (packet.json + one schema dir per segment) ready for generate_packet. Returns the output directory path.

gen = Generator()

out = gen.infer_packet(
    "./real/lending_package.pdf",   # one concatenated PDF (path or s3:// URI)
    name="lending-package",         # the packet name
    output_dir="./packets/lending-package",
    model=None,                     # vision-capable model; None -> "sonnet"
    boundaries="1-3,4,5-8",         # optional: fixed page ranges (else model-detected)
    on_question=None,               # optional clarifying dialogue (see below)
)

result = gen.generate_packet(out, scenario="First-time homebuyer in Portland, OR")

`gen.generate_from_samples(...)` / `gen.generate_batch_from_samples(...)`

One-shot convenience: infer a schema from example documents, then immediately generate from it. generate_from_samples → one GeneratedDoc; generate_batch_from_samples → a BatchResult. Both persist the inferred schema to output_dir (when given) so you keep an editable schema, not a throwaway.

gen = Generator()

# infer -> single
doc = gen.generate_from_samples(
    "./samples/invoice.pdf", name="invoice",
    scenario="IT consulting services",
    output_dir="./schemas/invoice",   # optional; keeps the inferred schema
)

# infer -> batch (count>1). Same diversity/seed/on_document knobs as generate_batch.
batch = gen.generate_batch_from_samples(
    "s3://my-bucket/invoices/", name="invoice",
    count=10, scenario="Regional US variety",
    output_dir="./schemas/invoice",
)

`gen.plan(...)` — any input to a schema

The unified front door. Takes any mix of inputs — free-text descriptions, example data files, formal schema definitions, real documents, ERD diagrams — auto-detects what each one is, and merges them into a single InferredSchema. That schema then drives either modality: pass it to generate_structured for tables or to generate/generate_batch for documents.

from seed_data import Generator

gen = Generator()

schema = gen.plan(
    "Customers place orders; each order has line items",  # free text
    "./samples/customers.csv",   # example data (CSV/XLS/XLSX)
    "./ddl/orders.sql",          # a formal schema (SQL DDL or JSON Schema)
    "./real/invoice.pdf",        # a document, read with a vision model
    name="retail",               # logical dataset name
    verbose=True,                # print progress
)

for entity in schema.entities:
    print(entity.entity_name, len(entity.fields))

inputs is variadic and at least one is required; each may be a path, a glob, an s3:// URI, or a bare free-text description. Returns an InferredSchema.

Each input is classified independently, so kinds can be combined freely in one call. The detectable types:

Input type	Detected from	Read as
`free_text`	anything with no recognized file extension	a prose description of the data
`example_data`	`.csv`, `.xls`, `.xlsx`	sample rows, profiled for columns, types and value ranges
`schema`	`.sql`, `.ddl`, or a `.json` file that exists on disk	SQL DDL or a JSON Schema document
`document`	`.pdf`, `.png`, `.jpg`, `.jpeg`, any `s3://` URI	a real document, read with a vision model
`erd`	`.dbml`, `.puml`, `.plantuml`, `.mmd`, `.mermaid`	ERD text — entities plus their relationships

Detection is by URI scheme and file extension. A .json argument is treated as a schema definition only if that file exists; otherwise it falls through to free text. Document, image and s3:// inputs are routed to the same vision path infer_schema uses; every other kind goes to the schema-extraction agent. Generator.available_input_types() returns the type names at runtime:

Generator.available_input_types()
# -> ["free_text", "example_data", "schema", "document", "erd"]

Profiling example_data needs pandas, from the [structured] extra. Free-text, document, schema, and ERD planning all work in the lean base install.

`gen.generate_structured(...)` — tabular data

Generate structured (tabular) data from an InferredSchema. Returns a StructuredResult. schema accepts a bundled schema name, a path to an InferredSchema JSON file, or an InferredSchema object (e.g. straight from plan). Output goes to the Generator's output_dir.

from seed_data import Generator

gen = Generator(output_dir="./output")

schema = gen.plan("Customers and their orders", name="retail")

result = gen.generate_structured(
    schema,              # bundled name, InferredSchema JSON path, or InferredSchema
    rows=100,            # target records per entity
    format="csv",        # "csv" | "parquet" | "excel" | "json"
    verbose=True,        # print progress
)

if result.success:
    print(result.output_paths)   # ["./output/customer.csv", "./output/order.csv"]
    print(result.row_counts)     # {"Customer": 100, "Order": 100}
    print(result.evaluation)     # {"diversity": ..., "fidelity": ..., ...}
else:
    print("failed:", result.error)

One file per entity is written into the output directory, named from the lowercased entity name with spaces replaced by underscores — customer.csv, purchase_order.csv. The extension follows format: .csv, .json, .xlsx (format="excel"), or .parquet. format="parquet" needs no extra install beyond [structured], which ships pyarrow.

Requires the [structured] extra

Structured generation needs pip install "seed-data[structured]" (pandas, openpyxl, pyarrow). Without it, the verb raises ImportError immediately, naming the extra to install. This is the one failure it raises rather than returning as a StructuredResult — a missing install is a caller mistake knowable before any work starts, not a generation outcome, and reporting it as success=False made it look like the pipeline had run and failed. The document pipeline never needs the extra.

`gen.plan_and_generate(...)` — end to end

Plan a schema from the inputs and generate from it in one call: no intermediate file, no second verb. output selects the modality; the destination directory is the Generator's output_dir.

from seed_data import Generator

gen = Generator(output_dir="./output")

result = gen.plan_and_generate(
    "Customers place orders; each order has line items",  # anything planning accepts
    output="structured",   # "structured" (tables) | "documents" (PDFs)
    name="retail",         # logical dataset name, passed to plan
    rows=100,              # structured only: target records per entity
    format="csv",          # structured only: csv | parquet | excel | json
    count=1,               # documents only: how many (>1 dispatches to batch)
    scenario="",           # documents only: free-text steering the content
    entity=None,           # documents only: which entity of a multi-entity schema
    augment=None,          # documents only: None -> Generator default
    verbose=True,          # print progress
)

inputs is variadic and at least one is required. The return type follows output and count:

`output`	`count`	Returns
`"structured"`	ignored	`StructuredResult`
`"documents"`	`1` (default)	`GeneratedDoc`
`"documents"`	`> 1`	`BatchResult`

Any other output value raises ValueError. Narrow the union by checking the modality you asked for:

result = gen.plan_and_generate("./samples/*.pdf", output="documents", count=5,
                 scenario="Midwest food distributors")

print(result.count_succeeded)          # BatchResult, because count > 1
for doc in result.succeeded:
    print(doc.pdf_path)

plan_and_generate is a convenience chain over plan plus one generation verb, so it exposes no seed or on_document hooks — call plan then generate_batch yourself when you need those.

Clarifying dialogue (`on_question`)

All four inference verbs accept an on_question callback. When provided, the model may ask you to clarify details it cannot determine from the document alone — whether a field that appears once is always present, a realistic value range, an ID format — and your answers shape the inferred schema. Without it (the default), inference runs fully non-interactively.

def ask(question: str) -> str:
    # Collect the answer however your host wants: stdin, a notebook widget,
    # a Slack DM, a web modal. Return "" to decline (the model uses its judgment).
    return input(f"{question}\n> ")

schema = gen.infer_schema("./samples/invoice.pdf", name="invoice", on_question=ask)

The callback is host-pluggable and safe: a None/empty answer or a raising callback degrades to "use your best judgment" rather than aborting inference. The CLI wires this to a terminal prompt via --allow-questions (interactive TTYs only).

Result & config types

`seed_data.stages.pipeline.GeneratedDoc`

Bases: BaseModel

Typed result of generating one document.

Source code in seed_data/stages/pipeline.py

class GeneratedDoc(BaseModel):
    """Typed result of generating one document."""
    success: bool
    doc_id: str
    doctype: str
    pdf_path: str | None = None
    data_json_path: str | None = None
    augmented_path: str | None = None
    verdict: str = "unknown"          # accepted | rejected | error | unknown
    score: int | None = None
    sha256: str | None = None
    size_bytes: int = 0
    execution_order: list[str] = Field(default_factory=list)
    token_usage: dict = Field(default_factory=lambda: {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0})
    error: str | None = None

    @property
    def data(self) -> dict | None:
        """The ground-truth JSON label for this document, loaded lazily.

        This is the paired label eval users want alongside the PDF. Returns None
        if the data file is missing.
        """
        if not self.data_json_path or not os.path.exists(self.data_json_path):
            return None
        with open(self.data_json_path) as f:
            return json.load(f)

`data` `property`

The ground-truth JSON label for this document, loaded lazily.

This is the paired label eval users want alongside the PDF. Returns None if the data file is missing.

`seed_data.api.BatchResult`

Bases: BaseModel

Typed result of a batch run — the per-document results plus a rollup.

Source code in seed_data/api.py

class BatchResult(BaseModel):
    """Typed result of a batch run — the per-document results plus a rollup."""
    count_requested: int
    count_succeeded: int
    count_failed: int
    documents: list[GeneratedDoc] = Field(default_factory=list)

    @property
    def succeeded(self) -> list[GeneratedDoc]:
        return [d for d in self.documents if d.success]

    @property
    def total_tokens(self) -> int:
        return sum(d.token_usage.get("totalTokens", 0) for d in self.documents)

`seed_data.api.StructuredResult`

Bases: BaseModel

Typed result of a structured-data generation run.

Sits alongside GeneratedDoc / BatchResult / PacketResult — the structured verb returns this, never a bare dict.

Source code in seed_data/api.py

class StructuredResult(BaseModel):
    """Typed result of a structured-data generation run.

    Sits alongside ``GeneratedDoc`` / ``BatchResult`` / ``PacketResult`` — the
    structured verb returns this, never a bare dict.
    """
    success: bool
    schema: InferredSchema
    output_paths: list[str] = Field(default_factory=list)   # written files
    format: str                                            # csv | parquet | excel | json
    row_counts: dict[str, int] = Field(default_factory=dict)  # per-entity row count
    evaluation: dict[str, float] = Field(default_factory=dict)  # metric scores
    token_usage: dict = Field(
        default_factory=lambda: {"inputTokens": 0, "outputTokens": 0, "totalTokens": 0}
    )
    error: str | None = None

`seed_data.stages.base.ModelConfig`

Bases: BaseModel

Which model each agent uses. Names resolve via seed_data.MODELS.

Source code in seed_data/stages/base.py

class ModelConfig(BaseModel):
    """Which model each agent uses. Names resolve via ``seed_data.MODELS``."""
    data: str = "sonnet"
    doc: str = "sonnet"
    critic: str = "haiku"
    aug: str = "sonnet"
    batch: str = "sonnet"

Schema — define a document type in code

A Schema is an in-code document type: a JSON Schema (the fields/structure) + generation guidance (prose steering data realism and visual style) + optional reference sample PDFs. It's the same thing a schema directory holds on disk (schema.json + *.md + samples/), but defined in Python — so you can call gen.generate(schema) without any files on disk.

Fields:

Field	Type	Notes
`name`	`str`	Authoritative — becomes the doctype title (overrides a pydantic model's class name).
`json_schema`	`dict \\| None`	A raw JSON Schema dict. Provide this or `model`.
`model`	`BaseModel` subclass `\\| None`	A pydantic model class, rendered to JSON Schema. Provide this or `json_schema`.
`generation_guidance`	`str`	Prose steering the generator and critic.
`sample_pdfs`	`list[str]`	Paths to existing reference PDFs; each must exist on disk.

You must provide exactly one of json_schema or model — supplying both or neither raises a ValidationError.

From a pydantic model

from pydantic import BaseModel
from seed_data import Generator, Schema

class Invoice(BaseModel):
    invoice_number: str
    vendor: str
    total: float

schema = Schema(
    name="invoice",                 # authoritative doctype title
    model=Invoice,                  # rendered to JSON Schema
    generation_guidance="Totals must equal the sum of line items; use net-30 terms.",
    sample_pdfs=[],                 # optional reference PDFs
)

gen = Generator()
doc = gen.generate(schema, scenario="Midwest food distributor")

From a raw JSON Schema dict

from seed_data import Generator, Schema

schema = Schema(
    name="invoice",
    json_schema={
        "type": "object",
        "properties": {
            "invoice_number": {"type": "string"},
            "vendor": {"type": "string"},
            "total": {"type": "number"},
        },
        "required": ["invoice_number", "vendor", "total"],
    },
    generation_guidance="Realistic vendor names; totals between $100 and $50,000.",
)

gen = Generator()
doc = gen.generate(schema)

Loading one from a directory

Schema.from_dir(path) loads a Schema from an on-disk schema directory (schema.json + *.md guidance + samples/):

from seed_data import Schema

schema = Schema.from_dir("./schemas/invoice")

`seed_data.schema.Schema`

Bases: BaseModel

An in-code document-type definition.

Provide exactly one of json_schema (a JSON Schema dict) or model (a pydantic model class, rendered to JSON Schema). generation_guidance is the prose steering the generator/critic; sample_pdfs are optional reference documents the doc critic compares against.

Source code in seed_data/schema/legacy.py

class Schema(BaseModel):
    """An in-code document-type definition.

    Provide exactly one of ``json_schema`` (a JSON Schema dict) or ``model`` (a
    pydantic model class, rendered to JSON Schema). ``generation_guidance`` is the
    prose steering the generator/critic; ``sample_pdfs`` are optional reference
    documents the doc critic compares against.
    """
    model_config = {"arbitrary_types_allowed": True}

    name: str
    json_schema: dict | None = None
    model: Type[BaseModel] | None = None
    generation_guidance: str = ""
    sample_pdfs: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def _one_source(self) -> "Schema":
        if (self.json_schema is None) == (self.model is None):
            raise ValueError(
                "Schema requires exactly one of `json_schema` or `model`."
            )
        for p in self.sample_pdfs:
            if not os.path.isfile(p):
                raise FileNotFoundError(f"sample_pdf not found: {p}")
        return self

    def to_schema_dict(self) -> dict:
        """The JSON Schema dict, with ``title`` set to ``name`` (the doctype).

        ``name`` is authoritative: a pydantic model's own title is its class name,
        which is rarely the doctype the user wants, so we override it.
        """
        schema = self.model.model_json_schema() if self.model is not None else dict(self.json_schema)
        schema["title"] = self.name
        return schema

    def resolve(self) -> tuple[dict, str, list[str]]:
        """Return ``(schema_dict, guidance_text, sample_pdfs)`` — the same shape
        as loading a schema directory, so the pipeline treats both uniformly."""
        return self.to_schema_dict(), self.generation_guidance, list(self.sample_pdfs)

    @classmethod
    def from_dir(cls, schema_dir: str) -> "Schema":
        """Load a Schema from a directory (schema.json + *.md + samples/)."""
        from seed_data.utils import load_schema_dir
        schema_dict, guidance, samples = load_schema_dir(schema_dir)
        name = str(schema_dict.get("title", os.path.basename(os.path.normpath(schema_dir))))
        return cls(name=name, json_schema=schema_dict,
                   generation_guidance=guidance, sample_pdfs=samples)

`from_dir(schema_dir)` `classmethod`

Load a Schema from a directory (schema.json + *.md + samples/).

Source code in seed_data/schema/legacy.py

@classmethod
def from_dir(cls, schema_dir: str) -> "Schema":
    """Load a Schema from a directory (schema.json + *.md + samples/)."""
    from seed_data.utils import load_schema_dir
    schema_dict, guidance, samples = load_schema_dir(schema_dir)
    name = str(schema_dict.get("title", os.path.basename(os.path.normpath(schema_dir))))
    return cls(name=name, json_schema=schema_dict,
               generation_guidance=guidance, sample_pdfs=samples)

`resolve()`

Return (schema_dict, guidance_text, sample_pdfs) — the same shape as loading a schema directory, so the pipeline treats both uniformly.

Source code in seed_data/schema/legacy.py

def resolve(self) -> tuple[dict, str, list[str]]:
    """Return ``(schema_dict, guidance_text, sample_pdfs)`` — the same shape
    as loading a schema directory, so the pipeline treats both uniformly."""
    return self.to_schema_dict(), self.generation_guidance, list(self.sample_pdfs)

`to_schema_dict()`

The JSON Schema dict, with title set to name (the doctype).

name is authoritative: a pydantic model's own title is its class name, which is rarely the doctype the user wants, so we override it.

Source code in seed_data/schema/legacy.py

def to_schema_dict(self) -> dict:
    """The JSON Schema dict, with ``title`` set to ``name`` (the doctype).

    ``name`` is authoritative: a pydantic model's own title is its class name,
    which is rarely the doctype the user wants, so we override it.
    """
    schema = self.model.model_json_schema() if self.model is not None else dict(self.json_schema)
    schema["title"] = self.name
    return schema

The canonical schema — `InferredSchema` / `EntitySchema` / `FieldDefinition`

Schema above is the legacy in-code document type, still fully supported and still what a schema directory loads into. Alongside it lives the canonical model shared by both modalities — it is what plan returns, what generate_structured consumes, and what generate/generate_batch accept via their entity= argument. Three nested classes, all importable from seed_data.schema:

Class	Holds
`InferredSchema`	`entities: list[EntitySchema]` — the whole dataset, one entry per entity/table. That is its only field.
`EntitySchema`	one entity: `entity_name`, `description`, `fields`, `relationships` (free-text notes), `structured_relationships` (typed foreign keys with cardinality), `generation_guidance` (prose realism/rendering rules), `reference_samples` (example records for few-shot generation).
`FieldDefinition`	one field: `name`, `type`, `description`, `nullable`, `required`, `unique`, `min_value`, `max_value`, `min_length`, `max_length`, `pattern`, `enum_values`, `default`, `distribution` (a statistical spec used by the structured sampler), `children` (sub-fields of a nested object/array field).

required and nullable are independent, and conflating them is the easiest mistake to make here:

required — must the key be present in the record?
nullable — may the value be null once the key is there?

All four combinations are meaningful. A required, nullable field is a key that always appears but whose value is null when the source document omits it — which is exactly what document ground-truth labels look like, and why one flag cannot stand in for the other. nullable defaults to False. required is bool | None and defaults to None, meaning "infer from nullable" (not nullable), so schemas written before required existed keep their original behaviour.

from seed_data.schema import EntitySchema, FieldDefinition, InferredSchema

schema = InferredSchema(entities=[
    EntitySchema(
        entity_name="Customer",
        description="Retail customers",
        fields=[
            FieldDefinition(name="customer_id", type="string", unique=True, required=True),
            # present on every record, but null when unknown
            FieldDefinition(name="phone", type="phone", required=True, nullable=True),
            FieldDefinition(name="tier", type="enum", enum_values=["basic", "gold"]),
        ],
        generation_guidance="US customers; realistic name/city pairings.",
    ),
])

Read a schema back off disk with seed_data.schema.io (from_schema_dir, from_json_schema) or hand generate_structured a path and let it resolve — an InferredSchema JSON dump is recognized by its top-level entities list, and anything else is read as a plain JSON Schema document.

`seed_data.schema.InferredSchema`

Bases: BaseModel

Source code in seed_data/schema/models.py

class InferredSchema(BaseModel):
    entities: list[EntitySchema] = Field(description="List of entity schemas with full definitions")

`seed_data.schema.EntitySchema`

Bases: BaseModel

Source code in seed_data/schema/models.py

class EntitySchema(BaseModel):
    entity_name: str = Field(description="Name of the entity/table")
    description: str = Field(default="", description="Description of what this entity represents")
    fields: list[FieldDefinition] = Field(description="List of field definitions")
    relationships: list[str] = Field(default_factory=list, description="Relationships to other entities (e.g., 'belongs_to: Customer')")
    structured_relationships: list[RelationshipDefinition] = Field(default_factory=list, description="Structured FK relationship definitions")
    # Extensions for document generation:
    generation_guidance: str = Field(default="", description="Free-text rendering/realism rules for this entity")
    reference_samples: list[dict] = Field(default_factory=list, description="Example records for few-shot generation")

    @field_validator("description", mode="before")
    @classmethod
    def coerce_description(cls, v):
        if v is None:
            return ""
        return v

`seed_data.schema.FieldDefinition`

Bases: _FieldDefinitionBase

Source code in seed_data/schema/models.py

class FieldDefinition(_FieldDefinitionBase):
    # Extension for document schemas: nested object/array fields carry their own
    # sub-fields here (e.g. an address object, or line-item array-of-objects).
    children: list["FieldDefinition"] | None = Field(
        default=None,
        description="Sub-fields for nested object/array types (JSON Schema properties/items)",
    )

Generator — Python API

Generator(...) — configure once

seed_data.api.Generator

generate(schema, *, scenario='', augment=None, entity=None, verbose=True)

generate_batch(schema, *, count, scenario, augment=None, entity=None, seed=None, on_document=None, verbose=True)

generate_packet(packet, *, count=1, scenario='', shuffle=False, doc_workers=1, augment=None)

infer_schema(inputs, *, name, model=None, max_docs=5, output_dir=None, on_question=None, verbose=True)

infer_packet(inputs, *, name, output_dir, model=None, boundaries=None, on_question=None, verbose=True)

generate_from_samples(inputs, *, name, scenario='', infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, verbose=True)

generate_batch_from_samples(inputs, *, name, count, scenario, infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, seed=None, on_document=None, verbose=True)

plan(*inputs, name='dataset', verbose=True)

ingest(*inputs, name='dataset', verbose=True)

generate_structured(schema, *, rows=100, format='csv', seed=None, verbose=True)

plan_and_generate(*inputs, output='structured', name='dataset', rows=100, format='csv', count=1, scenario='', entity=None, augment=None, seed=None, verbose=True)

run(*inputs, **kwargs)

available_schemas() staticmethod

available_packets() staticmethod

available_input_types() staticmethod

gen.generate(...) — one document

gen.generate_batch(...) — N documents from one brief

gen.generate_packet(...) — a coordinated packet

gen.infer_schema(...) — a Schema from example documents

gen.infer_packet(...) — a packet from one concatenated PDF

gen.generate_from_samples(...) / gen.generate_batch_from_samples(...)

gen.plan(...) — any input to a schema

gen.generate_structured(...) — tabular data

gen.plan_and_generate(...) — end to end

Clarifying dialogue (on_question)

Result & config types

seed_data.stages.pipeline.GeneratedDoc

data property

seed_data.api.BatchResult

seed_data.api.StructuredResult

seed_data.stages.base.ModelConfig

Schema — define a document type in code

From a pydantic model

From a raw JSON Schema dict

Loading one from a directory

seed_data.schema.Schema

from_dir(schema_dir) classmethod

resolve()

to_schema_dict()

The canonical schema — InferredSchema / EntitySchema / FieldDefinition

seed_data.schema.InferredSchema

seed_data.schema.EntitySchema

seed_data.schema.FieldDefinition

`Generator(...)` — configure once

`seed_data.api.Generator`

`generate(schema, *, scenario='', augment=None, entity=None, verbose=True)`

`generate_batch(schema, *, count, scenario, augment=None, entity=None, seed=None, on_document=None, verbose=True)`

`generate_packet(packet, *, count=1, scenario='', shuffle=False, doc_workers=1, augment=None)`

`infer_schema(inputs, *, name, model=None, max_docs=5, output_dir=None, on_question=None, verbose=True)`

`infer_packet(inputs, *, name, output_dir, model=None, boundaries=None, on_question=None, verbose=True)`

`generate_from_samples(inputs, *, name, scenario='', infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, verbose=True)`

`generate_batch_from_samples(inputs, *, name, count, scenario, infer_model=None, max_docs=5, output_dir=None, on_question=None, augment=None, seed=None, on_document=None, verbose=True)`

`plan(*inputs, name='dataset', verbose=True)`

`ingest(*inputs, name='dataset', verbose=True)`

`generate_structured(schema, *, rows=100, format='csv', seed=None, verbose=True)`

`plan_and_generate(*inputs, output='structured', name='dataset', rows=100, format='csv', count=1, scenario='', entity=None, augment=None, seed=None, verbose=True)`

`run(*inputs, **kwargs)`

`available_schemas()` `staticmethod`

`available_packets()` `staticmethod`

`available_input_types()` `staticmethod`

`gen.generate(...)` — one document

`gen.generate_batch(...)` — N documents from one brief

`gen.generate_packet(...)` — a coordinated packet

`gen.infer_schema(...)` — a Schema from example documents

`gen.infer_packet(...)` — a packet from one concatenated PDF

`gen.generate_from_samples(...)` / `gen.generate_batch_from_samples(...)`

`gen.plan(...)` — any input to a schema

`gen.generate_structured(...)` — tabular data

`gen.plan_and_generate(...)` — end to end

Clarifying dialogue (`on_question`)

`seed_data.stages.pipeline.GeneratedDoc`

`data` `property`

`seed_data.api.BatchResult`

`seed_data.api.StructuredResult`

`seed_data.stages.base.ModelConfig`

`seed_data.schema.Schema`

`from_dir(schema_dir)` `classmethod`

`resolve()`

`to_schema_dict()`

The canonical schema — `InferredSchema` / `EntitySchema` / `FieldDefinition`

`seed_data.schema.InferredSchema`

`seed_data.schema.EntitySchema`

`seed_data.schema.FieldDefinition`