Packet

seed_data.packet

Packet generation — coordinated sets of inter-related documents.

A packet is a collection of different document types that share context (e.g., a loan application containing a credit report, pay stubs, and a statement of intent — all for the same applicant).

This module handles

• Loading and validating packet configuration
• Resolving shared context (explicit or LLM-inferred)
• Generating individual sub-documents via the staged pipeline (seed_data.stages.pipeline.generate)
• Merging sub-document PDFs into a single packet PDF
• Emitting evaluation labels (document_class, page_indices, inference_result)

DocumentSpec dataclass

A single document type within a packet.

Source code in seed_data/packet.py

@dataclass
class DocumentSpec:
    """A single document type within a packet."""

    document_class: str
    schema_dir: str
    required: bool = True
    min_instances: int = 1
    max_instances: int = 1

PacketConfig dataclass

Parsed packet configuration.

Source code in seed_data/packet.py

@dataclass
class PacketConfig:
    """Parsed packet configuration."""

    name: str
    description: str
    documents: list[DocumentSpec]
    shared_context: dict[str, str] | None = None
    shared_context_hint: str = ""

PacketResult dataclass

Result of generating a single packet.

Source code in seed_data/packet.py

@dataclass
class PacketResult:
    """Result of generating a single packet."""

    packet_id: str
    merged_pdf: str | None = None
    sections: list[SectionResult] = field(default_factory=list)
    shared_context: dict[str, Any] = field(default_factory=dict)
    success: bool = True
    error: str | None = None

PlannedDocument dataclass

A single document instance to generate.

Source code in seed_data/packet.py

@dataclass
class PlannedDocument:
    """A single document instance to generate."""

    document_class: str
    schema_dir: str
    instance_index: int  # 0-based within this document type

SectionResult dataclass

Result for one sub-document section in the merged packet.

Source code in seed_data/packet.py

@dataclass
class SectionResult:
    """Result for one sub-document section in the merged packet."""

    document_class: str
    page_indices: list[int]
    inference_result: dict[str, Any]
    pdf_path: str | None = None
    data_json_path: str | None = None
    doc_id: str = ""
    success: bool = True

emit_classes_yaml(config, output_dir)

Write classes.yaml listing all document types in the packet.

Returns:

Type	Description
str	Path to the written classes.yaml.

Source code in seed_data/packet.py

def emit_classes_yaml(
    config: PacketConfig,
    output_dir: str,
) -> str:
    """Write classes.yaml listing all document types in the packet.

    Returns:
        Path to the written classes.yaml.
    """
    import yaml

    classes = sorted({doc.document_class for doc in config.documents})
    classes_path = os.path.join(output_dir, "classes.yaml")
    with open(classes_path, "w") as f:
        yaml.dump({"classes": classes}, f, default_flow_style=False)

    return classes_path

generate_packet(config, output_dir, packet_id=None, extra='', shared_context=None, shuffle=False, doc_workers=1, data_model='nova2-lite', doc_model='nova2-lite', critic_model='sonnet', aug_model='gpt-oss', context_model='nova2-lite', threshold=7, max_attempts=5, timeout=3600, augment=False, critic_samples=True, renderer='xhtml2pdf', enable_preview=False)

Generate a single packet: resolve context, generate docs, merge, emit labels.

Parameters:

Name	Type	Description	Default
config	PacketConfig	Parsed packet configuration.	required
output_dir	str	Root output directory for this packet batch.	required
packet_id	str | None	Unique ID for this packet (auto-generated if None).	None
extra	str	Scenario brief / extra instructions.	''
shared_context	dict[str, Any] | None	Pre-resolved shared context (skips resolution if provided).	None
shuffle	bool	Randomize sub-document order in merged PDF.	False
doc_workers	int	Parallel workers for sub-document generation within this packet.	1
data_model	str	Model for data generation.	'nova2-lite'
doc_model	str	Model for document generation.	'nova2-lite'
critic_model	str	Model for critique.	'sonnet'
aug_model	str	Model for augmentation.	'gpt-oss'
context_model	str	Model for shared context resolution.	'nova2-lite'
threshold	int	Quality threshold for critics.	7
max_attempts	int	Max retry attempts per document.	5
timeout	int	Timeout per document generation.	3600
augment	bool	Enable per-document augmentation.	False
critic_samples	bool	Use reference samples in critic.	True
renderer	str	PDF rendering engine.	'xhtml2pdf'
enable_preview	bool	Accepted for CLI compatibility; not used by the pipeline (no preview stage yet).	False

Returns:

Type	Description
PacketResult	PacketResult with merged PDF path, section results, and metadata.

Source code in seed_data/packet.py

def generate_packet(
    config: PacketConfig,
    output_dir: str,
    packet_id: str | None = None,
    extra: str = "",
    shared_context: dict[str, Any] | None = None,
    shuffle: bool = False,
    doc_workers: int = 1,
    data_model: str = "nova2-lite",
    doc_model: str = "nova2-lite",
    critic_model: str = "sonnet",
    aug_model: str = "gpt-oss",
    context_model: str = "nova2-lite",
    threshold: int = 7,
    max_attempts: int = 5,
    timeout: int = 3600,
    augment: bool = False,
    critic_samples: bool = True,
    renderer: str = "xhtml2pdf",
    enable_preview: bool = False,
) -> PacketResult:
    """Generate a single packet: resolve context, generate docs, merge, emit labels.

    Args:
        config: Parsed packet configuration.
        output_dir: Root output directory for this packet batch.
        packet_id: Unique ID for this packet (auto-generated if None).
        extra: Scenario brief / extra instructions.
        shared_context: Pre-resolved shared context (skips resolution if provided).
        shuffle: Randomize sub-document order in merged PDF.
        doc_workers: Parallel workers for sub-document generation within this packet.
        data_model: Model for data generation.
        doc_model: Model for document generation.
        critic_model: Model for critique.
        aug_model: Model for augmentation.
        context_model: Model for shared context resolution.
        threshold: Quality threshold for critics.
        max_attempts: Max retry attempts per document.
        timeout: Timeout per document generation.
        augment: Enable per-document augmentation.
        critic_samples: Use reference samples in critic.
        renderer: PDF rendering engine.
        enable_preview: Accepted for CLI compatibility; not used by the
            pipeline (no preview stage yet).

    Returns:
        PacketResult with merged PDF path, section results, and metadata.
    """
    packet_id = packet_id or uuid.uuid4().hex[:12]
    workspace_dir = os.path.join(output_dir, "artifacts", packet_id)
    os.makedirs(workspace_dir, exist_ok=True)

    # Step 1: Resolve shared context
    if shared_context is None:
        print(f"  Resolving shared context for packet {packet_id}...")
        shared_context = resolve_shared_context(config, extra=extra, model=context_model)

    # Save shared context for reproducibility
    context_path = os.path.join(workspace_dir, "shared_context.json")
    with open(context_path, "w") as f:
        json.dump(shared_context, f, indent=2)

    # Step 2: Build the document plan (resolve instance counts)
    doc_plan = _build_document_plan(config, shared_context)

    # Step 3: Generate each sub-document through the staged pipeline
    from seed_data.stages.base import ModelConfig
    models = ModelConfig(
        data=data_model, doc=doc_model, critic=critic_model,
        aug=aug_model, batch=context_model,
    )
    section_results = _generate_subdocuments(
        doc_plan=doc_plan,
        shared_context=shared_context,
        workspace_dir=workspace_dir,
        extra=extra,
        doc_workers=doc_workers,
        models=models,
        threshold=threshold,
        max_attempts=max_attempts,
        timeout=timeout,
        augment=augment,
        critic_samples=critic_samples,
        renderer=renderer,
    )

    # Step 4: Optionally shuffle order
    if shuffle:
        random.shuffle(section_results)

    # Step 5: Merge PDFs and compute page indices
    successful_sections = [s for s in section_results if s.success and s.pdf_path]
    if not successful_sections:
        return PacketResult(
            packet_id=packet_id,
            shared_context=shared_context,
            sections=section_results,
            success=False,
            error="No sub-documents were generated successfully.",
        )

    merged_pdf_name = f"{packet_id}.pdf"
    merged_pdf_path = os.path.join(output_dir, "input", merged_pdf_name)
    os.makedirs(os.path.dirname(merged_pdf_path), exist_ok=True)

    page_offset = 0
    for section in successful_sections:
        page_count = _count_pdf_pages(section.pdf_path)
        section.page_indices = list(range(page_offset, page_offset + page_count))
        page_offset += page_count

    _merge_pdfs(
        pdf_paths=[s.pdf_path for s in successful_sections],
        output_path=merged_pdf_path,
    )

    # Step 6: Emit baseline labels
    _emit_baseline_labels(
        sections=successful_sections,
        merged_pdf_name=merged_pdf_name,
        output_dir=output_dir,
    )

    # Step 7: Write single packet summary JSON
    _write_packet_summary(
        packet_id=packet_id,
        merged_pdf_name=merged_pdf_name,
        shared_context=shared_context,
        sections=successful_sections,
        workspace_dir=workspace_dir,
    )

    return PacketResult(
        packet_id=packet_id,
        merged_pdf=merged_pdf_path,
        sections=section_results,
        shared_context=shared_context,
        success=True,
    )

load_packet_config(packet_dir)

Load and validate a packet.json configuration.

Parameters:

Name	Type	Description	Default
packet_dir	str	Directory containing packet.json.	required

Returns:

Type	Description
PacketConfig	Parsed PacketConfig.

Source code in seed_data/packet.py

def load_packet_config(packet_dir: str) -> PacketConfig:
    """Load and validate a packet.json configuration.

    Args:
        packet_dir: Directory containing packet.json.

    Returns:
        Parsed PacketConfig.
    """
    config_path = os.path.join(packet_dir, "packet.json")
    with open(config_path) as f:
        raw = json.load(f)

    documents = [
        DocumentSpec(
            document_class=doc["document_class"],
            schema_dir=_resolve_schema_dir(doc["schema_dir"]),
            required=doc.get("required", True),
            min_instances=doc.get("min_instances", 1),
            max_instances=doc.get("max_instances", 1),
        )
        for doc in raw["documents"]
    ]

    return PacketConfig(
        name=raw["name"],
        description=raw.get("description", ""),
        documents=documents,
        shared_context=raw.get("shared_context"),
        shared_context_hint=raw.get("shared_context_hint", ""),
    )

resolve_shared_context(config, extra='', model='nova2-lite')

Resolve shared context fields for a packet.

If the packet config defines explicit shared_context fields, use those as the schema and ask the LLM to generate values. If not defined, the LLM analyzes all document schemas to infer common fields, then generates values.

Parameters:

Name	Type	Description	Default
config	PacketConfig	Parsed packet configuration.	required
extra	str	Extra instructions / scenario brief from the user.	''
model	str	Model key for the context-generation agent.	'nova2-lite'

Returns:

Type	Description
dict[str, Any]	Dictionary of shared context field names → generated values.

Source code in seed_data/packet.py

def resolve_shared_context(
    config: PacketConfig,
    extra: str = "",
    model: str = "nova2-lite",
) -> dict[str, Any]:
    """Resolve shared context fields for a packet.

    If the packet config defines explicit shared_context fields, use those
    as the schema and ask the LLM to generate values. If not defined, the
    LLM analyzes all document schemas to infer common fields, then generates
    values.

    Args:
        config: Parsed packet configuration.
        extra: Extra instructions / scenario brief from the user.
        model: Model key for the context-generation agent.

    Returns:
        Dictionary of shared context field names → generated values.
    """
    from pydantic import BaseModel, Field
    from strands import Agent

    # Gather schema summaries for the LLM
    schema_summaries = _collect_schema_summaries(config)

    if config.shared_context:
        return _generate_values_for_explicit_context(
            config, schema_summaries, extra, model,
        )

    return _infer_and_generate_context(
        config, schema_summaries, extra, model,
    )

write_packet_manifest(results, config, output_dir, command='', elapsed_s=0, **extra_metadata)

Write a packet_manifest.json summarizing the batch of packets.

Returns:

Type	Description
str	Path to the written manifest.

Source code in seed_data/packet.py

def write_packet_manifest(
    results: list[PacketResult],
    config: PacketConfig,
    output_dir: str,
    command: str = "",
    elapsed_s: float = 0,
    **extra_metadata,
) -> str:
    """Write a packet_manifest.json summarizing the batch of packets.

    Returns:
        Path to the written manifest.
    """
    import subprocess as _sp

    try:
        git_hash = _sp.check_output(
            ["git", "rev-parse", "HEAD"], stderr=_sp.DEVNULL, text=True,
        ).strip()
    except Exception:
        git_hash = "unknown"

    succeeded = sum(1 for r in results if r.success)

    manifest = {
        "packet_type": config.name,
        "description": config.description,
        "command": command,
        "git_hash": git_hash,
        "count_requested": len(results),
        "count_succeeded": succeeded,
        "elapsed_s": round(elapsed_s, 1),
        **extra_metadata,
        "packets": [
            {
                "packet_id": r.packet_id,
                "success": r.success,
                "merged_pdf": r.merged_pdf,
                "shared_context": r.shared_context,
                "error": r.error,
                "sections": [
                    {
                        "document_class": s.document_class,
                        "page_indices": s.page_indices,
                        "doc_id": s.doc_id,
                        "success": s.success,
                        "pdf_path": s.pdf_path,
                        "data_json_path": s.data_json_path,
                    }
                    for s in r.sections
                ],
            }
            for r in results
        ],
    }

    config_dir = os.path.join(output_dir, "config")
    os.makedirs(config_dir, exist_ok=True)
    manifest_path = os.path.join(config_dir, "packet_manifest.json")
    with open(manifest_path, "w") as f:
        json.dump(manifest, f, indent=2, default=str)

    return manifest_path