Building Custom Benchmarks & Evaluation Frameworks

Overview

Standard benchmarks (MMLU, HumanEval) are great for general comparison, but domain-specific benchmarks are essential for:

• Evaluating models on your actual use cases
• Detecting when standard benchmarks don't predict production success
• Fine-tuning and validation
• Tracking improvements over time
• Regulatory compliance (some domains require custom evaluation)

---

When to Build Custom Benchmarks

Standard Benchmarks Insufficient When

# Red flags:

# 1. Your domain is highly specialized
your_domain = "Medical AI"
# MMLU score ≠ Medical accuracy
# Medical expertise required for evaluation

# 2. Your task format is unusual
your_task = "Multi-turn conversation with context"
# Standard Q&A benchmarks don't apply
# Need conversation-specific metrics

# 3. Hallucination is critical
your_app = "Medical diagnosis"
# 5% hallucination in medical context = catastrophic
# Standard benchmarks tolerate this

# 4. Your evaluation is subjective
your_task = "Creative writing"
# Can't use exact match or BLEU score
# Need human raters or semantic similarity

# 5. Your data is proprietary
your_data = "Confidential customer conversations"
# Can't use public benchmarks
# Must evaluate on own data

---

Benchmark Architecture

Components

- ┌─────────────────────────────────────────┐
 - Custom Benchmark │
 - ┤
│ │
 - 1. Dataset │
 - Test cases (prompts, context) │
 - Expected outputs │
 - Category labels │
 - Difficulty levels │
│ │
 - 2. Evaluation Metrics │
 - Accuracy (exact match) │
 - Similarity (semantic) │
 - Safety (no harmful content) │
 - Domain-specific metrics │
│ │
 - 3. Evaluation Harness │
 - Model interface │
 - Prompt formatting │
 - Output parsing │
 - Metric calculation │
│ │
 - 4. Analysis & Reporting │
 - Score aggregation │
 - Breakdown by category │
 - Error analysis │
 - Visualization │
│ │
 - ┘

-

Step 1: Dataset Creation

Data Collection Strategies

Strategy 1: Existing Data

# Best option

def collect_from_production():
 """Gather representative examples from real usage"""

 data = {
 "examples": [],
 "metadata": {}
 }

 # From production logs
 for conversation in production_conversations:
 if conversation.user_satisfied: # Good example
 data["examples"].append({
 "prompt": conversation.user_message,
 "context": conversation.previous_context,
 "expected_response": conversation.assistant_response,
 "quality": "good",
 "user_rating": conversation.rating
 })

 # Advantages:
 # Representative of real use
 # Ground truth (user feedback)
 # Quick to create

 # Disadvantages:
 # May be biased (only successful cases)
 # Privacy concerns
 # Limited diversity

 return data

# Typical coverage
# 1000+ for production metrics

Strategy 2: Expert Annotation

# When no existing data

def create_by_expert_annotation():
 """Experts write test cases for their domain"""

 # Step 1: Define coverage areas
 domains = {
 "Customer Service": {
 "questions": ["Refund policy", "Shipping", "Returns"],
 "num_examples": 30
 },
 "Technical Support": {
 "questions": ["Installation", "Troubleshooting", "Configuration"],
 "num_examples": 30
 },
 "Billing": {
 "questions": ["Invoice", "Payment", "Subscription"],
 "num_examples": 20
 }
 }

 # Step 2: Experts write realistic examples
 dataset = []
 for domain, coverage in domains.items():
 for category in coverage["questions"]:
 for _ in range(coverage["num_examples"] // len(coverage["questions"])):
 example = {
 "prompt": expert.write_realistic_question(domain, category),
 "expected_response": expert.write_ideal_answer(),
 "domain": domain,
 "category": category,
 "difficulty": expert.estimate_difficulty(),
 "notes": expert.add_notes()
 }
 dataset.append(example)

 # Advantages:
 # High quality examples
 # Covers edge cases
 # Domain expertise

 # Disadvantages:
 # Expensive (expert time)
 # Time-consuming
 # May not match real distribution

 return dataset

# Typical effort
# Quality

Strategy 3: Hybrid Approach

# Best

def create_hybrid_dataset():
 """Use real data + expert curation"""

 dataset = []

 # 60% from production (representative)
 production_examples = collect_from_production()
 dataset.extend(production_examples[:600])

 # 20% expert-written (edge cases)
 edge_cases = create_by_expert_annotation() # Hard cases
 dataset.extend(edge_cases[:200])

 # 20% adversarial (corner cases)
 adversarial_examples = [
 {"prompt": "...extremely long prompt...", "expected": "..."},
 {"prompt": "...contradictory context...", "expected": "..."},
 {"prompt": "...edge case...", "expected": "..."},
]
 dataset.extend(adversarial_examples[:200])

 return dataset

# Advantages:
# Representative + comprehensive
# Good coverage of edge cases
# Quality + realism

Dataset Format

# Standard format for benchmark datasets

benchmark_dataset = {
 "name": "Support QA v1.0",
 "description": "Customer support Q&A benchmark",
 "version": "1.0",
 "created_date": "2024-08-09",
 "num_examples": 1000,

 "examples": [
 {
 "id": "support_001",
 "prompt": "How do I return an item?",
 "context": {
 "customer_tier": "premium",
 "purchase_date": "2024-08-01",
 "item": "shoes"
 },
 "expected_response": "Items can be returned within 30 days...",
 "metadata": {
 "category": "returns",
 "difficulty": "easy",
 "requires_context": True,
 "domain": "customer_service"
 }
 },
 #... more examples
],

 "statistics": {
 "by_category": {"returns": 300, "shipping": 400, "billing": 300},
 "by_difficulty": {"easy": 600, "medium": 300, "hard": 100},
 "avg_prompt_length": 45,
 "avg_response_length": 250
 }
}

-

Step 2: Define Evaluation Metrics

Exact Match / Accuracy

def exact_match_metric(predicted, expected):
 """Simple binary match"""
 return predicted.strip().lower() == expected.strip().lower()

# Issues:
# Too strict (typos, paraphrasing fail)
# Doesn't handle multiple valid answers

# Better
from difflib import SequenceMatcher

def fuzzy_match(predicted, expected, threshold=0.85):
 """Allow minor differences"""
 predicted = predicted.strip().lower()
 expected = expected.strip().lower()

 ratio = SequenceMatcher(None, predicted, expected).ratio()
 return ratio > threshold

# Example:
fuzzy_match("The capital is Paris", "The capital is paris") # True
fuzzy_match("Paris is the capital", "The capital is Paris") # True
fuzzy_match("Paris", "The capital of France is Paris") # False

Semantic Similarity

from sentence_transformers import SentenceTransformer, util
import torch

class SemanticSimilarityMetric:
 def __init__(self, model_name="all-MiniLM-L6-v2"):
 self.model = SentenceTransformer(model_name)

 def score(self, predicted, expected, threshold=0.85):
 """Compare semantic meaning, not exact text"""

 pred_embedding = self.model.encode(predicted)
 expected_embedding = self.model.encode(expected)

 similarity = util.pytorch_cos_sim(pred_embedding, expected_embedding)

 return float(similarity) > threshold

# Examples
metric = SemanticSimilarityMetric()

# Different wording, same meaning → True
metric.score(
 "The capital of France is Paris",
 "Paris is France's capital"
) # True

# Different meaning → False
metric.score(
 "Paris is in France",
 "Berlin is in Germany"
) # False

# Use case

BLEU Score (Translation, Paraphrase)

from nltk.translate.bleu_score import sentence_bleu
from nltk.tokenize import word_tokenize

class BLEUMetric:
 def score(self, predicted, expected, max_n=4):
 """Measure n-gram overlap"""

 reference = [word_tokenize(expected.lower())]
 candidate = word_tokenize(predicted.lower())

 # BLEU-4: Uses 1-gram to 4-gram overlap
 score = sentence_bleu(
 reference,
 candidate,
 weights=[0.25, 0.25, 0.25, 0.25] # Equal weight
)

 return score

# Example
metric = BLEUMetric()

bleu = metric.score(
 "The quick brown fox",
 "A quick brown fox"
)
# BLEU ≈ 0.75 (3/4 words overlap)

# Use case
# Good for
# Bad for

ROUGE Score (Summarization)

from rouge_score import rouge_scorer

class RougeMetric:
 def __init__(self):
 self.scorer = rouge_scorer.RougeScorer(
 ['rouge1', 'rougeL', 'rougeSuccinctness'],
 use_stemmer=True
)

 def score(self, predicted, expected):
 """Measure summary quality via overlap"""

 scores = self.scorer.score(expected, predicted)

 return {
 "rouge1": scores['rouge1'].fmeasure, # 1-gram overlap
 "rougeL": scores['rougeL'].fmeasure, # Longest common subseq
 }

# Example
metric = RougeMetric()

reference = """
The quick brown fox jumps over the lazy dog.
The dog was sleeping under a tree.
"""

summary = "The fox jumped over a sleeping dog."

scores = metric.score(reference, summary)
# ROUGE-1
# ROUGE-L

# Use case
# Good for

Domain-Specific Metrics

class MedicalAccuracyMetric:
 """Medical domain specific evaluation"""

 def score(self, predicted_diagnosis, expected_diagnosis):
 """
 Medical evaluation:
 - Must not hallucinate conditions
 - Must recommend professional help for serious conditions
 - Safety > recall
 """

 # Check for hallucinations (conditions that don't exist)
 hallucinated = self._check_hallucinations(predicted_diagnosis)
 if hallucinated:
 return 0.0 # Critical failure

 # Check safety (recommends professional for serious cases)
 if self._is_serious_condition(expected_diagnosis):
 if self._mentions_professional_help(predicted_diagnosis):
 return 0.8 # Good, safe
 else:
 return 0.0 # Critical failure

 # Standard accuracy for non-serious
 return 1.0 if self._matches(predicted_diagnosis, expected_diagnosis) else 0.0

 def _check_hallucinations(self, diagnosis):
 # Check against medical knowledge base
 return diagnosis not in self.valid_conditions

 def _is_serious_condition(self, diagnosis):
 return diagnosis in self.serious_conditions

 def _mentions_professional_help(self, diagnosis):
 keywords = ["doctor", "hospital", "specialist", "professional"]
 return any(kw in diagnosis.lower() for kw in keywords)

# Use case
# Key
# Hallucination = automatic fail

Multi-Metric Aggregation

class MultiMetricEvaluator:
 """Combine multiple metrics for holistic evaluation"""

 def __init__(self):
 self.metrics = {
 "exact_match": {"weight": 0.2, "score_fn": self.exact_match},
 "semantic_sim": {"weight": 0.3, "score_fn": self.semantic_similarity},
 "safety": {"weight": 0.5, "score_fn": self.safety_check}, # Higher weight
 }

 def evaluate(self, predicted, expected, context=None):
 """Calculate weighted score across metrics"""

 scores = {}
 for metric_name, metric_config in self.metrics.items():
 if metric_name == "safety":
 scores[metric_name] = metric_config["score_fn"](predicted, context)
 else:
 scores[metric_name] = metric_config["score_fn"](predicted, expected)

 # Weighted average
 final_score = sum(
 scores[name] * self.metrics[name]["weight"]
 for name in scores
)

 return {
 "final_score": final_score,
 "breakdown": scores,
 "threshold_pass": final_score > 0.8
 }

# Example weights
# Classification task
# Medical task
# Translation

-

Step 3: Evaluation Harness

Basic Framework

class BenchmarkHarness:
 """Execute benchmark on model"""

 def __init__(self, model, benchmark_data, metrics):
 self.model = model
 self.benchmark_data = benchmark_data
 self.metrics = metrics
 self.results = []

 def evaluate(self):
 """Run full benchmark"""

 for example in self.benchmark_data["examples"]:
 # Generate prediction
 predicted = self.model.generate(
 prompt=example["prompt"],
 context=example.get("context", {}),
 max_tokens=500
)

 # Score prediction
 scores = {}
 for metric_name, metric in self.metrics.items():
 scores[metric_name] = metric.score(
 predicted,
 example["expected_response"]
)

 # Store result
 result = {
 "id": example["id"],
 "predicted": predicted,
 "expected": example["expected_response"],
 "scores": scores,
 "metadata": example.get("metadata", {})
 }
 self.results.append(result)

 return self._aggregate_results()

 def _aggregate_results(self):
 """Calculate aggregated metrics"""

 aggregation = {}

 # Overall score
 overall_scores = [r["scores"]["overall"] for r in self.results]
 aggregation["overall_score"] = sum(overall_scores) / len(overall_scores)

 # By category
 for category in set(r["metadata"].get("category") for r in self.results):
 category_results = [
 r for r in self.results
 if r["metadata"].get("category") == category
]
 scores = [r["scores"]["overall"] for r in category_results]
 aggregation[f"score_by_{category}"] = sum(scores) / len(scores)

 return aggregation

# Usage
harness = BenchmarkHarness(
 model=my_llm,
 benchmark_data=custom_benchmark,
 metrics={
 "exact_match": ExactMatchMetric(),
 "semantic": SemanticSimilarityMetric(),
 "overall": MultiMetricEvaluator()
 }
)

results = harness.evaluate()
# Returns

Handling Multiple Valid Answers

class FlexibleBenchmark:
 """Support multiple valid responses"""

 def __init__(self, examples_with_multiple_valid):
 self.data = examples_with_multiple_valid

 def evaluate_example(self, predicted, expected_list):
 """Check against ANY of valid answers"""

 # Find best match
 scores = [
 self.compute_similarity(predicted, expected)
 for expected in expected_list
]

 # Take best score
 return max(scores)

# Example data
multi_valid = [
 {
 "prompt": "What is the capital of France?",
 "valid_responses": [
 "Paris",
 "Paris is the capital",
 "France's capital is Paris",
 "The capital of France is Paris"
]
 }
]

# Benefits:
# Allows paraphrases
# More realistic evaluation
# Handles subjectivity

-

Step 4: Analysis & Reporting

Error Analysis

def analyze_errors(results, benchmark_data):
 """Understand failure patterns"""

 errors = [r for r in results if r["scores"]["overall"] < 0.5]

 analysis = {
 "total_errors": len(errors),
 "error_rate": len(errors) / len(results),
 "by_category": {},
 "by_difficulty": {},
 "patterns": []
 }

 # Break down by category
 for error in errors:
 category = error["metadata"]["category"]
 if category not in analysis["by_category"]:
 analysis["by_category"][category] = 0
 analysis["by_category"][category] += 1

 # Break down by difficulty
 for error in errors:
 difficulty = error["metadata"]["difficulty"]
 if difficulty not in analysis["by_difficulty"]:
 analysis["by_difficulty"][difficulty] = 0
 analysis["by_difficulty"][difficulty] += 1

 # Find common failure patterns
 for error in errors[:10]: # Examine first 10 errors
 print(f"Error in: {error['metadata']['category']}")
 print(f"Prompt: {error['prompt'][:100]}")
 print(f"Expected: {error['expected'][:100]}")
 print(f"Got: {error['predicted'][:100]}")
 print("-")

 return analysis

# Output example:
# Error Rate
# By Category:
# Returns
# Shipping
# Billing

Visualization

import matplotlib.pyplot as plt
import pandas as pd

def visualize_benchmark_results(results):
 """Create benchmark dashboard"""

 # 1. Overall score
 scores = [r["scores"]["overall"] for r in results]
 plt.figure(figsize=(12, 4))

 plt.subplot(1, 3, 1)
 plt.hist(scores, bins=20, edgecolor='black')
 plt.axvline(sum(scores)/len(scores), color='red', label='Mean')
 plt.xlabel("Score")
 plt.ylabel("Frequency")
 plt.title("Score Distribution")
 plt.legend()

 # 2. By category
 plt.subplot(1, 3, 2)
 df = pd.DataFrame(results)
 category_scores = df.groupby("metadata.category")["overall"].mean()
 category_scores.plot(kind='bar')
 plt.ylabel("Average Score")
 plt.title("Performance by Category")
 plt.xticks(rotation=45)

 # 3. Metric breakdown
 plt.subplot(1, 3, 3)
 metric_names = list(results[0]["scores"].keys())
 metric_avgs = {
 metric: sum(r["scores"][metric] for r in results) / len(results)
 for metric in metric_names
 }
 plt.barh(list(metric_avgs.keys()), list(metric_avgs.values()))
 plt.xlabel("Average Score")
 plt.title("Metric Comparison")

 plt.tight_layout()
 plt.show()

---

Step 5: Statistical Significance

Comparing Models

import numpy as np
from scipy import stats

def compare_models(model_a_scores, model_b_scores, alpha=0.05):
 """Test if Model B is significantly better than Model A"""

 # 1. Descriptive stats
 mean_a = np.mean(model_a_scores)
 mean_b = np.mean(model_b_scores)

 print(f"Model A: {mean_a:.3f} ± {np.std(model_a_scores):.3f}")
 print(f"Model B: {mean_b:.3f} ± {np.std(model_b_scores):.3f}")
 print(f"Difference: {mean_b - mean_a:.3f}")

 # 2. Statistical test (paired t-test)
 t_stat, p_value = stats.ttest_rel(model_b_scores, model_a_scores)

 print(f"\nPaired t-test:")
 print(f"t-statistic: {t_stat:.3f}")
 print(f"p-value: {p_value:.4f}")

 if p_value < alpha:
 print(f"SIGNIFICANT: Model B is better (p < {alpha})")
 return True
 else:
 print(f"NOT SIGNIFICANT: Difference could be random")
 return False

# Example
model_a = [0.82, 0.85, 0.81, 0.84, 0.83, 0.80]
model_b = [0.85, 0.87, 0.83, 0.86, 0.85, 0.82]

compare_models(model_a, model_b)

# Output:
# Model A
# Model B
# Difference
# p-value
# SIGNIFICANT

Confidence Intervals

import numpy as np
from scipy import stats

def confidence_interval(scores, confidence=0.95):
 """Calculate CI for metric"""

 mean = np.mean(scores)
 se = stats.sem(scores) # Standard error

 ci = stats.t.interval(confidence, len(scores)-1, loc=mean, scale=se)

 print(f"Score: {mean:.3f}")
 print(f"95% CI: [{ci[0]:.3f}, {ci[1]:.3f}]")

 # Interpretation:
 # We're 95% confident the true score is in this range

 return ci

# Example
scores = [0.82, 0.85, 0.79, 0.84, 0.86] # etc
ci = confidence_interval(scores)

# Result
# Meaning

-

Step 6: Continuous Monitoring

Benchmark as Regression Test

class BenchmarkRegression:
 """Track metrics over time to detect degradation"""

 def __init__(self, baseline_results):
 self.baseline = baseline_results
 self.history = [baseline_results]

 def check_regression(self, new_results, threshold=0.02):
 """Alert if performance dropped > threshold"""

 baseline_score = self.baseline["overall"]
 new_score = new_results["overall"]

 drop = baseline_score - new_score

 if drop > threshold:
 print(f" REGRESSION DETECTED")
 print(f" Baseline: {baseline_score:.3f}")
 print(f" Current: {new_score:.3f}")
 print(f" Drop: {drop:.3f} ({drop/baseline_score*100:.1f}%)")

 self._trigger_alert()
 self._suggest_investigation()

 return False
 else:
 print(f"No regression detected")
 return True

 def _trigger_alert(self):
 """Alert team"""
 send_slack_message("#ml-alerts", "Model degradation detected!")

 def _suggest_investigation(self):
 """Suggest areas to investigate"""
 possible_causes = [
 "Recent code changes?",
 "Fine-tuning corrupted model?",
 "Data distribution change?",
 "Dependency update?",
 "GPU change?"
]
 print("Possible causes:")
 for cause in possible_causes:
 print(f" - {cause}")

# Usage
benchmark = BenchmarkRegression(baseline_results)
benchmark.check_regression(new_results)

-

Complete Example: E-Commerce QA Benchmark

# Full example with all pieces

class EcommerceQABenchmark:
 def __init__(self):
 self.data = self._load_data()
 self.evaluator = MultiMetricEvaluator()

 def _load_data(self):
 return {
 "examples": [
 {
 "id": "ecom_001",
 "prompt": "What's your shipping policy for international orders?",
 "expected_responses": [
 "We ship worldwide with flat $10 fee",
 "International shipping is $10",
 "Our international shipping costs $10"
],
 "metadata": {
 "category": "shipping",
 "difficulty": "easy"
 }
 },
 #... 99 more examples
]
 }

 def run_benchmark(self, model):
 """Evaluate model on benchmark"""

 results = []

 for example in self.data["examples"]:
 # Generate answer
 predicted = model.generate(example["prompt"])

 # Evaluate
 score = max(
 self.evaluator.score(predicted, valid)
 for valid in example["expected_responses"]
)

 results.append({
 "id": example["id"],
 "score": score,
 "category": example["metadata"]["category"]
 })

 # Aggregate
 overall = sum(r["score"] for r in results) / len(results)
 by_category = {}
 for category in ["shipping", "refund", "billing"]:
 category_results = [
 r["score"] for r in results
 if r["category"] == category
]
 by_category[category] = sum(category_results) / len(category_results)

 return {
 "overall": overall,
 "by_category": by_category,
 "num_examples": len(results)
 }

# Run it
benchmark = EcommerceQABenchmark()
results = benchmark.run_benchmark(my_llm)

print(f"Overall Score: {results['overall']:.3f}")
for cat, score in results['by_category'].items():
 print(f" {cat}: {score:.3f}")

---

Best Practices

Do This

1. Version your benchmark

benchmark_v1.0: Initial dataset
benchmark_v1.1: Fixed typos, added 10 examples
benchmark_v2.0: Completely new dataset

1. Document evaluation criteria

2. What counts as correct?

3. How are edge cases handled?

4. What assumptions are made?

5. Test the harness

6. Does it score correctly?

7. Does it handle failures?

8. Do results make sense?

9. Sample error analysis

10. Don't just report scores

11. Show failure patterns

12. Suggest improvements

13. Report confidence intervals

14. Not just point estimates

15. "85% ± 2%" not "85%"

Avoid This

1. Single benchmark for complex systems

2. Always use multiple metrics

3. Overfitting to benchmark

4. Fine-tune on test set → invalidates benchmark

5. Ignoring inter-rater agreement

6. If humans annotate, measure consistency

7. Using outdated data

8. Refresh test set periodically

9. Publishing without error analysis

10. Always show where you're failing

---

Tools & Libraries

Existing Benchmarking Frameworks

- LM-Eval-Harness: HuggingFace framework for standard benchmarks
- OpenCompass: Comprehensive LLM evaluation platform
- HELM: Holistic evaluation framework
- TruthfulQA: Measuring truthfulness
- Custom: Roll your own (often necessary)

Code Structure

your_benchmark
 - data
 - train.json # Training examples (for fine-tuning)
 - test.json # Test set (for evaluation)
 - metadata.yaml # Dataset documentation
 - metrics
 - accuracy.py
 - semantic.py
 - domain_specific.py
 - harness
 - evaluator.py
 - runner.py
 - results
 - baseline.json
 - model_a_2024_08_09.json
 - analysis.py # Error analysis, visualization

-

References

• 01 Research Benchmarks - Standard benchmarks reference
• 02 Production Metrics - Production evaluation
• HuggingFace Evaluate Library
• LM-Eval-Harness

-

Last Updated: 2026-08-09