Dspy Advanced Workflow

DSPy Advanced Workflow

🎯 Trigger Conditions

Use when asked about DSPy optimization, metric design, or the 7-step optimization loop.

📚 Prerequisites

• dspy package installed
• Training data available
• Evaluation metric defined

🛠️ The 7-Step Optimization Loop

Step 1: Define Your Objective

class GenerateAnswer(dspy.Signature):
    """Answer questions with accurate, concise responses."""
    context = dspy.InputField(desc="Relevant context passages")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="Accurate answer")

# Define evaluation metric
def answer_correctness(predicted, gold):
    return predicted.answer == gold.answer

Step 2: Prepare Training Data

import dspy

# Create training examples
train_data = [
    dspy.Example(
        context=["Paris is the capital of France."],
        question="What is the capital of France?",
        answer="Paris"
    ).with_inputs("context", "question"),
    # ... more examples
]

# Split into train/dev sets
train_set, dev_set = dspy.evaluate.reviews.split_dataset(train_data, 0.8)

Step 3: Choose Your Optimizer

# Common optimizers
from dspy import BootstrapFewShot, BootstrapFinetune, MIPROv2, Ensemble

optimizer = BootstrapFewShot(
    metric=answer_correctness,
    max_labeled_demos=10,
    max_rounds=10
)

# For larger datasets
optimizer = MIPROv2(
    metric=answer_correctness,
    num_threads=4,
    seed=42,
    use_cache=True
)

Step 4: Optimize

# Optimize the program
optimized_program = optimizer.optimize(
    program=original_program,
    trainset=train_set,
    valset=dev_set
)

Step 5: Evaluate

# Evaluate on dev set
evaluator = dspy.evaluate.Evaluate(
    devset=dev_set,
    metric=answer_correctness,
    num_threads=4,
    display_progress=True,
    display_table=5
)

results = evaluator(optimized_program)
print(f"Accuracy: {results['accuracy']:.2%}")

Step 6: Analyze & Debug

# Get detailed evaluation results
from dspy.evaluate import Evaluate

evaluator = Evaluate(devset=dev_set, metric=answer_correctness, num_threads=1, display_progress=True)
results = evaluator(optimized_program, display_table=0)

# Analyze failures
failures = [
    example for example in dev_set 
    if not answer_correctness(optimized_program(**example.inputs()), example)
]
print(f"Failures: {len(failures)}/{len(dev_set)}")

Step 7: Iterate

# Refine based on analysis
# - Add more training data
# - Adjust optimizer hyperparameters
# - Improve signatures
# - Try different optimizers

⚠️ Pitfalls

• Metric design: Poor metrics lead to poor optimization
• Overfitting: Monitor dev vs train performance
• Resource usage: Some optimizers are expensive
• Caching: Use cache wisely to avoid stale results

📖 References

• DSPy Optimization
• BootstrapFewShot
• MIPROv2