How to build an LLM-as-a-judge pipeline

Building a reliable LLM-as-a-judge pipeline is one of the highest-leverage investments you can make in an LLM product. Without systematic evaluation, you’re shipping blind — unable to catch regressions, compare models objectively, or know whether your prompt changes actually help. This guide walks through building a pipeline you can trust, from defining criteria to automating CI integration.

Step 1: Define your evaluation criteria as code

Start by codifying what “good” looks like. Vague vibes don’t scale. Write your criteria as structured rubrics with explicit scoring guidance. Each criterion should be independently scorable and map to a user-visible outcome.

# eval/criteria.py
from enum import Enum
from pydantic import BaseModel, Field
from typing import Literal

class Score(int, Enum):
    POOR = 1
    FAIR = 2
    GOOD = 3
    EXCELLENT = 4
    PERFECT = 5

class Criterion(BaseModel):
    name: str
    scoring_guide: dict[Score, str]  # maps score -> concrete behavior
    weight: float = 1.0

CRITERIA = [
    Criterion(
        name="instruction_following",
        description="Does the response satisfy all explicit constraints in the prompt?",
        scoring_guide={
            Score.POOR: "Ignores major constraints (format, length, forbidden content)",
            Score.FAIR: "Misses minor constraints or partially follows",
            Score.GOOD: "Follows all constraints with minor imperfections",
            Score.EXCELLENT: "Follows all constraints precisely",
            Score.PERFECT: "Flawless adherence, anticipates edge cases",
        },
        weight=1.5,
    ),
    Criterion(
        name="factual_accuracy",
        description="Are claims verifiably correct? No hallucinations.",
        scoring_guide={
            Score.POOR: "Fabricates facts, cites non-existent sources",
            Score.FAIR: "Some correct claims mixed with unverifiable ones",
            Score.GOOD: "Mostly accurate, minor imprecisions",
            Score.EXCELLENT: "Fully accurate, well-grounded",
            Score.PERFECT: "Accurate with precise citations and nuance",
        },
        weight=2.0,
    ),
    Criterion(
        name="reasoning_quality",
        description="Is the logic sound? Steps clearly explained?",
        scoring_guide={
            Score.POOR: "Non-sequiturs, circular reasoning, or no reasoning shown",
            Score.FAIR: "Jump to conclusions, gaps in logic",
            Score.GOOD: "Clear reasoning with minor gaps",
            Score.EXCELLENT: "Rigorous, well-structured reasoning",
            Score.PERFECT: "Exemplary reasoning, teaches the reader",
        },
        weight=1.0,
    ),
]

Store this in version control. When stakeholders debate quality, you point to the rubric — not opinions.

Step 2: Build a golden dataset with known ground truth

You need a representative sample of inputs where you know the correct answer or the ideal response characteristics. Aim for 50–200 examples covering your task distribution: edge cases, adversarial inputs, typical queries, and known failure modes.

# eval/golden_set.py
from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class GoldenExample:
    input: str
    reference_output: Optional[str] = None  # for exact-match tasks
    metadata: dict = None  # tags: "edge_case", "adversarial", "typical", etc.
    expected_scores: dict[str, int] = None  # criterion_name -> expected Score

GOLDEN_SET = [
    GoldenExample(
        input="Summarize the Q3 earnings call in 3 bullet points, max 20 words each.",
        reference_output=None,
        metadata={"tags": ["constraint_heavy", "format_sensitive"]},
        expected_scores={"instruction_following": 5, "factual_accuracy": 5},
    ),
    GoldenExample(
        input="What's the capital of Australia?",
        reference_output="Canberra",
        metadata={"tags": ["factual", "simple"]},
        expected_scores={"factual_accuracy": 5},
    ),
    GoldenExample(
        input="Write a Python function that returns the nth Fibonacci number using recursion.",
        reference_output=None,
        metadata={"tags": ["code", "correctness_critical"]},
        expected_scores={"instruction_following": 5, "factual_accuracy": 5, "reasoning_quality": 4},
    ),
    # ... 50+ more examples
]

def load_golden_set(path: str = "eval/golden.jsonl") -> list[GoldenExample]:
    examples = []
    with open(path) as f:
        for line in f:
            data = json.loads(line)
            examples.append(GoldenExample(**data))
    return examples

Commit this dataset. Treat it like test fixtures — it’s the contract your pipeline validates against.

Step 3: Implement the judge prompt with structured output

The judge prompt is itself a prompt engineering task. Use a system prompt that forces structured JSON output so you can parse scores programmatically. Include few-shot examples calibrated to your rubric.

# eval/judge.py
from pydantic import BaseModel, Field
from typing import Literal
import json

class JudgeScore(BaseModel):
    criterion: str
    score: Literal[1, 2, 3, 4, 5]
    reasoning: str = Field(description="2-3 sentences justifying the score")
    evidence: list[str] = Field(default_factory=list, description="Quotes from response supporting the score")

class JudgeOutput(BaseModel):
    scores: list[JudgeScore]
    overall_assessment: str

JUDGE_SYSTEM_PROMPT = """You are an expert evaluator. Score the model response against each criterion using the provided rubric.

Output ONLY valid JSON matching this schema:
{
  "scores": [
    {"criterion": "instruction_following", "score": 4, "reasoning": "...", "evidence": ["..."]}
  ],
  "overall_assessment": "Summary paragraph"
}

RUBRIC:
{criteria_json}

Few-shot examples:
{examples_json}
"""

def build_judge_prompt(criteria: list[Criterion], few_shots: list[dict]) -> str:
    criteria_json = json.dumps([{
        "name": c.name,
        "description": c.description,
        "scoring_guide": {str(k): v for k, v in c.scoring_guide.items()},
        "weight": c.weight,
    } for c in criteria], indent=2)
    
    examples_json = json.dumps(few_shots, indent=2)
    return JUDGE_SYSTEM_PROMPT.format(criteria_json=criteria_json, examples_json=examples_json)

Calibration tip: Spend 2–3 hours with 2–3 engineers independently scoring 20–30 examples using your rubric. Compute inter-rater agreement (Cohen’s kappa). If kappa < 0.7, your rubric is ambiguous — rewrite the scoring guides until humans agree. The judge will only be as consistent as your human calibration.

Step 4: Wire the evaluation loop

Now connect the pieces: load golden set, call the model under test, call the judge, aggregate results. Keep it synchronous and deterministic for CI; async/parallel comes later.

# eval/run.py
import asyncio
from dataclasses import dataclass
from typing import Callable
import openai  # or your preferred client
from .criteria import CRITERIA, Score
from .golden_set import GoldenExample, load_golden_set
from .judge import JudgeOutput, build_judge_prompt

@dataclass
class EvalResult:
    example: GoldenExample
    model_response: str
    judge_output: JudgeOutput
    passed: bool

async def evaluate_example(
    example: GoldenExample,
    model_fn: Callable[[str], str],
    judge_client: openai.AsyncOpenAI,
    judge_model: str,
    few_shots: list[dict],
) -> EvalResult:
    # 1. Get model response
    model_response = await model_fn(example.input)
    
    # 2. Build judge prompt
    judge_prompt = build_judge_prompt(CRITERIA, few_shots)
    user_prompt = f"INPUT:\n{example.input}\n\nRESPONSE TO EVALUATE:\n{model_response}"
    
    # 3. Call judge with structured output
    completion = await judge_client.beta.chat.completions.parse(
        model=judge_model,
        messages=[
            {"role": "system", "content": judge_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_format=JudgeOutput,
        temperature=0.0,
    )
    judge_output = completion.choices[0].message.parsed
    
    # 4. Determine pass/fail: weighted average >= threshold
    weighted_sum = sum(
        s.score * next(c.weight for c in CRITERIA if c.name == s.criterion)
        for s in judge_output.scores
    )
    total_weight = sum(c.weight for c in CRITERIA)
    weighted_avg = weighted_sum / total_weight
    passed = weighted_avg >= 3.5  # tune per project
    
    return EvalResult(
        example=example,
        model_response=model_response,
        judge_output=judge_output,
        passed=passed,
    )

async def run_eval_suite(
    model_fn: Callable[[str], str],
    judge_model: str = "gpt-4o",
    golden_path: str = "eval/golden.jsonl",
) -> list[EvalResult]:
    examples = load_golden_set(golden_path)
    few_shots = load_few_shots()  # your calibrated examples
    judge_client = openai.AsyncOpenAI()
    
    results = []
    for ex in examples:
        result = await evaluate_example(ex, model_fn, judge_client, judge_model, few_shots)
        results.append(result)
    return results

Step 5: Add regression detection and alerting

A pipeline that only runs locally is a toy. Integrate into CI so every PR gets evaluated. Fail the build if core metrics regress.

# .github/workflows/llm-eval.yml
name: LLM Evaluation
on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'eval/**'
      - 'src/llm/**'

jobs:
  evaluate:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements-eval.txt
      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          JUDGE_MODEL: gpt-4o
        run: |
          python -m eval.run --output eval_results.json
      - name: Check regression
        run: |
          python -m eval.check_regression eval_results.json
      - name: Upload artifacts
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: eval_results.json

The regression check compares against a baseline stored in the repo (or an artifact from main):

# eval/check_regression.py
import json
import sys
from pathlib import Path

BASELINE_PATH = Path("eval/baseline.json")

def load_baseline() -> dict:
    if BASELINE_PATH.exists():
        return json.loads(BASELINE_PATH.read_text())
    return {}

def compute_metrics(results: list[dict]) -> dict:
    total = len(results)
    passed = sum(1 for r in results if r["passed"])
    by_tag = {}
    for r in results:
        for tag in r["example"]["metadata"].get("tags", []):
            by_tag.setdefault(tag, {"total": 0, "passed": 0})
            by_tag[tag]["total"] += 1
            if r["passed"]:
                by_tag[tag]["passed"] += 1
    return {
        "overall_pass_rate": passed / total if total else 0,
        "by_tag": {k: v["passed"] / v["total"] for k, v in by_tag.items()},
    }

def main(results_path: str):
    results = json.loads(Path(results_path).read_text())
    current = compute_metrics(results)
    baseline = load_baseline()
    
    # Fail if overall pass rate drops > 5pp or any tag drops > 10pp
    overall_drop = baseline.get("overall_pass_rate", 1.0) - current["overall_pass_rate"]
    if overall_drop > 0.05:
        print(f"FAIL: Overall pass rate dropped {overall_drop:.1%}")
        sys.exit(1)
    
    for tag, rate in current["by_tag"].items():
        baseline_rate = baseline.get("by_tag", {}).get(tag, 1.0)
        if baseline_rate - rate > 0.10:
            print(f"FAIL: Tag '{tag}' pass rate dropped {baseline_rate - rate:.1%}")
            sys.exit(1)
    
    print("PASS: No significant regression")
    # Optionally update baseline on main branch
    if "--update-baseline" in sys.argv:
        BASELINE_PATH.write_text(json.dumps(current, indent=2))

if __name__ == "__main__":
    main(sys.argv[1])

Step 6: Instrument for observability

You’ll need to debug why a specific example failed. Log everything: inputs, model responses, judge reasoning, scores. Structure logs for querying.

# eval/logging.py
import structlog
from datetime import datetime
from .run import EvalResult

logger = structlog.get_logger()

def log_evaluation(result: EvalResult, run_id: str):
    logger.info(
        "eval_complete",
        run_id=run_id,
        example_id=id(result.example),
        input=result.example.input[:200],
        model_response=result.model_response[:500],
        judge_scores={s.criterion: s.score for s in result.judge_output.scores},
        judge_reasoning={s.criterion: s.reasoning for s in result.judge_output.scores},
        passed=result.passed,
        tags=result.example.metadata.get("tags", []),
        timestamp=datetime.utcnow().isoformat(),
    )

Ship these logs to your observability stack (Datadog, Honeycomb, Loki). When a regression alert fires, you can filter by tag, score dimension, or time window to find the pattern.

Step 7: Verify the pipeline works

Before trusting this in CI, run three validation passes:

1. Sanity check: Run against a known-good model (e.g., GPT-4o) and a known-bad model (e.g., a tiny local model or a deliberately broken prompt). The pipeline should clearly separate them.

# Expected: GPT-4o pass rate > 90%, broken model < 30%
python -m eval.run --model gpt-4o --output good.json
python -m eval.run --model broken_prompt --output bad.json
python -m eval.compare good.json bad.json

2. Judge consistency test: Re-run the same evaluation 5 times with temperature=0. The judge should produce identical scores each time. If not, your prompt or rubric is ambiguous.

# eval/consistency_check.py
async def check_judge_consistency(judge_model: str, n_runs: int = 5):
    example = load_golden_set()[0]
    model_response = "Fixed response to evaluate"
    judge_client = openai.AsyncOpenAI()
    few_shots = load_few_shots()
    
    scores_by_run = []
    for _ in range(n_runs):
        result = await evaluate_example(example, lambda _: model_response, judge_client, judge_model, few_shots)
        scores_by_run.append({s.criterion: s.score for s in result.judge_output.scores})
    
    # All runs should match exactly at temperature=0
    for criterion in scores_by_run[0]:
        values = [run[criterion] for run in scores_by_run]
        if len(set(values)) > 1:
            print(f"INCONSISTENT: {criterion} = {values}")
            return False
    print("CONSISTENT: All runs match")
    return True

3. Human spot-check: Randomly sample 20 evaluated examples. Read the input, model response, and judge reasoning. Do you agree with the scores? If you disagree > 15% of the time, your judge prompt or rubric needs work — not the model under test.

Step 8: Iterate on the rubric, not the judge model

A common trap: swapping judge models (GPT-4o → Claude → Gemini) hoping for better alignment. The judge model matters far less than rubric clarity. If human raters disagree, no LLM judge will be consistent.

Instead, treat rubric iteration as the primary quality lever:

• Add a new criterion when you discover a failure mode not captured
• Split a criterion when its scoring guide covers orthogonal behaviors
• Adjust weights when business priority shifts
• Version your rubric (criteria_v1.py, criteria_v2.py) and tag golden set versions to match

Operational notes

Cost control: Judge calls are cheap relative to production inference, but they add up at scale. Cache judge outputs keyed by (input, model_response, rubric_version). Invalidate on rubric changes.

Judge model choice: Use a strong reasoning model (GPT-4o, Claude 3.5 Sonnet) for the judge. Weaker models struggle with nuanced rubrics. The judge doesn’t need to be the same provider as your production model.

Parallelization: The evaluation loop is embarrassingly parallel. Use asyncio.gather with a semaphore (limit 10–20 concurrent) to saturate your rate limits without triggering 429s.

Fallback handling: If you route production traffic through a gateway that supports automatic fallback (like n4n.ai does across 240+ models), your eval pipeline should test the entire routing path — not just the primary model. A model that scores well but triggers fallback to a worse model under load is a regression.

What good looks like

After wiring this up, you should be able to:

• Open a PR that changes a system prompt, see the eval run in CI, and know within 5 minutes if it’s a net positive
• Query “show me all code tagged examples where factual_accuracy < 4 in the last 30 days”
• Onboard a new engineer who can read the rubric and independently score examples to within 1 point of the judge

That’s the pipeline. It’s not magic — it’s discipline encoded in code. Start with 20 golden examples and a 3-criterion rubric. Ship it to CI this week. Expand from there.