syndatalab 0.1.0

Production-grade synthetic data generation using LLMs with structured outputs, caching, checkpointing, and monitoring.

syndatalab

Production-grade synthetic data generation using LLMs with structured outputs, caching, checkpointing, and monitoring.

Installation

pip install syndatalab

# Install with specific provider support
pip install syndatalab[openai]
pip install syndatalab[anthropic]
pip install syndatalab[all]          # all providers
pip install syndatalab[all,quality]  # all providers + quality filters

Quick Start

import syndatalab as sd
from pydantic import BaseModel

class QAPair(BaseModel):
    question: str
    answer: str

class QAGenerator(sd.LLM):
    response_format = QAPair

    def prompt(self, input: dict) -> str:
        return f"Generate a Q&A pair about {input['topic']}"

    def parse(self, input: dict, response: QAPair) -> dict:
        return {"topic": input["topic"], "q": response.question, "a": response.answer}

gen = QAGenerator(model="gpt-4o-mini")
result = gen(dataset=[{"topic": "python"}, {"topic": "rust"}])

print(result.dataset)      # HuggingFace Dataset
print(result.stats)        # Token usage, cost, performance
print(result.run_id)       # Resume ID if interrupted

Features

• Multi-provider: OpenAI, Anthropic, LiteLLM, Ollama with automatic detection
• Provider failover: Automatic fallback chain across providers
• Structured outputs: Pydantic models with layered validation and auto-repair
• Checkpoint/resume: Crash recovery with deterministic run IDs
• Rate limiting: Dual token-bucket (RPM + TPM) with adaptive backoff
• Caching: SQLite-based with WAL mode, TTL, and atomic writes
• Quality filters: Deduplication, diversity scoring, custom filters
• TUI dashboard: Real-time progress, stats, cost tracking
• Audit trail: Per-request JSONL logging for debugging
• CLI tools: Cache management, run history, output inspection

Pipeline Chaining

gen1 = TopicGenerator(model="gpt-4o-mini")
gen2 = QAGenerator(model="gpt-4o-mini")
pipeline = gen1 | gen2
result = pipeline(dataset=[{"seed": "machine learning"}])

Quality Filtering

result = gen(
    dataset=inputs,
    filters=[
        sd.filters.min_length("answer", 50),
        sd.filters.no_duplicates("answer", threshold=0.85),
        sd.filters.custom(lambda row: "I cannot" not in row["answer"]),
    ],
)

CLI

syndatalab cache stats
syndatalab cache clear
syndatalab history
syndatalab inspect <run_id>

Configuration

Create ~/.syndatalab/config.toml:

[defaults]
model = "gpt-4o-mini"
max_retries = 3
max_concurrent_requests = 50
show_dashboard = true

[rate_limits]
max_requests_per_minute = 500
max_tokens_per_minute = 200000

[logging]
level = "INFO"

License

MIT