HinterBuild logoHinterBuild
AI Systems · 9 min read

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, author

Muhammad Abdul Sami

· 9 min read

  • LLM
  • LLM Serving
  • Evaluation
  • Cost Optimization

Table of Contents:

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.

ComponentPurposeKey Responsibilities
Dataset StoreVersion-controlled test casesStorage, retrieval, versioning, tagging
Runner EngineExecute test cases against modelsBatching, parallelization, retries, caching
Scoring EngineEvaluate output qualityMultiple metric support, thresholds, aggregation
Judge SystemAutomated evaluationLLM-as-judge, rule-based, hybrid approaches
ReporterResults visualizationDashboards, regression detection, trend analysis
CI IntegrationAutomated testingPre-commit, PR checks, deployment gates

Architecture Overview

python
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

python
# 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:

  1. Atomic changes — Dataset updates ship with the prompt/model changes they validate
  2. Code review — New test cases go through PR review like any code change
  3. Rollback capability — Bad test cases can be reverted with Git
  4. Blame tracking — See who added each test case and why
bash
# 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:

python
# 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:

python
# 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:

python
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

python
# 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

python
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:

python
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:

python
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:

python
# 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:

python
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?"):

python
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:

python
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:

python
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

python
# 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

python
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

yaml
# .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:

python
# .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:

python
# 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:

python
# 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

python
# 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:

python
# 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

python
# 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:

python
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:

python
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?

  1. Review failed tests monthly — if they're consistently failing, either fix the system or update the test
  2. Add new tests from production failures within 24 hours
  3. Version your datasets and track coverage by category
  4. 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:

  1. Non-deterministic model behavior — add temperature=0.0 to eval runs
  2. Rate limit errors — add retry logic with exponential backoff
  3. 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:

python
@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:

python
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:

  1. Update the baseline if the regression is acceptable (business decision to trade quality for speed/cost)
  2. Add regression to known issues list and prioritize fix in next sprint
  3. Implement gradual rollout and monitor metrics

Never silently ignore regressions.


Essential reading:

Related guides:

Production context:

Services:

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