Generate Synthetic Test Data

Synthetic Data Generation (SDG) expands or creates an evaluation dataset on the platform instead of on your laptop. You point the platform at a governed seed DatasetVersion, it applies a configured generation or augmentation pipeline on a worker, and it registers the result as a new augmented DatasetVersion with lineage back to the seed. That new version flows through the same quality-check, promote, and evaluate path as any hand-uploaded dataset — see Datasets and Golden Datasets.

Two situations call for it. When you already have a dataset but want more coverage or harder cases, augment it — the Image augmentation section perturbs drone imagery with a diffusion edit. When you have a document corpus but no evaluation set at all, generate one — the Q&A-pair generation section synthesises question–answer pairs for a RAG system.

SDG adapters

An SDG adapter is a named, versioned, workspace-scoped registration of a generation pipeline. Registering an adapter lets the platform invoke your pipeline by id; a run then binds one adapter to a source version, a target dataset, and a project. Registering an adapter and running it are done through the SDK (aip.register_sdg_adapter, aip.run_sdg) or REST (/sdg/adapters, /sdg/adapters/{id}/runs); the full parameter list is documented below.

Registration is Python-only. The simplest form wraps a custom pipeline class that implements run() and validate_chain(). Pass the class, not an instance. workspace_id is optional and auto-resolves from your membership, but pinning it explicitly matters on a multi-workspace server — an adapter registered into the wrong workspace makes a later run fail with Project '…' does not belong to the adapter's workspace. In the call below, pipeline receives the class itself rather than an instance, and workspace_id is pinned so the adapter and any project bound to it stay in the same workspace:

import os
import time

import pandas as pd
import aip_sdk as aip


class MySDGPipeline:
    def run(self, prompt, n=10):
        return [f"sample_{i}" for i in range(n)]

    def validate_chain(self):
        return True


adapter = aip.register_sdg_adapter(
    name="adversarial-gen",
    pipeline=MySDGPipeline,
    version="1.0",
    owner="data-team",
    workspace_id=ws.id,
)
print(adapter.id, adapter.name, adapter.version, adapter.workspace_id)

for a in aip.list_sdg_adapters(workspace_id=ws.id):
    print(a.id, a.name, a.version, a.owner or "—")

The two applied pipelines below are declarative JobConfig graphs from aip_sdg_core rather than hand-written classes. A JobConfig chains typed nodes — a seed input, one or more generation/augmentation steps, and an export sink — and is registered by posting its serialised form to /sdg/adapters with an explicit workspace_id. Use the configured client so the call carries your session, and keep the adapter and its target project in one workspace by passing workspace_id:

client = aip._context.get_default_client()

adapter = client.post(
    "/sdg/adapters",
    json={
        "name": f"my-sdg-{int(time.time())}",
        "version": "1.0",
        "workspace_id": ws.id,
        "job_config": sdg_job.model_dump_json(),
    },
)
print("SDG adapter:", adapter["id"])

Every run is the same three-argument shape regardless of pipeline: which adapter, which seed version to read, and which dataset to register the output onto. The arguments map directly: source_dataset_version_id is the governed seed rows, and target_dataset_id is where the augmented version lands. Capture the existing version IDs before calling aip.run_sdg(), which enqueues asynchronous work. aip.get_run(...).poll_status() waits up to 15 minutes and raises on failure, cancellation, or timeout; only a successful run continues to the output lookup:

pre_ids = {v.id for v in target_ds.versions()}
sdg_run = aip.run_sdg(
    adapter_id=adapter["id"],
    source_dataset_version_id=seed_version.id,
    target_dataset_id=target_ds.id,
    project_id=project.id,
)

status = aip.get_run(sdg_run.id).poll_status(interval=4, timeout=900)
print("SDG run status:", status)

The same run can be triggered over HTTP:

curl -sS -X POST "$AIP_API_URL/sdg/adapters/<adapter_id>/runs" \
  -H "Authorization: Bearer $AIP_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"project_id":"<project>","target_dataset_id":"<dataset>","source_dataset_version_id":"<version>"}'

When the run completes it registers a new version on the target dataset. Detect it by diffing the version list before and after, then read its lineage and provenance — lineage_parent_id points back at your seed, and run_id identifies the generation run. Keep the pre_ids captured before enqueueing above and match run_id so concurrent runs cannot contribute the wrong version:

new = [
    v for v in target_ds.versions()
    if v.id not in pre_ids and v.lineage.get("run_id") == sdg_run.id
]
if not new:
    raise RuntimeError(f"SDG run {sdg_run.id} completed without an output version")
augmented = max(new, key=lambda v: v.version)
print(augmented.stage, augmented.lineage.get("lineage_parent_id"))

Image augmentation (object detection)

This picks up from the object-detection guide: a gdi_image_v1 detection project, a YOLOv11 detector wired as a SUT, and a seed dataset of drone frames uploaded and mapped to the schema. The goal is to test the detector under a condition the seed barely covers — heavy weather — by synthesising perturbed frames from the real ones.

Establish the RAW baseline first. Before generating anything, run one hosted evaluation on the raw seed version so you have a number to compare against. Map the version, promote it to golden, publish an eval config, and run it exactly as in the object-detection guide, capturing the aggregate metrics. The promote below acknowledges the local quality-check warnings, after which you publish the config and run a hosted eval over od_v, reading the aggregates back from /runs/{id}/analysis:

od_v = od_ds.map_version(
    od_v.id,
    column_mapping={
        "image_id": "image_id",
        "label": "label",
        "predictions": "predictions",
    },
)
od_v = od_ds.promote(od_v.id, force=True, reason="Baseline: acknowledge local warnings")

selected_metrics = [
    "object_detection.precision",
    "object_detection.recall",
    "object_detection.f1",
    "object_detection.missed_detection_rate",
]

Build the augmentation pipeline. The graph reads the seed detection rows, runs a FLUX.2 Klein diffusion edit over each image, and exports the edited test cases back into gdi_image_v1 rows. FLUX is GPU-bound, so rather than require a CUDA GPU on the SDG worker, the flux_klein node offloads the edit to a generic ZeroGPU Hugging Face Space through the hf_space backend — the worker ships (config, rows) to the Space, which reconstructs the real module and runs the edit on the GPU. The scenario on the seed node stamps an edit prompt onto every row (FluxKleinConfig's prompt_template substitutes {scenario}), so "heavy snow" becomes the transformation applied to each frame. num_inference_steps is kept low here for a fast demo; raise it for higher-fidelity edits.

from aip_sdg_core import JobConfig
from aip_sdg_core.backends import HFSpaceBackendConfig
from aip_sdg_core.schemas.object_detection.augmentation.flux_klein import FluxKleinConfig
from aip_sdg_core.schemas.object_detection.export.test_case_to_gdi_od import TestCaseToGdiOdConfig
from aip_sdg_core.schemas.object_detection.input.dataset_version import (
    AnsweredObjectDetectionDatasetVersionInputConfig,
)
from aip_sdg_core.storage_configs import MinioStorageConfig

FLUX_SPACE_ID = "resaro/aip-sdg-flux-klein"

sdg_job = JobConfig(storage=MinioStorageConfig())
sdg_job.add_node(
    "seed",
    AnsweredObjectDetectionDatasetVersionInputConfig(scenario="heavy snow", limit=10),
)
sdg_job.add_node(
    "flux",
    FluxKleinConfig(
        num_inference_steps=2,
        backend=HFSpaceBackendConfig(space_id=FLUX_SPACE_ID),
    ),
    upstream="seed",
)
sdg_job.add_node(
    "out",
    TestCaseToGdiOdConfig(which="generated", aug_type="perturbation"),
    upstream="flux",
)

Keep limit at or above the platform's 10-row quality-check minimum so the augmented version can be promoted, and remember the Space cold-starts on the first call (the backend retries through startup), so allow a few minutes. The Space resaro/aip-sdg-flux-klein must be deployed and RUNNING, and the SDG worker must hold HF_TOKEN with access to the gated FLUX weights — the config carries no token by design.

Register, run, and fold the rows back. Point target_dataset_id at the same dataset as the seed so the augmented version lands alongside it. Register the JobConfig adapter pinned to the workspace and wait for successful generation. On completion, the worker registers a new augmented version whose lineage_parent_id is the seed — the edited frames carry the original bounding-box labels forward, so they are immediately scoreable. Select the new version belonging to this run and set it as the version you evaluate next:

adapter = aip.register_sdg_adapter(
    name=f"image-augmentation-{time.time_ns()}", job_config=sdg_job, workspace_id=od_project.workspace_id
)
pre_ids = {v.id for v in od_ds.versions()}
sdg_run = aip.run_sdg(
    adapter_id=adapter.id,
    source_dataset_version_id=od_v.id,
    target_dataset_id=od_ds.id,
    project_id=od_project.id,
)
aip.get_run(sdg_run.id).poll_status(interval=4, timeout=900)
new = [
    v for v in od_ds.versions()
    if v.id not in pre_ids and v.lineage.get("run_id") == sdg_run.id
]
if not new:
    raise RuntimeError(f"SDG run {sdg_run.id} completed without an output version")
augmented = max(new, key=lambda v: v.version)
print(f"v{augmented.version} stage={augmented.stage} parent={augmented.lineage.get('lineage_parent_id')}")
od_v = augmented

Re-evaluate. Run the identical map → promote → hosted-eval round on od_v, now the augmented version, and compare its object_detection.* aggregates against the RAW baseline. A drop in recall or a rise in missed_detection_rate under scenario="heavy snow" is exactly the weakness the augmentation was designed to surface. Because both rounds hit the same SUT with the same metrics, the two runs are directly comparable on the run pages — see Evaluation runs in the object-detection guide.

Q&A-pair generation (RAG/LLM)

Reach for generation when you have a document corpus but no evaluation set. This picks up from the RAG guide: a gdi_text_v1 project with task_type=single_turn_rag and AnythingLLM wired as the SUT, but with no Q&A pairs to test it on. The pipeline reads a chunk corpus, has an LLM write questions and reference answers grounded in those chunks, scores each pair with an open-book quality check, and registers the survivors as an augmented dataset:

NIST chunks (DatasetVersion)  ──►  ragas_single_hop  ──►  open-book QC  ──►  augmented DatasetVersion
   (one row per chunk)              (LLM + embeddings)    (scores pairs)     (query / answer / context)

Seed the corpus. Upload your chunks as a gdi_text_v1 / single_turn_rag dataset — one row per chunk — and take its latest version as the seed the generator reads:

chunks = pd.read_csv("nist_chunks.csv")
seed_ds = rag_project.upload_dataset(
    pd.DataFrame({"chunk": chunks["chunk"].astype(str)}),
    name=f"nist-chunks-{int(time.time())}",
)
seed_ver = seed_ds.latest_version()

Build the generation pipeline. The graph is seed → generation → quality control. RAGGenerationConfig with generator_type="ragas_single_hop" builds a knowledge graph over the chunks and synthesises grounded questions (topic_synthesis is a faster, lower-fidelity alternative); OpenBookQualityControlConfig answers each generated question with a model and judges it, filtering weak pairs. Both nodes call an LLM, so supply your key through the config:

from aip_sdg_core import JobConfig
from aip_sdg_core.rag_configs import RAGGenerationConfig
from aip_sdg_core.schemas.rag.input.dataset_version import RAGDatasetVersionInputConfig
from aip_sdg_core.schemas.rag.quality_control.open_book import OpenBookQualityControlConfig
from aip_sdg_core.storage_configs import MinioStorageConfig

key = os.environ["OPENAI_API_KEY"]

sdg_job = JobConfig(storage=MinioStorageConfig())
sdg_job.add_node("seed", RAGDatasetVersionInputConfig(limit=51, text_column=None))
sdg_job.add_node(
    "samples",
    RAGGenerationConfig(
        generator_type="ragas_single_hop",
        generator_params={
            "llm_model": "gpt-4o-mini",
            "embedding_model": "text-embedding-3-small",
            "language": "en",
            "llm_api_key": key,
            "embedding_api_key": key,
        },
        scenario="AI risk management knowledge base",
        n_questions=10,
        expected_filter_rate=0.5,
    ),
    upstream="seed",
)
sdg_job.add_node(
    "scored",
    OpenBookQualityControlConfig(
        answer_model="gpt-4o-mini",
        judge_model="gpt-4o-mini",
        answer_api_key=key,
        judge_api_key=key,
    ),
    upstream="samples",
)

Register, run, and collect the output. Create a separate target dataset for the generated pairs so the Q&A output stays distinct from the chunk corpus. Register the JobConfig adapter pinned to the workspace, run it with run_sdg() against seed_ver, and poll — building the knowledge graph over the whole corpus takes a few minutes. After successful completion, locate the new version from this run and pin the download to it:

adapter = aip.register_sdg_adapter(
    name=f"rag-qa-generation-{time.time_ns()}", job_config=sdg_job, workspace_id=rag_project.workspace_id
)
qa_ds = rag_project.upload_dataset(
    pd.DataFrame({"prompt": ["seed"], "expected_output": ["seed"]}),
    name=f"rag-sdg-qa-{int(time.time())}",
)
pre_ids = {v.id for v in qa_ds.versions()}
sdg_run = aip.run_sdg(
    adapter_id=adapter.id,
    source_dataset_version_id=seed_ver.id,
    target_dataset_id=qa_ds.id,
    project_id=rag_project.id,
)
aip.get_run(sdg_run.id).poll_status(interval=4, timeout=900)
new = [
    v for v in qa_ds.versions()
    if v.id not in pre_ids and v.lineage.get("run_id") == sdg_run.id
]
if not new:
    raise RuntimeError(f"SDG run {sdg_run.id} completed without an output version")
augmented = max(new, key=lambda v: v.version)
df_sdg_qa = aip.DatasetHandle(
    qa_ds.id, augmented.version, augmented.id, aip._context.get_default_client()
).pull()

Make it eval-ready and promote. The generator emits query / answer / context; the gdi_text_v1 evaluation columns are prompt / expected_output / reference_contexts. Rename them, ensure every row has an input_id (every eval metric requires it), then upload the patched frame as a fresh dataset, quality-check it, and promote to golden — the same flow as a hand-uploaded set in the RAG guide and Golden Datasets. reference_contexts is what rag.hit_at_k scores retrieval against, so preserve it as a list of strings — the mapping below coerces each context into a list of strings:

if "input_id" not in df_sdg_qa.columns:
    df_sdg_qa.insert(0, "input_id", [f"qa-{i:04d}" for i in range(len(df_sdg_qa))])
df_sdg_qa["prompt"] = df_sdg_qa.get("prompt", df_sdg_qa.get("query"))
df_sdg_qa["expected_output"] = df_sdg_qa.get("expected_output", df_sdg_qa.get("answer"))
df_sdg_qa["reference_contexts"] = df_sdg_qa["context"].apply(
    lambda x: [str(v) for v in x] if hasattr(x, "__iter__") and not isinstance(x, str) else [str(x)]
)

rag_ds = rag_project.upload_dataset(df_sdg_qa, name=f"rag-sdg-eval-{int(time.time())}")
rag_v = rag_ds.latest_version()
rag_golden = rag_ds.promote(rag_v.id, force=True, reason="SDG-generated QA pairs for evaluation.")

If the run completes but yields zero pairs, generation or QC filtered everything — raise n_questions or expected_filter_rate, or start from a larger corpus. With a golden Q&A set in hand, continue straight into the hosted evaluation in the RAG guide; the generated dataset behaves identically to one you uploaded yourself.