Synthetic Data › Implementation

# Install required packages
# uv pip install openai sentence-transformers scikit-learn numpy matplotlib

import random
from typing import Optional

# Simulated teacher model outputs (in production, replace with actual API calls)
SIMULATED_RESPONSES = {
    "instruction": [
        {
            "instruction": "Explain the concept of gradient descent in machine learning.",
            "response": "Gradient descent is an optimization algorithm that iteratively adjusts model parameters to minimize a loss function. At each step, it computes the gradient of the loss with respect to the parameters and moves in the opposite direction of the gradient by a small amount called the learning rate. The process repeats until the loss converges to a minimum. Variants like stochastic gradient descent (SGD) and Adam use subsets of data or adaptive learning rates to improve convergence speed.",
            "domain": "machine_learning",
            "difficulty": "intermediate",
        },
        {
            "instruction": "What is the difference between precision and recall?",
            "response": "Precision measures what fraction of the model's positive predictions are actually correct: TP / (TP + FP). Recall measures what fraction of actual positives the model correctly identified: TP / (TP + FN). High precision means few false alarms; high recall means few missed detections. The F1 score combines both into a single metric as their harmonic mean. In medical diagnosis, recall is usually prioritized to avoid missing true cases, while in spam filtering, precision may matter more to avoid blocking legitimate email.",
            "domain": "machine_learning",
            "difficulty": "intermediate",
        },
        {
            "instruction": "Describe the water cycle.",
            "response": "The water cycle describes the continuous movement of water through Earth's systems. Water evaporates from oceans, lakes, and land surfaces when heated by the sun. Water vapor rises into the atmosphere, cools, and condenses into clouds through a process called condensation. When enough water accumulates in clouds, it falls as precipitation: rain, snow, sleet, or hail. Precipitation either runs off into rivers and eventually back to the ocean, soaks into the ground to replenish groundwater, or is taken up by plants and returned to the atmosphere through transpiration.",
            "domain": "earth_science",
            "difficulty": "beginner",
        },
        {
            "instruction": "How does HTTPS encryption protect data in transit?",
            "response": "HTTPS uses TLS (Transport Layer Security) to encrypt data between your browser and the server. The process starts with a TLS handshake: the server presents its certificate to prove its identity, and the client and server negotiate an encryption cipher. They then use asymmetric cryptography to securely exchange a symmetric session key. All subsequent data is encrypted using that symmetric key, which is fast and efficient. This prevents eavesdroppers from reading traffic and ensures the server you're talking to is who it claims to be.",
            "domain": "computer_security",
            "difficulty": "intermediate",
        },
        {
            "instruction": "Write a Python function to check if a string is a palindrome.",
            "response": "def is_palindrome(s: str) -> bool:\n    # Normalize: lowercase and remove non-alphanumeric characters\n    cleaned = re.sub(r'[^a-z0-9]', '', s.lower())\n    return cleaned == cleaned[::-1]\n\n# Examples\nprint(is_palindrome('racecar'))   # True\nprint(is_palindrome('A man a plan a canal Panama'))  # True\nprint(is_palindrome('hello'))     # False",
            "domain": "programming",
            "difficulty": "beginner",
        },
        {
            "instruction": "What is photosynthesis and why is it important?",
            "response": "Photosynthesis is the process by which plants, algae, and some bacteria convert light energy, carbon dioxide, and water into glucose and oxygen. The reaction takes place in chloroplasts using the pigment chlorophyll. It matters because it is the primary source of oxygen in Earth's atmosphere and the foundation of almost all food chains on land. Without photosynthesis, the oxygen levels in the atmosphere would decline, and the chemical energy that supports most living organisms would disappear.",
            "domain": "biology",
            "difficulty": "beginner",
        },
        {
            "instruction": "Explain transformer attention in one paragraph.",
            "response": "Transformer attention allows each token in a sequence to selectively gather information from all other tokens. For each position, the model computes a query vector, and each other position contributes a key vector and a value vector. The attention score between two positions is the dot product of their query and key vectors, scaled by the square root of the dimension to prevent magnitude blowup, then passed through softmax to produce weights. These weights are used to compute a weighted sum of the value vectors, producing a context-aware representation for each position.",
            "domain": "machine_learning",
            "difficulty": "advanced",
        },
        {
            "instruction": "What caused the French Revolution?",
            "response": "The French Revolution (1789-1799) arose from a convergence of financial crisis, social inequality, and Enlightenment ideals. France was effectively bankrupt after funding the American Revolution and years of expensive wars. The tax system was deeply unfair: nobles and clergy were largely exempt while commoners bore most of the burden. A series of bad harvests in the late 1780s caused widespread food shortages and rising bread prices, pushing the urban poor to crisis. Enlightenment philosophers like Rousseau and Voltaire had spread ideas about individual rights and the illegitimacy of absolute monarchy, providing intellectual fuel for the uprising.",
            "domain": "history",
            "difficulty": "intermediate",
        },
    ]
}


# Simulate generating synthetic examples from a teacher model
def simulate_teacher_generation(
    instructions: list[str], domain_filter: Optional[str] = None
) -> list[dict]:
    """
    In production, this function would call an API.
    Here we return pre-defined examples filtered by domain.
    """
    examples = SIMULATED_RESPONSES["instruction"]
    if domain_filter:
        examples = [e for e in examples if e["domain"] == domain_filter]
    # Shuffle to simulate non-determinism
    random.seed(42)
    result = random.choices(examples, k=len(instructions))
    return result

def check_length(
    response: str, min_words: int = 20, max_words: int = 500
) -> bool:
    words = len(response.split())
    return min_words <= words <= max_words


def check_repetition(
    response: str, ngram_size: int = 5, threshold: float = 0.3
) -> bool:
    """Returns True if repetition is below threshold (response is NOT repetitive)."""
    words = response.lower().split()
    if len(words) < ngram_size:
        return True
    ngrams = [
        tuple(words[i : i + ngram_size])
        for i in range(len(words) - ngram_size + 1)
    ]
    unique_ratio = len(set(ngrams)) / len(ngrams)
    return unique_ratio >= (1.0 - threshold)


def check_no_refusal(response: str) -> bool:
    """Returns True if the response does not contain refusal phrases."""
    refusal_phrases = [
        "as an ai language model",
        "i cannot provide",
        "i'm not able to",
        "i don't have the ability",
        "i am unable to assist",
    ]
    response_lower = response.lower()
    return not any(phrase in response_lower for phrase in refusal_phrases)


def rule_based_filter(examples: list[dict]) -> tuple[list[dict], dict]:
    """Apply all rule-based filters and return passing examples + stats."""
    passed = []
    stats = {
        "total": len(examples),
        "failed_length": 0,
        "failed_repetition": 0,
        "failed_refusal": 0,
    }

    for ex in examples:
        response = ex["response"]
        if not check_length(response):
            stats["failed_length"] += 1
            continue
        if not check_repetition(response):
            stats["failed_repetition"] += 1
            continue
        if not check_no_refusal(response):
            stats["failed_refusal"] += 1
            continue
        passed.append(ex)

    stats["passed"] = len(passed)
    return passed, stats

import re


def score_response_heuristic(instruction: str, response: str) -> float:
    """
    Heuristic quality scorer (simulates model-based scoring).
    Returns a score from 0.0 to 1.0.
    In production, replace with a judge model API call.
    """
    score = 0.0
    response_words = len(response.split())
    instr_words = len(instruction.split())

    # Length relative to instruction complexity
    if response_words > instr_words * 3:
        score += 0.25
    elif response_words > instr_words:
        score += 0.15

    # Presence of substantive content signals
    if any(c.isdigit() for c in response):
        score += 0.10  # Contains specific numbers/data
    if ":" in response:
        score += 0.10  # Contains structured content
    if response[0].isupper() and response[-1] in ".!?":
        score += 0.10  # Proper sentence structure

    # Vocabulary richness
    words = re.sub(r"[^a-z\s]", "", response.lower()).split()
    if len(words) > 0:
        vocab_richness = len(set(words)) / len(words)
        score += min(0.35, vocab_richness * 0.5)

    return min(1.0, score)


def model_based_filter(
    examples: list[dict], threshold: float = 0.5
) -> tuple[list[dict], list[float]]:
    """Score examples and filter by threshold."""
    scored = []
    scores = []
    for ex in examples:
        score = score_response_heuristic(ex["instruction"], ex["response"])
        scores.append(score)
        if score >= threshold:
            scored.append(ex)
    return scored, scores

from collections import Counter

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity


def compute_diversity_metrics(examples: list[dict]) -> dict:
    """
    Compute diversity metrics for a set of examples.
    Returns average pairwise distance and self-BLEU proxy.
    """
    if len(examples) < 2:
        return {
            "avg_pairwise_distance": 0.0,
            "vocabulary_diversity": 0.0,
            "domain_entropy": 0.0,
        }

    texts = [ex["instruction"] + " " + ex["response"] for ex in examples]

    # TF-IDF-based embedding diversity
    vectorizer = TfidfVectorizer(max_features=500, stop_words="english")
    try:
        tfidf_matrix = vectorizer.fit_transform(texts)
        similarity_matrix = cosine_similarity(tfidf_matrix)
        # Average pairwise distance (1 - similarity) for upper triangle
        n = len(texts)
        distances = []
        for i in range(n):
            for j in range(i + 1, n):
                distances.append(1.0 - similarity_matrix[i, j])
        avg_distance = np.mean(distances) if distances else 0.0
    except Exception:
        avg_distance = 0.0

    # Vocabulary diversity: unique words / total words across all texts
    all_words = " ".join(texts).lower().split()
    vocab_diversity = len(set(all_words)) / len(all_words) if all_words else 0.0

    # Domain entropy
    domains = [ex.get("domain", "unknown") for ex in examples]
    domain_counts = Counter(domains)
    total = sum(domain_counts.values())
    entropy = -sum(
        (c / total) * np.log2(c / total + 1e-9) for c in domain_counts.values()
    )
    max_entropy = np.log2(len(domain_counts)) if len(domain_counts) > 1 else 1.0
    domain_entropy = entropy / max_entropy if max_entropy > 0 else 0.0

    return {
        "avg_pairwise_distance": avg_distance,
        "vocabulary_diversity": vocab_diversity,
        "domain_entropy": domain_entropy,
        "n_domains": len(domain_counts),
        "domain_distribution": dict(domain_counts),
    }

# Step 1: Simulate generation from a teacher model
seed_instructions = [
    "Explain the concept of gradient descent in machine learning.",
    "What is the difference between precision and recall?",
    "Describe the water cycle.",
    "How does HTTPS encryption protect data in transit?",
    "Write a Python function to check if a string is a palindrome.",
    "What is photosynthesis and why is it important?",
    "Explain transformer attention in one paragraph.",
    "What caused the French Revolution?",
]

generated_examples = simulate_teacher_generation(seed_instructions)

# Step 2: Rule-based filtering
rule_passed, rule_stats = rule_based_filter(generated_examples)

# Step 3: Model-based quality scoring and filtering
quality_passed, quality_scores = model_based_filter(rule_passed, threshold=0.5)

# Step 4: Diversity analysis
diversity_before = compute_diversity_metrics(generated_examples)
diversity_after = compute_diversity_metrics(quality_passed)

# Step 5: Summary statistics
pipeline_summary = {
    "generated": len(generated_examples),
    "after_rule_filter": len(rule_passed),
    "after_quality_filter": len(quality_passed),
    "rule_filter_stats": rule_stats,
    "diversity_before": diversity_before,
    "diversity_after": diversity_after,
    "avg_quality_score": np.mean(quality_scores) if quality_scores else 0.0,
    "min_quality_score": np.min(quality_scores) if quality_scores else 0.0,
    "max_quality_score": np.max(quality_scores) if quality_scores else 0.0,
}

import numpy as np


def softmax(logits: np.ndarray, temperature: float = 1.0) -> np.ndarray:
    """Compute softmax with temperature scaling."""
    scaled = logits / temperature
    exp_logits = np.exp(
        scaled - np.max(scaled)
    )  # Subtract max for numerical stability
    return exp_logits / exp_logits.sum()


def kl_divergence(p: np.ndarray, q: np.ndarray, eps: float = 1e-9) -> float:
    """KL divergence KL(p || q). Measures how much p differs from q."""
    p = p + eps
    q = q + eps
    return float(np.sum(p * np.log(p / q)))


def distillation_loss(
    student_logits: np.ndarray,
    teacher_logits: np.ndarray,
    true_label_idx: int,
    temperature: float = 2.0,
    alpha: float = 0.5,
) -> dict:
    """
    Compute the combined distillation loss.

    Args:
        student_logits: Student model's raw output scores (before softmax)
        teacher_logits: Teacher model's raw output scores
        true_label_idx: Index of the correct label
        temperature: Softening temperature (T > 1 reveals dark knowledge)
        alpha: Weight for distillation loss vs hard label loss

    Returns:
        Dictionary with loss components
    """
    vocab_size = len(student_logits)

    # Hard label cross-entropy loss (standard supervised learning)
    student_probs = softmax(student_logits, temperature=1.0)
    hard_ce_loss = -np.log(student_probs[true_label_idx] + 1e-9)

    # Soft targets from teacher at temperature T
    teacher_soft = softmax(teacher_logits, temperature=temperature)
    student_soft = softmax(student_logits, temperature=temperature)

    # KL divergence between teacher and student soft distributions
    # Multiplied by T^2 to compensate for the temperature scaling of gradients
    soft_kl_loss = (temperature**2) * kl_divergence(teacher_soft, student_soft)

    # Combined loss
    combined_loss = (1 - alpha) * hard_ce_loss + alpha * soft_kl_loss

    return {
        "hard_ce_loss": hard_ce_loss,
        "soft_kl_loss": soft_kl_loss,
        "combined_loss": combined_loss,
        "teacher_soft_probs": teacher_soft,
        "student_soft_probs": student_soft,
    }


# Example: 4-token vocabulary, teacher is more confident than student
np.random.seed(42)
vocab_size = 4
true_label = 0  # Correct token index

# Teacher: very confident about token 0, some mass on token 1
teacher_logits = np.array([3.5, 1.2, -0.5, -1.0])

# Student: less confident (smaller model, earlier in training)
student_logits = np.array([1.8, 1.5, 0.3, -0.2])

# Compute losses at different temperatures
results_by_temp = {}
for temp in [1.0, 2.0, 4.0]:
    results_by_temp[temp] = distillation_loss(
        student_logits, teacher_logits, true_label, temperature=temp, alpha=0.5
    )