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: Use real data from production/past projects
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: 100-500 examples for quick eval
# 1000+ for production metrics
Strategy 2: Expert Annotation¶
# When no existing data: Create new dataset
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: 2-4 hours per 100 examples
# Quality: Very high, but biased toward expert expectations
Strategy 3: Hybrid Approach¶
# Best: Combine production data + expert annotation
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: Normalized match with fuzzy matching
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: Paraphrases, minor variations
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: Translation quality
metric = BLEUMetric()
bleu = metric.score(
"The quick brown fox",
"A quick brown fox"
)
# BLEU ≈ 0.75 (3/4 words overlap)
# Use case: Translation, paraphrase evaluation
# Good for: Capturing n-gram overlap
# Bad for: Semantic equivalence (doesn't understand meaning)
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: Document summarization
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: 0.67 (word overlap)
# ROUGE-L: 0.60 (subsequence overlap)
# Use case: Summarization, abstractive QA
# Good for: Capturing content overlap
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: Medical LLMs
# Key: Safety > recall
# 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: Accuracy 80%, Confidence 20%
# Medical task: Safety 50%, Accuracy 30%, No Hallucination 20%
# Translation: BLEU 40%, Semantic 40%, Fluency 20%
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: {"overall_score": 0.84, "score_by_returns": 0.88, ...}
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: 16%
# By Category:
# Returns: 8 errors (10% of category)
# Shipping: 4 errors (5% of category)
# Billing: 8 errors (25% of category) ← Focus here
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: 0.825 ± 0.017
# Model B: 0.847 ± 0.018
# Difference: 0.022
# p-value: 0.032
# ✓ 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: Benchmark with 100 examples
scores = [0.82, 0.85, 0.79, 0.84, 0.86] # etc
ci = confidence_interval(scores)
# Result: Score 0.832 [0.798, 0.866]
# Meaning: True performance probably between 79.8% and 86.6%
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: Run daily/weekly
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¶
-
Version your benchmark
benchmark_v1.0: Initial dataset benchmark_v1.1: Fixed typos, added 10 examples benchmark_v2.0: Completely new dataset -
Document evaluation criteria
- What counts as correct?
- How are edge cases handled?
-
What assumptions are made?
-
Test the harness
- Does it score correctly?
- Does it handle failures?
-
Do results make sense?
-
Sample error analysis
- Don't just report scores
- Show failure patterns
-
Suggest improvements
-
Report confidence intervals
- Not just point estimates
- "85% ± 2%" not "85%"
❌ Avoid This¶
- Single benchmark for complex systems
-
Always use multiple metrics
-
Overfitting to benchmark
-
Fine-tune on test set → invalidates benchmark
-
Ignoring inter-rater agreement
-
If humans annotate, measure consistency
-
Using outdated data
-
Refresh test set periodically
-
Publishing without error analysis
- 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