Module 7: Prompt Evaluation
7. Automated Evaluation Pipelines
Description
Automating evaluation end-to-end: from loading the dataset to generating reports. Scheduling with cron/CI. Trend tracking with a metrics history. Alerts when quality drops. A dashboard with visualizations.
What Is an Evaluation Pipeline?
An automated evaluation pipeline is the system that runs evaluations regularly with no manual intervention. It's the equivalent of a regression test suite, but for LLMs.
Manual pipeline (no automation):
- Whenever someone remembers to do it
- Results in spreadsheets or notes
- No history, no trends
- The team doesn't find out when it breaks
Automated pipeline:
Dataset → Run prompts → Evaluate → Store results → Report → Alert if needed
↑ ↓
Cron/CI ←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←←← Slack/Email
Pipeline Architecture
┌─────────────────────────────────────────────────────┐
│ EVALUATION PIPELINE │
│ │
│ 1. LOADER → Loads the golden set │
│ 2. RUNNER → Runs the prompt(s) │
│ 3. EVALUATOR → Computes the metrics │
│ 4. COMPARATOR → Compares against baseline │
│ 5. REPORTER → Generates a markdown report │
│ 6. NOTIFIER → Sends alerts on failure │
│ 7. STORAGE → Persists the history │
└─────────────────────────────────────────────────────┘
Complete Implementation
Base Pipeline
import json
import time
import asyncio
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Callable, Any
from openai import OpenAI, AsyncOpenAI
client = OpenAI()
async_client = AsyncOpenAI()
@dataclass
class EvaluationRun:
"""Result of a full pipeline run."""
run_id: str
prompt_name: str
prompt_version: str
timestamp: str
golden_set_size: int
metrics: dict[str, float] = field(default_factory=dict)
failures: list[dict] = field(default_factory=list)
metadata: dict = field(default_factory=dict)
duration_seconds: float = 0.0
estimated_cost: float = 0.0
class EvaluationPipeline:
"""
Modular, extensible evaluation pipeline.
It supports:
- Multiple metrics (accuracy, faithfulness, format, latency)
- Async evaluation for better performance
- Comparison against a baseline
- Persistent history
- Automatic alerts
"""
def __init__(
self,
golden_set_path: str,
history_path: str = "eval_history.jsonl",
baseline_path: str = "baseline.json",
notifications: bool = True
):
self.golden_set_path = Path(golden_set_path)
self.history_path = Path(history_path)
self.baseline_path = Path(baseline_path)
self.notifications = notifications
# Load the golden set
with open(self.golden_set_path) as f:
self.golden_set = json.load(f)
print(f"Pipeline initialized: {len(self.golden_set)} examples in the golden set")
async def _run_prompt_async(
self,
prompt_template: str,
input_text: str,
semaphore: asyncio.Semaphore
) -> tuple[str, dict]:
"""Runs a prompt asynchronously with concurrency control."""
async with semaphore:
start = time.time()
response = await async_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": prompt_template.format(input=input_text)
}],
temperature=0
)
return (
response.choices[0].message.content.strip(),
{
"latency_ms": (time.time() - start) * 1000,
"tokens": response.usage.total_tokens,
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens
}
)
def _calculate_accuracy(self, outputs: list[str]) -> float:
"""Computes accuracy against the golden set's expected_outputs."""
if not outputs:
return 0.0
correct = sum(
output.strip().lower() == str(ex["expected_output"]).strip().lower()
for output, ex in zip(outputs, self.golden_set)
)
return correct / len(outputs)
def _calculate_format_compliance(
self,
outputs: list[str],
output_format: str = "text"
) -> float:
"""Computes format compliance against the expected format."""
if not outputs:
return 0.0
compliant = 0
for output in outputs:
if output_format == "json":
try:
json.loads(output)
compliant += 1
except json.JSONDecodeError:
pass
elif output_format == "text":
compliant += 1 # Text is always valid
elif output_format in ["POSITIVE", "NEGATIVE", "NEUTRAL"]:
if output.strip().upper() in ["POSITIVE", "NEGATIVE", "NEUTRAL"]:
compliant += 1
return compliant / len(outputs)
async def _calculate_faithfulness_batch(
self,
outputs: list[str],
sample_rate: float = 0.2
) -> float:
"""
Computes faithfulness with LLM-as-judge on a sample.
It only evaluates sample_rate of the outputs to cut cost.
"""
import random
indices = random.sample(
range(len(outputs)),
int(len(outputs) * sample_rate) + 1
)
scores = []
semaphore = asyncio.Semaphore(5)
async def evaluate_faithfulness(idx):
ex = self.golden_set[idx]
output = outputs[idx]
prompt = f"""Does the OUTPUT use only information from the INPUT?
INPUT: {ex['input']}
OUTPUT: {output}
Answer "1" (faithful) or "0" (invents). Only the number."""
resp_text, _ = await self._run_prompt_async(prompt, "", semaphore)
return 1.0 if "1" in resp_text else 0.0
tasks = [evaluate_faithfulness(i) for i in indices]
scores = await asyncio.gather(*tasks)
return sum(scores) / len(scores) if scores else 0.5
def _identify_failures(
self,
outputs: list[str],
max_failures: int = 20
) -> list[dict]:
"""Identifies the cases where the prompt failed."""
failures = []
for i, (output, ex) in enumerate(zip(outputs, self.golden_set)):
expected = str(ex["expected_output"]).strip().lower()
actual = output.strip().lower()
if expected != actual:
failures.append({
"id": ex.get("id", str(i)),
"input": str(ex["input"])[:100],
"expected": str(ex["expected_output"]),
"actual": output[:100],
"category": ex.get("category", "unknown"),
"difficulty": ex.get("difficulty", "unknown")
})
if len(failures) >= max_failures:
break
return failures
async def run(
self,
prompt_template: str,
prompt_name: str,
prompt_version: str,
metrics: list[str] = None,
max_concurrent: int = 10
) -> EvaluationRun:
"""
Runs the full evaluation pipeline.
metrics: List of metrics to compute.
Options: "accuracy", "format", "faithfulness", "latency", "cost"
"""
if metrics is None:
metrics = ["accuracy", "format", "latency", "cost"]
run_id = f"{prompt_name}_{prompt_version}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
total_start = time.time()
print(f"\n{'='*50}")
print(f"Running evaluation: {prompt_name} {prompt_version}")
print(f"Golden set: {len(self.golden_set)} examples")
print(f"{'='*50}")
# 1. Run every prompt in parallel
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [
self._run_prompt_async(prompt_template, str(ex["input"]), semaphore)
for ex in self.golden_set
]
print(f"Running {len(tasks)} prompts in parallel (max_concurrent={max_concurrent})...")
results = await asyncio.gather(*tasks)
outputs = [r[0] for r in results]
metadatas = [r[1] for r in results]
# 2. Compute the metrics
print("Computing metrics...")
computed_metrics = {}
if "accuracy" in metrics:
computed_metrics["accuracy"] = self._calculate_accuracy(outputs)
if "format" in metrics:
output_format = self.golden_set[0].get("expected_format", "text") if self.golden_set else "text"
computed_metrics["format"] = self._calculate_format_compliance(outputs, output_format)
if "faithfulness" in metrics:
print(" Computing faithfulness (LLM-as-judge, 20% sample)...")
computed_metrics["faithfulness"] = await self._calculate_faithfulness_batch(outputs)
if "latency" in metrics:
latencies = [m["latency_ms"] for m in metadatas]
sorted_lat = sorted(latencies)
n = len(sorted_lat)
computed_metrics["latency_p50"] = sorted_lat[n // 2]
computed_metrics["latency_p95"] = sorted_lat[int(n * 0.95)]
computed_metrics["latency_mean"] = sum(latencies) / n
if "cost" in metrics:
total_tokens = sum(m["tokens"] for m in metadatas)
# GPT-4o-mini: ~$0.15/1M input tokens, ~$0.60/1M output tokens
prompt_tokens = sum(m["prompt_tokens"] for m in metadatas)
completion_tokens = sum(m["completion_tokens"] for m in metadatas)
cost = prompt_tokens * 0.15 / 1e6 + completion_tokens * 0.60 / 1e6
computed_metrics["total_tokens"] = total_tokens
computed_metrics["cost_usd"] = cost
computed_metrics["cost_per_request"] = cost / len(outputs)
# 3. Identify the failures
failures = self._identify_failures(outputs)
# 4. Build the run result
duration = time.time() - total_start
run = EvaluationRun(
run_id=run_id,
prompt_name=prompt_name,
prompt_version=prompt_version,
timestamp=datetime.now().isoformat(),
golden_set_size=len(self.golden_set),
metrics=computed_metrics,
failures=failures,
duration_seconds=duration,
estimated_cost=computed_metrics.get("cost_usd", 0.0)
)
# 5. Persist to the history
self._save_history(run)
print(f"\n✅ Evaluation finished in {duration:.1f}s")
self._print_summary(run)
return run
def _save_history(self, run: EvaluationRun) -> None:
"""Persists the run to the JSONL history."""
with open(self.history_path, "a") as f:
f.write(json.dumps(asdict(run)) + "\n")
def _print_summary(self, run: EvaluationRun) -> None:
"""Prints a summary to the console."""
print(f"\n📊 Metrics ({run.prompt_name} {run.prompt_version}):")
for metric, value in run.metrics.items():
if isinstance(value, float):
print(f" {metric:25s}: {value:.4f}")
else:
print(f" {metric:25s}: {value}")
if run.failures:
print(f"\n❌ Failures: {len(run.failures)}/{run.golden_set_size}")
for failure in run.failures[:3]: # Show only the first 3
print(f" [{failure['id']}] Expected: {failure['expected']} | Got: {failure['actual'][:40]}")Trend Tracking
The history lets you spot trends: is quality getting better or worse over time?
class TrendTracker:
"""Analyzes trends in the evaluation history."""
def __init__(self, history_path: str = "eval_history.jsonl"):
self.history_path = Path(history_path)
def load_history(
self,
prompt_name: str,
last_n: int = 30
) -> list[dict]:
"""Loads the last N runs of a prompt."""
if not self.history_path.exists():
return []
runs = []
with open(self.history_path) as f:
for line in f:
run = json.loads(line.strip())
if run.get("prompt_name") == prompt_name:
runs.append(run)
return runs[-last_n:]
def analyze_trend(
self,
prompt_name: str,
metric: str = "accuracy",
window: int = 7 # last N runs
) -> dict:
"""
Analyzes a metric's trend over time.
Uses linear regression to detect whether it's improving/degrading.
"""
from scipy import stats as scipy_stats
history = self.load_history(prompt_name, last_n=window)
if len(history) < 2:
return {"trend": "INSUFFICIENT_DATA", "n": len(history)}
values = []
for run in history:
m = run.get("metrics", {}).get(metric)
if m is not None:
values.append(float(m))
if len(values) < 2:
return {"trend": "METRIC_NOT_AVAILABLE"}
x = list(range(len(values)))
slope, intercept, r_value, p_value, std_err = scipy_stats.linregress(x, values)
trend = "STABLE"
if slope > 0.005 and p_value < 0.05:
trend = "IMPROVING ↑"
elif slope < -0.005 and p_value < 0.05:
trend = "DEGRADING ↓"
return {
"trend": trend,
"slope": slope,
"r_squared": r_value ** 2,
"p_value": p_value,
"last_value": values[-1],
"first_value": values[0],
"total_delta": values[-1] - values[0],
"n_points": len(values),
"values": values
}
def generate_trend_report(self, prompt_name: str) -> str:
"""Generates a trend report for every metric."""
lines = [
f"# Trend Report: {prompt_name}",
f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
""
]
metrics = ["accuracy", "faithfulness", "format", "latency_p95"]
for metric in metrics:
analysis = self.analyze_trend(prompt_name, metric)
if "trend" not in analysis:
continue
emoji = {"IMPROVING ↑": "📈", "DEGRADING ↓": "📉", "STABLE": "➡️"}.get(
analysis["trend"], "❓"
)
lines.append(f"## {emoji} {metric.upper()}")
lines.append(f"- Trend: **{analysis['trend']}**")
if "last_value" in analysis:
last = analysis["last_value"]
if metric in ["accuracy", "faithfulness", "format"]:
lines.append(f"- Last value: {last:.2%}")
else:
lines.append(f"- Last value: {last:.1f}ms")
if "total_delta" in analysis:
delta = analysis["total_delta"]
lines.append(f"- Delta (total): {delta:+.4f}")
lines.append("")
return "\n".join(lines)Alerting System
class AlertManager:
"""Manages alerts when metrics fall below their thresholds."""
def __init__(self, thresholds: dict[str, float] | None = None):
# Default thresholds — customize them for your use case
self.thresholds = thresholds or {
"accuracy": 0.85, # Alert if accuracy < 85%
"faithfulness": 0.80, # Alert if faithfulness < 80%
"format": 0.95, # Alert if format compliance < 95%
"latency_p95": 5000.0, # Alert if p95 > 5000ms
"cost_per_request": 0.01 # Alert if cost > $0.01/request
}
self.active_alerts: list[dict] = []
def verify(self, run: EvaluationRun) -> list[dict]:
"""
Checks whether any metric crosses its threshold.
Returns the list of alerts generated.
"""
alerts = []
for metric, threshold in self.thresholds.items():
value = run.metrics.get(metric)
if value is None:
continue
# For latency and cost: alert if it goes ABOVE the threshold (higher is worse)
# For the rest: alert if it goes BELOW the threshold (lower is worse)
if metric in ["latency_p95", "cost_per_request"]:
if value > threshold:
alerts.append({
"run_id": run.run_id,
"metric": metric,
"value": value,
"threshold": threshold,
"type": "ABOVE_THRESHOLD",
"message": f"⚠️ {metric}={value:.4f} exceeds threshold {threshold}"
})
else:
if value < threshold:
alerts.append({
"run_id": run.run_id,
"metric": metric,
"value": value,
"threshold": threshold,
"type": "BELOW_THRESHOLD",
"message": f"🚨 {metric}={value:.2%} below the threshold {threshold:.2%}"
})
self.active_alerts.extend(alerts)
return alerts
def notify_slack(self, alerts: list[dict], webhook_url: str) -> None:
"""Sends alerts to Slack via webhook."""
if not alerts:
return
import urllib.request
import json
text = f"🚨 *Evaluation Pipeline Alerts*\n\n"
for alert in alerts:
text += f"• {alert['message']}\n"
text += f" Run ID: `{alert['run_id']}`\n\n"
payload = {"text": text}
data = json.dumps(payload).encode()
req = urllib.request.Request(
webhook_url,
data=data,
headers={"Content-Type": "application/json"}
)
try:
urllib.request.urlopen(req, timeout=5)
print(f"✅ {len(alerts)} alerts sent to Slack")
except Exception as e:
print(f"❌ Error sending to Slack: {e}")
def notify_email(
self,
alerts: list[dict],
to: str,
from_email: str,
smtp_config: dict
) -> None:
"""Sends alerts over email."""
if not alerts:
return
import smtplib
from email.mime.text import MIMEText
body = "Evaluation Pipeline alerts:\n\n"
for alert in alerts:
body += f"• {alert['message']}\n"
msg = MIMEText(body)
msg["Subject"] = f"🚨 Alert: {len(alerts)} metrics out of threshold"
msg["From"] = from_email
msg["To"] = to
with smtplib.SMTP(smtp_config["host"], smtp_config["port"]) as server:
server.sendmail(from_email, to, msg.as_string())
print(f"✅ Alert email sent to {to}")Automatic Report
def generate_markdown_report(
run: EvaluationRun,
baseline_metrics: dict | None = None,
trends: dict | None = None
) -> str:
"""Generates a full markdown report for the evaluation run."""
timestamp = datetime.fromisoformat(run.timestamp).strftime("%Y-%m-%d %H:%M:%S")
lines = [
f"# Evaluation Report: {run.prompt_name}",
f"**Version:** {run.prompt_version} ",
f"**Date:** {timestamp} ",
f"**Run ID:** `{run.run_id}` ",
"",
f"## 📊 Main Metrics",
"",
"| Metric | Value |" + (" Baseline | Delta |" if baseline_metrics else ""),
"|---------|-------|" + (" ---------|-------|" if baseline_metrics else ""),
]
for metric, value in run.metrics.items():
if isinstance(value, float):
val_str = f"{value:.2%}" if value <= 1.0 else f"{value:.1f}"
else:
val_str = str(value)
if baseline_metrics and metric in baseline_metrics:
baseline_val = baseline_metrics[metric]
delta = value - baseline_val if isinstance(value, float) else 0
delta_str = f"{delta:+.2%}" if abs(delta) <= 1 else f"{delta:+.1f}"
status = "✅" if delta >= 0 else "⚠️" if delta > -0.05 else "🚨"
lines.append(f"| {metric} | {val_str} | {baseline_val:.2%} | {status} {delta_str} |")
else:
lines.append(f"| {metric} | {val_str} |")
# Trends
if trends:
lines.extend(["", "## 📈 Trends (last 7 runs)", ""])
for metric, t in trends.items():
if isinstance(t, dict) and "trend" in t:
lines.append(f"- **{metric}:** {t['trend']}")
# Failures
if run.failures:
lines.extend([
"",
f"## ❌ Failed Cases ({len(run.failures)}/{run.golden_set_size})",
"",
"| ID | Input | Expected | Got | Category |",
"|----|-------|----------|----------|-----------|",
])
for failure in run.failures[:10]: # At most 10 in the report
short_input = failure["input"][:40].replace("|", "\\|")
lines.append(
f"| {failure['id']} | {short_input}... | `{failure['expected']}` | `{failure['actual'][:30]}` | {failure['category']} |"
)
else:
lines.extend(["", "## ✅ No Failures", ""])
# Metadata
lines.extend([
"",
"## ℹ️ Metadata",
f"- **Duration:** {run.duration_seconds:.1f}s",
f"- **Estimated cost:** ${run.estimated_cost:.4f}",
f"- **Golden set:** {run.golden_set_size} examples",
])
return "\n".join(lines)Scheduling: Running on a Schedule
With Cron (Linux/Mac)
# Edit the crontab: crontab -e
# Format: minute hour day_of_month month day_of_week command
# Run the evaluation pipeline every day at 2am
0 2 * * * cd /app && python run_evaluation.py >> /var/log/eval_pipeline.log 2>&1
# Run every Monday at 9am (weekly report)
0 9 * * 1 cd /app && python run_weekly_report.pyRunner Script (run_evaluation.py)
#!/usr/bin/env python3
"""Main script for running the evaluation pipeline."""
import asyncio
import sys
from pathlib import Path
# Configuration
GOLDEN_SET_PATH = "datasets/sentiment_classifier.json"
PROMPT_FILE = "prompts/classifier_v2.txt"
PROMPT_NAME = "sentiment_classifier"
PROMPT_VERSION = "v2.1"
SLACK_WEBHOOK = "https://hooks.slack.com/services/xxx/yyy/zzz" # Optional
async def main():
print(f"Starting evaluation pipeline: {PROMPT_NAME} {PROMPT_VERSION}")
# Load the prompt
with open(PROMPT_FILE) as f:
prompt_template = f.read()
# Initialize the pipeline
pipeline = EvaluationPipeline(
golden_set_path=GOLDEN_SET_PATH,
history_path="eval_history.jsonl"
)
# Run the evaluation
run = await pipeline.run(
prompt_template=prompt_template,
prompt_name=PROMPT_NAME,
prompt_version=PROMPT_VERSION,
metrics=["accuracy", "format", "faithfulness", "latency", "cost"]
)
# Check the alerts
alert_mgr = AlertManager(thresholds={
"accuracy": 0.88,
"faithfulness": 0.80,
"format": 0.95
})
alerts = alert_mgr.verify(run)
if alerts and SLACK_WEBHOOK:
alert_mgr.notify_slack(alerts, SLACK_WEBHOOK)
# Generate the report
tracker = TrendTracker()
trends = {}
for metric in ["accuracy", "faithfulness"]:
trends[metric] = tracker.analyze_trend(PROMPT_NAME, metric)
report = generate_markdown_report(run, trends=trends)
# Save the report
report_path = f"reports/{run.run_id}.md"
Path("reports").mkdir(exist_ok=True)
with open(report_path, "w") as f:
f.write(report)
print(f"\nReport saved: {report_path}")
# Exit code 1 if there are critical alerts (for CI/CD)
if any(a["type"] == "BELOW_THRESHOLD" for a in alerts):
print(f"\n❌ {len(alerts)} critical alerts — exit code 1")
sys.exit(1)
print("\n✅ Pipeline completed successfully")
if __name__ == "__main__":
asyncio.run(main())A Simple Dashboard with Tables
def generate_text_dashboard(
prompt_name: str,
history_path: str = "eval_history.jsonl",
last_n: int = 10
) -> str:
"""
Generates a text/markdown dashboard with the last N runs.
"""
tracker = TrendTracker(history_path)
history = tracker.load_history(prompt_name, last_n)
if not history:
return f"No data for {prompt_name}"
lines = [
f"# Dashboard: {prompt_name}",
f"Last {len(history)} runs",
"",
"| Date | Version | Accuracy | Faithfulness | Format | Cost |",
"|-------|---------|----------|-------------|--------|-------|",
]
for run_data in history:
timestamp = run_data.get("timestamp", "")[:16]
version = run_data.get("prompt_version", "?")
metrics = run_data.get("metrics", {})
accuracy = f"{metrics.get('accuracy', 0):.2%}"
faith = f"{metrics.get('faithfulness', 0):.2%}" if "faithfulness" in metrics else "N/A"
format_c = f"{metrics.get('format', 0):.2%}" if "format" in metrics else "N/A"
cost = f"${metrics.get('cost_usd', 0):.4f}" if "cost_usd" in metrics else "N/A"
lines.append(f"| {timestamp} | {version} | {accuracy} | {faith} | {format_c} | {cost} |")
return "\n".join(lines)Troubleshooting
Problem 1: Slow pipeline with a large golden set
Symptom: Evaluating 500 examples takes 15+ minutes.
Solution:
# Use higher concurrency (check the API's rate limits)
run = await pipeline.run(
prompt_template=prompt,
prompt_name="classifier",
prompt_version="v2",
max_concurrent=20 # Raise it from the default of 10
)
# For very large golden sets: split into batches
async def evaluate_in_batches(golden_set, batch_size=100):
for i in range(0, len(golden_set), batch_size):
batch = golden_set[i:i+batch_size]
print(f"Processing batch {i//batch_size + 1}/{len(golden_set)//batch_size + 1}")
# Evaluate the batch...
await asyncio.sleep(1) # Pause between batchesProblem 2: Very noisy alerts (lots of false positives)
Symptom: You get alerts every day even though the system is fine.
Cause: Thresholds that are too strict, or the model's natural variation.
Solution:
# 1. Use a sliding window (average of the last N runs)
# instead of comparing a single run
def threshold_with_window(history: list[dict], metric: str, window: int = 3) -> float:
"""Averages the last `window` runs to smooth out variation."""
last_runs = history[-window:]
values = [run.get("metrics", {}).get(metric, 0) for run in last_runs]
return sum(values) / len(values) if values else 0
# 2. Only alert if it drops for 3 consecutive runs (not just 1)
# 3. Tune the thresholds based on real historical dataProblem 3: The history grows too large
Symptom: eval_history.jsonl is hundreds of megabytes.
Solution:
def compress_history(
history_path: str = "eval_history.jsonl",
max_runs: int = 90 # Keep the last 90 runs (~3 months)
) -> None:
"""Keeps only the last max_runs entries in the history."""
with open(history_path) as f:
lines = f.readlines()
if len(lines) <= max_runs:
return
# Back it up before truncating
backup_path = f"{history_path}.bak"
with open(backup_path, "w") as f:
f.writelines(lines)
# Keep only the last N
with open(history_path, "w") as f:
f.writelines(lines[-max_runs:])
print(f"History compressed: {len(lines)} → {max_runs} runs")Exercises
Exercise 1: Build a minimum viable pipeline
Implement a pipeline that evaluates a classification prompt against 20 examples and saves the results to JSONL.
See solution
import json
import time
from datetime import datetime
from openai import OpenAI
client = OpenAI()
def minimal_pipeline(
prompt_template: str,
golden_set: list[dict],
prompt_name: str = "my_prompt",
history_path: str = "eval_history.jsonl"
) -> dict:
"""Minimum viable pipeline."""
start = time.time()
outputs = []
for ex in golden_set:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt_template.format(input=ex["input"])}],
temperature=0
)
outputs.append(response.choices[0].message.content.strip())
# Compute accuracy
correct = sum(
o.lower() == str(e["expected_output"]).lower()
for o, e in zip(outputs, golden_set)
)
accuracy = correct / len(golden_set)
# Save the result
result = {
"timestamp": datetime.now().isoformat(),
"prompt_name": prompt_name,
"golden_set_size": len(golden_set),
"metrics": {"accuracy": accuracy},
"duration_s": time.time() - start
}
with open(history_path, "a") as f:
f.write(json.dumps(result) + "\n")
print(f"✅ Accuracy: {accuracy:.2%} | Saved to {history_path}")
return result
# Usage:
PROMPT = "Classify as POSITIVE, NEGATIVE or NEUTRAL: {input}. Only the category."
golden = [
{"input": "Excellent product", "expected_output": "POSITIVE"},
{"input": "Terrible service", "expected_output": "NEGATIVE"},
{"input": "The package arrived", "expected_output": "NEUTRAL"},
]
result = minimal_pipeline(PROMPT, golden)Exercise 2: Add trend tracking
Given the following history, implement a function that detects whether accuracy is improving or degrading:
See solution
def analyze_simple_trend(history: list[dict], metric: str = "accuracy") -> str:
"""Analyzes a simple trend using the last 5 values."""
from scipy import stats
values = [r.get("metrics", {}).get(metric) for r in history]
values = [v for v in values if v is not None][-5:] # Last 5
if len(values) < 3:
return "INSUFFICIENT_DATA"
x = list(range(len(values)))
slope, _, _, p_value, _ = stats.linregress(x, values)
if abs(slope) < 0.002 or p_value > 0.1:
return f"STABLE ({values[-1]:.2%})"
elif slope > 0:
return f"IMPROVING ↑ ({values[0]:.2%} → {values[-1]:.2%})"
else:
return f"DEGRADING ↓ ({values[0]:.2%} → {values[-1]:.2%})"
# Test with a simulated history:
history = [
{"metrics": {"accuracy": 0.88}},
{"metrics": {"accuracy": 0.89}},
{"metrics": {"accuracy": 0.91}},
{"metrics": {"accuracy": 0.92}},
{"metrics": {"accuracy": 0.93}},
]
print(analyze_simple_trend(history)) # IMPROVING ↑ (88% → 93%)Summary
- Pipeline: Dataset → Run prompts → Evaluate → Store → Report → Alert — automated
- Async: Parallelize the API calls to evaluate large golden sets in seconds instead of minutes
- History: Persistent JSONL with timestamps for trend tracking
- Trend tracking: Linear regression over the history to catch gradual degradation
- Alerts: Per-metric thresholds with notification to Slack/email/CI
- Scheduling: Cron for daily/weekly evaluation; CI/CD for every commit
- Dashboard: History tables for team visibility
Additional resources
- LangSmith — Managed evaluation pipeline
- Weights & Biases — Experiment tracking with dashboards
- Prometheus + Grafana — Monitoring and alerting for production
- Apache Airflow — Scheduling complex pipelines
- asyncio Documentation — For parallelization