Building an LLM Evaluation Suite from Scratch
Building an LLM Evaluation Suite from Scratch guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable,.
Muhammad Abdul Sami
· 9 min read
- LLM
- LLM Serving
- Evaluation
- Cost Optimization
Table of Contents:
- Why Build Your Own Eval Suite
- Core Components of an Eval Framework
- Dataset Management and Versioning
- Scoring Engine Architecture
- Automated Judge Implementation
- Regression Detection System
- CI/CD Integration Patterns
- Real Production Examples
- Performance at Scale
- Frequently Asked Questions
Why Build Your Own Eval Suite
Short answer: Building an LLM evaluation suite from scratch gives you control over metrics, datasets, and integration with your deployment pipeline — something off-the-shelf frameworks can't match for production AI systems.
After shipping AI agent evaluation systems for 15+ companies at HinterBuild, the pattern is clear: teams that build custom eval infrastructure detect regressions 3-5 days faster than teams using generic solutions. When you're deploying prompt changes weekly and model updates monthly, that speed difference prevents user-facing incidents.
Key Takeaways:
- Custom eval suites align perfectly with your success metrics and domain-specific requirements
- Full control over dataset versioning prevents silent test set degradation
- Automated regression detection catches quality drops before production deployment
- Integration with your CI/CD pipeline enables evaluation-driven development workflows
- Performance optimization at eval time saves hours per test run at scale
The team at a fintech startup spent two months using a popular eval framework before building their own. Why? Their business logic required validating not just output correctness but also tool call sequencing, permission checks, and audit trail formatting. Generic frameworks measured accuracy. Their custom suite measured deployability. After switching, they caught 23 regressions in the first month that would have reached production.
This guide covers building a production-grade LLM evaluation suite with dataset management, multiple scoring engines, automated judging, regression detection, and CI/CD integration patterns you can deploy this week.
Core Components of an Eval Framework
A complete LLM eval suite from scratch requires six integrated components. Each serves a specific purpose in the evaluation workflow.
| Component | Purpose | Key Responsibilities |
|---|---|---|
| Dataset Store | Version-controlled test cases | Storage, retrieval, versioning, tagging |
| Runner Engine | Execute test cases against models | Batching, parallelization, retries, caching |
| Scoring Engine | Evaluate output quality | Multiple metric support, thresholds, aggregation |
| Judge System | Automated evaluation | LLM-as-judge, rule-based, hybrid approaches |
| Reporter | Results visualization | Dashboards, regression detection, trend analysis |
| CI Integration | Automated testing | Pre-commit, PR checks, deployment gates |
Architecture Overview
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from enum import Enum
class EvalStatus(Enum):
PASSED = "passed"
FAILED = "failed"
ERROR = "error"
SKIPPED = "skipped"
@dataclass
class TestCase:
"""Single evaluation test case"""
id: str
category: str
input_prompt: str
expected_output: Optional[str]
expected_tool_calls: Optional[List[Dict[str, Any]]]
metadata: Dict[str, Any]
tags: List[str]
@dataclass
class EvalResult:
"""Result of running one test case"""
test_id: str
status: EvalStatus
actual_output: str
scores: Dict[str, float] # metric_name -> score
latency_ms: float
token_count: int
cost_usd: float
error_message: Optional[str]
@dataclass
class EvalRun:
"""Complete evaluation run metadata"""
run_id: str
timestamp: str
model_name: str
prompt_version: str
results: List[EvalResult]
pass_rate: float
avg_latency_ms: float
total_cost_usd: float
This dataclass hierarchy provides type safety and makes it easy to serialize results to JSON or persist to databases. Every production eval suite needs these primitives.
Design Principles
1. Reproducibility First — Lock model versions, prompt templates, temperature, and dataset versions. Every eval run must be exactly reproducible.
2. Fast Feedback Loops — Optimize for speed. Developers should get eval results in under 5 minutes for rapid iteration.
3. Flexible Metrics — Support multiple scoring approaches: exact match, semantic similarity, LLM-as-judge, custom validators.
4. Fail-Fast Behavior — Surface critical failures immediately rather than waiting for full suite completion.
5. Cost Awareness — Track API costs per run. Evaluation can get expensive at scale — monitor and optimize.
Dataset Management and Versioning
The dataset store is the foundation of your eval suite. Poor dataset management leads to flaky tests and silent quality degradation.
Dataset Schema Design
# eval_suite/dataset.py
import json
from pathlib import Path
from typing import List, Optional
from datetime import datetime
class Dataset:
"""Versioned evaluation dataset with tagging and filtering"""
def __init__(self, name: str, version: str, base_path: Path):
self.name = name
self.version = version
self.base_path = base_path
self.cases: List[TestCase] = []
def load(self) -> None:
"""Load dataset from disk"""
dataset_file = self.base_path / f"{self.name}-{self.version}.json"
if not dataset_file.exists():
raise FileNotFoundError(f"Dataset not found: {dataset_file}")
with open(dataset_file, 'r') as f:
data = json.load(f)
self.cases = [
TestCase(
id=case['id'],
category=case['category'],
input_prompt=case['input_prompt'],
expected_output=case.get('expected_output'),
expected_tool_calls=case.get('expected_tool_calls'),
metadata=case.get('metadata', {}),
tags=case.get('tags', [])
)
for case in data['test_cases']
]
def filter_by_tags(self, tags: List[str]) -> List[TestCase]:
"""Filter cases by tag intersection"""
return [
case for case in self.cases
if any(tag in case.tags for tag in tags)
]
def filter_by_category(self, category: str) -> List[TestCase]:
"""Filter cases by category"""
return [case for case in self.cases if case.category == category]
def add_case(self, case: TestCase) -> None:
"""Add new test case"""
if any(c.id == case.id for c in self.cases):
raise ValueError(f"Test case {case.id} already exists")
self.cases.append(case)
def save(self) -> None:
"""Persist dataset to disk"""
dataset_file = self.base_path / f"{self.name}-{self.version}.json"
data = {
'name': self.name,
'version': self.version,
'created_at': datetime.utcnow().isoformat(),
'test_cases': [
{
'id': case.id,
'category': case.category,
'input_prompt': case.input_prompt,
'expected_output': case.expected_output,
'expected_tool_calls': case.expected_tool_calls,
'metadata': case.metadata,
'tags': case.tags
}
for case in self.cases
]
}
with open(dataset_file, 'w') as f:
json.dump(data, f, indent=2)
Version Control Strategy
Store datasets in Git alongside code. This ensures:
- Atomic changes — Dataset updates ship with the prompt/model changes they validate
- Code review — New test cases go through PR review like any code change
- Rollback capability — Bad test cases can be reverted with Git
- Blame tracking — See who added each test case and why
# Example dataset directory structure
evals/
├── datasets/
│ ├── refund_policy_v1.0.json
│ ├── refund_policy_v1.1.json
│ ├── tool_calling_v2.0.json
│ └── edge_cases_v1.0.json
├── results/
│ └── runs/
│ ├── run_20260914_120000.json
│ └── run_20260914_130000.json
└── reports/
└── regression_report_2026_09_14.html
Dataset Evolution Patterns
Additive growth — Most common pattern. Add new cases for discovered failures without removing old ones:
# Add production failure as new test case
def capture_production_failure(
failure_log: Dict[str, Any],
dataset: Dataset
) -> None:
"""Convert production failure into test case"""
case = TestCase(
id=f"prod_failure_{failure_log['incident_id']}",
category="production_failures",
input_prompt=failure_log['prompt'],
expected_output=failure_log['correct_output'],
expected_tool_calls=failure_log.get('correct_tool_calls'),
metadata={
'incident_id': failure_log['incident_id'],
'date': failure_log['timestamp'],
'severity': failure_log['severity']
},
tags=['production', 'regression_prevention']
)
dataset.add_case(case)
dataset.save()
Version bumps — When test semantics change significantly (e.g., policy update changes correct answers), create a new major version:
# evals/migrate_dataset.py
def migrate_v1_to_v2(old_dataset: Dataset) -> Dataset:
"""Migrate test cases to new policy version"""
new_dataset = Dataset(
name=old_dataset.name,
version="v2.0",
base_path=old_dataset.base_path
)
for case in old_dataset.cases:
# Update expected outputs based on new policy
updated_case = case
if case.category == "refund_policy":
updated_case.expected_output = apply_new_policy_rules(
case.input_prompt
)
new_dataset.add_case(updated_case)
new_dataset.save()
return new_dataset
Dataset Quality Checks
Run validation before committing dataset changes:
def validate_dataset(dataset: Dataset) -> List[str]:
"""Check dataset quality and return issues"""
issues = []
# Check for duplicate IDs
ids = [case.id for case in dataset.cases]
if len(ids) != len(set(ids)):
issues.append("Duplicate test case IDs found")
# Check for empty prompts
empty_prompts = [c.id for c in dataset.cases if not c.input_prompt.strip()]
if empty_prompts:
issues.append(f"Empty prompts in cases: {empty_prompts}")
# Check for missing expected outputs or tool calls
no_expectations = [
c.id for c in dataset.cases
if not c.expected_output and not c.expected_tool_calls
]
if no_expectations:
issues.append(
f"Cases with no expected outputs or tool calls: {no_expectations}"
)
# Check for orphaned tags
all_tags = set()
for case in dataset.cases:
all_tags.update(case.tags)
# Warn if tags appear in < 3 cases (might be typos)
tag_counts = {
tag: sum(1 for c in dataset.cases if tag in c.tags)
for tag in all_tags
}
rare_tags = [tag for tag, count in tag_counts.items() if count < 3]
if rare_tags:
issues.append(f"Rare tags (< 3 uses): {rare_tags}")
return issues
This catches common mistakes before they pollute your eval suite. Run as a pre-commit hook.
Scoring Engine Architecture
The scoring engine evaluates model outputs against expected results. Support multiple metric types for different evaluation needs.
Metric Interface
# eval_suite/metrics.py
from abc import ABC, abstractmethod
from typing import Dict, Any
class Metric(ABC):
"""Base class for all evaluation metrics"""
@property
@abstractmethod
def name(self) -> str:
"""Metric identifier"""
pass
@abstractmethod
def score(
self,
actual: str,
expected: str,
context: Dict[str, Any]
) -> float:
"""
Compute metric score
Args:
actual: Model output
expected: Ground truth
context: Additional info (test case metadata, etc.)
Returns:
Score in [0.0, 1.0] where 1.0 is perfect
"""
pass
Built-in Metrics
import re
from difflib import SequenceMatcher
class ExactMatchMetric(Metric):
"""Binary exact match"""
@property
def name(self) -> str:
return "exact_match"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
return 1.0 if actual.strip() == expected.strip() else 0.0
class FuzzyMatchMetric(Metric):
"""Fuzzy string matching using sequence similarity"""
@property
def name(self) -> str:
return "fuzzy_match"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
return SequenceMatcher(None, actual, expected).ratio()
class ContainsMetric(Metric):
"""Check if output contains expected substring"""
@property
def name(self) -> str:
return "contains"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
return 1.0 if expected.lower() in actual.lower() else 0.0
class RegexMetric(Metric):
"""Regex pattern matching"""
@property
def name(self) -> str:
return "regex_match"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
# expected should be a regex pattern
return 1.0 if re.search(expected, actual) else 0.0
class JSONValidityMetric(Metric):
"""Check if output is valid JSON"""
@property
def name(self) -> str:
return "json_validity"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
try:
json.loads(actual)
return 1.0
except json.JSONDecodeError:
return 0.0
Semantic Similarity with Embeddings
For cases where meaning matters more than exact wording:
from openai import OpenAI
import numpy as np
class SemanticSimilarityMetric(Metric):
"""Cosine similarity of embeddings"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
@property
def name(self) -> str:
return "semantic_similarity"
def _get_embedding(self, text: str) -> np.ndarray:
"""Get text embedding from OpenAI"""
response = self.client.embeddings.create(
input=text,
model="text-embedding-3-small"
)
return np.array(response.data[0].embedding)
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
actual_emb = self._get_embedding(actual)
expected_emb = self._get_embedding(expected)
# Cosine similarity
similarity = np.dot(actual_emb, expected_emb) / (
np.linalg.norm(actual_emb) * np.linalg.norm(expected_emb)
)
# Map from [-1, 1] to [0, 1]
return (similarity + 1) / 2
For more on embeddings, see our embeddings explained complete guide.
Composite Scoring
Combine multiple metrics with weights:
class CompositeMetric(Metric):
"""Weighted combination of multiple metrics"""
def __init__(self, metrics: List[Metric], weights: List[float]):
if len(metrics) != len(weights):
raise ValueError("Metrics and weights must have same length")
if not np.isclose(sum(weights), 1.0):
raise ValueError("Weights must sum to 1.0")
self.metrics = metrics
self.weights = weights
@property
def name(self) -> str:
return "composite"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
scores = [
metric.score(actual, expected, context)
for metric in self.metrics
]
return sum(s * w for s, w in zip(scores, self.weights))
Example usage:
# For customer support responses, weight semantic similarity and tone
composite = CompositeMetric(
metrics=[
SemanticSimilarityMetric(api_key=os.getenv("OPENAI_API_KEY")),
ToneMetric(), # Custom metric checking professional tone
LengthConstraintMetric(max_tokens=200)
],
weights=[0.6, 0.3, 0.1]
)
Tool Call Validation
For AI agents with tool calling:
class ToolCallMetric(Metric):
"""Validate tool call sequence and arguments"""
@property
def name(self) -> str:
return "tool_call_accuracy"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
# actual and expected are JSON strings of tool calls
try:
actual_calls = json.loads(actual)
expected_calls = json.loads(expected)
except json.JSONDecodeError:
return 0.0
if len(actual_calls) != len(expected_calls):
# Partial credit based on overlap
min_len = min(len(actual_calls), len(expected_calls))
matches = sum(
1 for i in range(min_len)
if self._calls_match(actual_calls[i], expected_calls[i])
)
return matches / max(len(actual_calls), len(expected_calls))
# Full sequence match
matches = sum(
1 for ac, ec in zip(actual_calls, expected_calls)
if self._calls_match(ac, ec)
)
return matches / len(expected_calls)
def _calls_match(self, actual: Dict, expected: Dict) -> bool:
"""Check if two tool calls match"""
if actual.get('name') != expected.get('name'):
return False
# Check required arguments
expected_args = expected.get('arguments', {})
actual_args = actual.get('arguments', {})
for key, val in expected_args.items():
if key not in actual_args or actual_args[key] != val:
return False
return True
This metric is critical for evaluating agents that make multiple tool calls in sequence, like those described in our AI agent architecture patterns.
Automated Judge Implementation
LLM-as-judge scales human evaluation to thousands of test cases. Key insight: you need multiple judge strategies for different test types.
Simple Binary Judge
For yes/no questions (e.g., "Does the response follow the refund policy?"):
class BinaryJudge(Metric):
"""LLM-as-judge for binary yes/no evaluation"""
def __init__(self, api_key: str, model: str = "gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
@property
def name(self) -> str:
return "llm_binary_judge"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
"""
expected: The evaluation criteria (question to answer)
actual: The model output to judge
"""
prompt = f"""You are evaluating an AI system's response.
Evaluation Criteria: {expected}
AI Response:
{actual}
Does the response meet the criteria? Answer only YES or NO.
Answer:"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
max_tokens=10
)
answer = response.choices[0].message.content.strip().upper()
return 1.0 if answer == "YES" else 0.0
Rubric-Based Judge
For nuanced evaluation across multiple criteria:
class RubricJudge(Metric):
"""LLM-as-judge with multi-criteria rubric"""
def __init__(self, api_key: str, rubric: Dict[str, float], model: str = "gpt-4o"):
"""
rubric: Dict mapping criterion name to weight
Example: {"accuracy": 0.4, "tone": 0.3, "completeness": 0.3}
"""
self.client = OpenAI(api_key=api_key)
self.model = model
self.rubric = rubric
@property
def name(self) -> str:
return "llm_rubric_judge"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
rubric_text = "\n".join([
f"- {criterion} (weight: {weight})"
for criterion, weight in self.rubric.items()
])
prompt = f"""You are evaluating an AI system's response against multiple criteria.
Reference Answer: {expected}
AI Response: {actual}
Evaluation Rubric:
{rubric_text}
For each criterion, assign a score from 0-10. Return your evaluation as JSON:
{{
"criterion_name": {{"score": X, "reasoning": "..."}},
...
}}
Evaluation:"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
response_format={"type": "json_object"}
)
try:
evaluation = json.loads(response.choices[0].message.content)
# Compute weighted average
total_score = 0.0
for criterion, weight in self.rubric.items():
if criterion in evaluation:
total_score += (evaluation[criterion]['score'] / 10.0) * weight
return total_score
except (json.JSONDecodeError, KeyError) as e:
print(f"Judge parsing error: {e}")
return 0.0
Calibration Against Human Labels
LLM judges drift from human judgment. Calibrate regularly:
def calibrate_judge(
judge: Metric,
calibration_set: List[tuple[str, str, float]] # (actual, expected, human_score)
) -> Dict[str, Any]:
"""
Measure judge agreement with human labels
Returns metrics:
- accuracy: % of cases where judge matches human within threshold
- correlation: Pearson correlation with human scores
- avg_difference: Mean absolute difference
"""
judge_scores = []
human_scores = []
for actual, expected, human_score in calibration_set:
judge_score = judge.score(actual, expected, context={})
judge_scores.append(judge_score)
human_scores.append(human_score)
# Agreement within 0.1 threshold
agreements = sum(
1 for js, hs in zip(judge_scores, human_scores)
if abs(js - hs) <= 0.1
)
accuracy = agreements / len(calibration_set)
# Pearson correlation
correlation = np.corrcoef(judge_scores, human_scores)[0, 1]
# Mean absolute error
avg_difference = np.mean([
abs(js - hs) for js, hs in zip(judge_scores, human_scores)
])
return {
'accuracy': accuracy,
'correlation': correlation,
'avg_difference': avg_difference,
'num_samples': len(calibration_set)
}
Run this monthly and regenerate judge prompts if agreement drops below 0.85.
For more patterns, see our dedicated LLM-as-judge evaluation guide.
Regression Detection System
Detect when new prompt/model changes break existing functionality. This is the most valuable component of your eval suite.
Baseline Management
# eval_suite/regression.py
from pathlib import Path
import json
from typing import Optional
class RegressionDetector:
"""Detect score regressions against baseline"""
def __init__(self, baseline_dir: Path):
self.baseline_dir = baseline_dir
self.baseline_dir.mkdir(parents=True, exist_ok=True)
def save_baseline(self, run: EvalRun, name: str = "production") -> None:
"""Save eval run as baseline"""
baseline_file = self.baseline_dir / f"{name}.json"
baseline_data = {
'run_id': run.run_id,
'timestamp': run.timestamp,
'model_name': run.model_name,
'prompt_version': run.prompt_version,
'pass_rate': run.pass_rate,
'avg_latency_ms': run.avg_latency_ms,
'per_test_scores': {
result.test_id: result.scores
for result in run.results
}
}
with open(baseline_file, 'w') as f:
json.dump(baseline_data, f, indent=2)
def detect_regressions(
self,
current_run: EvalRun,
baseline_name: str = "production",
threshold: float = 0.05 # 5% drop is regression
) -> Dict[str, Any]:
"""
Compare current run against baseline
Returns:
- regressions: List of test IDs that regressed
- improvements: List of test IDs that improved
- summary: Overall metrics
"""
baseline_file = self.baseline_dir / f"{baseline_name}.json"
if not baseline_file.exists():
return {'error': f'Baseline {baseline_name} not found'}
with open(baseline_file, 'r') as f:
baseline = json.load(f)
baseline_scores = baseline['per_test_scores']
regressions = []
improvements = []
for result in current_run.results:
if result.test_id not in baseline_scores:
continue # New test, skip
baseline_score = self._aggregate_score(baseline_scores[result.test_id])
current_score = self._aggregate_score(result.scores)
diff = current_score - baseline_score
if diff < -threshold: # Dropped by more than threshold
regressions.append({
'test_id': result.test_id,
'baseline_score': baseline_score,
'current_score': current_score,
'diff': diff
})
elif diff > threshold: # Improved by more than threshold
improvements.append({
'test_id': result.test_id,
'baseline_score': baseline_score,
'current_score': current_score,
'diff': diff
})
return {
'baseline_pass_rate': baseline['pass_rate'],
'current_pass_rate': current_run.pass_rate,
'pass_rate_diff': current_run.pass_rate - baseline['pass_rate'],
'regressions': regressions,
'improvements': improvements,
'num_regressions': len(regressions),
'num_improvements': len(improvements)
}
def _aggregate_score(self, scores: Dict[str, float]) -> float:
"""Aggregate multiple metric scores into single value"""
# Use primary metric if defined, otherwise average
if 'primary' in scores:
return scores['primary']
return sum(scores.values()) / len(scores) if scores else 0.0
Automated Regression Reporting
def generate_regression_report(detection_result: Dict[str, Any]) -> str:
"""Generate human-readable regression report"""
lines = ["# Evaluation Regression Report\n"]
# Summary
lines.append("## Summary\n")
lines.append(f"**Baseline Pass Rate:** {detection_result['baseline_pass_rate']:.1%}")
lines.append(f"**Current Pass Rate:** {detection_result['current_pass_rate']:.1%}")
lines.append(f"**Change:** {detection_result['pass_rate_diff']:+.1%}\n")
# Regressions
if detection_result['regressions']:
lines.append("## ⚠️ Regressions Detected\n")
lines.append(f"**{detection_result['num_regressions']} test(s) regressed**\n")
for reg in detection_result['regressions']:
lines.append(f"### Test: `{reg['test_id']}`")
lines.append(f"- Baseline: {reg['baseline_score']:.2f}")
lines.append(f"- Current: {reg['current_score']:.2f}")
lines.append(f"- **Drop: {reg['diff']:.2f}**\n")
else:
lines.append("## ✅ No Regressions\n")
# Improvements
if detection_result['improvements']:
lines.append(f"## 🎉 Improvements ({detection_result['num_improvements']})\n")
for imp in detection_result['improvements'][:5]: # Top 5
lines.append(f"- `{imp['test_id']}`: +{imp['diff']:.2f}")
return "\n".join(lines)
CI/CD Integration Example
# .github/workflows/eval.yml
name: LLM Evaluation
on:
pull_request:
branches: [main]
paths:
- 'prompts/**'
- 'agents/**'
- 'evals/**'
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install -e eval_suite/
- name: Run evaluation suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
python -m eval_suite.run \
--dataset evals/datasets/production_v1.0.json \
--output results/pr_${{ github.event.pull_request.number }}.json
- name: Detect regressions
run: |
python -m eval_suite.regression \
--current results/pr_${{ github.event.pull_request.number }}.json \
--baseline results/baseline_production.json \
--output regression_report.md
- name: Comment PR with results
uses: actions/github-script@v6
with:
script: |
const fs = require('fs');
const report = fs.readFileSync('regression_report.md', 'utf8');
github.rest.issues.createComment({
issue_number: context.issue.number,
owner: context.repo.owner,
repo: context.repo.repo,
body: report
});
- name: Fail on regressions
run: |
python -m eval_suite.check_threshold \
--report regression_report.md \
--max-regressions 0
This workflow runs on every PR that touches prompts or agent code, detects regressions, posts results as a PR comment, and blocks merge if quality dropped.
For more CI/CD patterns, see our AI evals in CI/CD guide.
CI/CD Integration Patterns
Production teams run evals at multiple points in the development lifecycle.
Pre-Commit Hooks
Fast smoke test before code commits:
# .pre-commit-config.yaml
repos:
- repo: local
hooks:
- id: eval-smoke-test
name: LLM Eval Smoke Test
entry: python -m eval_suite.run --dataset evals/smoke_test.json --fast
language: system
pass_filenames: false
stages: [commit]
Smoke test runs 10-20 critical cases in under 60 seconds. Catches obvious breakage before push.
PR Checks
Comprehensive eval on PR creation:
# scripts/run_pr_eval.py
import sys
import subprocess
from pathlib import Path
def run_pr_evaluation(pr_number: int) -> bool:
"""
Run full eval suite and detect regressions
Returns True if all checks pass, False otherwise
"""
# Run full evaluation
result = subprocess.run([
'python', '-m', 'eval_suite.run',
'--dataset', 'evals/datasets/production_v1.0.json',
'--output', f'results/pr_{pr_number}.json',
'--parallel', '10'
])
if result.returncode != 0:
print("Evaluation run failed")
return False
# Detect regressions
result = subprocess.run([
'python', '-m', 'eval_suite.regression',
'--current', f'results/pr_{pr_number}.json',
'--baseline', 'results/baseline_production.json',
'--threshold', '0.05',
'--output', 'regression_report.md'
])
if result.returncode != 0:
print("Regression check failed")
return False
# Check if any regressions were found
with open('regression_report.md', 'r') as f:
report = f.read()
if '⚠️ Regressions Detected' in report:
print("Quality regressions detected - see report")
return False
return True
if __name__ == '__main__':
pr_number = int(sys.argv[1])
success = run_pr_evaluation(pr_number)
sys.exit(0 if success else 1)
Deployment Gates
Run final validation before production deploy:
# scripts/deployment_gate.py
import sys
from eval_suite.runner import run_evaluation
from eval_suite.regression import RegressionDetector
def deployment_gate_check() -> bool:
"""
Final check before production deployment
Returns True if safe to deploy, False to block
"""
# Run evaluation on staging
staging_run = run_evaluation(
dataset_path='evals/datasets/production_v1.0.json',
model='staging-model',
environment='staging'
)
# Check pass rate
if staging_run.pass_rate < 0.95:
print(f"Pass rate {staging_run.pass_rate:.1%} below 95% threshold")
return False
# Check for regressions
detector = RegressionDetector(baseline_dir=Path('results/baselines'))
regressions = detector.detect_regressions(
current_run=staging_run,
baseline_name='production',
threshold=0.03 # Stricter threshold for production
)
if regressions['num_regressions'] > 0:
print(f"{regressions['num_regressions']} regressions detected")
return False
# Check latency
if staging_run.avg_latency_ms > 2000: # 2 second threshold
print(f"Average latency {staging_run.avg_latency_ms}ms exceeds threshold")
return False
print("All deployment gate checks passed ✓")
return True
if __name__ == '__main__':
success = deployment_gate_check()
sys.exit(0 if success else 1)
Real Production Examples
Example 1: Customer Support Agent Eval
# examples/support_agent_eval.py
from eval_suite.core import Dataset, TestCase
from eval_suite.runner import EvalRunner
from eval_suite.metrics import (
SemanticSimilarityMetric,
ContainsMetric,
CompositeMetric,
BinaryJudge
)
# Build dataset
dataset = Dataset(name="support_agent", version="v1.0", base_path=Path("evals/datasets"))
# Add test cases covering common scenarios
test_cases = [
TestCase(
id="refund_policy_basic",
category="refund_policy",
input_prompt="Customer purchased item 3 days ago, wants refund. Item is unopened.",
expected_output="Eligible for full refund within 30-day window.",
metadata={'policy_version': '2026-Q1'},
tags=['refund', 'policy', 'critical']
),
TestCase(
id="refund_policy_exception",
category="refund_policy",
input_prompt="Customer purchased item 35 days ago, item is defective.",
expected_output="Outside normal window but defective - escalate to supervisor for exception approval.",
metadata={'policy_version': '2026-Q1'},
tags=['refund', 'policy', 'edge_case']
),
# ... 200+ more cases
]
for case in test_cases:
dataset.add_case(case)
dataset.save()
# Configure metrics
judge = BinaryJudge(api_key=os.getenv("OPENAI_API_KEY"))
semantic = SemanticSimilarityMetric(api_key=os.getenv("OPENAI_API_KEY"))
contains = ContainsMetric()
composite = CompositeMetric(
metrics=[semantic, judge],
weights=[0.6, 0.4]
)
# Run evaluation
runner = EvalRunner(
model="gpt-4o",
metrics=[composite],
system_prompt=load_support_agent_prompt()
)
results = runner.run(dataset=dataset)
print(f"Pass rate: {results.pass_rate:.1%}")
print(f"Avg latency: {results.avg_latency_ms:.0f}ms")
print(f"Total cost: ${results.total_cost_usd:.2f}")
Example 2: RAG System Evaluation
For RAG pipelines:
# examples/rag_eval.py
from eval_suite.metrics import Metric
class RAGFaithfulnessMetric(Metric):
"""Check if response is grounded in retrieved context"""
@property
def name(self) -> str:
return "rag_faithfulness"
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
retrieved_docs = context.get('retrieved_docs', [])
if not retrieved_docs:
return 0.0
# Use LLM to check if response is supported by docs
docs_text = "\n\n".join([doc['content'] for doc in retrieved_docs])
prompt = f"""Check if the following response is fully supported by the provided documents.
Documents:
{docs_text}
Response: {actual}
Is the response fully supported by the documents with no hallucinated information?
Answer YES or NO.
Answer:"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.0
)
answer = response.choices[0].message.content.strip().upper()
return 1.0 if answer == "YES" else 0.0
# Use in eval
rag_test = TestCase(
id="product_spec_query",
category="retrieval",
input_prompt="What is the maximum load capacity of Model X-200?",
expected_output="500 kg",
metadata={
'retrieved_docs': [
{'id': 'spec_123', 'content': 'Model X-200 specifications: max load 500kg, ...'},
# ... more docs
]
},
tags=['rag', 'factual']
)
For more RAG evaluation patterns, see our RAG evaluation guide and RAGAS deep dive.
Performance at Scale
As your eval suite grows to thousands of test cases, performance optimization becomes critical.
Parallel Execution
# eval_suite/runner.py
import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List
class EvalRunner:
"""Execute evaluation runs with parallelization"""
def __init__(
self,
model: str,
metrics: List[Metric],
system_prompt: str,
max_workers: int = 10
):
self.model = model
self.metrics = metrics
self.system_prompt = system_prompt
self.max_workers = max_workers
def run(self, dataset: Dataset) -> EvalRun:
"""Run evaluation with parallel execution"""
import time
start_time = time.time()
# Run tests in parallel
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(self._run_test_case, case): case
for case in dataset.cases
}
results = []
for future in as_completed(futures):
result = future.result()
results.append(result)
# Progress indicator
print(f"Completed {len(results)}/{len(dataset.cases)}", end='\r')
# Compute summary metrics
passed = sum(1 for r in results if r.status == EvalStatus.PASSED)
pass_rate = passed / len(results) if results else 0.0
avg_latency = sum(r.latency_ms for r in results) / len(results)
total_cost = sum(r.cost_usd for r in results)
return EvalRun(
run_id=f"run_{int(time.time())}",
timestamp=datetime.utcnow().isoformat(),
model_name=self.model,
prompt_version="v1.0",
results=results,
pass_rate=pass_rate,
avg_latency_ms=avg_latency,
total_cost_usd=total_cost
)
def _run_test_case(self, case: TestCase) -> EvalResult:
"""Execute single test case"""
import time
start = time.time()
try:
# Call model
actual_output = self._call_model(case.input_prompt)
# Score with all metrics
scores = {}
for metric in self.metrics:
scores[metric.name] = metric.score(
actual=actual_output,
expected=case.expected_output or "",
context={'metadata': case.metadata}
)
# Determine pass/fail
primary_score = scores.get('primary', sum(scores.values()) / len(scores))
status = EvalStatus.PASSED if primary_score >= 0.7 else EvalStatus.FAILED
latency_ms = (time.time() - start) * 1000
return EvalResult(
test_id=case.id,
status=status,
actual_output=actual_output,
scores=scores,
latency_ms=latency_ms,
token_count=len(actual_output.split()), # Rough estimate
cost_usd=self._estimate_cost(case.input_prompt, actual_output),
error_message=None
)
except Exception as e:
return EvalResult(
test_id=case.id,
status=EvalStatus.ERROR,
actual_output="",
scores={},
latency_ms=(time.time() - start) * 1000,
token_count=0,
cost_usd=0.0,
error_message=str(e)
)
def _call_model(self, prompt: str) -> str:
"""Make API call to model"""
# Implementation depends on your model provider
pass
def _estimate_cost(self, input_prompt: str, output: str) -> float:
"""Estimate API cost"""
# Rough token counts
input_tokens = len(input_prompt.split()) * 1.3 # ~1.3 tokens per word
output_tokens = len(output.split()) * 1.3
# GPT-4o pricing (example)
cost_per_1k_input = 0.005
cost_per_1k_output = 0.015
cost = (
(input_tokens / 1000) * cost_per_1k_input +
(output_tokens / 1000) * cost_per_1k_output
)
return cost
With 10 parallel workers, a 1,000-case eval that would take 2 hours sequentially completes in ~15 minutes.
Caching Strategies
Cache expensive operations:
import hashlib
from functools import lru_cache
class CachedEvalRunner(EvalRunner):
"""Eval runner with result caching"""
def __init__(self, *args, cache_dir: Path, **kwargs):
super().__init__(*args, **kwargs)
self.cache_dir = cache_dir
self.cache_dir.mkdir(parents=True, exist_ok=True)
def _run_test_case(self, case: TestCase) -> EvalResult:
"""Execute test case with caching"""
# Generate cache key from inputs
cache_key = self._cache_key(case)
cache_file = self.cache_dir / f"{cache_key}.json"
# Check cache
if cache_file.exists():
with open(cache_file, 'r') as f:
cached = json.load(f)
return EvalResult(**cached)
# Run test
result = super()._run_test_case(case)
# Save to cache
with open(cache_file, 'w') as f:
json.dump(result.__dict__, f)
return result
def _cache_key(self, case: TestCase) -> str:
"""Generate deterministic cache key"""
key_input = f"{self.model}:{case.input_prompt}:{case.id}"
return hashlib.sha256(key_input.encode()).hexdigest()
This prevents re-running unchanged tests when only a subset of your eval suite is affected by code changes.
Cost Optimization
Large eval suites can cost $50-500 per run. Optimize:
def smart_eval_strategy(
dataset: Dataset,
changed_categories: List[str]
) -> List[TestCase]:
"""
Only run tests affected by code changes
If prompts changed for 'refund_policy' category,
only run those tests instead of full suite
"""
if not changed_categories:
# No targeted changes, run full suite
return dataset.cases
# Run affected categories + always-run critical tests
affected_cases = []
for case in dataset.cases:
if case.category in changed_categories or 'critical' in case.tags:
affected_cases.append(case)
return affected_cases
# Usage in CI
changed_files = get_changed_files_from_git()
changed_categories = infer_affected_categories(changed_files)
cases_to_run = smart_eval_strategy(dataset, changed_categories)
print(f"Running {len(cases_to_run)}/{len(dataset.cases)} tests")
This can reduce CI eval costs by 60-80% while maintaining quality coverage.
Primary references: official documentation, official documentation, official documentation, official documentation.
Frequently Asked Questions
How many test cases do I need in my eval suite?
Start with 50-100 cases covering critical paths and known failure modes. Grow the suite organically by adding every production failure as a new test case. Mature eval suites have 500-2,000 cases.
Quality matters more than quantity. 100 well-designed cases beat 1,000 redundant ones.
Should I use exact match or LLM-as-judge for scoring?
Use exact match for structured outputs (JSON, tool calls, specific formats). Use LLM-as-judge for natural language where multiple correct answers exist.
For critical decisions, use both: exact match as primary signal, LLM-as-judge to catch edge cases where exact match is too strict.
How do I prevent eval suite staleness?
- Review failed tests monthly — if they're consistently failing, either fix the system or update the test
- Add new tests from production failures within 24 hours
- Version your datasets and track coverage by category
- Run calibration checks quarterly to ensure judges align with human judgment
What pass rate threshold should I use?
For production deployment gates: 95-98% pass rate. Lower than 95% indicates systemic issues. Higher than 98% might mean your tests are too easy.
For experimental features in development: 80-90% is acceptable as you iterate.
How do I handle flaky tests?
Flaky tests (pass sometimes, fail other times) usually indicate:
- Non-deterministic model behavior — add temperature=0.0 to eval runs
- Rate limit errors — add retry logic with exponential backoff
- Poorly specified expected outputs — tighten test criteria or use ranges instead of exact values
Tag flaky tests and run them 3 times, taking majority vote.
Can I evaluate agents with multi-turn conversations?
Yes. Structure test cases as sequences:
@dataclass
class MultiTurnTestCase:
id: str
turns: List[Dict[str, str]] # [{"user": "...", "expected_assistant": "..."}, ...]
expected_tool_calls: List[List[Dict]] # Tool calls per turn
Score each turn independently and aggregate. See our multi-turn conversation evaluation guide for full patterns.
How do I measure hallucination rate in production?
Add faithfulness metrics that check if outputs are grounded in provided context. For RAG systems, verify every claim in the output appears in retrieved documents.
For more, see measuring hallucination rate in production.
Should I run evals in CI on every commit?
Run fast smoke tests (10-20 critical cases, <2 min) on every commit. Run full suite on PR creation and before production deploy.
Balance speed and coverage. Developers won't wait for 30-minute eval runs before committing.
How do I evaluate multi-modal outputs (text + images)?
Extend the scoring interface to support multi-modal inputs. Use vision models as judges for image outputs. Example:
class ImageQualityMetric(Metric):
def score(self, actual: str, expected: str, context: Dict[str, Any]) -> float:
# actual is image path or URL
# Use GPT-4V or Claude 3.5 to judge quality
pass
What if my eval suite finds a regression but I want to ship anyway?
Document the decision with reasoning. Either:
- Update the baseline if the regression is acceptable (business decision to trade quality for speed/cost)
- Add regression to known issues list and prioritize fix in next sprint
- Implement gradual rollout and monitor metrics
Never silently ignore regressions.
Related Resources
Essential reading:
- LLM-as-Judge with Claude Evaluation Patterns
- Detecting Prompt Regression and Quality Drops
- RAGAS Deep Dive: Faithfulness and Relevancy Metrics
- Evaluation-Driven Development for AI Systems
Related guides:
- LLM Evaluation: How to Test Models
- RAG Evaluation: Measuring Retrieval Quality
- AI Evals in CI/CD with GitHub Actions
- Tracing LLMs with OpenTelemetry
Production context:
- Why RAG Pipelines Return Garbage (And How to Fix It)
- LLM Hallucination: Causes and Fixes
- Structured Output from LLMs: JSON Every Time
Services:
- AI Agent Development Services — We build production AI systems with built-in evaluation pipelines
- Observability & Monitoring — Full-stack observability for AI systems, including eval metrics dashboarding
Conclusion
- Define the contract and baseline before choosing tools.
- Design bounded failure handling and an explicit degraded mode.
- Gate rollout on correctness, latency, reliability, and cost.
- Preserve a tested rollback path and an owned runbook.
Discuss your implementation with our Building an LLM Evaluation Suite from Scratch engineers.
Free consultation
Book a free consultation call on LLM evaluation & testing frameworks
30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.
Book a meeting
Keep reading
Related articles
LLM-as-Judge with Claude: Complete Evaluation Pattern Guide
Learn llm-as-judge with claude through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.
Read post
LLM Evaluation: How to Test Models Before Production (Guide)
LLM Evaluation guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.
Read post
Triton vs vLLM: LLM Serving Framework Comparison for
Triton vs vLLM guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.
Read post
LLM Tracing with OpenTelemetry: Complete Observability Guide
Learn llm tracing with opentelemetry through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.
Read post
