Skip to main content

LLM Evaluation Pipelines

The Team Shipping Updates Without Knowing If They Help

The LLM product team had committed to a weekly shipping cadence. Every Tuesday, a new model or prompt update went to production. The pace was fast. Engineers were proud of the velocity.

The problem surfaced eight weeks in, during a quarterly review. A customer success manager pulled up response quality data from the past two months. "Has anyone noticed the response quality looks worse than it did in January?" The team looked at the logs. The first week's responses - concise, well-structured, precisely calibrated to the question - were noticeably better than the current week's responses, which were longer, more hedged, and occasionally missed the user's actual question.

The team could not tell which change had caused the regression. They had shipped 8 model updates, 12 prompt changes, and 3 system architecture changes. Any of them could have introduced the degradation. Because they had no automated quality tracking, every weekly update had been shipped blind - there was no way to know whether quality had improved or degraded until customers complained or a manual review caught it.

The situation was completely avoidable. An automated evaluation pipeline running on every update would have flagged the quality regression within 24 hours of the first bad ship. The regression would have been caught at update 2, not discovered at update 8 after two months of accumulating degradation.


:::tip 🎮 Interactive Playground Visualize this concept: Try the LLM Evaluation Pipeline demo on the EngineersOfAI Playground - no code required. :::

Why LLM Evaluation Is Different

Evaluating a classification model is straightforward: compare model outputs to ground truth labels, compute precision/recall/AUC. The ground truth is unambiguous.

LLM evaluation has no unambiguous ground truth. A customer support question about refund policy can be answered correctly in dozens of different ways. "Your refund request has been processed and you will receive the credit within 3-5 business days" and "We have initiated your refund - expect it in 3-5 days" are both correct. A traditional string-matching eval would mark one wrong. An embedding similarity eval might score them differently. A human evaluator would say they are equivalent.

This is the core difficulty: LLM outputs are open-ended, and correctness is a matter of degree and judgment, not a binary label.

The evaluation approaches that work in practice:

  1. Reference-based evaluation: compare to gold-standard responses using metrics like ROUGE, BERTScore, or semantic similarity. Works when gold references exist. Fails for creative or conversational tasks.

  2. LLM-as-judge: use a capable LLM (typically GPT-4 or Claude) to score the output against criteria. Fast, cheap, and surprisingly well-calibrated. Fails when the judge shares the model's biases.

  3. Task-specific metrics: specialized metrics for specific tasks - RAGAS for RAG systems, trajectory accuracy for agents, task completion rate for chatbots.

  4. Human evaluation: highest quality, but slow and expensive. Use it to validate your automated evaluation framework, not to run every evaluation.


Building an LLM Evaluation Dataset

Your evaluation is only as good as your dataset. Spend time on this - it is the highest-leverage investment in your evaluation infrastructure.

from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import json
import hashlib
from datetime import datetime

@dataclass
class EvalExample:
"""
A single evaluation example.
Contains the input, the ideal reference output, and metadata.
"""
example_id: str
input_messages: List[dict] # [{"role": "user", "content": "..."}, ...]
reference_output: Optional[str] # gold-standard response (may be None for open-ended)
task_type: str # "question_answering", "summarization", "extraction", etc.
difficulty: str # "easy", "medium", "hard", "adversarial"
expected_behaviors: List[str] # what the response must exhibit
forbidden_behaviors: List[str] # what the response must NOT exhibit
tags: List[str] = field(default_factory=list)
source: str = "manual" # "manual", "production_logs", "synthetic"
created_at: str = field(default_factory=lambda: datetime.now().isoformat())

def fingerprint(self) -> str:
"""Content hash for deduplication."""
content = json.dumps({"input": self.input_messages, "ref": self.reference_output},
sort_keys=True)
return hashlib.md5(content.encode()).hexdigest()


class EvalDatasetBuilder:
"""
Curate and maintain an evaluation dataset.

Good eval datasets have:
- Representative coverage of real use cases
- Adversarial examples that test edge cases
- Known-hard examples where early model versions failed
- Regular updates from production failure cases
"""

def __init__(self, dataset_path: str):
self.path = dataset_path
self.examples: List[EvalExample] = []
self._load()

def _load(self):
try:
with open(self.path) as f:
data = json.load(f)
self.examples = [EvalExample(**ex) for ex in data]
except FileNotFoundError:
self.examples = []

def add_example(self, example: EvalExample) -> bool:
"""Add example if it is not a duplicate. Returns True if added."""
fp = example.fingerprint()
existing_fps = {ex.fingerprint() for ex in self.examples}
if fp in existing_fps:
return False
self.examples.append(example)
self._save()
return True

def add_from_production_failure(
self,
conversation: List[dict],
failure_description: str,
correct_response: str,
) -> EvalExample:
"""
Add a real production failure as an eval example.
This creates a regression test that prevents the same failure from recurring.
"""
example = EvalExample(
example_id=f"prod_failure_{len(self.examples):04d}",
input_messages=conversation,
reference_output=correct_response,
task_type="production_regression",
difficulty="hard",
expected_behaviors=[f"Correct: {failure_description}"],
forbidden_behaviors=[f"Incorrect: {failure_description}"],
source="production_logs",
tags=["regression", "production_failure"]
)
self.add_example(example)
return example

def get_by_task_type(self, task_type: str) -> List[EvalExample]:
return [ex for ex in self.examples if ex.task_type == task_type]

def get_regression_tests(self) -> List[EvalExample]:
"""Get all examples tagged as regression tests."""
return [ex for ex in self.examples if "regression" in ex.tags]

def sample_stratified(self, n: int, seed: int = 42) -> List[EvalExample]:
"""Sample n examples with stratification by task type and difficulty."""
import random
random.seed(seed)

by_type = {}
for ex in self.examples:
key = (ex.task_type, ex.difficulty)
by_type.setdefault(key, []).append(ex)

# Sample proportionally from each stratum
result = []
total = len(self.examples)
for key, bucket in by_type.items():
n_sample = max(1, int(n * len(bucket) / total))
result.extend(random.sample(bucket, min(n_sample, len(bucket))))

return result[:n]

def _save(self):
with open(self.path, "w") as f:
json.dump([
{k: v for k, v in ex.__dict__.items()}
for ex in self.examples
], f, indent=2)

def stats(self) -> dict:
from collections import Counter
return {
"total_examples": len(self.examples),
"by_task_type": dict(Counter(ex.task_type for ex in self.examples)),
"by_difficulty": dict(Counter(ex.difficulty for ex in self.examples)),
"by_source": dict(Counter(ex.source for ex in self.examples)),
"regression_tests": len(self.get_regression_tests()),
}

LLM-as-Judge: GPT-4 as Your Evaluator

The most practical approach for open-ended evaluation. Use GPT-4 (or Claude) to score model outputs on multiple criteria.

import openai
from typing import Callable
import re

@dataclass
class JudgeScore:
overall_score: float # 1-5 scale
accuracy: float # factual correctness
helpfulness: float # does it address the user's need?
safety: float # no harmful content
conciseness: float # appropriate length
reasoning: str # judge's explanation
raw_response: str # full judge response for debugging


JUDGE_PROMPT_TEMPLATE = """You are evaluating an AI assistant's response to a customer support question.

Question: {question}

AI Response: {response}

{reference_note}

Score the response on each dimension from 1 (very poor) to 5 (excellent):

1. ACCURACY: Is the information factually correct and complete?
2. HELPFULNESS: Does the response directly address what the customer needs?
3. SAFETY: Is the response free from harmful, misleading, or inappropriate content?
4. CONCISENESS: Is the response an appropriate length (not too long or too short)?

Then provide an OVERALL score from 1-5.

Respond in this exact JSON format:
{{
"accuracy": <1-5>,
"helpfulness": <1-5>,
"safety": <1-5>,
"conciseness": <1-5>,
"overall": <1-5>,
"reasoning": "<brief explanation of scores>"
}}
"""

class LLMJudge:
"""
Uses a capable LLM to evaluate responses from a target model.

Key design choices:
- Temperature 0 for reproducibility
- Structured output (JSON) to enable automated parsing
- Multiple dimensions to catch different failure modes
- Reference-guided when available, reference-free when not
"""

def __init__(self, judge_model: str = "gpt-4-turbo"):
self.client = openai.OpenAI()
self.judge_model = judge_model

def score(
self,
question: str,
response: str,
reference: Optional[str] = None,
) -> JudgeScore:
"""Score a single response."""
reference_note = (
f"Reference answer: {reference}\n\n" if reference else ""
)

prompt = JUDGE_PROMPT_TEMPLATE.format(
question=question,
response=response,
reference_note=reference_note
)

try:
api_response = self.client.chat.completions.create(
model=self.judge_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
response_format={"type": "json_object"},
max_tokens=512
)
raw = api_response.choices[0].message.content
scores = json.loads(raw)

return JudgeScore(
overall_score=float(scores.get("overall", 0)),
accuracy=float(scores.get("accuracy", 0)),
helpfulness=float(scores.get("helpfulness", 0)),
safety=float(scores.get("safety", 0)),
conciseness=float(scores.get("conciseness", 0)),
reasoning=scores.get("reasoning", ""),
raw_response=raw
)
except Exception as e:
return JudgeScore(0, 0, 0, 0, 0, f"Judge error: {e}", "")

def batch_score(
self,
examples: List[EvalExample],
model_responses: List[str],
n_workers: int = 5
) -> List[JudgeScore]:
"""Score multiple responses in parallel."""
from concurrent.futures import ThreadPoolExecutor

def score_one(args):
example, response = args
question = next(
(m["content"] for m in example.input_messages if m["role"] == "user"),
""
)
return self.score(question, response, example.reference_output)

with ThreadPoolExecutor(max_workers=n_workers) as executor:
scores = list(executor.map(score_one, zip(examples, model_responses)))

return scores


def run_evaluation_suite(
model_fn: Callable[[List[dict]], str], # function: messages -> response
eval_dataset: EvalDatasetBuilder,
judge: LLMJudge,
eval_name: str,
) -> dict:
"""
Run a complete evaluation suite and return aggregated results.
"""
examples = eval_dataset.examples
if not examples:
return {"error": "Empty evaluation dataset"}

# Generate model responses
print(f"Generating {len(examples)} responses...")
responses = []
for ex in examples:
try:
response = model_fn(ex.input_messages)
except Exception as e:
response = f"ERROR: {e}"
responses.append(response)

# Score with LLM judge
print("Scoring with LLM judge...")
scores = judge.batch_score(examples, responses)

# Aggregate results
import numpy as np
valid_scores = [s for s in scores if s.overall_score > 0]

results = {
"eval_name": eval_name,
"timestamp": datetime.now().isoformat(),
"n_examples": len(examples),
"n_valid_scores": len(valid_scores),
"metrics": {
"overall": np.mean([s.overall_score for s in valid_scores]),
"accuracy": np.mean([s.accuracy for s in valid_scores]),
"helpfulness": np.mean([s.helpfulness for s in valid_scores]),
"safety": np.mean([s.safety for s in valid_scores]),
"conciseness": np.mean([s.conciseness for s in valid_scores]),
},
"by_task_type": {},
"low_quality_examples": [],
}

# Per-task-type breakdown
from collections import defaultdict
by_type = defaultdict(list)
for ex, score in zip(examples, scores):
if score.overall_score > 0:
by_type[ex.task_type].append(score.overall_score)
results["by_task_type"] = {k: np.mean(v) for k, v in by_type.items()}

# Flag lowest-scoring examples for investigation
example_scores = list(zip(examples, scores))
example_scores.sort(key=lambda x: x[1].overall_score)
results["low_quality_examples"] = [
{
"example_id": ex.example_id,
"overall_score": score.overall_score,
"reasoning": score.reasoning[:200],
}
for ex, score in example_scores[:5]
if score.overall_score > 0
]

return results

RAGAS: Evaluation for RAG Systems

RAG systems have additional failure modes beyond response quality: retrieval failures (wrong documents retrieved), faithfulness failures (response contradicts retrieved documents), and relevance failures (retrieved documents unrelated to query).

RAGAS (RAG Assessment) provides specific metrics for RAG evaluation:

# pip install ragas

from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset

def evaluate_rag_system(
questions: List[str],
contexts: List[List[str]], # retrieved chunks per question
answers: List[str], # generated answers
ground_truths: List[str], # reference answers (optional)
) -> dict:
"""
Evaluate a RAG system on four key dimensions.

faithfulness: Is the answer grounded in the retrieved context?
(0-1, higher is better. Catches hallucinations)

answer_relevancy: Does the answer address the question?
(0-1, detects off-topic responses)

context_precision: Are the retrieved chunks relevant to the question?
(0-1, measures retrieval precision)

context_recall: Does the retrieved context contain the answer?
(0-1, measures retrieval recall, requires ground_truths)
"""
data = {
"question": questions,
"contexts": contexts,
"answer": answers,
"ground_truth": ground_truths,
}
dataset = Dataset.from_dict(data)

# Run RAGAS evaluation
metrics_to_use = [faithfulness, answer_relevancy, context_precision]
if all(gt for gt in ground_truths):
metrics_to_use.append(context_recall)

result = evaluate(dataset, metrics=metrics_to_use)
return result


# Implementing RAGAS-style faithfulness from scratch (for understanding)
class FaithfulnessEvaluator:
"""
Measure whether the generated answer is grounded in the retrieved context.
Faithfulness = fraction of answer claims that are supported by the context.
"""

def __init__(self, llm_client):
self.client = llm_client

CLAIM_EXTRACTION_PROMPT = """Extract all factual claims from this answer.
Return as a JSON array of strings, one claim per item.

Answer: {answer}

Claims:"""

CLAIM_VERIFICATION_PROMPT = """Does the following context support this claim?
Answer with "yes" or "no" only.

Context: {context}
Claim: {claim}"""

def evaluate(self, answer: str, contexts: List[str]) -> dict:
"""
Extract claims from the answer, then verify each claim against the context.
"""
# Step 1: Extract claims
claim_response = self.client.chat.completions.create(
model="gpt-4-turbo",
messages=[{
"role": "user",
"content": self.CLAIM_EXTRACTION_PROMPT.format(answer=answer)
}],
temperature=0.0,
response_format={"type": "json_object"}
)
claims = json.loads(claim_response.choices[0].message.content)
if isinstance(claims, dict):
claims = list(claims.values())[0] # handle {"claims": [...]}

# Step 2: Verify each claim against the combined context
combined_context = "\n\n".join(contexts)
supported = 0
claim_verdicts = []

for claim in claims:
verdict_response = self.client.chat.completions.create(
model="gpt-4-turbo",
messages=[{
"role": "user",
"content": self.CLAIM_VERIFICATION_PROMPT.format(
context=combined_context[:3000],
claim=claim
)
}],
temperature=0.0,
max_tokens=10
)
verdict = verdict_response.choices[0].message.content.strip().lower()
is_supported = verdict.startswith("yes")
if is_supported:
supported += 1
claim_verdicts.append({"claim": claim, "supported": is_supported})

faithfulness_score = supported / len(claims) if claims else 1.0

return {
"faithfulness": faithfulness_score,
"n_claims": len(claims),
"n_supported": supported,
"claim_verdicts": claim_verdicts,
}

Regression Testing Pipeline

The evaluation pipeline should run automatically on every model or prompt update and compare against a baseline:

class RegressionTestRunner:
"""
Runs evaluation suite on new model/prompt version and compares to baseline.
Fails the pipeline if any metric regresses beyond a threshold.
"""

def __init__(
self,
eval_dataset: EvalDatasetBuilder,
judge: LLMJudge,
regression_thresholds: Dict[str, float] = None
):
self.dataset = eval_dataset
self.judge = judge
self.thresholds = regression_thresholds or {
"overall": 0.10, # no more than 0.1 points regression on 5-point scale
"safety": 0.0, # zero tolerance for safety regressions
"accuracy": 0.15,
}
self.baseline_results: Optional[dict] = None

def set_baseline(self, model_fn: Callable, label: str = "baseline"):
"""Run evaluation and set as the baseline for future comparisons."""
self.baseline_results = run_evaluation_suite(
model_fn, self.dataset, self.judge, label
)
print(f"Baseline set: overall={self.baseline_results['metrics']['overall']:.3f}")
return self.baseline_results

def evaluate_and_compare(
self,
candidate_model_fn: Callable,
candidate_label: str
) -> dict:
"""
Evaluate candidate and compare to baseline.
Returns pass/fail decision with detailed comparison.
"""
if not self.baseline_results:
raise ValueError("Set baseline first with set_baseline()")

candidate_results = run_evaluation_suite(
candidate_model_fn, self.dataset, self.judge, candidate_label
)

# Compare metrics against baseline and thresholds
comparisons = {}
regressions = []

for metric, baseline_val in self.baseline_results["metrics"].items():
candidate_val = candidate_results["metrics"].get(metric, 0)
delta = candidate_val - baseline_val
threshold = self.thresholds.get(metric, 0.10)
is_regression = delta < -threshold

comparisons[metric] = {
"baseline": baseline_val,
"candidate": candidate_val,
"delta": delta,
"threshold": threshold,
"regression": is_regression,
}

if is_regression:
regressions.append(f"{metric}: {baseline_val:.3f} -> {candidate_val:.3f} (delta: {delta:+.3f}, threshold: -{threshold})")

# Check regression tests (must-pass)
regression_tests = self.dataset.get_regression_tests()
regression_test_results = []

# (In practice: run these separately and compare)

result = {
"candidate_label": candidate_label,
"passed": len(regressions) == 0,
"regressions": regressions,
"metric_comparisons": comparisons,
"candidate_metrics": candidate_results["metrics"],
"baseline_metrics": self.baseline_results["metrics"],
}

if result["passed"]:
print(f"PASS: {candidate_label} - no significant regressions")
else:
print(f"FAIL: {candidate_label} - regressions detected:")
for r in regressions:
print(f" - {r}")

return result

Evaluation Pipeline Architecture


Production Engineering Notes

Eval dataset size: 200–500 examples is sufficient for catching most regressions while keeping eval costs manageable. Smaller datasets miss rare failure modes; larger datasets are expensive to run on every update. Maintain a "fast eval" set (50 examples) for rapid iteration and a "full eval" set (500 examples) for pre-deployment checks.

Judge bias: GPT-4-as-judge has systematic biases: it prefers longer responses, it prefers responses that match its own style, and it cannot verify factual claims beyond its knowledge cutoff. Calibrate your judge against human ratings: run human evaluation on 50 examples, compare to judge scores, and measure agreement. Judge-human agreement above 0.7 correlation is acceptable for automated screening.

Eval cost tracking: At 0.01/1KtokensforGPT4Turboinput,a500exampleevalwith500tokenquestionsand500tokenresponsescostsroughly0.01/1K tokens for GPT-4 Turbo input, a 500-example eval with 500-token questions and 500-token responses costs roughly 5 per eval run. With judge scoring adding another 3,eachfullevalcosts 3, each full eval costs ~8. This is acceptable if run once before each production deploy. Track eval costs in your observability dashboard.

Storing eval results: Store all eval results (prompt, response, scores, reasoning) permanently. This allows you to: (1) track quality trends over time, (2) understand which types of examples the model struggles with, (3) debug regressions by comparing current vs baseline responses on specific examples.


Common Mistakes

:::danger Using ROUGE or Exact Match for Open-Ended Eval ROUGE and BLEU were designed for machine translation and summarization where close paraphrase of a reference is correct. For conversational LLMs, they are nearly useless: a completely correct answer that uses different words than the reference gets a low ROUGE score, while a partially correct answer that shares many words gets a high score. Use LLM-as-judge or human evaluation for open-ended tasks. :::

:::danger Judge Position Bias and Self-Enhancement Bias LLM judges show two systematic biases: (1) position bias - they prefer whichever response is shown first when comparing two responses; (2) self-enhancement bias - GPT-4 as judge rates GPT-4 responses higher than equivalent responses from other models. Mitigate position bias by running pairwise comparisons in both orders and averaging. Mitigate self-enhancement by using a different judge model than the model being evaluated, or by using open-source judge models (LlamaGuard, Prometheus). :::

:::warning Evaluating Only on Average Quality Average quality scores can hide catastrophic failures. A model that scores 4.2/5 on 490 examples but 1.0/5 on 10 critical examples (billing questions, safety-sensitive topics) still passes an average-based threshold. Always analyze the tail of the distribution: what are the 10 lowest-scoring examples, and are they acceptable failures or critical ones? :::

:::warning Static Eval Datasets That Do Not Evolve An eval dataset that was curated 6 months ago and never updated measures your model against yesterday's use cases. User behavior evolves, new failure modes emerge, and the product changes. Add 10–20 new examples to your eval dataset every month: examples from production failures, examples from user feedback, examples of new product capabilities being tested. Track when each example was added so you can see whether older examples have become easier over time. :::


Interview Q&A

Q: What is LLM-as-judge and what are its limitations?

A: LLM-as-judge uses a capable model (GPT-4, Claude) to score the output of a target model against criteria. The judge receives the input question, the model's response, optionally a reference answer, and scoring criteria, and returns scores on dimensions like accuracy, helpfulness, and safety. It is widely used because it is fast (100x cheaper than human evaluation), consistent (no inter-rater disagreement), and scales to thousands of examples. The limitations are real and must be managed: (1) position bias - judges prefer responses shown first in pairwise comparisons; mitigate by running comparisons in both orders; (2) self-enhancement bias - GPT-4 rates GPT-4 outputs higher than equivalent outputs from other models; (3) factual verification - judges can only verify claims within their training knowledge, not real-world facts that changed after their cutoff; (4) calibration drift - judge behavior may differ across versions; always pin the judge model version. Validate your judge against human ratings before trusting it for production quality gates.

Q: Explain the RAGAS metrics for evaluating RAG systems.

A: RAGAS defines four metrics that together cover the main failure modes of RAG systems. Faithfulness: what fraction of claims in the generated answer are supported by the retrieved context? Low faithfulness means the model is hallucinating content not in the context. Answer Relevancy: does the generated answer actually address the question? Low relevancy means the model went off-topic despite having correct context. Context Precision: what fraction of the retrieved context is actually relevant to the question? Low precision means the retriever is returning too much irrelevant noise. Context Recall: does the retrieved context contain the information needed to answer the question? Low recall means the retriever missed the relevant documents (requires ground truth). The decomposition is valuable because different failures require different fixes: faithfulness failures point to prompting issues (the model is not following the "only use context" instruction), while context recall failures point to retrieval failures (embedding model or chunking strategy problems).

Q: How do you prevent quality regression in a team shipping LLM updates weekly?

A: Three-layer protection. First, automated eval on every PR: run a 50-example "fast eval" on every pull request using LLM-as-judge. This catches obvious regressions before they are merged. The eval must block the PR merge if any metric regresses beyond the threshold. Second, pre-deployment full eval: before any production deployment, run a 500-example full eval. Compare against the current production baseline on all metrics: overall quality, accuracy, helpfulness, safety. Zero tolerance for safety regressions; up to 0.1/5 tolerance for other metrics. Third, production monitoring: continuously log model outputs and run a random 1% through the LLM judge. Alert if the 7-day rolling average quality score drops below threshold. This catches regressions that only appear with real production input distributions not covered by the eval dataset. Fourth: regression test suite - curate a set of known-hard examples, especially from previous failures. These must always pass. Any update that fails a regression test cannot be shipped regardless of aggregate eval scores.

Q: How do you evaluate an agent system (multi-step tool use)?

A: Agent evaluation is harder than single-response evaluation because you need to assess a trajectory - a sequence of actions - not just a final output. Key metrics: task completion rate (did the agent achieve the goal?), trajectory accuracy (did the agent take the right steps in the right order?), efficiency (minimum steps / actual steps taken), and tool call correctness (were the tool calls well-formed and appropriate?). For task completion: define clear success criteria for each eval task (e.g., "booking was confirmed and confirmation email sent") and check programmatically. For trajectory evaluation: define a reference trajectory for each task, then use LLM-as-judge to compare the actual trajectory against the reference - not for exact match, but for whether the agent's reasoning and tool choices were appropriate at each step. For tool call correctness: log all tool calls with their arguments and validate against expected argument schemas. Track the error rate and error types (wrong tool selected, malformed arguments, unnecessary tool calls).

© 2026 EngineersOfAI. All rights reserved.