Building an End-to-End Judge Pipeline: Dataset-Level Aggregation and Confidence-Based Escalation

This curriculum's llm-evaluation topic already shows how to run a single pairwise judge comparison, with position-swapping to cancel position bias. A PRODUCTION judge pipeline needs three additional pieces that single-comparison example doesn't cover: running across an entire dataset (not one example), aggregating those per-example results into something actionable, and deciding which specific results are trustworthy enough to act on without human review.

Dataset-level aggregation means running the judge across every example in an evaluation set (potentially hundreds or thousands of examples), then summarizing the results in ways that are actually useful for decision-making — not just a giant list of individual scores. Useful aggregate views include: the overall average score (and its distribution, using the calibration-adjacent framing from eval-metrics-fundamentals's own calibration subtopic — is the score distribution itself trustworthy, or does it cluster suspiciously at round numbers suggesting the judge isn't discriminating carefully); a breakdown by category or difficulty tier, so you can see WHERE quality is weakest rather than just one blended number; and a comparison against a PREVIOUS run's aggregate scores, to detect regressions before they ship (directly connecting to the drift-detection material in this curriculum's llm-observability topic).

Confidence-based escalation is the piece that makes an automated judge pipeline genuinely trustworthy rather than a black box you hope is right: rather than treating every judge verdict as equally final, flag specific results for human review when the judge shows signs of low confidence in its own verdict — for instance, using the previous subtopic's probability-weighted scoring, a score that lands right on a decision boundary (like 2.9 when your pass/fail threshold is 3.0) is a much weaker signal than a score of 4.8, and deserves a second look before being trusted automatically. Another useful confidence signal: running the SAME comparison multiple times (or with position-swapping, per the existing subtopic's bias mitigation) and flagging cases where the judge DISAGREES with itself across repeated runs — genuine disagreement across repeated judgments is a strong signal the case is inherently ambiguous and warrants a human decision rather than blind trust in whichever verdict happened to come back first.

Put together, this turns 'LLM-as-judge' from a single clever prompting trick into an actual production evaluation SYSTEM: it runs at scale across a real dataset, produces aggregate views that support real decisions (ship or don't ship, which category needs attention), and — critically — knows its own limits well enough to route the genuinely uncertain cases to a human rather than silently guessing on everything with equal unwarranted confidence.

# A dataset-level judge pipeline: run over many examples, aggregate
# results meaningfully, and escalate low-confidence/disagreeing cases.

def judge_single_example(example: dict) -> dict:
    """Stand-in for a real probability-weighted G-Eval-style judge call
    (previous subtopic) -- returns a score plus a repeated-run check."""
    score = example["simulated_score"]
    repeated_run_score = example["simulated_repeated_run_score"]
    return {"id": example["id"], "category": example["category"],
            "score": score, "repeated_run_score": repeated_run_score}

def needs_human_escalation(result: dict, pass_threshold: float = 3.0,
                           boundary_margin: float = 0.2, disagreement_margin: float = 0.5) -> bool:
    """Two independent confidence signals: near the decision boundary,
    OR disagreement between repeated judge runs."""
    near_boundary = abs(result["score"] - pass_threshold) < boundary_margin
    disagreement = abs(result["score"] - result["repeated_run_score"]) > disagreement_margin
    return near_boundary or disagreement

def run_judge_pipeline(dataset: list[dict]) -> dict:
    results = [judge_single_example(ex) for ex in dataset]
    escalated = [r for r in results if needs_human_escalation(r)]
    auto_approved = [r for r in results if r not in escalated]

    by_category = {}
    for r in results:
        by_category.setdefault(r["category"], []).append(r["score"])
    category_averages = {cat: sum(scores) / len(scores) for cat, scores in by_category.items()}

    return {
        "total_examples": len(results),
        "overall_average": sum(r["score"] for r in results) / len(results),
        "category_averages": category_averages,
        "auto_approved_count": len(auto_approved),
        "escalated_to_human": [r["id"] for r in escalated],
    }

eval_dataset = [
    {"id": "ex1", "category": "summarization", "simulated_score": 4.8, "simulated_repeated_run_score": 4.7},
    {"id": "ex2", "category": "summarization", "simulated_score": 2.9, "simulated_repeated_run_score": 3.0},  # near boundary
    {"id": "ex3", "category": "qa",            "simulated_score": 4.5, "simulated_repeated_run_score": 2.8},  # disagreement
    {"id": "ex4", "category": "qa",            "simulated_score": 4.6, "simulated_repeated_run_score": 4.5},
]

report = run_judge_pipeline(eval_dataset)
for key, value in report.items():
    print(f"{key}: {value}")