Generator — Python API

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

from seed_data import Generator, ModelConfig, Schema, GeneratedDoc, BatchResult

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`

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

`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", ...]

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",
        *,
        scenario: str = "",
        augment: bool | 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, or
                - a :class:`Schema` object defined in code (JSON schema or a
                  pydantic model, plus generation guidance).
            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.
            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,
        )
        from seed_data.schema import Schema
        if isinstance(schema, Schema):
            return _pipeline_generate(resolved=schema.resolve(), **common)
        return _pipeline_generate(schema_dir=self._resolve_schema(schema), **common)

    def generate_batch(
        self,
        schema: "str | Schema",
        *,
        count: int,
        scenario: str,
        augment: bool | 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, or a ``Schema``.

        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.
            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
        from seed_data.schema import Schema

        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,
        )
        if isinstance(schema, Schema):
            docs = _batch(resolved=schema.resolve(), **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,
            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,
        )

    # -- 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)

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

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

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

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

Generate a single document. Returns a typed GeneratedDoc.

Parameters:

Name	Type	Description	Default
`schema`	`'str \| Schema'`	One of — - a bundled schema name (e.g. `"invoice"`), - a path to a schema directory, or - a :class:`Schema` object defined in code (JSON schema or a pydantic model, plus generation guidance).	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`
`verbose`	`bool`	Print stage progress.	`True`

Source code in seed_data/api.py

def generate(
    self,
    schema: "str | Schema",
    *,
    scenario: str = "",
    augment: bool | 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, or
            - a :class:`Schema` object defined in code (JSON schema or a
              pydantic model, plus generation guidance).
        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.
        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,
    )
    from seed_data.schema import Schema
    if isinstance(schema, Schema):
        return _pipeline_generate(resolved=schema.resolve(), **common)
    return _pipeline_generate(schema_dir=self._resolve_schema(schema), **common)

`generate_batch(schema, *, count, scenario, augment=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, or a Schema.

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
`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",
    *,
    count: int,
    scenario: str,
    augment: bool | 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, or a ``Schema``.

    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.
        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
    from seed_data.schema import Schema

    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,
    )
    if isinstance(schema, Schema):
        docs = _batch(resolved=schema.resolve(), **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,
        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,
    )

`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)

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

Returns a GeneratedDoc. schema accepts a bundled schema name, a path to a schema directory, or a Schema object.

from seed_data import Generator, ModelConfig

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

doc = gen.generate(
    "invoice",                              # bundled name, dir path, or a Schema
    scenario="Midwest food distributor, net-30 terms",  # what to generate this run
    augment=None,                           # None -> use the Generator's default; True/False overrides
    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, or a Schema
    count=10,
    scenario="Distributors across the US Midwest, varied totals and terms",
    augment=None,                    # None -> Generator default; True/False overrides
    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",
)

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.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.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.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.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.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

Generator — Python API

Generator(...) — configure once

seed_data.api.Generator

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

generate_batch(schema, *, count, scenario, augment=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)

available_schemas() staticmethod

available_packets() 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(...)

Clarifying dialogue (on_question)

Result & config types

seed_data.stages.pipeline.GeneratedDoc

data property

seed_data.api.BatchResult

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()

`Generator(...)` — configure once

`seed_data.api.Generator`

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

`generate_batch(schema, *, count, scenario, augment=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)`

`available_schemas()` `staticmethod`

`available_packets()` `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(...)`

Clarifying dialogue (`on_question`)

`seed_data.stages.pipeline.GeneratedDoc`

`data` `property`

`seed_data.api.BatchResult`

`seed_data.stages.base.ModelConfig`

`seed_data.schema.Schema`

`from_dir(schema_dir)` `classmethod`

`resolve()`

`to_schema_dict()`