synthetic-data-generation

Synthetic Data Generation with SDG Hub

Generate synthetic data using composable blocks and flows. Blocks are processing units that transform datasets; flows chain blocks into pipelines defined in YAML.

Core concept: dataset -> Block_1 -> Block_2 -> Block_3 -> enriched_dataset

Choose Your Approach

Approach	When to Use
Pre-built flow	Standard pipeline exists for your task (QA generation, text analysis, red-teaming, RAG eval, MCP distillation)
Custom Python	Quick experiments, ad-hoc generation, custom logic
Custom YAML flow	Reusable pipeline, team sharing, complex multi-block workflows
Agent-based	Need external agent frameworks (Langflow, LangGraph) or MCP tool-use in your pipeline

Approach A: Pre-Built Flows

Step 1: Discover flows

# play.py
from sdg_hub import FlowRegistry

# List all flows
for f in FlowRegistry.list_flows():
    print(f"- {f['name']} (tags: {f.get('tags', [])})")

# Search by tag
FlowRegistry.search_flows(tag="qa-generation")

Consult references/pre_built_flows.md for the full catalog with descriptions and required inputs.

Step 2: Load and inspect

from sdg_hub import Flow, FlowRegistry

path = FlowRegistry.get_flow_path("Flow Name or ID")
flow = Flow.from_yaml(path)
flow.print_info()

# Check what dataset columns are needed
reqs = flow.get_dataset_requirements()
if reqs:
    print(f"Required columns: {reqs.required_columns}")

Step 3: Configure model

import os

flow.set_model_config(
    model="openai/gpt-4o-mini",
    api_key=os.environ.get("OPENAI_API_KEY")
)

# For local models (vLLM, Ollama)
flow.set_model_config(
    model="meta-llama/Llama-3.3-70B-Instruct",
    api_base="http://localhost:8000/v1",
    api_key="EMPTY"
)

See references/model_configs.md for all supported providers (OpenAI, Anthropic, Azure, vLLM, Ollama, Together, Groq, Bedrock, etc.).

Step 4: Prepare data and dry run

import pandas as pd

df = pd.DataFrame({"document": ["Your text here..."]})

# Validate dataset against flow requirements
errors = flow.validate_dataset(df)
if errors:
    print(f"Fix these: {errors}")

# Dry run with 2 samples -- do this before every full run
dry = flow.dry_run(df, sample_size=2)
print(f"Success: {dry['execution_successful']}")
for block in dry['blocks_executed']:
    print(f"  {block['block_name']}: {block['execution_time_seconds']:.2f}s")