dspy-evaluation-suite

name	dspy-evaluation-suite
description	Comprehensive evaluation metrics and testing framework for DSPy programs
allowed-tools	["Read","Write","Glob","Grep"]

DSPy Evaluation Suite

Goal

Systematically evaluate DSPy programs using built-in and custom metrics with parallel execution.

When to Use

• Measuring program performance before/after optimization
• Comparing different program variants
• Establishing baselines
• Validating production readiness

Inputs

Input	Type	Description
program	dspy.Module	Program to evaluate
devset	list[dspy.Example]	Evaluation examples
metric	callable	Scoring function
num_threads	int	Parallel threads

Outputs

Output	Type	Description
score	float	Average metric score
results	list	Per-example results

Workflow

Phase 1: Setup Evaluator

from dspy.evaluate import Evaluate

evaluator = Evaluate(
    devset=devset,
    metric=my_metric,
    num_threads=8,
    display_progress=True
)

Phase 2: Run Evaluation

score = evaluator(my_program)
print(f"Score: {score:.2%}")

Built-in Metrics

answer_exact_match

import dspy

# Normalized, case-insensitive comparison
metric = dspy.evaluate.answer_exact_match

SemanticF1

LLM-based semantic evaluation:

from dspy.evaluate import SemanticF1

semantic = SemanticF1()
score = semantic(example, prediction)

Custom Metrics

Basic Metric

def exact_match(example, pred, trace=None):
    """Returns bool, int, or float."""
    return example.answer.lower().strip() == pred.answer.lower().strip()

Multi-Factor Metric

def quality_metric(example, pred, trace=None):
    """Score based on multiple factors."""
    score = 0.0
    
    # Correctness (50%)
    if example.answer.lower() in pred.answer.lower():
        score += 0.5
    
    # Conciseness (25%)
    if len(pred.answer.split()) <= 20:
        score += 0.25
    
    # Has reasoning (25%)
    if hasattr(pred, 'reasoning') and pred.reasoning:
        score += 0.25
    
    return score

GEPA-Compatible Metric

def feedback_metric(example, pred, trace=None):
    """Returns (score, feedback) for GEPA optimizer."""
    correct = example.answer.lower() in pred.answer.lower()
    
    if correct:
        return 1.0, "Correct answer provided."
    else:
        return 0.0, f"Expected '{example.answer}', got '{pred.answer}'"

Production Example

import dspy
from dspy.evaluate import Evaluate, SemanticF1
import json
import logging
from typing import Optional
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class EvaluationResult:
    score: float
    num_examples: int
    correct: int
    incorrect: int
    errors: int

def comprehensive_metric(example, pred, trace=None) -> float:
    """Multi-dimensional evaluation metric."""
    scores = []
    
    # 1. Correctness
    if hasattr(example, 'answer') and hasattr(pred, 'answer'):
        correct = example.answer.lower().strip() in pred.answer.lower().strip()
        scores.append(1.0 if correct else 0.0)
    
    # 2. Completeness (answer not empty or error)
    if hasattr(pred, 'answer'):
        complete = len(pred.answer.strip()) > 0 and "error" not in pred.answer.lower()
        scores.append(1.0 if complete  )
    
    
     (pred, ):
        has_reasoning = ((pred.reasoning)) > 
        scores.append(  has_reasoning  )
    
     (scores) / (scores)  scores  

 :
     ():
        .devset = devset
        .num_threads = num_threads
    
     () -> EvaluationResult:
        
        metric = metric  comprehensive_metric
        
        evaluator = Evaluate(
            devset=.devset,
            metric=metric,
            num_threads=.num_threads,
            display_progress=,
            return_all_scores=
        )
        
        score, results = evaluator(program)
        
        correct = (  r  results  r >= )
        errors = (  r  results  r == )
        
         EvaluationResult(
            score=score,
            num_examples=(.devset),
            correct=correct,
            incorrect=(.devset) - correct - errors,
            errors=errors
        )
    
     () -> :
        
        results = {}
        
         name, program  programs.items():
            logger.info()
            results[name] = .evaluate(program, metric)
        
        
        ranked = (results.items(), key= x: x[].score, reverse=)
        
        ()
         rank, (name, result)  (ranked, ):
            ()
        
         results
    
     ():
        
        result = .evaluate(program, metric)
        
        report = {
            : {
                : result.score,
                : result.num_examples,
                : result.correct,
                : result.correct / result.num_examples
            },
            : {
                : .num_threads,
                : (.devset)
            }
        }
        
         (output_path, )  f:
            json.dump(report, f, indent=)
        
        logger.info()
         report


suite = EvaluationSuite(devset, num_threads=)


result = suite.evaluate(my_program)
()


results = suite.compare({
    : baseline_program,
    : optimized_program,
    : finetuned_program
})

Best Practices

1. Hold out test data - Never optimize on evaluation set
2. Multiple metrics - Combine correctness, quality, efficiency
3. Statistical significance - Use enough examples (100+)
4. Track over time - Version control evaluation results

Limitations

• Metrics are task-specific; no universal measure
• SemanticF1 requires LLM calls (cost)
• Parallel evaluation can hit rate limits
• Edge cases may not be captured