Компиляция и оптимизация запросов

Что означает оптимизация в DSPy

Определение функции метрики

# Metric: exact match on answer field
def exact_match_metric(example, prediction, trace=None):
    """
    example: a training example with .answer
    prediction: the module's output with .answer
    Returns 1.0 if correct, 0.0 otherwise
    """
    expected = example.answer.strip().lower()
    predicted = prediction.answer.strip().lower()
    return float(expected == predicted)

# Metric: F1 score for token overlap (common in QA)
def token_f1_metric(example, prediction, trace=None):
    gold_tokens = set(example.answer.lower().split())
    pred_tokens = set(prediction.answer.lower().split())
    if not pred_tokens:
        return 0.0
    precision = len(gold_tokens & pred_tokens) / len(pred_tokens)
    recall = len(gold_tokens & pred_tokens) / len(gold_tokens)
    if precision + recall == 0:
        return 0.0
    return 2 * precision * recall / (precision + recall)

Подготовка обучающей выборки

import dspy

# Build training examples
trainset = [
    dspy.Example(
        question='What is the capital of Germany?',
        answer='Berlin'
    ).with_inputs('question'),

    dspy.Example(
        question='Who wrote Romeo and Juliet?',
        answer='William Shakespeare'
    ).with_inputs('question'),

    dspy.Example(
        question='What year did World War II end?',
        answer='1945'
    ).with_inputs('question'),
    # ... add more examples
]

print(f'Training set size: {len(trainset)} examples')
print(trainset[0].question, '->', trainset[0].answer)

Оптимизатор BootstrapFewShot

import dspy
from dspy.teleprompt import BootstrapFewShot

# Define your program
class QA(dspy.Signature):
    """Answer factual questions."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

program = dspy.ChainOfThought(QA)

# Set up the optimizer
optimizer = BootstrapFewShot(
    metric=exact_match_metric,
    max_bootstrapped_demos=4,   # Up to 4 few-shot examples per predictor
    max_labeled_demos=4,        # Use labeled examples directly if available
)

# Compile!
compiled_program = optimizer.compile(program, trainset=trainset)
print('Compilation complete')

Оптимизатор MIPRO

import dspy
from dspy.teleprompt import MIPROv2

class QA(dspy.Signature):
    """Answer factual questions."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

program = dspy.ChainOfThought(QA)

# MIPRO: optimizes both instructions AND few-shot examples
optimizer = MIPROv2(
    metric=exact_match_metric,
    auto='medium',      # 'light' / 'medium' / 'heavy' for optimization budget
    num_threads=4,      # Parallel evaluation threads
)

compiled_program = optimizer.compile(
    program,
    trainset=trainset,
    num_trials=20,       # Number of candidate prompts to evaluate
)
print('MIPRO compilation complete')

Как выглядит скомпилированный промпт

import dspy

# After compiling, inspect the optimized state
compiled_program = dspy.ChainOfThought('question -> answer')
# (Assume this was returned by optimizer.compile(...))

# Inspect the demos that were found
for demo in compiled_program.demos:
    print('Input:', demo.question)
    print('Reasoning:', demo.get('reasoning', 'N/A'))
    print('Answer:', demo.answer)
    print('---')

# Save the compiled state
compiled_program.save('compiled_qa_program.json')
print('Saved compiled program')

Интерфейс teleprompter.compile()

from dspy.teleprompt import BootstrapFewShot, MIPROv2, COPRO

# All optimizers use the same interface:
# compiled = optimizer.compile(program, trainset=trainset)

# BootstrapFewShot: fast, uses successful traces as demos
opt1 = BootstrapFewShot(metric=exact_match_metric)

# MIPRO: slower, optimizes instructions too
opt2 = MIPROv2(metric=exact_match_metric, auto='light')

# COPRO: coordinate descent over instruction proposals
opt3 = COPRO(metric=exact_match_metric, depth=3)

# Swap between them with one line change:
compiled = opt1.compile(program, trainset=trainset)
# or: compiled = opt2.compile(program, trainset=trainset)

BootstrapFewShotWithRandomSearch

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

# Split data into train and validation
trainset = examples[:40]
devset = examples[40:60]

optimizer = BootstrapFewShotWithRandomSearch(
    metric=exact_match_metric,
    max_bootstrapped_demos=4,
    num_candidate_programs=8,  # Try 8 different demo sets
    num_threads=4,
)

compiled_program = optimizer.compile(
    program,
    trainset=trainset,
    valset=devset,  # Picks the best program based on validation
)
print('Best program selected from 8 candidates')

Затраты на компиляцию

Проверка скомпилированной программы

import dspy

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

# Compare uncompiled vs compiled
uncompiled_score = evaluate(uncompiled_program)
compiled_score = evaluate(compiled_program)

print(f'Uncompiled accuracy: {uncompiled_score:.1%}')
print(f'Compiled accuracy:   {compiled_score:.1%}')
print(f'Improvement: +{compiled_score - uncompiled_score:.1%}')

Объединение всех элементов

import dspy
from dspy.teleprompt import BootstrapFewShot

# 1. Configure LM
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini', api_key='sk-...'))

# 2. Define signature and module
class QA(dspy.Signature):
    """Answer questions accurately."""
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

program = dspy.ChainOfThought(QA)

# 3. Compile
optimizer = BootstrapFewShot(metric=exact_match_metric)
compiled = optimizer.compile(program, trainset=trainset)

# 4. Save
compiled.save('production_qa.json')
print('Production program ready')

Проверка знаний: BootstrapFewShot

Повторение: компиляция и оптимизация