Utils

seed_data.utils

Shared utilities for the orchestrator.

bundled_schemas_dir()

Path to the schemas/ directory shipped inside the installed package.

Source code in seed_data/utils.py

def bundled_schemas_dir() -> str:
    """Path to the schemas/ directory shipped inside the installed package."""
    return os.path.join(os.path.dirname(__file__), "schemas")

clone_schema_library(dest)

Copy the bundled schema library out to a local, editable directory.

Users install seed-data from PyPI, so the built-in schemas live inside site-packages where they cannot be edited comfortably. This copies the whole library to dest so users can tweak schemas and point --schema-dir at them.

Returns the path the schemas were copied to. Raises if dest already exists.

Source code in seed_data/utils.py

def clone_schema_library(dest: str) -> str:
    """Copy the bundled schema library out to a local, editable directory.

    Users install seed-data from PyPI, so the built-in schemas live inside
    site-packages where they cannot be edited comfortably. This copies the whole
    library to ``dest`` so users can tweak schemas and point --schema-dir at them.

    Returns the path the schemas were copied to. Raises if ``dest`` already exists.
    """
    src = bundled_schemas_dir()
    if not os.path.isdir(src):
        raise FileNotFoundError(f"Bundled schema library not found at {src}")
    if os.path.exists(dest):
        raise FileExistsError(
            f"Destination already exists: {dest}\n"
            f"Choose a different path or remove it first."
        )
    # ignore samples/ (local-only reference PDFs, gitignored) and caches
    shutil.copytree(
        src, dest,
        ignore=shutil.ignore_patterns("samples", "__pycache__", ".DS_Store"),
    )
    return dest

load_schema_dir(schema_dir)

Load schema.json, *.md steering docs, and sample PDFs from a directory.

Returns:

Type	Description
tuple[dict, str, list[str]]	(schema_dict, steering_text, sample_pdf_paths)

Source code in seed_data/utils.py

def load_schema_dir(schema_dir: str) -> tuple[dict, str, list[str]]:
    """Load schema.json, *.md steering docs, and sample PDFs from a directory.

    Returns:
        (schema_dict, steering_text, sample_pdf_paths)
    """
    json_files = glob.glob(os.path.join(schema_dir, "*.json"))
    if len(json_files) != 1:
        raise FileNotFoundError(
            f"Expected exactly 1 JSON schema in {schema_dir}, found {len(json_files)}"
        )
    with open(json_files[0]) as f:
        schema = json.load(f)

    md_files = sorted(glob.glob(os.path.join(schema_dir, "*.md")))
    steering_parts = []
    for md in md_files:
        with open(md) as f:
            steering_parts.append(f.read())

    samples_dir = os.path.join(schema_dir, "samples")
    sample_pdfs = sorted(glob.glob(os.path.join(samples_dir, "*.pdf"))) if os.path.isdir(samples_dir) else []

    return schema, "\n\n---\n\n".join(steering_parts), sample_pdfs

make_model(model_key, thinking_budget=0, role='', session=None, temperature=None)

Create a BedrockModel from a MODELS key or raw model ID.

model_key may be a key in MODELS or a raw Bedrock model ID (for region portability, e.g. an EU/GovCloud inference profile). session is an optional boto3 Session for in-process use (containers, Lambda, AgentCore); if omitted, one is resolved from the environment via get_boto_session.

temperature is optional so the structured/ingest agents — which each run at their own sampling temperature — can come through this factory rather than constructing BedrockModel themselves. This is the only place boto_session is wired, so a call site that bypasses it silently ignores the caller's session.

Source code in seed_data/utils.py

def make_model(model_key: str, thinking_budget: int = 0, role: str = "",
               session=None, temperature: float | None = None) -> BedrockModel:
    """Create a BedrockModel from a MODELS key or raw model ID.

    ``model_key`` may be a key in ``MODELS`` or a raw Bedrock model ID (for region
    portability, e.g. an EU/GovCloud inference profile). ``session`` is an optional
    boto3 Session for in-process use (containers, Lambda, AgentCore); if omitted,
    one is resolved from the environment via ``get_boto_session``.

    ``temperature`` is optional so the structured/ingest agents — which each run at
    their own sampling temperature — can come through this factory rather than
    constructing ``BedrockModel`` themselves. This is the only place ``boto_session``
    is wired, so a call site that bypasses it silently ignores the caller's session.
    """
    import warnings

    entry = MODELS.get(model_key)
    if entry:
        model_id = entry["model_id"]
        max_tokens = entry["max_tokens"]
    else:
        model_id = model_key
        max_tokens = 8192

    # Warn if Anthropic models are used for generation (not critique)
    if "anthropic" in model_id and role in ("data", "doc"):
        warnings.warn(
            f"⚠ Anthropic model '{model_key}' selected for {role} generation. "
            "Ensure your use of this model for synthetic data generation complies "
            "with Anthropic's Acceptable Use Policy and your organization's terms of service.",
            stacklevel=2,
        )

    kwargs = {
        "model_id": model_id,
        "boto_session": session or get_boto_session(),
        "boto_client_config": BOTO_CONFIG,
        "max_tokens": max_tokens,
    }

    if temperature is not None:
        kwargs["temperature"] = temperature

    # Some models don't support streaming with tool use
    if entry and entry.get("streaming") is False:
        kwargs["streaming"] = False

    if thinking_budget > 0 and "anthropic" in model_id:
        kwargs["additional_request_fields"] = {
            "thinking": {"type": "enabled", "budget_tokens": thinking_budget}
        }
    return BedrockModel(**kwargs)

safe_path_segment(name, fallback)

Reduce a model-supplied name to a single safe path segment.

Any name that reaches the filesystem having come from a model is untrusted: ../../../../tmp/pwned joined onto an output directory escapes it, and os.makedirs will happily create the result. Keeps only the basename, maps anything outside [alnum]-_ to -, strips leading/trailing dots and dashes, and substitutes fallback when nothing usable is left (an empty segment would write into the parent instead of a subdirectory).

Shared by packet_infer (for schema_dir_name, at inference time) and packet (for document_class, at generation time). Both are needed: sanitizing only at inference leaves a hand-written or previously-generated packet.json unguarded, so the check belongs at every point a name becomes a path.

Source code in seed_data/utils.py

def safe_path_segment(name: str, fallback: str) -> str:
    """Reduce a model-supplied name to a single safe path segment.

    Any name that reaches the filesystem having come from a model is untrusted:
    ``../../../../tmp/pwned`` joined onto an output directory escapes it, and
    ``os.makedirs`` will happily create the result. Keeps only the basename, maps
    anything outside ``[alnum]-_`` to ``-``, strips leading/trailing dots and
    dashes, and substitutes ``fallback`` when nothing usable is left (an empty
    segment would write into the parent instead of a subdirectory).

    Shared by ``packet_infer`` (for ``schema_dir_name``, at inference time) and
    ``packet`` (for ``document_class``, at generation time). Both are needed:
    sanitizing only at inference leaves a hand-written or previously-generated
    ``packet.json`` unguarded, so the check belongs at every point a name becomes
    a path.
    """
    base = os.path.basename(name.strip().replace("\\", "/").rstrip("/"))
    cleaned = "".join(c if (c.isalnum() or c in "-_") else "-" for c in base).strip("-.")
    return cleaned or fallback