feat: add paired offline LLM response comparison

README.md

@@ -158,10 +158,15 @@ streamlit run app/app.py
 
 ## Production evolution
 
+Paired offline comparison is available with `python -m scripts.compare_runs baseline.json candidate.json`.
+Each file maps the dataset's case IDs to saved responses. The report holds on missing,
+unexpected or failed candidate cases, and flags an aggregate score decrease for review.
+It evaluates supplied responses; it does not call a model or claim live provider validation.
+
 Natural extensions include:
 
 - multi-turn benchmark cases;
-- prompt/model baseline-vs-candidate comparison;
+- live prompt/model comparison using the paired offline gate;
 - latency and cost budgets;
 - RAG faithfulness checks;
 - historical run storage;

promptpulse/comparison.py

@@ -0,0 +1,75 @@
+"""Paired, offline baseline-versus-candidate evaluation."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Mapping
+
+from .evaluation import evaluate_response
+
+
+@dataclass(frozen=True)
+class ComparisonReport:
+    decision: str
+    baseline_mean: float
+    candidate_mean: float
+    regressed_cases: tuple[str, ...]
+    failed_candidate_cases: tuple[str, ...]
+    missing_cases: tuple[str, ...]
+    unexpected_cases: tuple[str, ...]
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+def compare_runs(
+    dataset: list[dict[str, Any]],
+    baseline: Mapping[str, str],
+    candidate: Mapping[str, str],
+    *,
+    mean_tolerance: float = 0.02,
+) -> ComparisonReport:
+    if not dataset or mean_tolerance < 0:
+        raise ValueError("A non-empty dataset and non-negative tolerance are required.")
+    cases = {row["id"]: row for row in dataset}
+    if len(cases) != len(dataset):
+        raise ValueError("Dataset case IDs must be unique.")
+    expected = set(cases)
+    missing = tuple(sorted((expected - baseline.keys()) | (expected - candidate.keys())))
+    unexpected = tuple(sorted((baseline.keys() | candidate.keys()) - expected))
+    aligned = sorted(expected & baseline.keys() & candidate.keys())
+    baseline_scores: list[float] = []
+    candidate_scores: list[float] = []
+    regressions: list[str] = []
+    failures: list[str] = []
+
+    for case_id in aligned:
+        row = cases[case_id]
+        answers = (baseline[case_id], candidate[case_id])
+        if any(not isinstance(answer, str) or not answer.strip() for answer in answers):
+            raise ValueError(f"Case {case_id}: both responses must be non-empty strings.")
+        results = [
+            evaluate_response(
+                user_query=row["user_query"],
+                reference_context=row["reference_context"],
+                response=answer,
+                required_terms=row.get("required_terms", []),
+                forbidden_terms=row.get("forbidden_terms", []),
+            )
+            for answer in answers
+        ]
+        baseline_scores.append(results[0].pulse_score)
+        candidate_scores.append(results[1].pulse_score)
+        if results[0].passed and not results[1].passed:
+            regressions.append(case_id)
+        if not results[1].passed:
+            failures.append(case_id)
+
+    base_mean = round(sum(baseline_scores) / len(aligned), 3) if aligned else 0.0
+    cand_mean = round(sum(candidate_scores) / len(aligned), 3) if aligned else 0.0
+    decision = "HOLD" if missing or unexpected or failures else (
+        "INVESTIGATE" if cand_mean < base_mean - mean_tolerance else "PASS"
+    )
+    return ComparisonReport(
+        decision, base_mean, cand_mean, tuple(regressions), tuple(failures), missing, unexpected
+    )

scripts/compare_runs.py

@@ -0,0 +1,27 @@
+"""Compare two JSON objects mapping dataset case IDs to saved responses."""
+
+import argparse
+import json
+from pathlib import Path
+
+from promptpulse.comparison import compare_runs
+from promptpulse.data import load_dataset
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("baseline", type=Path)
+    parser.add_argument("candidate", type=Path)
+    args = parser.parse_args()
+    baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
+    candidate = json.loads(args.candidate.read_text(encoding="utf-8"))
+    if not isinstance(baseline, dict) or not isinstance(candidate, dict):
+        parser.error("Both run files must be JSON objects keyed by case ID.")
+    report = compare_runs(load_dataset(), baseline, candidate)
+    print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
+    if report.decision != "PASS":
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tests/test_comparison.py

@@ -0,0 +1,32 @@
+from promptpulse.comparison import compare_runs
+from promptpulse.data import load_dataset
+
+
+def _answers():
+    rows = load_dataset()
+    return rows, {row["id"]: row["expected_answer"] for row in rows}
+
+
+def test_same_cases_and_answers_pass():
+    rows, answers = _answers()
+    report = compare_runs(rows, answers, answers)
+    assert report.decision == "PASS"
+    assert report.regressed_cases == ()
+
+
+def test_missing_case_holds_even_if_remaining_answers_are_good():
+    rows, answers = _answers()
+    candidate = dict(answers)
+    candidate.pop(rows[0]["id"])
+    report = compare_runs(rows, answers, candidate)
+    assert report.decision == "HOLD"
+    assert report.missing_cases == (rows[0]["id"],)
+
+
+def test_policy_regression_holds_despite_other_good_cases():
+    rows, answers = _answers()
+    candidate = dict(answers)
+    candidate["refund-policy"] = "Unused products can be refunded within 60 days."
+    report = compare_runs(rows, answers, candidate)
+    assert report.decision == "HOLD"
+    assert "refund-policy" in report.regressed_cases