DSPyパイプラインの評価

import dspy

# Build a devset of labeled examples
devset = [
    dspy.Example(question='What is 7 * 8?', answer='56').with_inputs('question'),
    dspy.Example(question='Name the largest planet.', answer='Jupiter').with_inputs('question'),
    # ... more examples
]

# Create evaluator
evaluate = dspy.Evaluate(
    devset=devset,
    metric=exact_match_metric,  # Your metric function
    num_threads=4,              # Parallel evaluation
    display_progress=True,      # Show progress bar
    display_table=True,         # Show per-example results
)

# Run
score = evaluate(my_program)
print(f'Overall score: {score:.1%}')

import dspy

def exact_match_metric(example, prediction, trace=None):
    return float(
        example.answer.strip().lower() == prediction.answer.strip().lower()
    )

def contains_metric(example, prediction, trace=None):
    """Check if expected answer appears anywhere in prediction."""
    return float(example.answer.lower() in prediction.answer.lower())

def length_penalized_metric(example, prediction, trace=None):
    """Reward correct answers, penalize overly long ones."""
    correct = float(example.answer.lower() in prediction.answer.lower())
    length_ok = float(len(prediction.answer.split()) <= 20)
    return correct * (0.8 + 0.2 * length_ok)

# Use any of these as the metric parameter
evaluate = dspy.Evaluate(devset=devset, metric=contains_metric)

import dspy

def quality_metric(example, prediction, trace=None):
    """
    Multi-factor metric with pass/fail threshold.
    Returns float 0.0 to 1.0.
    During compilation (trace is not None), DSPy uses this to decide
    which traces to bootstrap as demos.
    """
    score = 0.0

    # Factor 1: Factual correctness (0.6 weight)
    if example.answer.lower() in prediction.answer.lower():
        score += 0.6

    # Factor 2: Conciseness (0.4 weight)
    word_count = len(prediction.answer.split())
    if word_count <= 15:
        score += 0.4
    elif word_count <= 30:
        score += 0.2

    # During optimization: only use examples scoring >= 0.6
    if trace is not None:
        return score >= 0.6

    return score

import random

# All labeled examples
all_examples = load_examples()  # Returns list of dspy.Example
random.shuffle(all_examples)

total = len(all_examples)
train_end = int(total * 0.6)
dev_end   = int(total * 0.8)

trainset = all_examples[:train_end]    # 60% for optimization
devset   = all_examples[train_end:dev_end]  # 20% for validation
testset  = all_examples[dev_end:]      # 20% held out

print(f'Train: {len(trainset)}, Dev: {len(devset)}, Test: {len(testset)}')

import dspy

evaluate = dspy.Evaluate(
    devset=testset,
    metric=exact_match_metric,
    num_threads=4,
    display_progress=True,
)

# Baseline: unoptimized program
baseline_score = evaluate(unoptimized_program)
print(f'Baseline (no optimization): {baseline_score:.1%}')

# BootstrapFewShot compiled
bs_score = evaluate(bootstrap_compiled_program)
print(f'BootstrapFewShot compiled:  {bs_score:.1%}')

# MIPRO compiled
mipro_score = evaluate(mipro_compiled_program)
print(f'MIPRO compiled:             {mipro_score:.1%}')

# Pick the winner
print(f'Best improvement: +{max(bs_score, mipro_score) - baseline_score:.1%}')

import dspy
import time

devset = [...]  # 200 examples

# Sequential evaluation
start = time.time()
evaluate_seq = dspy.Evaluate(devset=devset, metric=metric, num_threads=1)
score_seq = evaluate_seq(program)
print(f'Sequential: {time.time()-start:.0f}s')

# Parallel evaluation (4 threads)
start = time.time()
evaluate_par = dspy.Evaluate(devset=devset, metric=metric, num_threads=4)
score_par = evaluate_par(program)
print(f'Parallel (4 threads): {time.time()-start:.0f}s')
# Typically ~4x faster — same score, less wait time

import dspy

evaluate = dspy.Evaluate(
    devset=devset,
    metric=exact_match_metric,
    num_threads=2,
    display_progress=True,
    display_table=10,  # Show first 10 rows of results table
    return_outputs=True,  # Return (score, outputs) tuple
)

score, outputs = evaluate(program, return_all_scores=True)

# Find failing examples
failures = [
    (ex, pred, s)
    for ex, pred, s in outputs
    if s == 0.0
]
print(f'Failures: {len(failures)}/{len(devset)}')
for ex, pred, _ in failures[:3]:
    print(f'Q: {ex.question}')
    print(f'Expected: {ex.answer}')
    print(f'Got: {pred.answer}')

import dspy

class GradeAnswer(dspy.Signature):
    """Grade whether the predicted answer is correct given the reference."""
    question: str = dspy.InputField()
    reference_answer: str = dspy.InputField()
    predicted_answer: str = dspy.InputField()
    is_correct: bool = dspy.OutputField(
        desc='True if the predicted answer is semantically correct'
    )

grader = dspy.Predict(GradeAnswer)

def llm_graded_metric(example, prediction, trace=None):
    result = grader(
        question=example.question,
        reference_answer=example.answer,
        predicted_answer=prediction.answer,
    )
    return float(result.is_correct)

# Use this metric when answers can vary in phrasing
evaluate = dspy.Evaluate(devset=devset, metric=llm_graded_metric)

import json
import dspy

def run_and_save_evaluation(program, program_name, testset, metric):
    evaluate = dspy.Evaluate(
        devset=testset,
        metric=metric,
        num_threads=4,
    )
    score = evaluate(program)

    # Save score to history file
    history_file = 'eval_history.json'
    try:
        with open(history_file) as f:
            history = json.load(f)
    except FileNotFoundError:
        history = []

    history.append({'program': program_name, 'score': score})
    with open(history_file, 'w') as f:
        json.dump(history, f, indent=2)

    print(f'{program_name}: {score:.1%}')
    return score