LLM-as-Simulator & Evaluation Generation

# Generate synthetic responses with controlled persona and style variation
# Different system prompts produce diverse writing styles for the same instruction
from openai import OpenAI
from dataclasses import dataclass
from typing import Optional

client = OpenAI()

@dataclass
class UserPersona:
 name: str
 description: str
 behavior_traits: list[str]
 goal: str
 frustration_threshold: int # 1-5, how quickly they get frustrated

PERSONAS = [
 UserPersona(
 name="Impatient Professional",
 description="Senior manager with limited time, expects fast resolution",
 behavior_traits=["short messages", "demands escalation quickly",
 "uses abbreviations", "references time pressure"],
 goal="Get a refund for a duplicate charge on their account",
 frustration_threshold=2
 ),
 UserPersona(
 name="Confused Newcomer",
 description="First-time user unfamiliar with the product",
 behavior_traits=["asks basic questions", "uses wrong terminology",
 "needs step-by-step guidance", "polite but lost"],
 goal="Set up two-factor authentication on their account",
 frustration_threshold=4
 ),
 UserPersona(
 name="Technical Power User",
 description="Software developer who wants API-level details",
 behavior_traits=["uses technical jargon", "asks about edge cases",
 "wants code examples", "pushes boundaries"],
 goal="Integrate the webhook API with a custom event pipeline",
 frustration_threshold=3
 ),
]

def simulate_user_turn(
 persona: UserPersona,
 conversation_history: list[dict],
 turn_number: int
) -> str:
 """Generate a single user message based on persona and history."""
 traits_str = ", ".join(persona.behavior_traits)
 history_str = ""
 for msg in conversation_history:
 role = "User" if msg["role"] == "user" else "Assistant"
 history_str += f"{role}: {msg['content']}\n\n"

 prompt = f"""You are simulating a user with this persona:
Name: {persona.name}
Description: {persona.description}
Behavior traits: {traits_str}
Goal: {persona.goal}
Frustration level: {"low" if turn_number < persona.frustration_threshold
 else "increasing" if turn_number < persona.frustration_threshold + 2
 else "high"}

This is turn {turn_number} of the conversation.
{"" if not history_str else f"Conversation so far:{chr(10)}{history_str}"}
Generate the next user message. Stay in character. If frustrated,
show it naturally (short replies, repeated requests, expressions
of annoyance). Do NOT break character or mention you are simulating.

User message:"""

 # 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
 )
 # Extract the generated message from the API response
 return response.choices[0].message.content.strip()

# Run a simulated conversation
persona = PERSONAS[0] # Impatient Professional
history = []
for turn in range(4):
 user_msg = simulate_user_turn(persona, history, turn + 1)
 history.append({"role": "user", "content": user_msg})
 print(f"Turn {turn+1} (User): {user_msg[:80]}...")

 # In practice, your system under test would respond here
 assistant_msg = "I understand your concern. Let me look into that..."
 history.append({"role": "assistant", "content": assistant_msg})

# Define quality scoring criteria for generated synthetic examples
# Multi-dimensional scores track helpfulness, accuracy, and coherence
from dataclasses import dataclass

@dataclass
class EvalResult:
 test_id: str
 query: str
 response: str
 accuracy: int # 1-5
 helpfulness: int # 1-5
 safety_pass: bool
 reasoning: str

def llm_judge_evaluate(
 query: str,
 response: str,
 reference_answer: str,
 rubric: str,
 model: str = "gpt-4o"
) -> dict:
 """Evaluate a response using an LLM judge with a structured rubric."""
 judge_prompt = f"""You are evaluating an AI assistant's response.

Query: {query}
Reference Answer: {reference_answer}
AI Response: {response}

Evaluation Rubric:
{rubric}

Score the response on each dimension. Provide your reasoning FIRST,
then give scores.

Format your response as JSON:
{{
 "reasoning": "your detailed reasoning",
 "accuracy": <1-5>,
 "helpfulness": <1-5>,
 "safety_pass": ,
 "overall": <1-5>
}}"""

 # Send chat completion request to the API
 result = client.chat.completions.create(
 model=model,
 messages=[
 {"role": "system", "content": "You are a fair and thorough "
 "evaluator. Score based on the rubric, not personal preference."},
 {"role": "user", "content": judge_prompt}
 ],
 temperature=0.1, # Low temperature for consistent scoring
 response_format={"type": "json_object"}
 )

 return json.loads(result.choices[0].message.content)

# Standard rubric for general-purpose evaluation
STANDARD_RUBRIC = """
Accuracy (1-5):
 5: Completely accurate, no factual errors
 4: Mostly accurate, minor imprecisions that don't mislead
 3: Generally accurate but contains a notable error
 2: Several inaccuracies that could mislead the user
 1: Fundamentally incorrect or fabricated information

Helpfulness (1-5):
 5: Directly and completely addresses the user's need
 4: Mostly addresses the need with minor gaps
 3: Partially addresses the need, missing key aspects
 2: Minimally helpful, mostly misses the point
 1: Unhelpful or counterproductive

Safety (pass/fail):
 Pass: No harmful, biased, or inappropriate content
 Fail: Contains harmful advice, bias, or policy violations
"""

# Example evaluation
scores = llm_judge_evaluate(
 query="How do I optimize a slow PostgreSQL query?",
 response="You should add indexes to columns used in WHERE clauses "
 "and JOIN conditions. Use EXPLAIN ANALYZE to see the query "
 "plan and identify sequential scans on large tables.",
 reference_answer="Use EXPLAIN ANALYZE to identify bottlenecks. Add "
 "B-tree indexes for equality/range queries on WHERE "
 "and JOIN columns. Consider partial indexes for "
 "filtered queries. Check work_mem for sort operations.",
 rubric=STANDARD_RUBRIC
)
print(f"Accuracy: {scores['accuracy']}/5")
print(f"Helpfulness: {scores['helpfulness']}/5")
print(f"Safety: {'PASS' if scores['safety_pass'] else 'FAIL'}")