Principles of Synthetic Data Generation

# Load and inspect training data distributions using pandas
# Understanding class balance and text length helps guide generation strategy
import pandas as pd

# Cost comparison: Human annotation vs. LLM-generated synthetic data
cost_data = {
 "Method": [
 "Expert annotation (complex NLP)",
 "Crowd annotation (simple classification)",
 "GPT-4o synthetic generation",
 "GPT-4o-mini synthetic generation",
 "Llama 3.1 70B (self-hosted)"
 ],
 "Cost per Example": ["$5.00 - $20.00", "$0.10 - $0.50", "$0.005 - $0.02",
 "$0.001 - $0.005", "$0.0005 - $0.002"],
 "Speed (examples/hour)": ["10-30", "50-200", "1,000-5,000",
 "5,000-20,000", "2,000-10,000"],
 "Quality Control": ["Inter-annotator agreement", "Majority vote",
 "LLM-as-judge + sampling", "LLM-as-judge + sampling",
 "LLM-as-judge + sampling"],
 "Best For": ["Gold eval sets", "Large simple tasks",
 "Complex instruction data", "High-volume generation",
 "Privacy-sensitive domains"]
}

df = pd.DataFrame(cost_data)
print(df.to_string(index=False))

# Define the generate_preference_pair function
# This handles the core processing logic
def generate_preference_pair(instruction: str) -> dict:
 """Generate a preference pair: one good response and one flawed response."""

 # Generate the high-quality (chosen) response
 chosen_resp = client.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "system", "content": "You are an expert assistant. "
 "Provide a thorough, accurate, and helpful response."},
 {"role": "user", "content": instruction}
 ],
 temperature=0.7
 )

 # Generate the lower-quality (rejected) response with induced flaws
 rejected_resp = client.chat.completions.create(
 model="gpt-4o",
 messages=[
 {"role": "system", "content": "You are a mediocre assistant. "
 "Provide a response that is somewhat helpful but has issues: "
 "it may be vague, miss key details, include minor inaccuracies, "
 "or lack proper structure. Do NOT be obviously wrong."},
 {"role": "user", "content": instruction}
 ],
 temperature=1.0
 )

 return {
 "prompt": instruction,
 "chosen": chosen_resp.choices[0].message.content,
 "rejected": rejected_resp.choices[0].message.content
 }

# Example usage
pair = generate_preference_pair(
 "Explain the difference between L1 and L2 regularization."
)
print(f"Chosen length: {len(pair['chosen'])} chars")
print(f"Rejected length: {len(pair['rejected'])} chars")

# Define data structures for synthetic data generation configuration
# Specify target distributions, quality thresholds, and domain constraints
from dataclasses import dataclass
from typing import List

@dataclass
class QualityMetrics:
 """Metrics for evaluating synthetic data quality."""
 diversity_score: float # 0-1: Variety across topics, formats, styles
 accuracy_score: float # 0-1: Factual correctness (sampled + verified)
 consistency_score: float # 0-1: No contradictions, stable formatting
 naturalness_score: float # 0-1: Resembles human-written text

 @property
 def composite_score(self) -> float:
 """Weighted composite: accuracy matters most for training data."""
 weights = {
 "diversity": 0.25,
 "accuracy": 0.35,
 "consistency": 0.20,
 "naturalness": 0.20
 }
 return (
 weights["diversity"] * self.diversity_score +
 weights["accuracy"] * self.accuracy_score +
 weights["consistency"] * self.consistency_score +
 weights["naturalness"] * self.naturalness_score
 )

 def passes_threshold(self, min_score: float = 0.7) -> bool:
 """Check if all individual dimensions meet minimum threshold."""
 return all(
 score >= min_score
 for score in [
 self.diversity_score,
 self.accuracy_score,
 self.consistency_score,
 self.naturalness_score
 ]
 )

# Example evaluation
metrics = QualityMetrics(
 diversity_score=0.82,
 accuracy_score=0.91,
 consistency_score=0.78,
 naturalness_score=0.85
)
print(f"Composite score: {metrics.composite_score:.3f}")
print(f"Passes 0.7 threshold: {metrics.passes_threshold()}")
print(f"Passes 0.8 threshold: {metrics.passes_threshold(0.8)}")

# Deduplicate synthetic examples using content hashing
# Remove near-duplicates to ensure diversity in the generated dataset
import hashlib
from collections import Counter

def measure_diversity(texts: list[str]) -> dict:
 """Measure lexical and structural diversity of generated texts."""
 # Unique n-gram ratio (type-token ratio for bigrams)
 all_bigrams = []
 for text in texts:
 words = text.lower().split()
 bigrams = [f"{w1} {w2}" for w1, w2 in zip(words, words[1:])]
 all_bigrams.extend(bigrams)

 bigram_counts = Counter(all_bigrams)
 unique_ratio = len(bigram_counts) / max(len(all_bigrams), 1)

 # Sentence length variance
 sent_lengths = []
 for text in texts:
 sentences = text.split(".")
 sent_lengths.extend(len(s.split()) for s in sentences if s.strip())

 length_variance = (
 sum((l - sum(sent_lengths) / len(sent_lengths)) ** 2
 for l in sent_lengths) / max(len(sent_lengths), 1)
 )

 # Near-duplicate detection via hashing first 100 chars
 hashes = [hashlib.md5(t[:100].encode()).hexdigest() for t in texts]
 unique_starts = len(set(hashes)) / max(len(hashes), 1)

 return {
 "unique_bigram_ratio": round(unique_ratio, 4),
 "sentence_length_variance": round(length_variance, 2),
 "unique_opening_ratio": round(unique_starts, 4),
 "num_texts": len(texts)
 }

# Compare human vs. synthetic data diversity
human_texts = [
 "The quick brown fox jumps over the lazy dog near the stream.",
 "I have been working on this project since last Tuesday morning.",
 "Why do cats always land on their feet? It is a common question.",
 "The budget for Q3 looks tight; we need to cut infrastructure costs."
]
synthetic_texts = [
 "Certainly! Here is a comprehensive overview of the topic at hand.",
 "Certainly! Let me provide a detailed explanation of the concept.",
 "Certainly! I would be happy to explain this topic in detail.",
 "Certainly! Here is a thorough analysis of the subject matter."
]

print("Human data:", measure_diversity(human_texts))
print("Synthetic data:", measure_diversity(synthetic_texts))