Skip to content

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 
21
22
23
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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
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)

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.

Source code in seed_data/utils.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def make_model(model_key: str, thinking_budget: int = 0, role: str = "",
               session=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``.
    """
    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,
    }

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