
    """
Pattern 151: Synthetic Data Generation

Description:
    Generates synthetic training and test data for agent systems using LLMs.
    Creates diverse, high-quality examples with schema validation, quality
    metrics, and coverage analysis.

Components:
    - Data Schema Definitions: Structured format specifications
    - Generation Strategies: Various approaches for data creation
    - Quality Validators: Check data quality and validity
    - Coverage Analyzers: Ensure diverse data distribution
    - Augmentation Engine: Enhance existing datasets

Use Cases:
    - Training data creation for fine-tuning
    - Test case generation for QA
    - Data augmentation for small datasets
    - Edge case generation for robustness testing
    - Privacy-preserving data synthesis

Benefits:
    - Reduces manual data labeling effort
    - Generates diverse test scenarios
    - Creates privacy-safe datasets
    - Enables systematic coverage
    - Scales data creation efficiently

Trade-offs:
    - Generated data may lack real-world complexity
    - Quality depends on LLM capabilities
    - May introduce biases from LLM
    - Validation overhead
    - Cost of generation at scale

LangChain Implementation:
    Uses ChatOpenAI for generation with structured output parsing,
    Pydantic models for validation, and custom metrics.
"""

import os
import json
import time
from typing import List, Dict, Any, Optional, Set, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
from collections import defaultdict
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field, validator

load_dotenv()


class DataType(Enum):
    """Types of synthetic data"""
    QA_PAIR = "qa_pair"
    CLASSIFICATION = "classification"
    CONVERSATION = "conversation"
    INSTRUCTION = "instruction"
    SENTIMENT = "sentiment"
    ENTITY_EXTRACTION = "entity_extraction"
    SUMMARIZATION = "summarization"


class QualityMetric(Enum):
    """Quality metrics for generated data"""
    COHERENCE = "coherence"
    RELEVANCE = "relevance"
    DIVERSITY = "diversity"
    COMPLEXITY = "complexity"
    VALIDITY = "validity"


# Pydantic Models for Structured Generation

class QAPair(BaseModel):
    """Question-Answer pair"""
    question: str = Field(description="The question")
    answer: str = Field(description="The answer")
    context: Optional[str] = Field(default=None, description="Optional context")
    difficulty: str = Field(default="medium", description="easy, medium, or hard")
    category: Optional[str] = Field(default=None, description="Question category")


class ClassificationExample(BaseModel):
    """Text classification example"""
    text: str = Field(description="The text to classify")
    label: str = Field(description="The classification label")
    confidence: float = Field(default=1.0, description="Confidence score 0-1")
    explanation: Optional[str] = Field(default=None, description="Explanation for label")


class ConversationTurn(BaseModel):
    """Single conversation turn"""
    speaker: str = Field(description="Speaker identifier")
    utterance: str = Field(description="What was said")
    intent: Optional[str] = Field(default=None, description="Speaker intent")


class Conversation(BaseModel):
    """Multi-turn conversation"""
    turns: List[ConversationTurn] = Field(description="Conversation turns")
    topic: Optional[str] = Field(default=None, description="Conversation topic")
    outcome: Optional[str] = Field(default=None, description="Conversation outcome")


@dataclass
class GenerationConfig:
    """Configuration for data generation"""
    data_type: DataType
    count: int = 10
    domain: Optional[str] = None
    difficulty: str = "mixed"  # easy, medium, hard, mixed
    diversity_target: float = 0.7  # Target diversity score
    quality_threshold: float = 0.6  # Minimum quality score
    include_edge_cases: bool = True
    seed_examples: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class QualityScores:
    """Quality scores for generated data"""
    coherence: float = 0.0
    relevance: float = 0.0
    diversity: float = 0.0
    complexity: float = 0.0
    validity: float = 0.0
    
    def overall_score(self) -> float:
        """Calculate overall quality score"""
        return (self.coherence + self.relevance + self.diversity + 
                self.complexity + self.validity) / 5
    
    def to_dict(self) -> Dict[str, float]:
        """Convert to dictionary"""
        return {
            "coherence": self.coherence,
            "relevance": self.relevance,
            "diversity": self.diversity,
            "complexity": self.complexity,
            "validity": self.validity,
            "overall": self.overall_score()
        }


class SyntheticDataGenerator:
    """
    Agent that generates synthetic training and test data.
    Produces high-quality, diverse datasets with validation.
    """
    
    def __init__(self, temperature: float = 0.9):
        """Initialize the synthetic data generator"""
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=temperature)
        self.validator_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.3)
        
        # Track generated data for diversity
        self.generated_texts: Set[str] = set()
        self.category_counts: Dict[str, int] = defaultdict(int)
        
        # Generation prompts
        self.qa_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a data generation expert. Generate high-quality question-answer pairs.
Domain: {domain}
Difficulty: {difficulty}
Requirements: {requirements}

Generate diverse, realistic questions with accurate answers.
Return as JSON with fields: question, answer, context, difficulty, category"""),
            ("user", "Generate {count} question-answer pairs.")
        ])
        
        self.classification_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a data generation expert. Generate text classification examples.
Domain: {domain}
Labels: {labels}
Requirements: {requirements}

Generate realistic text examples with correct labels.
Return as JSON with fields: text, label, confidence, explanation"""),
            ("user", "Generate {count} classification examples.")
        ])
        
        self.conversation_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a conversation generation expert. Create realistic multi-turn conversations.
Domain: {domain}
Number of turns: {num_turns}
Requirements: {requirements}

Generate natural, coherent conversations.
Return as JSON with structure: turns (list of speaker/utterance), topic, outcome"""),
            ("user", "Generate {count} conversations.")
        ])
        
        self.quality_eval_prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate the quality of generated data on multiple dimensions.
Rate each dimension from 0.0 to 1.0:
- Coherence: Logical consistency and clarity
- Relevance: Relevance to the domain/task
- Diversity: Uniqueness and variety
- Complexity: Appropriate complexity level
- Validity: Correctness and accuracy

Return JSON with scores for each dimension."""),
            ("user", "Evaluate this data:\n{data}")
        ])
    
    def generate_dataset(self, config: GenerationConfig) -> Dict[str, Any]:
        """
        Generate synthetic dataset according to configuration
        
        Args:
            config: Generation configuration
            
        Returns:
            Dictionary with generated data and metrics
        """
        start_time = time.time()
        
        print(f"\n🔄 Generating {config.count} {config.data_type.value} examples...")
        if config.domain:
            print(f"   Domain: {config.domain}")
        print(f"   Difficulty: {config.difficulty}")
        
        # Generate data based on type
        if config.data_type == DataType.QA_PAIR:
            examples = self._generate_qa_pairs(config)
        elif config.data_type == DataType.CLASSIFICATION:
            examples = self._generate_classifications(config)
        elif config.data_type == DataType.CONVERSATION:
            examples = self._generate_conversations(config)
        else:
            examples = self._generate_generic(config)
        
        # Validate and score quality
        quality_scores = self._evaluate_quality(examples, config)
        
        # Analyze diversity
        diversity_metrics = self._analyze_diversity(examples)
        
        # Filter by quality threshold
        filtered_examples = self._filter_by_quality(examples, quality_scores, 
                                                    config.quality_threshold)
        
        generation_time = time.time() - start_time
        
        return {
            "examples": filtered_examples,
            "count": len(filtered_examples),
            "quality_scores": quality_scores,
            "diversity_metrics": diversity_metrics,
            "generation_time": generation_time,
            "config": {
                "data_type": config.data_type.value,
                "target_count": config.count,
                "domain": config.domain,
                "difficulty": config.difficulty
            }
        }
    
    def _generate_qa_pairs(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate question-answer pairs"""
        examples = []
        
        # Generate in batches for diversity
        batch_size = min(5, config.count)
        num_batches = (config.count + batch_size - 1) // batch_size
        
        for batch in range(num_batches):
            batch_count = min(batch_size, config.count - len(examples))
            
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.qa_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general knowledge",
                    "difficulty": config.difficulty,
                    "requirements": requirements,
                    "count": batch_count
                })
                
                # Parse JSON response
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list):
                    examples.extend(parsed)
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
        
        return examples[:config.count]
    
    def _generate_classifications(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate classification examples"""
        examples = []
        
        # Default labels if not specified
        labels = ["positive", "negative", "neutral"]
        if config.domain:
            labels = self._infer_labels_from_domain(config.domain)
        
        batch_size = min(5, config.count)
        num_batches = (config.count + batch_size - 1) // batch_size
        
        for batch in range(num_batches):
            batch_count = min(batch_size, config.count - len(examples))
            
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.classification_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general text classification",
                    "labels": ", ".join(labels),
                    "requirements": requirements,
                    "count": batch_count
                })
                
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list):
                    examples.extend(parsed)
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Batch {batch + 1} generation error: {str(e)}")
        
        return examples[:config.count]
    
    def _generate_conversations(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate multi-turn conversations"""
        examples = []
        
        num_turns = 5  # Default conversation length
        
        for i in range(config.count):
            requirements = self._get_generation_requirements(config, examples)
            
            try:
                chain = self.conversation_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "domain": config.domain or "general conversation",
                    "num_turns": num_turns,
                    "requirements": requirements,
                    "count": 1
                })
                
                parsed = self._parse_json_response(result)
                if isinstance(parsed, list) and parsed:
                    examples.append(parsed[0])
                elif isinstance(parsed, dict):
                    examples.append(parsed)
                    
            except Exception as e:
                print(f"   ⚠️  Example {i + 1} generation error: {str(e)}")
        
        return examples
    
    def _generate_generic(self, config: GenerationConfig) -> List[Dict[str, Any]]:
        """Generate generic examples"""
        generic_prompt = ChatPromptTemplate.from_messages([
            ("system", "Generate {count} diverse examples of {data_type} in {domain} domain. Return as JSON."),
            ("user", "Generate the examples.")
        ])
        
        chain = generic_prompt | self.llm | StrOutputParser()
        result = chain.invoke({
            "count": config.count,
            "data_type": config.data_type.value,
            "domain": config.domain or "general"
        })
        
        parsed = self._parse_json_response(result)
        return parsed if isinstance(parsed, list) else [parsed]
    
    def _get_generation_requirements(self, config: GenerationConfig, 
                                   existing: List[Dict[str, Any]]) -> str:
        """Generate requirements for next batch"""
        requirements = []
        
        if config.diversity_target > 0.5:
            requirements.append("Ensure high diversity from previous examples")
        
        if config.include_edge_cases and len(existing) > 5:
            requirements.append("Include edge cases and unusual scenarios")
        
        # Category balance
        if self.category_counts:
            underrep_categories = [cat for cat, count in self.category_counts.items() 
                                  if count  List[str]:
        """Infer appropriate labels from domain"""
        domain_lower = domain.lower()
        
        if "sentiment" in domain_lower:
            return ["positive", "negative", "neutral"]
        elif "emotion" in domain_lower:
            return ["joy", "sadness", "anger", "fear", "surprise"]
        elif "intent" in domain_lower:
            return ["question", "command", "statement", "request"]
        elif "urgency" in domain_lower:
            return ["urgent", "normal", "low_priority"]
        else:
            return ["category_a", "category_b", "category_c"]
    
    def _parse_json_response(self, response: str) -> Any:
        """Parse JSON from LLM response"""
        try:
            # Try direct JSON parse
            return json.loads(response)
        except json.JSONDecodeError:
            # Extract JSON from markdown code blocks
            if "```json" in response:
                start = response.find("```json") + 7
                end = response.find("```", start)
                json_str = response[start:end].strip()
                return json.loads(json_str)
            elif "```" in response:
                start = response.find("```") + 3
                end = response.find("```", start)
                json_str = response[start:end].strip()
                return json.loads(json_str)
            else:
                # Try to find JSON array or object
                for char in ['{', '[']:
                    if char in response:
                        start = response.find(char)
                        return json.loads(response[start:])
                
                return []
    
    def _evaluate_quality(self, examples: List[Dict[str, Any]], 
                         config: GenerationConfig) -> List[QualityScores]:
        """Evaluate quality of generated examples"""
        quality_scores = []
        
        # Sample for evaluation (evaluate all if small dataset)
        sample_size = min(len(examples), 10)
        sample_indices = list(range(0, len(examples), max(1, len(examples) // sample_size)))
        
        for idx in sample_indices[:sample_size]:
            example = examples[idx]
            
            try:
                chain = self.quality_eval_prompt | self.validator_llm | StrOutputParser()
                result = chain.invoke({"data": json.dumps(example, indent=2)})
                
                scores_dict = self._parse_json_response(result)
                
                scores = QualityScores(
                    coherence=scores_dict.get("coherence", 0.7),
                    relevance=scores_dict.get("relevance", 0.7),
                    diversity=scores_dict.get("diversity", 0.7),
                    complexity=scores_dict.get("complexity", 0.7),
                    validity=scores_dict.get("validity", 0.7)
                )
                quality_scores.append(scores)
                
            except Exception as e:
                # Default scores on error
                quality_scores.append(QualityScores(
                    coherence=0.7, relevance=0.7, diversity=0.7,
                    complexity=0.7, validity=0.7
                ))
        
        # Extend scores to all examples (reuse evaluated scores)
        while len(quality_scores)  Dict[str, Any]:
        """Analyze diversity of generated examples"""
        if not examples:
            return {"diversity_score": 0.0}
        
        # Extract texts for similarity analysis
        texts = []
        for ex in examples:
            if "question" in ex:
                texts.append(ex["question"])
            elif "text" in ex:
                texts.append(ex["text"])
            elif "turns" in ex:
                texts.append(" ".join(t.get("utterance", "") for t in ex["turns"]))
        
        # Calculate pairwise uniqueness
        unique_ratio = len(set(texts)) / len(texts) if texts else 0
        
        # Category distribution
        categories = [ex.get("category", "unknown") for ex in examples]
        category_dist = {cat: categories.count(cat) / len(categories) 
                        for cat in set(categories)}
        
        # Length variance
        lengths = [len(t) for t in texts]
        avg_length = sum(lengths) / len(lengths) if lengths else 0
        length_variance = sum((l - avg_length) ** 2 for l in lengths) / len(lengths) if lengths else 0
        
        return {
            "diversity_score": unique_ratio,
            "unique_examples": len(set(texts)),
            "total_examples": len(texts),
            "category_distribution": category_dist,
            "avg_length": avg_length,
            "length_variance": length_variance
        }
    
    def _filter_by_quality(self, examples: List[Dict[str, Any]], 
                          quality_scores: List[QualityScores],
                          threshold: float) -> List[Dict[str, Any]]:
        """Filter examples by quality threshold"""
        filtered = []
        
        for example, scores in zip(examples, quality_scores):
            if scores.overall_score() >= threshold:
                filtered.append(example)
        
        if len(filtered)  List[Dict[str, Any]]:
        """
        Augment existing dataset with variations
        
        Args:
            existing_data: Existing examples to augment
            augmentation_factor: How many variations per example
            
        Returns:
            Augmented dataset
        """
        print(f"\n🔄 Augmenting {len(existing_data)} examples (factor: {augmentation_factor})...")
        
        augmented = existing_data.copy()
        
        augment_prompt = ChatPromptTemplate.from_messages([
            ("system", """Create {count} variations of the following example while maintaining its core meaning and structure.
Make the variations diverse but semantically similar.
Return as JSON array."""),
            ("user", "Original example:\n{example}")
        ])
        
        for example in existing_data:
            try:
                chain = augment_prompt | self.llm | StrOutputParser()
                result = chain.invoke({
                    "count": augmentation_factor,
                    "example": json.dumps(example, indent=2)
                })
                
                variations = self._parse_json_response(result)
                if isinstance(variations, list):
                    augmented.extend(variations[:augmentation_factor])
                    
            except Exception as e:
                print(f"   ⚠️  Augmentation error: {str(e)}")
        
        print(f"   ✅ Augmented to {len(augmented)} examples")
        return augmented
    
    def generate_report(self, results: Dict[str, Any]) -> str:
        """Generate comprehensive generation report"""
        report = [
            "\n" + "="*70,
            "SYNTHETIC DATA GENERATION REPORT",
            "="*70,
            f"\nConfiguration:",
            f"  Data Type: {results['config']['data_type']}",
            f"  Target Count: {results['config']['target_count']}",
            f"  Domain: {results['config']['domain'] or 'General'}",
            f"  Difficulty: {results['config']['difficulty']}",
            f"\nGeneration Results:",
            f"  Generated: {results['count']} examples",
            f"  Generation Time: {results['generation_time']:.2f}s",
            f"  Rate: {results['count']/results['generation_time']:.1f} examples/sec"
        ]
        
        # Quality metrics
        if results['quality_scores']:
            avg_scores = {
                "coherence": sum(s.coherence for s in results['quality_scores']) / len(results['quality_scores']),
                "relevance": sum(s.relevance for s in results['quality_scores']) / len(results['quality_scores']),
                "diversity": sum(s.diversity for s in results['quality_scores']) / len(results['quality_scores']),
                "complexity": sum(s.complexity for s in results['quality_scores']) / len(results['quality_scores']),
                "validity": sum(s.validity for s in results['quality_scores']) / len(results['quality_scores'])
            }
            
            report.extend([
                "\nQuality Scores (0.0-1.0):",
                f"  Coherence: {avg_scores['coherence']:.3f}",
                f"  Relevance: {avg_scores['relevance']:.3f}",
                f"  Diversity: {avg_scores['diversity']:.3f}",
                f"  Complexity: {avg_scores['complexity']:.3f}",
                f"  Validity: {avg_scores['validity']:.3f}",
                f"  Overall: {sum(avg_scores.values())/len(avg_scores):.3f}"
            ])
        
        # Diversity metrics
        div = results['diversity_metrics']
        report.extend([
            "\nDiversity Metrics:",
            f"  Diversity Score: {div['diversity_score']:.3f}",
            f"  Unique Examples: {div['unique_examples']}/{div['total_examples']}",
            f"  Average Length: {div['avg_length']:.0f} characters"
        ])
        
        if 'category_distribution' in div:
            report.append("  Category Distribution:")
            for cat, ratio in sorted(div['category_distribution'].items(), 
                                    key=lambda x: x[1], reverse=True):
                report.append(f"    - {cat}: {ratio*100:.1f}%")
        
        report.append("="*70)
        return "\n".join(report)


def demonstrate_synthetic_data_generation():
    """Demonstrate synthetic data generation"""
    print("="*70)
    print("Pattern 151: Synthetic Data Generation")
    print("="*70)
    
    generator = SyntheticDataGenerator()
    
    # Test 1: QA Pairs
    print("\n" + "="*70)
    print("TEST 1: Generate Question-Answer Pairs")
    print("="*70)
    
    qa_config = GenerationConfig(
        data_type=DataType.QA_PAIR,
        count=10,
        domain="Python programming",
        difficulty="mixed",
        quality_threshold=0.6
    )
    
    qa_results = generator.generate_dataset(qa_config)
    
    print(f"\n✅ Generated {qa_results['count']} QA pairs")
    print("\nSample Examples:")
    for i, example in enumerate(qa_results['examples'][:3], 1):
        print(f"\n{i}. Q: {example.get('question', 'N/A')}")
        print(f"   A: {example.get('answer', 'N/A')[:100]}...")
        if 'difficulty' in example:
            print(f"   Difficulty: {example['difficulty']}")
    
    print(generator.generate_report(qa_results))
    
    # Test 2: Classification Examples
    print("\n" + "="*70)
    print("TEST 2: Generate Classification Examples")
    print("="*70)
    
    class_config = GenerationConfig(
        data_type=DataType.CLASSIFICATION,
        count=8,
        domain="sentiment analysis",
        difficulty="medium",
        quality_threshold=0.6
    )
    
    class_results = generator.generate_dataset(class_config)
    
    print(f"\n✅ Generated {class_results['count']} classification examples")
    print("\nSample Examples:")
    for i, example in enumerate(class_results['examples'][:3], 1):
        print(f"\n{i}. Text: {example.get('text', 'N/A')[:80]}...")
        print(f"   Label: {example.get('label', 'N/A')}")
        if 'confidence' in example:
            print(f"   Confidence: {example['confidence']:.2f}")
    
    # Test 3: Conversations
    print("\n" + "="*70)
    print("TEST 3: Generate Conversations")
    print("="*70)
    
    conv_config = GenerationConfig(
        data_type=DataType.CONVERSATION,
        count=5,
        domain="customer support",
        quality_threshold=0.6
    )
    
    conv_results = generator.generate_dataset(conv_config)
    
    print(f"\n✅ Generated {conv_results['count']} conversations")
    if conv_results['examples']:
        print("\nSample Conversation:")
        example = conv_results['examples'][0]
        if 'turns' in example:
            for turn in example['turns'][:4]:
                speaker = turn.get('speaker', 'Unknown')
                utterance = turn.get('utterance', '')
                print(f"   {speaker}: {utterance}")
        if 'topic' in example:
            print(f"   Topic: {example['topic']}")
    
    # Test 4: Data Augmentation
    print("\n" + "="*70)
    print("TEST 4: Data Augmentation")
    print("="*70)
    
    seed_data = [
        {"question": "What is Python?", "answer": "A programming language", "category": "basics"},
        {"question": "How to define a function?", "answer": "Use the def keyword", "category": "syntax"}
    ]
    
    augmented = generator.augment_dataset(seed_data, augmentation_factor=2)
    print(f"\n✅ Augmented {len(seed_data)} examples to {len(augmented)} examples")
    
    # Summary
    print("\n" + "="*70)
    print("SUMMARY")
    print("="*70)
    print("""
Synthetic Data Generation Pattern provides:

1. Multiple Data Types:
   - Question-Answer pairs
   - Classification examples
   - Multi-turn conversations
   - Custom formats

2. Quality Control:
   - Automated quality evaluation
   - Coherence, relevance, validity checks
   - Configurable quality thresholds
   - Filtering low-quality examples

3. Diversity Assurance:
   - Uniqueness tracking
   - Category balance
   - Length variation
   - Edge case inclusion

4. Augmentation Capabilities:
   - Variation generation
   - Dataset expansion
   - Semantic preservation

5. Production Features:
   - Batch generation
   - Domain specialization
   - Difficulty levels
   - Comprehensive reporting

This pattern is essential for creating training data, test cases,
and expanding datasets with privacy-safe synthetic examples.
    """)


if __name__ == "__main__":
    demonstrate_synthetic_data_generation()

Agentic Design Pattern: Synthetic Data

Pattern 151: Synthetic Data – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Synthetic Data pattern implementation.

Pattern Description

Pattern 151: Synthetic Data Generation

Description: Generates synthetic training and test data for agent systems using LLMs. Creates diverse, high-quality examples with schema validation, quality metrics, and coverage analysis.

Components: – Data Schema Definitions: Structured format specifications – Generation Strategies: Various approaches for data creation – Quality Validators: Check data quality and validity – Coverage Analyzers: Ensure diverse data distribution – Augmentation Engine: Enhance existing datasets

Use Cases: – Training data creation for fine-tuning – Test case generation for QA – Data augmentation for small datasets – Edge case generation for robustness testing – Privacy-preserving data synthesis

Benefits: – Reduces manual data labeling effort – Generates diverse test scenarios – Creates privacy-safe datasets – Enables systematic coverage – Scales data creation efficiently

Trade-offs: – Generated data may lack real-world complexity – Quality depends on LLM capabilities – May introduce biases from LLM – Validation overhead – Cost of generation at scale

LangChain Implementation: Uses ChatOpenAI for generation with structured output parsing, Pydantic models for validation, and custom metrics.

File Statistics

Total Lines: 772
Classes: 9
Functions: 1
Constants: 0

Module Documentation

"""
Pattern 151: Synthetic Data Generation

Description:
    Generates synthetic training and test data for agent systems using LLMs.
    Creates diverse, high-quality examples with schema validation, quality
    metrics, and coverage analysis.

Components:
    - Data Schema Definitions: Structured format specifications
    - Generation Strategies: Various approaches for data creation
    - Quality Validators: Check data quality and validity
    - Coverage Analyzers: Ensure diverse data distribution
    - Augmentation Engine: Enhance existing datasets

Use Cases:
    - Training data creation for fine-tuning
    - Test case generation for QA
    - Data augmentation for small datasets
    - Edge case generation for robustness testing
    - Privacy-preserving data synthesis

Benefits:
    - Reduces manual data labeling effort
    - Generates diverse test scenarios
    - Creates privacy-safe datasets
    - Enables systematic coverage
    - Scales data creation efficiently

Trade-offs:
    - Generated data may lack real-world complexity
    - Quality depends on LLM capabilities
    - May introduce biases from LLM
    - Validation overhead
    - Cost of generation at scale

LangChain Implementation:
    Uses ChatOpenAI for generation with structured output parsing,
    Pydantic models for validation, and custom metrics.
"""

Purpose: This module implements the Synthetic Data pattern, providing a structured approach to agent design and implementation.

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import json
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from pydantic.BaseModel
from pydantic.Field
from pydantic.validator

Purpose: Additional dependencies for specialized functionality.

Classes

Class: `DataType`

Lines 60-68

Inherits from: Enum

Purpose:

Types of synthetic data

Class: `QualityMetric`

Lines 71-77

Inherits from: Enum

Purpose:

Quality metrics for generated data

Class: `QAPair`

Lines 82-88

Inherits from: BaseModel

Purpose:

Question-Answer pair

Class: `ClassificationExample`

Lines 91-96

Inherits from: BaseModel

Purpose:

Text classification example

Class: `ConversationTurn`

Lines 99-103

Inherits from: BaseModel

Purpose:

Single conversation turn

Class: `Conversation`

Lines 106-110

Inherits from: BaseModel

Purpose:

Multi-turn conversation

Class: `GenerationConfig`

Lines 114-123

Purpose:

Configuration for data generation

Class: `QualityScores`

Lines 127-149

Purpose:

Quality scores for generated data

Methods:

`overall_score(self)`

Lines 135-138

Purpose:

Calculate overall quality score

Parameters:

Returns: float

`to_dict(self)`

Lines 140-149

Purpose:

Convert to dictionary

Parameters:

Returns: <ast.Subscript object at 0x000001A06D852450>

Class: `SyntheticDataGenerator`

Lines 152-631

Purpose:

Agent that generates synthetic training and test data.
Produces high-quality, diverse datasets with validation.

Methods:

`init(self, temperature)`

Lines 158-212

Purpose:

Initialize the synthetic data generator

Parameters: – temperature: Sampling temperature (0-1)