Build a Prompt Comparison and Evaluation Pipeline
Project Goal Build a reusable pipeline that compares two prompt versions on the same test cases and produces a decision-ready report. Lessons 66 and 114–119 explained the ideas: measuring quality, paired comparison, uncertainty, and regression testing. This project turns them into one tool that:
Runs every variant on every test case and stores all outputs with metadata (Lesson 67) Scores each output with a check function you provide, returning pass/fail and a reason Reports pass rates with 95% intervals (Lesson 66) Shows a paired comparison: which cases only one variant passed (Lesson 118) Lists every failure with its reason, so you can see why each variant failed Can be tested without API calls Requirements Requirement Details Python 3.10 or later Packages pip install openai API key OPENAI_API_KEY environment variable Reuse llm.py from Project 128 Project Structure prompt_eval/ ├── llm.py # call_model(instructions, prompt) ├── pipeline.py # run, summarize, compare, report, save ├── cases.jsonl # test cases: id, input, expected (Lesson 115) ├── prompts/ # v1.txt, v2.txt: the instructions being compared └── main.py # run the comparison, save results and report Step 1: Running a Variant pipeline.py (first part)
import json import math import time from datetime import datetime, timezone
def run_variant(name: str, instructions: str, cases: list[dict], call_model, check) -> list[dict]: """Run one prompt variant on every case and record the output and check result.""" records = [] for case in cases: start = time.perf_counter() output = call_model(instructions, case["input"]) passed, note = check(output, case) records.append({"variant": name, "case_id": case["id"], "output": output, "passed": passed, "note": note, "seconds": round(time.perf_counter() - start, 3)}) return records Design Decisions The check function is a parameter. The pipeline does not know what "correct" means for your task; you supply a function that returns (passed, note). The same pipeline can then evaluate classifiers, extractors, or summarizers. Every output is stored, not only pass/fail, so you can read failures and re-score later with improved checks (Lesson 114). Timing is recorded per case, so you can compare latency between variants. Step 2: Summaries, Intervals, and Paired Results def wilson_interval(successes: int, n: int, z: float = 1.96) -> tuple[float, float]: if n == 0: return 0.0, 0.0 p = successes / n denominator = 1 + z2 / n centre = (p + z2 / (2 * n)) / denominator margin = z math.sqrt(p (1 - p) / n + z**2 / (4 n*2)) / denominator return max(0.0, centre - margin), min(1.0, centre + margin)
def summarize(records: list[dict]) -> dict: passed = sum(r["passed"] for r in records) low, high = wilson_interval(passed, len(records)) return {"passed": passed, "total": len(records), "rate": round(passed / len(records), 2), "interval": f"{low:.0%}-{high:.0%}"}
def paired(records_a: list[dict], records_b: list[dict]) -> dict: a = {r["case_id"]: r["passed"] for r in records_a} b = {r["case_id"]: r["passed"] for r in records_b} return {"only_A": sorted(c for c in a if a[c] and not b[c]), "only_B": sorted(c for c in a if b[c] and not a[c]), "both": sum(a[c] and b[c] for c in a), "neither": sum(not a[c] and not b[c] for c in a)} Code Explanation wilson_interval() is the confidence interval from Lesson 66, which stays sensible for small samples. summarize() gives each variant's pass count, rate, and interval. paired() compares results case by case: the cases only variant A passed, the cases only variant B passed, and the counts of agreements. Step 3: The Report def report(name_a: str, name_b: str, records_a: list[dict], records_b: list[dict]) -> str: sa, sb, pair = summarize(records_a), summarize(records_b), paired(records_a, records_b) lines = [ "# Prompt Comparison", "", f"Run: {datetime.now(timezone.utc):%Y-%m-%d %H:%M} UTC", "", "| Variant | Passed | Rate | 95% interval |", "|---|---:|---:|---|", f"| {name_a} | {sa['passed']}/{sa['total']} | {sa['rate']:.0%} | {sa['interval']} |", f"| {name_b} | {sb['passed']}/{sb['total']} | {sb['rate']:.0%} | {sb['interval']} |", "", f"Only {name_a} passed: {', '.join(pair['only_A']) or 'none'}", f"Only {name_b} passed: {', '.join(pair['only_B']) or 'none'}", "", ] failures = [r for r in records_a + records_b if not r["passed"]] if failures: lines += ["## Failures", ""] + [f"- {r['variant']} / {r['case_id']}: {r['note']}" for r in failures] return "\n".join(lines) + "\n"
def save_jsonl(records: list[dict], path: str) -> None: with open(path, "w", encoding="utf-8") as file: for record in records: file.write(json.dumps(record, ensure_ascii=False) + "\n") The report is Markdown, so it can be read in any editor, attached to a pull request (Lesson 67), or displayed in a repository. save_jsonl() stores the full records for later analysis.
Step 4: A Check Function For a ticket classifier, a check function might be:
LABELS = {"Billing", "Access", "Technical", "Other"}
def check_label(output: str, case: dict) -> tuple[bool, str]: label = output.strip().strip(".").capitalize() if label not in LABELS: return False, f"invalid output {output!r}" if label != case["expected"]: return False, f"expected {case['expected']}, got {label}" return True, "" The note distinguishes format failures from wrong answers, which need different fixes (Lesson 114). For other tasks, write checks that use the techniques from Lessons 104, 116, and 117: grounding, rubrics, or required points.
Step 5: Testing Without the API cases = [ {"id": "t1", "input": "Charged twice for React", "expected": "Billing"}, {"id": "t2", "input": "Paid but course locked", "expected": "Access"}, {"id": "t3", "input": "Video freezes", "expected": "Technical"}, {"id": "t4", "input": "Weekend batches?", "expected": "Other"}, {"id": "t5", "input": "Coupon invalid", "expected": "Billing"}, ]
ANSWERS_A = {"Charged": "Billing", "Paid": "Billing", "Video": "Technical", "Weekend": "Other", "Coupon": "Billing"} ANSWERS_B = {"Charged": "Billing", "Paid": "Access", "Video": "Technical.", "Weekend": "General question", "Coupon": "Billing"}
records_a = run_variant("v1", "...", cases, lambda i, text: ANSWERS_A[text.split()[0]], check_label) records_b = run_variant("v2", "...", cases, lambda i, text: ANSWERS_B[text.split()[0]], check_label) print(report("v1", "v2", records_a, records_b)) Output (the run time will show your current date and time):
Prompt Comparison
Run: 2026-11-01 15:45 UTC
| Variant | Passed | Rate | 95% interval |
|---|---|---|---|
| v1 | 4/5 | 80% | 38%-96% |
| v2 | 4/5 | 80% | 38%-96% |
Only v1 passed: t4 Only v2 passed: t2
Failures
- v1 / t2: expected Access, got Billing
- v2 / t4: invalid output 'General question' What the Report Reveals Equal rates hide different behavior. Both variants pass 4 of 5, but they fail different cases. Different kinds of failure. v1 makes a boundary error, the "paid but locked" rule (Lesson 129). v2 fixes that but produces an invalid label for a general question, a format problem. Uncertainty is visible. With 5 cases, the 95% interval runs from 38% to 96%. Five cases cannot show which variant is better; you need many more (Lesson 66). Normalization works. v2's "Technical." with a period still passes, because the check normalizes it. A sensible next step here is a v3 that combines v2's boundary rule with a stricter output rule, followed by evaluation on a much larger test set.
Step 6: The Command-Line Tool main.py
import json import sys from pathlib import Path
from llm import call_model from pipeline import check_label, report, run_variant, save_jsonl
name_a, name_b = sys.argv[1], sys.argv[2] # e.g. v1 v2 cases = [json.loads(line) for line in Path("cases.jsonl").read_text(encoding="utf-8").splitlines() if line.strip()]
results = {} for name in (name_a, name_b): instructions = Path(f"prompts/{name}.txt").read_text(encoding="utf-8") results[name] = run_variant(name, instructions, cases, call_model, check_label) save_jsonl(results[name], f"results_{name}.jsonl")
Path("comparison.md").write_text(report(name_a, name_b, results[name_a], results[name_b]), encoding="utf-8") print("Saved comparison.md and results_*.jsonl") Running It python main.py v1 v2 Put check_label in pipeline.py, or in a separate checks.py module if you evaluate several tasks.
Extending the Pipeline Extension How Repeated runs Run each case several times and report pass rates per case, to detect instability (Lesson 120) Model judge Write a check function that calls a rubric judge (Lesson 116), and validate the judge against people Critical cases Mark cases as critical in cases.jsonl and block when any critical case fails (Lesson 118) Cost and latency Record token usage from API responses and report averages per variant Continuous integration Run the pipeline when prompt files change, and attach the report to the pull request Common Mistakes Mistake 1: Comparing Only Totals What it is: Choosing the variant with the higher pass rate.
Why it happens: One number is easy to compare.
What it causes: Important cases broken by the "better" variant go unnoticed.
How to fix it: Read the paired comparison and the failure list.
Mistake 2: Deciding on Tiny Test Sets What it is: Declaring a winner after a handful of cases.
Why it happens: Building test sets takes time.
What it causes: Decisions based on chance.
How to fix it: Report intervals, and grow the test set until differences are clear (Lesson 119).
Mistake 3: Discarding Outputs What it is: Storing only pass/fail.
Why it happens: Scores seem to be all that matters.
What it causes: You cannot investigate failures or re-score with better checks.
How to fix it: Save every output with its variant, case, and timing.
Best Practices Separate the pipeline from the check. One pipeline, many tasks.
Run all variants on identical cases. Fair, paired comparisons.
Store everything. Outputs, notes, timing, and versions.
Report intervals and paired results. Not just averages.
Explain failures by type. Format, boundary, and content problems need different fixes.
Try It Yourself Exercise 1: A Real Comparison Create cases.jsonl with at least 40 labeled tickets and two prompt files that differ in one rule. Run the pipeline and decide, from the report, whether the change should be kept.
Exercise 2: A Different Task Write a check function for the summarizer from Project 130 that fails a summary if it contains numbers not in the source or exceeds the word limit. Compare two summarization prompts with the pipeline.
Exercise 3: Repeated Runs Extend run_variant() with a repeats parameter that runs each case several times and records the pass rate per case. Add a section to the report listing unstable cases, those passing in some runs but not all.