Lab 05: Build an Eval Harness
Goal
Build a systematic evaluation harness that compares multiple models on your specific use case. Establish a baseline and track quality over time.
Prerequisites
pip install openai asyncio statistics
Exercise 1: Manual Eval Baseline
Before automating, do manual evaluation to understand what "good" looks like.
# lab05/01_manual_eval.py
from openai import OpenAI
client = OpenAI()
# Define your use case — customize this for your domain
TEST_CASES = [
{
"id": "tc_001",
"description": "KV cache explanation",
"messages": [
{"role": "system", "content": "You are a concise ML engineering instructor."},
{"role": "user", "content": "Explain the KV cache in 2-3 sentences."}
],
"must_contain": ["key", "value", "cache"],
"must_not_contain": ["I don't know"],
"max_length": 300,
"ideal_answer": "The KV cache stores computed key and value tensors from previous tokens during autoregressive generation. This avoids recomputing them on each decoding step, trading memory for compute. Without it, inference time grows quadratically with sequence length."
},
{
"id": "tc_002",
"description": "Quantization trade-off",
"messages": [
{"role": "system", "content": "You are a concise ML engineering instructor."},
{"role": "user", "content": "What is the main tradeoff in model quantization?"}
],
"must_contain": ["memory", "quality"],
"must_not_contain": ["I'm not sure"],
"max_length": 250,
"ideal_answer": "Quantization reduces model size and inference speed by lowering numerical precision (e.g., FP16 to INT4), but at the cost of some accuracy. The main trade-off is memory/speed vs. output quality."
},
{
"id": "tc_003",
"description": "RAG vs. fine-tuning",
"messages": [
{"role": "system", "content": "You are a concise ML engineering instructor."},
{"role": "user", "content": "When should I use RAG instead of fine-tuning?"}
],
"must_contain": ["knowledge", "data"],
"must_not_contain": ["always fine-tune"],
"max_length": 400,
"ideal_answer": "Use RAG when you need to ground the model in frequently updated knowledge or private documents. Use fine-tuning when you need to change the model's style, format, or task-specific behavior. RAG is better for facts; fine-tuning is better for behavior."
},
# Add 7+ more for a useful eval set
]
def rule_based_score(answer: str, test_case: dict) -> dict:
"""Score based on rules defined in the test case."""
score = 1.0
issues = []
# Length check
if len(answer) > test_case["max_length"]:
score -= 0.2
issues.append(f"Too long ({len(answer)} chars, max {test_case['max_length']})")
# Must contain
for term in test_case.get("must_contain", []):
if term.lower() not in answer.lower():
score -= 0.2
issues.append(f"Missing required term: '{term}'")
# Must not contain
for term in test_case.get("must_not_contain", []):
if term.lower() in answer.lower():
score -= 0.3
issues.append(f"Contains forbidden phrase: '{term}'")
return {"score": max(0, score), "issues": issues}
# Run manual eval
model = "gpt-4o-mini"
print(f"Evaluating {model}\n{'='*50}")
results = []
for tc in TEST_CASES:
response = client.chat.completions.create(
model=model,
messages=tc["messages"],
max_tokens=500,
temperature=0
)
answer = response.choices[0].message.content
scored = rule_based_score(answer, tc)
results.append({
"id": tc["id"],
"description": tc["description"],
"score": scored["score"],
"issues": scored["issues"],
"answer_preview": answer[:100]
})
status = "✓" if scored["score"] >= 0.8 else "✗"
print(f"{status} [{tc['id']}] {tc['description']}")
print(f" Score: {scored['score']:.2f}")
if scored["issues"]:
print(f" Issues: {', '.join(scored['issues'])}")
import statistics
scores = [r["score"] for r in results]
print(f"\n{'='*50}")
print(f"Model: {model}")
print(f"Avg score: {statistics.mean(scores):.3f}")
print(f"Pass rate (≥0.8): {sum(1 for s in scores if s >= 0.8) / len(scores):.1%}")
Exercise 2: LLM-as-Judge
# lab05/02_llm_judge.py
from openai import OpenAI
import json
client = OpenAI()
def llm_judge(question: str, answer: str, criteria: str, reference: str = None) -> dict:
"""Use GPT-4o as a judge to score an answer."""
ref_section = f"\nReference answer:\n{reference}" if reference else ""
prompt = f"""You are an expert judge evaluating answers to technical questions about LLMs and ML engineering.
Question: {question}
{ref_section}
Answer to evaluate: {answer}
Evaluation criteria: {criteria}
Provide:
1. score: integer from 1-5 (5=excellent, 1=poor)
2. reasoning: one sentence explaining the score
3. key_issue: the main problem if score < 4, otherwise null
Respond in valid JSON only."""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0
)
try:
return json.loads(response.choices[0].message.content)
except:
return {"score": 3, "reasoning": "Parse error", "key_issue": "Could not parse judge response"}
# Test the judge
example = {
"question": "What is the KV cache in LLMs?",
"answer": "The KV cache stores precomputed key and value matrices from the attention mechanism. This avoids redundant computation during token generation, significantly speeding up inference.",
"criteria": "Is the explanation technically accurate? Does it mention what is cached and why it helps performance?",
"reference": "The KV cache caches key and value tensors from previous tokens in the attention computation, eliminating the need to recompute them at each decoding step."
}
result = llm_judge(**example)
print(json.dumps(result, indent=2))
Exercise 3: Async Multi-Model Comparison
# lab05/03_compare_models.py
import asyncio, json, time, statistics
from openai import AsyncOpenAI
client = AsyncOpenAI()
MODELS = ["gpt-4o-mini", "gpt-4o"]
TEST_CASES = [
{
"id": "tc_001",
"messages": [
{"role": "system", "content": "You are a concise ML engineering instructor."},
{"role": "user", "content": "Explain the KV cache in 2-3 sentences."}
],
"criteria": "Technically accurate. Mentions what is cached (keys/values) and why (avoid recomputation).",
"ideal": "Caches key/value tensors from attention to avoid recomputing them each decode step."
},
{
"id": "tc_002",
"messages": [
{"role": "system", "content": "You are a concise ML engineering instructor."},
{"role": "user", "content": "What is the difference between RAG and fine-tuning?"}
],
"criteria": "Correctly distinguishes RAG (retrieval at query time) from fine-tuning (modifying weights). Gives guidance on when to use each.",
"ideal": "RAG retrieves documents at inference time; fine-tuning updates model weights. Use RAG for knowledge, fine-tuning for behavior/style."
},
]
async def evaluate_single(model: str, tc: dict) -> dict:
start = time.time()
response = await client.chat.completions.create(
model=model,
messages=tc["messages"],
max_tokens=500,
temperature=0
)
answer = response.choices[0].message.content
latency = time.time() - start
# Score with LLM judge
judge_response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"Rate 1-5: {tc['criteria']}\n\nAnswer: {answer}\n\nRespond with just the score number."
}],
max_tokens=5,
temperature=0
)
try:
score = int(judge_response.choices[0].message.content.strip())
except:
score = 3
return {
"model": model,
"tc_id": tc["id"],
"score": score,
"latency_ms": latency * 1000,
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
"answer_preview": answer[:100]
}
async def compare_models():
tasks = [
evaluate_single(model, tc)
for model in MODELS
for tc in TEST_CASES
]
results = await asyncio.gather(*tasks)
# Aggregate by model
by_model = {}
for r in results:
model = r["model"]
if model not in by_model:
by_model[model] = []
by_model[model].append(r)
print("\n# Model Comparison\n")
print(f"{'Model':<25} {'Avg Score':<12} {'P50 Latency':<15} {'Avg Cost Est.'}")
print("-" * 70)
# Rough pricing
PRICES = {
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4o": {"input": 2.50, "output": 10.00},
}
for model, model_results in sorted(by_model.items(), key=lambda x: statistics.mean(r["score"] for r in x[1]), reverse=True):
scores = [r["score"] for r in model_results]
latencies = [r["latency_ms"] for r in model_results]
prices = PRICES.get(model, {"input": 1.0, "output": 4.0})
total_input = sum(r["input_tokens"] for r in model_results)
total_output = sum(r["output_tokens"] for r in model_results)
cost = (total_input / 1e6 * prices["input"]) + (total_output / 1e6 * prices["output"])
print(f"{model:<25} {statistics.mean(scores):<12.2f} {statistics.median(latencies):<15.0f}ms ${cost:.6f}")
asyncio.run(compare_models())
Exercise 4: Track Eval Over Time
# lab05/04_track_evals.py
import json, sqlite3, time
from datetime import datetime
# Create eval history database
conn = sqlite3.connect("eval_history.db")
conn.execute("""
CREATE TABLE IF NOT EXISTS eval_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT,
model TEXT,
timestamp TEXT,
avg_score REAL,
pass_rate REAL,
p50_latency_ms REAL,
results_json TEXT
)
""")
conn.commit()
def save_eval_run(model: str, results: list[dict]):
"""Save eval results to history database."""
import uuid
run_id = str(uuid.uuid4())[:8]
scores = [r["score"] for r in results]
latencies = [r.get("latency_ms", 0) for r in results]
import statistics
conn.execute("""
INSERT INTO eval_runs (run_id, model, timestamp, avg_score, pass_rate, p50_latency_ms, results_json)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
run_id,
model,
datetime.now().isoformat(),
statistics.mean(scores),
sum(1 for s in scores if s >= 4) / len(scores),
statistics.median(latencies) if latencies else 0,
json.dumps(results)
))
conn.commit()
return run_id
def get_eval_history(model: str) -> list[dict]:
"""Get historical eval runs for a model."""
cursor = conn.execute(
"SELECT run_id, timestamp, avg_score, pass_rate, p50_latency_ms FROM eval_runs WHERE model = ? ORDER BY timestamp",
(model,)
)
return [dict(zip(["run_id", "timestamp", "avg_score", "pass_rate", "latency_ms"], row)) for row in cursor.fetchall()]
def check_regression(model: str, new_score: float, threshold: float = 0.1) -> bool:
"""Return True if new score is significantly below historical average."""
history = get_eval_history(model)
if len(history) < 2:
return False
# Compare against rolling average of last 3 runs
recent_avg = sum(r["avg_score"] for r in history[-3:]) / min(3, len(history))
return new_score < recent_avg - threshold
# Example usage
mock_results = [
{"score": 4, "latency_ms": 320, "tc_id": "tc_001"},
{"score": 5, "latency_ms": 280, "tc_id": "tc_002"},
{"score": 3, "latency_ms": 350, "tc_id": "tc_003"},
]
run_id = save_eval_run("gpt-4o-mini", mock_results)
print(f"Saved eval run: {run_id}")
history = get_eval_history("gpt-4o-mini")
print(f"\nEval history for gpt-4o-mini ({len(history)} runs):")
for run in history:
print(f" {run['timestamp'][:19]} | score={run['avg_score']:.2f} | pass={run['pass_rate']:.1%}")
Deliverables
- Manual eval baseline run (Exercise 1)
- LLM judge working on 3+ test cases (Exercise 2)
- Model comparison complete for 2+ models (Exercise 3)
- Eval history tracking set up (Exercise 4)
- Final report: which model would you deploy and why?
Reflection Questions
- What is the difference between rule-based scoring and LLM-as-judge?
- What are the biases of LLM-as-judge and how can you mitigate them?
- How many test cases do you need for a reliable eval set?
- A prompt change improves your eval score by 15% but hurts production satisfaction. What happened?
- How would you implement an automated gate that blocks deployment if eval score drops more than 10%?