LLM Data Generator in Machine Learning

import openai
import json
import random
import time
from typing import List, Dict, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed

client = openai.OpenAI()  # Uses OPENAI_API_KEY env var

# Seed tasks: 8-15 diverse, high-quality human-written examples
SEED_TASKS = [
    {
        "instruction": "Write a Python function that checks if a string is a valid palindrome, ignoring spaces and punctuation.",
        "input": "",
        "output": "def is_palindrome(s: str) -> bool:\n    cleaned = ''.join(c.lower() for c in s if c.isalnum())\n    return cleaned == cleaned[::-1]\n\n# Examples:\n# is_palindrome('A man, a plan, a canal: Panama') -> True\n# is_palindrome('hello') -> False"
    },
    {
        "instruction": "Classify the sentiment of the following product review as positive, negative, or neutral.",
        "input": "The delivery was super fast but the product quality is disappointing. Wouldn't buy again.",
        "output": "Negative. While the reviewer acknowledges fast delivery (positive aspect), the overall sentiment is negative due to disappointing product quality and stated intent not to repurchase."
    },
    {
        "instruction": "Explain the concept of database sharding to a junior developer.",
        "input": "",
        "output": "Database sharding is splitting a large database into smaller, faster pieces called shards. Imagine a library so big that one building can't hold all the books -- so you build multiple buildings, each holding books for certain categories. Each shard holds a subset of the data (e.g., users A-M in shard 1, N-Z in shard 2). This improves performance because queries only search one shard instead of the entire database. The tricky part is choosing a good shard key and handling queries that span multiple shards."
    },
]

GENERATION_PROMPT = """Below are examples of instruction-input-output triples for training an AI assistant.

{examples}

Generate a NEW, creative instruction-input-output triple that is DIFFERENT from the examples above.
Requirements:
- The instruction should be clear, specific, and self-contained
- Cover a different topic or skill than the examples
- The output should be detailed and high-quality
- Vary the format: some with input, some without
- Include diverse task types: coding, analysis, explanation, creative writing, math, etc.

Respond with ONLY a valid JSON object:
{{"instruction": "...", "input": "...", "output": "..."}}"""

def generate_single_example(
    task_pool: List[Dict],
    model: str = "gpt-4o-mini",
    temperature: float = 0.9,
) -> Optional[Dict]:
    """Generate one instruction-response pair using Self-Instruct."""
    # Sample 3-5 diverse examples from the pool
    n_demos = min(random.randint(3, 5), len(task_pool))
    demos = random.sample(task_pool, n_demos)
    
    examples_text = "\n\n".join([
        f"Example {i+1}:\n"
        f"Instruction: {d['instruction']}\n"
        f"Input: {d.get('input', '')}\n"
        f"Output: {d['output']}"
        for i, d in enumerate(demos)
    ])
    
    prompt = GENERATION_PROMPT.format(examples=examples_text)
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            max_tokens=2048,
        )
        content = response.choices[0].message.content.strip()
        
        # Handle markdown code blocks
        if content.startswith("```"):
            content = content.split("```")[1]
            if content.startswith("json"):
                content = content[4:]
        
        parsed = json.loads(content)
        
        # Validate required fields
        if not all(k in parsed for k in ["instruction", "output"]):
            return None
        if len(parsed["instruction"]) < 10 or len(parsed["output"]) < 20:
            return None
        
        parsed.setdefault("input", "")
        return parsed
        
    except (json.JSONDecodeError, KeyError, IndexError):
        return None

def generate_dataset(
    n_examples: int = 1000,
    model: str = "gpt-4o-mini",
    max_workers: int = 10,
    output_path: str = "synthetic_data.json",
) -> List[Dict]:
    """Generate a synthetic instruction dataset using Self-Instruct."""
    task_pool = SEED_TASKS.copy()
    generated = []
    failed = 0
    
    print(f"Generating {n_examples} examples with {model}...")
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all generation tasks
        futures = []
        for _ in range(int(n_examples * 1.3)):  # Over-generate by 30% for filtering
            futures.append(
                executor.submit(generate_single_example, task_pool, model)
            )
        
        for future in as_completed(futures):
            result = future.result()
            if result is not None:
                generated.append(result)
                task_pool.append(result)  # Grow the pool for diversity
                
                if len(generated) % 100 == 0:
                    print(f"  Generated: {len(generated)}/{n_examples}")
                
                if len(generated) >= n_examples:
                    break
            else:
                failed += 1
    
    print(f"\nGeneration complete: {len(generated)} examples ({failed} failed)")
    
    with open(output_path, "w") as f:
        json.dump(generated, f, indent=2, ensure_ascii=False)
    
    return generated

# Generate 1000 synthetic instruction pairs
dataset = generate_dataset(n_examples=1000, model="gpt-4o-mini")
print(f"Estimated cost: ~${len(dataset) * 0.003:.2f}")

import openai
import json
import random
from typing import List, Dict, Optional

client = openai.OpenAI()

# Evolution operator prompts (from WizardLM methodology)
EVOL_OPERATORS = {
    "deepen": """Rewrite the following instruction to require deeper thinking and multi-step reasoning.
Add complexity by requiring analysis, comparison, or synthesis of multiple concepts.

Original: {instruction}

Rewrite the instruction to be MORE COMPLEX (deeper reasoning required).
Respond with ONLY the rewritten instruction, nothing else.""",

    "broaden": """Rewrite the following instruction to cover a broader scope or relate to additional topics.
Expand the scope while keeping it answerable.

Original: {instruction}

Rewrite the instruction to be BROADER in scope.
Respond with ONLY the rewritten instruction, nothing else.""",

    "constrain": """Add specific constraints or requirements to the following instruction.
Examples: word limits, format requirements, specific perspectives, language style.

Original: {instruction}

Rewrite the instruction with ADDED CONSTRAINTS (format, length, perspective, etc.).
Respond with ONLY the rewritten instruction, nothing else.""",

    "concretize": """Make the following instruction more specific and concrete.
Replace general concepts with specific examples, real-world scenarios, or particular technologies.

Original: {instruction}

Rewrite the instruction to be MORE SPECIFIC and CONCRETE.
Respond with ONLY the rewritten instruction, nothing else.""",
}

def evolve_instruction(
    instruction: str,
    operator: Optional[str] = None,
    model: str = "gpt-4o-mini",
) -> Optional[str]:
    """Apply an evolution operator to an instruction."""
    if operator is None:
        operator = random.choice(list(EVOL_OPERATORS.keys()))
    
    prompt = EVOL_OPERATORS[operator].format(instruction=instruction)
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=512,
        )
        evolved = response.choices[0].message.content.strip()
        
        # Sanity checks
        if len(evolved) < 10 or len(evolved) > 2000:
            return None
        if evolved.lower() == instruction.lower():
            return None  # No evolution happened
        
        return evolved
    except Exception:
        return None

def generate_response(
    instruction: str,
    model: str = "gpt-4o",
) -> Optional[str]:
    """Generate a high-quality response for an evolved instruction."""
    system_msg = (
        "You are a helpful AI assistant. Provide detailed, accurate, "
        "and well-structured responses. Use examples where appropriate."
    )
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system_msg},
                {"role": "user", "content": instruction},
            ],
            temperature=0.3,  # Lower temp for response quality
            max_tokens=2048,
        )
        return response.choices[0].message.content.strip()
    except Exception:
        return None

def evol_instruct_pipeline(
    seed_instructions: List[str],
    n_rounds: int = 3,
    model_evolve: str = "gpt-4o-mini",
    model_respond: str = "gpt-4o",
) -> List[Dict]:
    """Run multi-round Evol-Instruct on seed instructions."""
    results = []
    current_instructions = seed_instructions.copy()
    
    for round_num in range(n_rounds):
        print(f"\n=== Evolution Round {round_num + 1}/{n_rounds} ===")
        evolved_instructions = []
        
        for i, instr in enumerate(current_instructions):
            evolved = evolve_instruction(instr, model=model_evolve)
            if evolved:
                evolved_instructions.append(evolved)
                
                # Generate response with stronger model
                response = generate_response(evolved, model=model_respond)
                if response:
                    results.append({
                        "instruction": evolved,
                        "input": "",
                        "output": response,
                        "evolution_round": round_num + 1,
                        "source_instruction": instr,
                    })
            
            if (i + 1) % 10 == 0:
                print(f"  Processed {i + 1}/{len(current_instructions)}")
        
        current_instructions = evolved_instructions
        print(f"  Round {round_num + 1}: {len(evolved_instructions)} evolved instructions")
    
    print(f"\nTotal examples generated: {len(results)}")
    return results

# Example usage
seeds = [
    "Explain what a REST API is.",
    "Write a function to sort a list.",
    "What is the difference between SQL and NoSQL?",
    "Summarize the benefits of cloud computing.",
]

evolved_data = evol_instruct_pipeline(seeds, n_rounds=3)

# Save results
with open("evolved_instructions.json", "w") as f:
    json.dump(evolved_data, f, indent=2, ensure_ascii=False)

print(f"\nSample evolved instruction (round 3):")
round3 = [d for d in evolved_data if d["evolution_round"] == 3]
if round3:
    print(f"  {round3[0]['instruction'][:150]}...")

import openai
import json
import numpy as np
from typing import List, Dict, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
from collections import defaultdict

client = openai.OpenAI()

JUDGE_PROMPT = """You are evaluating the quality of an instruction-response pair for training an AI assistant.

Rate this example on THREE criteria (1-5 each):
1. **Instruction Quality**: Is the instruction clear, specific, and non-trivial?
2. **Response Accuracy**: Is the response correct, complete, and helpful?
3. **Educational Value**: Would training on this example improve an AI assistant?

Instruction: {instruction}
Input: {input}
Response: {response}

Respond with ONLY a JSON object:
{{"instruction_quality": <1-5>, "response_accuracy": <1-5>, "educational_value": <1-5>, "overall": <1-5>, "reason": "<brief explanation>"}}"""

def score_example(example: Dict, model: str = "gpt-4o-mini") -> Tuple[Dict, Dict]:
    """Score a single example using LLM-as-judge."""
    prompt = JUDGE_PROMPT.format(
        instruction=example["instruction"],
        input=example.get("input", ""),
        response=example["output"],
    )
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            max_tokens=300,
        )
        scores = json.loads(response.choices[0].message.content)
        return example, scores
    except Exception:
        return example, {"overall": 0, "reason": "scoring_failed"}

def rouge_l_similarity(text1: str, text2: str) -> float:
    """Compute ROUGE-L F1 similarity between two texts."""
    words1 = text1.lower().split()
    words2 = text2.lower().split()
    
    if not words1 or not words2:
        return 0.0
    
    # LCS computation
    m, n = len(words1), len(words2)
    dp = [[0] * (n + 1) for _ in range(m + 1)]
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if words1[i-1] == words2[j-1]:
                dp[i][j] = dp[i-1][j-1] + 1
            else:
                dp[i][j] = max(dp[i-1][j], dp[i][j-1])
    
    lcs_len = dp[m][n]
    precision = lcs_len / n if n > 0 else 0
    recall = lcs_len / m if m > 0 else 0
    
    if precision + recall == 0:
        return 0.0
    return 2 * precision * recall / (precision + recall)

def deduplicate(
    examples: List[Dict],
    threshold: float = 0.7,
) -> List[Dict]:
    """Remove near-duplicate instructions using ROUGE-L similarity."""
    deduplicated = []
    
    for example in examples:
        is_duplicate = False
        for existing in deduplicated:
            sim = rouge_l_similarity(
                example["instruction"], existing["instruction"]
            )
            if sim > threshold:
                is_duplicate = True
                break
        
        if not is_duplicate:
            deduplicated.append(example)
    
    return deduplicated

def filter_and_deduplicate(
    dataset: List[Dict],
    min_score: int = 4,
    rouge_threshold: float = 0.7,
    max_workers: int = 15,
) -> List[Dict]:
    """Full quality filtering and deduplication pipeline."""
    print(f"Starting with {len(dataset)} examples")
    
    # Stage 1: Basic length and format filters
    length_filtered = [
        ex for ex in dataset
        if 10 < len(ex["instruction"]) < 2000
        and 20 < len(ex["output"]) < 10000
        and not ex["instruction"].strip().startswith("As an AI")
    ]
    print(f"After length/format filter: {len(length_filtered)}")
    
    # Stage 2: LLM-as-judge quality scoring
    scored = []
    score_distribution = defaultdict(int)
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(score_example, ex): ex
            for ex in length_filtered
        }
        for future in as_completed(futures):
            example, scores = future.result()
            overall = scores.get("overall", 0)
            score_distribution[overall] += 1
            
            if overall >= min_score:
                example["quality_scores"] = scores
                scored.append(example)
    
    print(f"After quality filter (>={min_score}): {len(scored)}")
    print(f"Score distribution: {dict(sorted(score_distribution.items()))}")
    
    # Stage 3: Deduplication
    final = deduplicate(scored, threshold=rouge_threshold)
    print(f"After deduplication (ROUGE-L < {rouge_threshold}): {len(final)}")
    
    retention_rate = len(final) / len(dataset) * 100
    print(f"\nOverall retention: {retention_rate:.1f}%")
    
    return final

# Usage
raw_dataset = json.load(open("synthetic_data.json"))
clean_dataset = filter_and_deduplicate(raw_dataset, min_score=4)
json.dump(clean_dataset, open("clean_synthetic_data.json", "w"), indent=2)

import openai
import json
from typing import List, Dict, Optional
from pydantic import BaseModel, Field
from enum import Enum

client = openai.OpenAI()

# Define schema for domain-specific generation
class DifficultyLevel(str, Enum):
    easy = "easy"
    medium = "medium"
    hard = "hard"

class DomainExample(BaseModel):
    instruction: str = Field(description="The user's question or task")
    context: str = Field(default="", description="Optional context or background")
    response: str = Field(description="Detailed expert response")
    domain: str = Field(description="Specific sub-domain")
    difficulty: DifficultyLevel
    tags: List[str] = Field(description="Topic tags for categorization")

# Domain-specific generation for Indian e-commerce customer support
DOMAIN_CONFIG = {
    "name": "Indian E-commerce Customer Support",
    "sub_domains": [
        "order tracking", "returns and refunds", "payment issues",
        "product quality", "delivery complaints", "account management",
        "COD (cash on delivery)", "EMI and financing", "seller disputes",
    ],
    "system_prompt": """You are generating training data for an AI customer support
assistant for an Indian e-commerce platform (similar to Flipkart/Amazon India).

Generate realistic customer queries and expert responses that:
- Use natural Indian English (include common phrases like "kindly", "do the needful")
- Reference INR amounts, Indian addresses, PIN codes
- Cover real scenarios: COD issues, UPI payment failures, festival sale problems
- Include both polite and frustrated customer tones
- Responses should be empathetic, solution-oriented, and policy-aware""",
    "few_shot_examples": [
        {
            "instruction": "My order #FKT892341 was supposed to arrive yesterday but tracking shows it's stuck in Bangalore hub for 3 days. I paid extra for express delivery. This is very frustrating.",
            "context": "Customer ordered electronics worth INR 15,999 with express delivery (INR 99 extra). Order placed 5 days ago.",
            "response": "I completely understand your frustration, and I sincerely apologize for the delay with your order #FKT892341. Express delivery delays are unacceptable, and I want to help resolve this right away.\n\nI can see your package is currently at our Bangalore hub. Let me take these steps:\n1. I'm escalating this to our logistics team for priority dispatch today\n2. I'm initiating a full refund of the INR 99 express delivery charge\n3. You'll receive an updated tracking link within 2 hours\n\nIf the package doesn't reach you within 24 hours, I'll also apply a INR 500 store credit to your account as compensation. Would you like me to proceed with this?",
            "domain": "delivery complaints",
            "difficulty": "medium",
            "tags": ["express delivery", "logistics delay", "refund", "escalation"]
        }
    ]
}

def generate_domain_examples(
    config: Dict,
    n_examples: int = 100,
    model: str = "gpt-4o",
) -> List[Dict]:
    """Generate domain-specific training examples with schema validation."""
    results = []
    schema_str = json.dumps(DomainExample.model_json_schema(), indent=2)
    
    for i in range(n_examples):
        sub_domain = config["sub_domains"][i % len(config["sub_domains"])]
        difficulty = ["easy", "medium", "hard"][i % 3]
        
        few_shot = json.dumps(config["few_shot_examples"][0], indent=2)
        
        prompt = f"""Generate a realistic {config['name']} training example.

Sub-domain: {sub_domain}
Difficulty: {difficulty}

Here's an example of the format:
{few_shot}

Generate a NEW example that is different from the above.
The output MUST be valid JSON matching this schema:
{schema_str}

Respond with ONLY the JSON object."""
        
        try:
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": config["system_prompt"]},
                    {"role": "user", "content": prompt},
                ],
                temperature=0.8,
                max_tokens=1500,
            )
            
            content = response.choices[0].message.content.strip()
            if content.startswith("```"):
                content = content.split("```")[1]
                if content.startswith("json"):
                    content = content[4:]
                content = content.strip()
            
            parsed = json.loads(content)
            validated = DomainExample(**parsed)
            results.append(validated.model_dump())
            
            if (i + 1) % 20 == 0:
                print(f"Generated {i + 1}/{n_examples} examples")
                
        except Exception as e:
            print(f"  Failed example {i + 1}: {e}")
            continue
    
    print(f"\nSuccessfully generated {len(results)}/{n_examples} examples")
    return results

# Generate domain-specific dataset
dataset = generate_domain_examples(DOMAIN_CONFIG, n_examples=100)

with open("ecommerce_support_data.json", "w") as f:
    json.dump(dataset, f, indent=2, ensure_ascii=False)

# Analyze distribution
from collections import Counter
domains = Counter(ex["domain"] for ex in dataset)
print(f"\nDomain distribution: {dict(domains)}")

# DataDreamer configuration for synthetic data generation
# pip install datadreamer

from datadreamer import DataDreamer
from datadreamer.llms import OpenAI
from datadreamer.steps import (
    DataFromPrompt,
    ProcessWithPrompt,
    FilterByScore,
)

with DataDreamer("./output"):
    # Step 1: Generate instructions
    llm = OpenAI(model_name="gpt-4o-mini")
    
    instructions = DataFromPrompt(
        name="Generate Instructions",
        llm=llm,
        n=5000,
        prompt=(
            "Generate a unique instruction for an AI assistant. "
            "Cover diverse topics: coding, writing, analysis, math, science. "
            "Output ONLY the instruction, nothing else."
        ),
        temperature=0.9,
    )
    
    # Step 2: Generate responses with stronger model
    strong_llm = OpenAI(model_name="gpt-4o")
    
    responses = ProcessWithPrompt(
        name="Generate Responses",
        llm=strong_llm,
        inputs=instructions,
        prompt="{instruction}",
        temperature=0.3,
    )
    
    # Step 3: Quality filtering
    filtered = FilterByScore(
        name="Quality Filter",
        llm=llm,
        inputs=responses,
        min_score=4,
        scoring_prompt=(
            "Rate this instruction-response pair 1-5 for quality. "
            "Respond with only a number."
        ),
    )
    
    # Export
    filtered.export_to_hf_dataset("./synthetic_dataset")