HinterBuild logoHinterBuild
Back to blog
AI Systems · 9 min read

Synthetic Data Generation for LLM Evals

Synthetic Data Generation for LLM Evals guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 9 min read

  • LLM
  • LLM Serving
  • Evaluation
  • Cost Optimization

Table of Contents:

Why Synthetic Eval Data: The Real-World Problem

Short answer: Hand-labeling evaluation data doesn't scale. Synthetic generation creates thousands of high-quality test cases in hours, not months, enabling continuous evaluation of production AI systems.

A fintech startup needed to evaluate their AI agent across 40 edge cases, 15 languages, and 200 intent variations. Manual annotation would take 3 months and $50K. We built a synthetic generation pipeline that produced 8,000 validated test cases in 4 days for $200 in LLM costs.

Key Takeaways:

  • Synthetic data enables testing at scale—1000x faster than manual annotation
  • Constraint-based synthesis ensures coverage of edge cases
  • LLM-as-generator produces realistic variations with proper prompting
  • Adversarial generation uncovers failure modes early
  • Quality validation prevents garbage data from polluting evals
  • Bias mitigation ensures representative test sets

For evaluation-driven development, synthetic data accelerates iteration cycles from weeks to hours.

The Generation Pipeline: Architecture

Production synthetic data generation follows a multi-stage pipeline with validation gates.

python
from dataclasses import dataclass
from typing import List, Dict, Any
from datetime import datetime, timezone
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

@dataclass
class SyntheticExample:
    """Single synthetic test case."""
    input_text: str
    expected_output: str
    metadata: Dict[str, Any]
    difficulty: str  # easy, medium, hard
    tags: List[str]
    generated_at: str
    validation_score: float = 0.0
    
    def __post_init__(self):
        if not self.generated_at:
            self.generated_at = datetime.now(timezone.utc).isoformat()

class SyntheticDataPipeline:
    """Multi-stage synthetic data generation."""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.validators = []
        self.generators = []
    
    async def generate_dataset(
        self,
        num_examples: int,
        constraints: Dict[str, Any],
    ) -> List[SyntheticExample]:
        """Generate synthetic evaluation dataset."""
        examples = []
        raw_examples = await self._generate_raw(num_examples, constraints)
        print(f"✓ Generated {len(raw_examples)} raw examples")
        
        # Stage 2: Validate quality
        validated = await self._validate_batch(raw_examples)
        print(f"✓ Validated {len(validated)} examples (rejected {len(raw_examples) - len(validated)})")
        
        # Stage 3: Check diversity
        diverse = await self._ensure_diversity(validated, target_diversity=0.85)
        print(f"✓ Ensured diversity: {len(diverse)} unique examples")
        
        # Stage 4: Balance difficulty
        balanced = await self._balance_difficulty(diverse)
        print(f"✓ Balanced difficulty distribution")
        
        return balanced
    
    async def _generate_raw(
        self,
        num_examples: int,
        constraints: Dict[str, Any],
    ) -> List[SyntheticExample]:
        """Generate raw examples in parallel."""
        tasks = [
            self._generate_single(i, constraints)
            for i in range(num_examples)
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Filter out failures
        return [r for r in results if isinstance(r, SyntheticExample)]
    
    async def _generate_single(
        self,
        index: int,
        constraints: Dict[str, Any],
    ) -> SyntheticExample:
        """Generate single synthetic example."""
        # Implementation in next section
        pass
    
    async def _validate_batch(
        self,
        examples: List[SyntheticExample],
    ) -> List[SyntheticExample]:
        """Validate quality of generated examples."""
        validated = []
        
        for example in examples:
            score = await self._quality_score(example)
            example.validation_score = score
            
            if score >= self.config.get("quality_threshold", 0.7):
                validated.append(example)
        
        return validated
    
    async def _quality_score(self, example: SyntheticExample) -> float:
        """Calculate quality score for example."""
        checks = [
            self._check_coherence(example),
            self._check_format(example),
            self._check_realism(example),
        ]
        
        scores = await asyncio.gather(*checks)
        return sum(scores) / len(scores)
    
    async def _check_coherence(self, example: SyntheticExample) -> float:
        """Check if input and output are coherent."""
        prompt = f"""Rate coherence (0.0-1.0):
Input: {example.input_text}
Output: {example.expected_output}

Return JSON: {{"score": float}}"""

        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
        )
        
        import json
        return json.loads(response.choices[0].message.content)["score"]
    
    async def _check_format(self, example: SyntheticExample) -> float:
        """Validate format requirements."""
        if not example.input_text or not example.expected_output:
            return 0.0
        
        if len(example.input_text) < 10 or len(example.expected_output) < 5:
            return 0.3
        
        return 1.0
    
    async def _check_realism(self, example: SyntheticExample) -> float:
        """Check if example resembles real-world data."""
        # Placeholder—implement domain-specific checks
        return 0.8
    
    async def _ensure_diversity(
        self,
        examples: List[SyntheticExample],
        target_diversity: float,
    ) -> List[SyntheticExample]:
        """Remove near-duplicates."""
        from sklearn.feature_extraction.text import TfidfVectorizer
        from sklearn.metrics.pairwise import cosine_similarity
        import numpy as np
        
        # Embed inputs
        vectorizer = TfidfVectorizer()
        texts = [ex.input_text for ex in examples]
        embeddings = vectorizer.fit_transform(texts)
        
        # Compute pairwise similarities
        similarities = cosine_similarity(embeddings)
        
        # Keep diverse examples
        keep = []
        for i, example in enumerate(examples):
            # Check if too similar to kept examples
            if not keep:
                keep.append(example)
                continue
            
            kept_indices = [examples.index(k) for k in keep]
            max_sim = max(similarities[i][j] for j in kept_indices)
            
            if max_sim < (1 - target_diversity):
                keep.append(example)
        
        return keep
    
    async def _balance_difficulty(
        self,
        examples: List[SyntheticExample],
    ) -> List[SyntheticExample]:
        """Balance difficulty distribution."""
        from collections import Counter
        
        difficulty_dist = Counter(ex.difficulty for ex in examples)
        target_per_level = len(examples) // 3
        
        balanced = []
        for difficulty in ["easy", "medium", "hard"]:
            level_examples = [ex for ex in examples if ex.difficulty == difficulty]
            
            # Sample to target
            import random
            sampled = random.sample(
                level_examples,
                min(target_per_level, len(level_examples)),
            )
            balanced.extend(sampled)
        
        return balanced

# Usage
pipeline = SyntheticDataPipeline(config={"quality_threshold": 0.7})

constraints = {
    "domain": "customer_support",
    "intent_types": ["refund", "billing", "technical"],
    "difficulty_levels": ["easy", "medium", "hard"],
}

dataset = await pipeline.generate_dataset(num_examples=1000, constraints=constraints)
print(f"\n✓ Final dataset: {len(dataset)} examples")

This pipeline ensures quality at every stage, preventing low-quality examples from reaching production evals.

Connect to LLM evaluation suite for end-to-end testing.

LLM-as-Generator Pattern

Use LLMs to generate variations of base examples with constraints.

python
from typing import Optional

class LLMSyntheticGenerator:
    """Generate synthetic examples using LLM."""
    
    async def generate_with_constraints(
        self,
        base_example: str,
        constraints: Dict[str, Any],
        num_variations: int = 10,
    ) -> List[SyntheticExample]:
        """Generate constrained variations."""
        examples = []
        
        generation_prompt = self._build_generation_prompt(
            base_example,
            constraints,
        )
        
        for i in range(num_variations):
            example = await self._generate_single_variation(
                generation_prompt,
                constraints,
            )
            
            if example:
                examples.append(example)
        
        return examples
    
    def _build_generation_prompt(
        self,
        base_example: str,
        constraints: Dict[str, Any],
    ) -> str:
        """Build generation prompt with constraints."""
        constraint_text = "\n".join(
            f"- {key}: {value}" for key, value in constraints.items()
        )
        
        return f"""Generate a NEW example similar to this base case but with variations:

BASE EXAMPLE:
{base_example}

CONSTRAINTS:
{constraint_text}

REQUIREMENTS:
1. Change wording and structure but maintain intent
2. Introduce realistic typos if difficulty=hard
3. Vary formality level
4. Keep semantic meaning similar
5. Output must be realistic user input

Return JSON:
{{
  "input": "user input text",
  "expected_output": "correct system response",
  "difficulty": "easy|medium|hard",
  "tags": ["tag1", "tag2"]
}}"""
    
    async def _generate_single_variation(
        self,
        prompt: str,
        constraints: Dict[str, Any],
    ) -> Optional[SyntheticExample]:
        """Generate single variation."""
        try:
            response = await client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}],
                response_format={"type": "json_object"},
                temperature=0.9,  # Higher for diversity
            )
            
            import json
            data = json.loads(response.choices[0].message.content)
            
            return SyntheticExample(
                input_text=data["input"],
                expected_output=data["expected_output"],
                difficulty=data.get("difficulty", "medium"),
                tags=data.get("tags", []),
                metadata=constraints,
                generated_at=datetime.now(timezone.utc).isoformat(),
            )
        
        except Exception as e:
            print(f"Generation failed: {e}")
            return None

# Usage
generator = LLMSyntheticGenerator()

base_example = """Input: I need to cancel my subscription
Output: I'll help you cancel. Can I ask why you're leaving?"""

constraints = {
    "domain": "customer_support",
    "intent": "cancellation",
    "difficulty": "medium",
    "tone": "frustrated",
}

variations = await generator.generate_with_constraints(
    base_example,
    constraints,
    num_variations=20,
)

print(f"Generated {len(variations)} variations:")
for var in variations[:3]:
    print(f"\n  Input: {var.input_text[:60]}...")
    print(f"  Difficulty: {var.difficulty}")

Temperature tuning: Use 0.9-1.1 for diversity, 0.3-0.5 for consistency.

For few-shot prompting, seed with 3-5 high-quality examples.

Constraint-Based Synthesis

Systematically cover edge cases with constraint specifications.

python
from itertools import product
from typing import Iterator

class ConstraintBasedGenerator:
    """Generate examples to cover constraint combinations."""
    
    def __init__(self):
        self.constraint_space = {}
    
    def define_constraints(
        self,
        constraint_space: Dict[str, List[Any]],
    ) -> None:
        """Define constraint dimensions."""
        self.constraint_space = constraint_space
    
    def generate_combinations(self) -> Iterator[Dict[str, Any]]:
        """Generate all constraint combinations."""
        keys = list(self.constraint_space.keys())
        values = list(self.constraint_space.values())
        
        for combination in product(*values):
            yield dict(zip(keys, combination))
    
    async def generate_for_constraints(
        self,
        constraints: Dict[str, Any],
    ) -> SyntheticExample:
        """Generate example satisfying constraints."""
        prompt = f"""Generate a realistic example satisfying these constraints:

{self._format_constraints(constraints)}

Return JSON:
{{
  "input": "realistic user input",
  "expected_output": "correct response",
  "rationale": "why this satisfies constraints"
}}"""
        
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
        )
        
        import json
        data = json.loads(response.choices[0].message.content)
        
        return SyntheticExample(
            input_text=data["input"],
            expected_output=data["expected_output"],
            metadata={
                **constraints,
                "rationale": data["rationale"],
            },
            difficulty=constraints.get("difficulty", "medium"),
            tags=list(constraints.keys()),
            generated_at=datetime.now(timezone.utc).isoformat(),
        )
    
    def _format_constraints(self, constraints: Dict[str, Any]) -> str:
        """Format constraints for prompt."""
        return "\n".join(f"- {k}: {v}" for k, v in constraints.items())
    
    async def generate_full_coverage(self) -> List[SyntheticExample]:
        """Generate examples covering all constraint combinations."""
        examples = []
        
        for i, constraints in enumerate(self.generate_combinations()):
            print(f"Generating combination {i+1}...")
            example = await self.generate_for_constraints(constraints)
            examples.append(example)
        
        return examples

# Usage: systematic edge case coverage
generator = ConstraintBasedGenerator()

# Define constraint space
generator.define_constraints({
    "intent": ["refund", "cancel", "upgrade"],
    "tone": ["polite", "angry", "confused"],
    "length": ["short", "long"],
    "difficulty": ["easy", "hard"],
})

# Total combinations: 3 × 3 × 2 × 2 = 36
print(f"Total combinations: {sum(1 for _ in generator.generate_combinations())}")

# Generate full coverage
coverage_dataset = await generator.generate_full_coverage()
print(f"\n✓ Generated {len(coverage_dataset)} examples covering all combinations")

# Verify coverage
from collections import Counter
intent_coverage = Counter(ex.metadata["intent"] for ex in coverage_dataset)
print(f"\nIntent coverage: {dict(intent_coverage)}")

Coverage guarantee: Every constraint combination represented at least once.

For agent testing, generate edge cases for tool calling scenarios.

Adversarial Generation

Find failure modes early with adversarial examples.

python
class AdversarialGenerator:
    """Generate adversarial test cases."""
    
    async def generate_adversarial(
        self,
        target_system: str,
        num_examples: int = 50,
    ) -> List[SyntheticExample]:
        """Generate examples designed to break target system."""
        adversarial_examples = []
        
        attack_strategies = [
            self._generate_prompt_injection(),
            self._generate_ambiguous_input(),
            self._generate_edge_case_input(),
            self._generate_contradiction(),
            self._generate_out_of_distribution(),
        ]
        
        for strategy in attack_strategies:
            examples = await strategy(target_system, num_examples // len(attack_strategies))
            adversarial_examples.extend(examples)
        
        return adversarial_examples
    
    async def _generate_prompt_injection(
        self,
        target_system: str,
        num: int,
    ) -> List[SyntheticExample]:
        """Generate prompt injection attempts."""
        prompt = f"""Generate {num} prompt injection attempts for a {target_system} system.

Examples should try to:
1. Override system instructions
2. Extract sensitive information
3. Cause unintended behavior

Return JSON array:
[
  {{
    "input": "injection attempt",
    "expected_output": "safe rejection response",
    "attack_type": "instruction_override"
  }},
  ...
]"""
        
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
        )
        
        import json
        data = json.loads(response.choices[0].message.content)
        
        return [
            SyntheticExample(
                input_text=item["input"],
                expected_output=item["expected_output"],
                metadata={"attack_type": item["attack_type"]},
                difficulty="hard",
                tags=["adversarial", "security"],
                generated_at=datetime.now(timezone.utc).isoformat(),
            )
            for item in data.get("examples", [])
        ]
    
    async def _generate_ambiguous_input(
        self,
        target_system: str,
        num: int,
    ) -> List[SyntheticExample]:
        """Generate highly ambiguous inputs."""
        prompt = f"""Generate {num} ambiguous inputs for {target_system}.

Requirements:
- Multiple valid interpretations
- Unclear intent
- Missing context
- Contradictory statements

Return JSON array with input, expected_output, ambiguity_type."""
        
        # Similar implementation
        return []
    
    async def _generate_edge_case_input(
        self,
        target_system: str,
        num: int,
    ) -> List[SyntheticExample]:
        """Generate edge cases."""
        examples = []
        
        edge_cases = [
            "",  # Empty input
            "a" * 10000,  # Very long input
            "🎉" * 100,  # Unicode spam
            "\n\n\n",  # Whitespace only
            "NULL",  # SQL injection attempt
        ]
        
        for edge_input in edge_cases:
            examples.append(
                SyntheticExample(
                    input_text=edge_input,
                    expected_output="Error: invalid input",
                    metadata={"edge_case": "boundary_condition"},
                    difficulty="hard",
                    tags=["edge_case"],
                    generated_at=datetime.now(timezone.utc).isoformat(),
                )
            )
        
        return examples
    
    async def _generate_contradiction(
        self,
        target_system: str,
        num: int,
    ) -> List[SyntheticExample]:
        """Generate contradictory inputs."""
        return []
    
    async def _generate_out_of_distribution(
        self,
        target_system: str,
        num: int,
    ) -> List[SyntheticExample]:
        """Generate out-of-distribution examples."""
        return []

# Usage
adv_generator = AdversarialGenerator()

adversarial_dataset = await adv_generator.generate_adversarial(
    target_system="customer support chatbot",
    num_examples=100,
)

print(f"Generated {len(adversarial_dataset)} adversarial examples:")
for ex in adversarial_dataset[:3]:
    print(f"\n  Attack: {ex.metadata.get('attack_type', 'unknown')}")
    print(f"  Input: {ex.input_text[:60]}...")

Test early: Run adversarial evals before each deployment.

For agent security, generate tool calling attacks.

Quality Validation Framework

Automated quality checks prevent bad data from reaching evals.

python
from typing import Callable
import re

class QualityValidator:
    """Multi-stage quality validation."""
    
    def __init__(self):
        self.checks: List[Callable] = []
        self._register_default_checks()
    
    def _register_default_checks(self) -> None:
        """Register default quality checks."""
        self.checks = [
            self._check_length,
            self._check_coherence_score,
            self._check_format_validity,
            self._check_no_placeholder_text,
            self._check_language_quality,
        ]
    
    async def validate(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, List[str]]:
        """Run all validation checks."""
        failures = []
        
        for check in self.checks:
            passed, message = await check(example)
            if not passed:
                failures.append(message)
        
        is_valid = len(failures) == 0
        return is_valid, failures
    
    async def _check_length(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, str]:
        """Check reasonable length."""
        if len(example.input_text) < 5:
            return False, "Input too short"
        
        if len(example.expected_output) < 3:
            return False, "Output too short"
        
        if len(example.input_text) > 5000:
            return False, "Input too long"
        
        return True, ""
    
    async def _check_coherence_score(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, str]:
        """Check input/output coherence."""
        prompt = f"""Rate coherence (0-10):
Input: {example.input_text}
Output: {example.expected_output}

Return JSON: {{"score": int, "reason": "brief explanation"}}"""
        
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
        )
        
        import json
        data = json.loads(response.choices[0].message.content)
        
        if data["score"] < 7:
            return False, f"Low coherence: {data['reason']}"
        
        return True, ""
    
    async def _check_format_validity(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, str]:
        """Check format requirements."""
        if not example.difficulty in ["easy", "medium", "hard"]:
            return False, f"Invalid difficulty: {example.difficulty}"
        
        if not example.tags:
            return False, "Missing tags"
        
        return True, ""
    
    async def _check_no_placeholder_text(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, str]:
        """Check for placeholder/template text."""
        placeholders = [
            r"\[.*?\]",  # [placeholder]
            r"\{.*?\}",  # {variable}
            r"<.*?>",    # <tag>
            r"TODO",
            r"XXX",
        ]
        
        text = example.input_text + " " + example.expected_output
        
        for pattern in placeholders:
            if re.search(pattern, text):
                return False, f"Contains placeholder: {pattern}"
        
        return True, ""
    
    async def _check_language_quality(
        self,
        example: SyntheticExample,
    ) -> tuple[bool, str]:
        """Check language quality."""
        # Placeholder—implement language model scoring
        return True, ""

# Usage with pipeline
validator = QualityValidator()

validated_examples = []
rejected_examples = []

for example in raw_generated_examples:
    is_valid, failures = await validator.validate(example)
    
    if is_valid:
        validated_examples.append(example)
    else:
        rejected_examples.append((example, failures))
        print(f"✗ Rejected: {failures}")

print(f"\n✓ Validated: {len(validated_examples)}")
print(f"✗ Rejected: {len(rejected_examples)}")

Rejection rate: Expect 20-40% rejection in early iterations.

For evaluation without ground truth, validate consistency.

Bias Detection and Mitigation

Ensure representative test sets across demographics and scenarios.

python
class BiasDetector:
    """Detect and mitigate bias in synthetic datasets."""
    
    async def analyze_bias(
        self,
        dataset: List[SyntheticExample],
    ) -> Dict[str, Any]:
        """Analyze dataset for biases."""
        analysis = {
            "demographic_representation": await self._check_demographics(dataset),
            "scenario_diversity": await self._check_scenarios(dataset),
            "language_patterns": await self._check_language(dataset),
            "difficulty_balance": self._check_difficulty(dataset),
        }
        
        return analysis
    
    async def _check_demographics(
        self,
        dataset: List[SyntheticExample],
    ) -> Dict[str, Any]:
        """Check demographic representation."""
        # Use NER to identify mentions
        demographic_mentions = {
            "gender": [],
            "age": [],
            "location": [],
        }
        
        for example in dataset:
            # Extract demographic markers
            text = example.input_text + " " + example.expected_output
            
            # Simplified—use proper NER in production
            if any(word in text.lower() for word in ["he", "him", "his"]):
                demographic_mentions["gender"].append("male")
            if any(word in text.lower() for word in ["she", "her", "hers"]):
                demographic_mentions["gender"].append("female")
        
        from collections import Counter
        return {k: dict(Counter(v)) for k, v in demographic_mentions.items()}
    
    async def _check_scenarios(
        self,
        dataset: List[SyntheticExample],
    ) -> Dict[str, int]:
        """Check scenario diversity."""
        from collections import Counter
        
        tags = [tag for ex in dataset for tag in ex.tags]
        return dict(Counter(tags))
    
    async def _check_language(
        self,
        dataset: List[SyntheticExample],
    ) -> Dict[str, Any]:
        """Check language patterns."""
        # Check formality, complexity, etc.
        return {"formality": "mixed", "complexity": "balanced"}
    
    def _check_difficulty(
        self,
        dataset: List[SyntheticExample],
    ) -> Dict[str, int]:
        """Check difficulty balance."""
        from collections import Counter
        
        difficulties = [ex.difficulty for ex in dataset]
        return dict(Counter(difficulties))
    
    async def mitigate_bias(
        self,
        dataset: List[SyntheticExample],
        target_balance: Dict[str, float],
    ) -> List[SyntheticExample]:
        """Mitigate identified biases."""
        # Oversample underrepresented groups
        # Undersample overrepresented groups
        
        analysis = await self.analyze_bias(dataset)
        
        print("Current distribution:", analysis)
        print("Target balance:", target_balance)
        
        # Implementation: rebalancing logic
        return dataset  # Placeholder

# Usage
detector = BiasDetector()

bias_analysis = await detector.analyze_bias(dataset)
print("Bias analysis:", bias_analysis)

# Mitigate if needed
balanced_dataset = await detector.mitigate_bias(
    dataset,
    target_balance={"difficulty": {"easy": 0.33, "medium": 0.34, "hard": 0.33}},
)

Diversity metrics: Track representation across key dimensions.

For multi-agent systems, test agent interactions.

Dataset Versioning

Track dataset versions with metadata and provenance.

python
import hashlib
from pathlib import Path
import json

class DatasetVersionManager:
    """Version control for synthetic datasets."""
    
    def __init__(self, storage_path: Path):
        self.storage_path = storage_path
        self.storage_path.mkdir(parents=True, exist_ok=True)
    
    def save_dataset(
        self,
        dataset: List[SyntheticExample],
        version: str,
        metadata: Dict[str, Any],
    ) -> str:
        """Save versioned dataset."""
        # Compute hash
        dataset_hash = self._compute_hash(dataset)
        
        # Save data
        version_path = self.storage_path / f"v{version}"
        version_path.mkdir(exist_ok=True)
        
        # Save examples
        data_file = version_path / "dataset.jsonl"
        with data_file.open("w") as f:
            for example in dataset:
                f.write(json.dumps({
                    "input": example.input_text,
                    "output": example.expected_output,
                    "metadata": example.metadata,
                    "difficulty": example.difficulty,
                    "tags": example.tags,
                }) + "\n")
        
        # Save metadata
        meta_file = version_path / "metadata.json"
        full_metadata = {
            **metadata,
            "version": version,
            "hash": dataset_hash,
            "size": len(dataset),
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
        
        with meta_file.open("w") as f:
            json.dump(full_metadata, f, indent=2)
        
        print(f"✓ Saved dataset v{version} ({len(dataset)} examples)")
        return dataset_hash
    
    def load_dataset(self, version: str) -> List[SyntheticExample]:
        """Load versioned dataset."""
        data_file = self.storage_path / f"v{version}" / "dataset.jsonl"
        
        examples = []
        with data_file.open("r") as f:
            for line in f:
                data = json.loads(line)
                examples.append(SyntheticExample(
                    input_text=data["input"],
                    expected_output=data["output"],
                    metadata=data["metadata"],
                    difficulty=data["difficulty"],
                    tags=data["tags"],
                    generated_at=data["metadata"].get("generated_at", ""),
                ))
        
        return examples
    
    def _compute_hash(self, dataset: List[SyntheticExample]) -> str:
        """Compute dataset hash."""
        content = "".join(
            example.input_text + example.expected_output
            for example in dataset
        )
        return hashlib.sha256(content.encode()).hexdigest()[:8]

# Usage
version_manager = DatasetVersionManager(storage_path=Path("./eval_datasets"))

# Save dataset
metadata = {
    "generation_method": "llm_synthetic",
    "quality_threshold": 0.7,
    "num_constraints": 36,
}

dataset_hash = version_manager.save_dataset(
    dataset=final_dataset,
    version="1.0.0",
    metadata=metadata,
)

# Load later
loaded = version_manager.load_dataset(version="1.0.0")
print(f"Loaded {len(loaded)} examples from v1.0.0")

Version on every generation run for reproducibility.

For CI/CD evals, pin dataset versions.

Production Deployment

Deploy generation pipeline with monitoring and cost controls.

python
class ProductionSyntheticPipeline:
    """Production-ready synthetic generation."""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.pipeline = SyntheticDataPipeline(config)
        self.validator = QualityValidator()
        self.bias_detector = BiasDetector()
        self.version_manager = DatasetVersionManager(Path("./datasets"))
    
    async def generate_production_dataset(
        self,
        requirements: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Generate production dataset with full validation."""
        print("Starting production generation...")
        
        # Generate
        dataset = await self.pipeline.generate_dataset(
            num_examples=requirements["num_examples"],
            constraints=requirements["constraints"],
        )
        
        # Validate quality
        validated = []
        for example in dataset:
            is_valid, failures = await self.validator.validate(example)
            if is_valid:
                validated.append(example)
        
        print(f"✓ Quality validation: {len(validated)}/{len(dataset)} passed")
        
        # Check bias
        bias_analysis = await self.bias_detector.analyze_bias(validated)
        print(f"✓ Bias analysis: {bias_analysis}")
        
        # Version and save
        version = requirements.get("version", "1.0.0")
        dataset_hash = self.version_manager.save_dataset(
            validated,
            version,
            metadata={
                "requirements": requirements,
                "bias_analysis": bias_analysis,
            },
        )
        
        return {
            "dataset": validated,
            "version": version,
            "hash": dataset_hash,
            "statistics": {
                "total_generated": len(dataset),
                "quality_validated": len(validated),
                "rejection_rate": 1 - (len(validated) / len(dataset)),
            },
        }

# Deploy
production_pipeline = ProductionSyntheticPipeline(config={
    "quality_threshold": 0.75,
    "max_cost_usd": 500,
})

result = await production_pipeline.generate_production_dataset(
    requirements={
        "num_examples": 2000,
        "constraints": {
            "domain": "customer_support",
            "intent_types": ["refund", "billing", "technical", "general"],
        },
        "version": "2.0.0",
    },
)

print(f"\n✓ Production dataset ready:")
print(f"  Version: {result['version']}")
print(f"  Hash: {result['hash']}")
print(f"  Size: {len(result['dataset'])} examples")
print(f"  Rejection rate: {result['statistics']['rejection_rate']:.1%}")

Deploy with cloud infrastructure and observability.

Primary references: official documentation, official documentation, official documentation, official documentation.

Synthetic Data Generation for LLM Evals Decision Table

DecisionPrefer the simpler path whenAdd operational complexity when
ArchitectureOne component can own the contract and stateIndependent scaling or fault isolation is required
RolloutOffline replay covers the meaningful casesLive behavior requires shadow traffic and a canary
RecoveryA failed operation is safe to repeatPartial effects require idempotency or reconciliation
MeasurementOne service objective represents user impactQuality, latency, and cost need separate gates

Operating Synthetic Data Generation for LLM Evals as a System

The implementation is only one part of Synthetic Data Generation for LLM Evals. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Synthetic Data Generation for LLM Evals behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Synthetic Data Generation for LLM Evals engineering support.

Frequently Asked Questions

How accurate is synthetic data compared to real data?

Well-generated synthetic data achieves 85-95% of real data quality. The key is proper validation—low-quality synthetic data is worse than no data. Use human review on 5-10% of generated examples to validate.

What's the cost per synthetic example?

$0.0001-0.001 per example depending on model and complexity. GPT-4o generation costs ~$0.0005/example, gpt-4o-mini costs ~$0.00005/example. Bulk generation of 10K examples: $5-10.

Should I use synthetic or real data?

Both. Start with synthetic for coverage and speed, then augment with real production data. Ideal ratio: 70% synthetic, 30% real production cases.

How do I prevent model memorization?

Use different models for generation and evaluation. Generate with GPT-4o, evaluate with Claude Sonnet. This prevents memorization-based false positives.

Can I generate multilingual synthetic data?

Yes, but validate per-language quality. LLMs perform better on high-resource languages (English, Spanish) than low-resource languages (Swahili, Pashto). Budget 2x validation effort for low-resource languages.

How often should I regenerate synthetic datasets?

Regenerate when your system changes significantly—new features, updated models, changed business logic. Monthly regeneration for active development, quarterly for stable systems.

Conclusion

Synthetic data generation enables evaluation at scale:

  • Pipeline approach ensures quality through validation gates
  • LLM-as-generator creates realistic variations efficiently
  • Constraint-based synthesis guarantees edge case coverage
  • Adversarial generation uncovers failure modes early
  • Quality validation prevents bad data from polluting evals
  • Bias detection ensures representative test sets

Synthetic generation is 1000x faster than manual annotation.

At HinterBuild, we build synthetic data pipelines for production AI systems:

Contact us for synthetic data consulting.

Free consultation

Book a free consultation call on synthetic data for LLM testing

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Keep reading

Related articles

AI Systems · 10 min read

Triton vs vLLM: LLM Serving Framework Comparison for

Triton vs vLLM guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.

Read post

AI Systems · 9 min read

LLM Tracing with OpenTelemetry: Complete Observability Guide

Learn llm tracing with opentelemetry through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

AI Systems · 10 min read

PII Detection and Scrubbing in LLM Pipelines

PII Detection and Scrubbing in LLM Pipelines guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Read post

AI Systems · 11 min read

OWASP Top 10 for LLM Applications: Complete Security Guide

Learn owasp top 10 for llm applications through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post