LLM-Powered Data Generation Pipelines

# Generate diverse instructions using the Self-Instruct pipeline
# Seed tasks bootstrap generation; the LLM expands the instruction set
import json
import random
from openai import OpenAI

client = OpenAI()

# Seed instructions (in practice, use 150-200 diverse examples)
SEED_INSTRUCTIONS = [
 "Write a Python function that reverses a linked list.",
 "Explain the difference between TCP and UDP protocols.",
 "Summarize the key principles of object-oriented programming.",
 "Convert the following CSV data into a JSON format.",
 "What are the pros and cons of microservices architecture?"
]

def self_instruct_generate(
 seed_pool: list[str],
 num_examples: int = 8,
 model: str = "gpt-4o"
) -> dict:
 """Generate a new instruction-response pair using Self-Instruct."""
 # Step 1: Sample from the seed pool
 sampled = random.sample(seed_pool, min(num_examples, len(seed_pool)))
 examples_text = "\n".join(f"{i+1}. {inst}" for i, inst in enumerate(sampled))

 # Step 2: Generate a new instruction
 gen_prompt = f"""Here are {len(sampled)} example task instructions:

{examples_text}

Generate a completely NEW and DIFFERENT task instruction that:
- Is distinct from all the examples above
- Is specific and actionable
- Can be answered in a single response
- Covers a different topic or skill

New instruction:"""

 gen_response = client.chat.completions.create(
 model=model,
 messages=[{"role": "user", "content": gen_prompt}],
 temperature=1.0,
 max_tokens=200
 )
 new_instruction = gen_response.choices[0].message.content.strip()

 # Step 3: Generate the response
 resp_response = client.chat.completions.create(
 model=model,
 messages=[
 {"role": "system", "content": "Provide a thorough, accurate, "
 "and well-structured response to the following instruction."},
 {"role": "user", "content": new_instruction}
 ],
 temperature=0.7,
 max_tokens=1024
 )
 response_text = resp_response.choices[0].message.content.strip()

 return {
 "instruction": new_instruction,
 "response": response_text,
 "source": "self-instruct",
 "seed_count": len(sampled)
 }

# Generate a batch
pool = SEED_INSTRUCTIONS.copy()
generated = []
for i in range(3):
 pair = self_instruct_generate(pool)
 generated.append(pair)
 pool.append(pair["instruction"]) # Add back to pool
 print(f"Generated {i+1}: {pair['instruction'][:60]}...")

# Generate multi-turn synthetic conversations around a topic and persona
# Each turn builds on the previous context to create realistic dialogue
def generate_conversation(
 topic: str,
 persona: str,
 num_turns: int = 4,
 model: str = "gpt-4o"
) -> list[dict]:
 """Generate a multi-turn conversation with natural follow-ups."""
 system_msg = f"""You are simulating a realistic conversation between a
user and an AI assistant. The user has the following persona: {persona}

Topic: {topic}

Generate a natural {num_turns}-turn conversation where:
- Each user message builds on the previous assistant response
- The user asks increasingly specific follow-up questions
- The assistant provides detailed, helpful responses
- The conversation feels natural, not scripted

Format each turn as:
USER: [message]
ASSISTANT: [response]"""

 # Send chat completion request to the API
 response = client.chat.completions.create(
 model=model,
 messages=[{"role": "user", "content": system_msg}],
 temperature=0.85,
 max_tokens=2048
 )

 # Parse turns from the generated conversation
 text = response.choices[0].message.content
 turns = []
 current_role = None
 current_text = []

 for line in text.split("\n"):
 if line.startswith("USER:"):
 if current_role:
 turns.append({"role": current_role,
 "content": "\n".join(current_text).strip()})
 current_role = "user"
 current_text = [line.replace("USER:", "").strip()]
 elif line.startswith("ASSISTANT:"):
 if current_role:
 turns.append({"role": current_role,
 "content": "\n".join(current_text).strip()})
 current_role = "assistant"
 current_text = [line.replace("ASSISTANT:", "").strip()]
 else:
 current_text.append(line)

 if current_role:
 turns.append({"role": current_role,
 "content": "\n".join(current_text).strip()})

 return turns

# Generate diverse conversations
conversations = [
 generate_conversation(
 "optimizing PostgreSQL queries",
 "junior backend developer with 1 year experience"
 ),
 generate_conversation(
 "building a recommendation system",
 "data scientist transitioning from academia to industry"
 ),
]
for i, conv in enumerate(conversations):
 print(f"Conversation {i+1}: {len(conv)} turns")
 for turn in conv[:2]:
 print(f" {turn['role']}: {turn['content'][:60]}...")

# Implement Evol-Instruct to evolve simple instructions into complex ones
# Successive transformations add constraints, reasoning steps, and edge cases
import itertools

PERSONA_DIMENSIONS = {
 "expertise": ["beginner", "intermediate", "senior", "expert"],
 "role": [
 "software engineer", "data scientist", "product manager",
 "student", "researcher", "DevOps engineer"
 ],
 "communication_style": [
 "concise and direct",
 "detailed and thorough",
 "casual and conversational",
 "formal and precise"
 ],
 "context": [
 "working on a startup MVP",
 "maintaining a legacy enterprise system",
 "preparing for a technical interview",
 "writing a research paper",
 "building a side project"
 ]
}

def build_persona(dimensions: dict) -> str:
 """Construct a persona description from dimension choices."""
 return (
 f"A {dimensions['expertise']}-level {dimensions['role']} who "
 f"communicates in a {dimensions['communication_style']} manner. "
 f"Currently {dimensions['context']}."
 )

def generate_with_persona(persona: str, topic: str) -> dict:
 """Generate an instruction from a specific persona's perspective."""
 prompt = f"""You are role-playing as the following persona:
{persona}

Given this persona, write a realistic question or task instruction
that this person would actually ask about: {topic}

The question should reflect the persona's expertise level,
communication style, and current context. Be authentic.

Question:"""

 # Send chat completion request to the API
 response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{"role": "user", "content": prompt}],
 temperature=0.9,
 max_tokens=200
 )

 return {
 "persona": persona,
 "topic": topic,
 # Extract the generated message from the API response
 "instruction": response.choices[0].message.content.strip()
 }

# Generate diverse data by sampling persona combinations
def sample_personas(n: int = 10) -> list[dict]:
 """Sample n diverse persona combinations."""
 personas = []
 for _ in range(n):
 dims = {
 key: random.choice(values)
 for key, values in PERSONA_DIMENSIONS.items()
 }
 personas.append(dims)
 return personas

persona_samples = sample_personas(5)
for dims in persona_samples:
 persona = build_persona(dims)
 result = generate_with_persona(persona, "database indexing")
 print(f"Persona: {persona[:60]}...")
 print(f" Q: {result['instruction'][:70]}...")
 print()

# Define the domain_grounded_generation function
# This handles the core processing logic
def domain_grounded_generation(
 document: str,
 domain: str,
 num_pairs: int = 3,
 model: str = "gpt-4o"
) -> list[dict]:
 """Generate QA pairs grounded in a domain document."""
 prompt = f"""You are an expert in {domain}. Given the following document,
generate {num_pairs} question-answer pairs that test understanding of the
key concepts. Each question should:
- Be answerable from the document content
- Range from factual recall to analytical reasoning
- Use proper domain terminology
- Be relevant to a practitioner in this field

Document:
{document[:3000]}

Generate exactly {num_pairs} pairs in this format:
Q1: [question]
A1: [detailed answer with references to the document]
Q2: ...
A2: ..."""

 # Send chat completion request to the API
 response = client.chat.completions.create(
 model=model,
 messages=[{"role": "user", "content": prompt}],
 temperature=0.7,
 max_tokens=2048
 )

 # Parse QA pairs
 text = response.choices[0].message.content
 pairs = []
 lines = text.split("\n")
 current_q, current_a = None, None

 for line in lines:
 if line.startswith("Q") and ":" in line[:4]:
 if current_q and current_a:
 pairs.append({"question": current_q, "answer": current_a})
 current_q = line.split(":", 1)[1].strip()
 current_a = None
 elif line.startswith("A") and ":" in line[:4]:
 current_a = line.split(":", 1)[1].strip()

 if current_q and current_a:
 pairs.append({"question": current_q, "answer": current_a})

 return pairs

# Example: Generate from a technical document
sample_doc = """
PostgreSQL uses a cost-based query optimizer that evaluates multiple
execution plans and selects the one with the lowest estimated cost.
The optimizer considers sequential scan cost, index scan cost, join
strategies (nested loop, hash join, merge join), and statistics
collected by ANALYZE. The work_mem parameter controls how much memory
is available for sort operations before spilling to disk.
"""

pairs = domain_grounded_generation(sample_doc, "database engineering")
for p in pairs:
 print(f"Q: {p['question'][:70]}...")
 print(f"A: {p['answer'][:70]}...")
 print()