Guia Completo de DSPy — Guia Avançado

🧠O que é DSPy e Por que Usar

DSPy (Declarative Self-Improving Python) é um framework do Stanford NLP Group que trata o desenvolvimento de sistemas de IA como um problema de programação, não de prompt engineering manual. Em vez de ajustar strings de texto frágeis, você escreve código modular que descreve o quê o sistema deve fazer — e o DSPy cuida de descobrir como fazê-lo com o LLM.

Prompt Engineering Manual vs. DSPy

# Antes - prompt fragil, quebra ao mudar de modelo
prompt = """You are a helpful assistant.
Question: {question}
Context: {context}
Be concise and accurate.
Answer:"""

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": prompt.format(...)}]
)

# Depois - declarativo, portavel, otimizavel
import dspy

class AnswerQuestion(dspy.Signature):
    """Answer questions with short, accurate answers."""
    context: str = dspy.InputField()
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()

predictor = dspy.Predict(AnswerQuestion)
result = predictor(context="...", question="...")
print(result.answer)

Por que usar DSPy

⚙️Instalação e Configuração

Instalação

pip install -U dspy

# Com suporte a ChromaDB e integracoes de retrieval
pip install dspy chromadb openai anthropic

Configurar o Language Model (LM)

import dspy

# OpenAI
lm = dspy.LM('openai/gpt-4o-mini')

# Anthropic Claude
lm = dspy.LM('anthropic/claude-sonnet-4-6')

# Google Gemini
lm = dspy.LM('google/gemini-2.5-flash')

# Ollama local
lm = dspy.LM('ollama/llama3.3', api_base='http://localhost:11434')

# Parametros de geracao
lm = dspy.LM(
    'openai/gpt-4o-mini',
    temperature=0.7,
    max_tokens=1000,
    cache=True,  # cacheia respostas identicas
)

# Aplicar globalmente
dspy.configure(lm=lm)

Configurar o Retrieval Model (RM)

import dspy

# ColBERTv2 hospedado pela Stanford (Wikipedia)
rm = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(lm=lm, rm=rm)

# ChromaDB local
import chromadb
from dspy.retrieval import ChromadbRM

client = chromadb.Client()
rm = ChromadbRM(collection_name="docs", client=client)
dspy.configure(rm=rm)

✍️Signatures — Declarando Inputs e Outputs

Uma Signature define a transformação que o LLM deve realizar: quais campos recebe e quais deve produzir. É o contrato entre o seu código e o modelo.

Sintaxe Shorthand (inline)

# Forma mais simples
'question -> answer'

# Com tipos explicitos
'context: str, question: str -> answer: str'

# Com lista
'context: list[str], question: str -> answer: str'

# Com float de confianca
'text -> label: str, confidence: float'

# Com literal type (enum)
'question, choices: list[str] -> selection: int, reasoning: str'

# Uso direto em modulos
qa = dspy.Predict('question -> answer')
result = qa(question='O que e machine learning?')
print(result.answer)

Sintaxe Class-Based (recomendada para produção)

import dspy
from typing import Literal

class GenerateAnswer(dspy.Signature):
    """Answer questions with short factoid answers."""
    context: str = dspy.InputField(
        desc="may contain relevant facts about the question"
    )
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(
        desc="often between 1 and 5 words"
    )

# Tipos com Literal (enum estrito)
class EmotionClassifier(dspy.Signature):
    """Classify the emotion of a sentence."""
    sentence: str = dspy.InputField()
    emotion: Literal['joy', 'sadness', 'anger', 'fear', 'surprise'] = dspy.OutputField()
    reasoning: str = dspy.OutputField()

🧱Modules — Blocos de Construção

Módulos são os componentes que executam Signatures com estratégias de inferência diferentes.

Exemplos de cada módulo

import dspy

# Predict - basico
qa = dspy.Predict('question -> answer')
print(qa(question='Capital da Franca?').answer)  # -> "Paris"

# ChainOfThought - raciocinio intermediario automatico
cot = dspy.ChainOfThought('question -> answer')
r = cot(question='Por que o ceu e azul?')
print(r.reasoning)  # campo adicionado automaticamente
print(r.answer)

# ReAct - agente com tools
def calcular(expressao: str) -> str:
    """Avalia uma expressao matematica."""
    return str(eval(expressao))

def buscar_web(query: str) -> str:
    """Busca informacoes na web."""
    return f"Resultados para '{query}'..."  # mock

react = dspy.ReAct('question -> answer', tools=[calcular, buscar_web])
r = react(question='Quanto e 2**10 + 100?')
print(r.answer)

# ProgramOfThought - resolve gerando codigo Python
pot = dspy.ProgramOfThought('problem -> answer: int')
r = pot(problem='Qual a soma de 1 a 100?')
print(r.answer)  # -> 5050

# MultiChainComparison - melhor de 3
mcc = dspy.MultiChainComparison('question -> answer', M=3)
r = mcc(question='Explique fotossintese')
print(r.answer)

# Retrieve - busca semantica
retriever = dspy.Retrieve(k=3)
results = retriever('What is DSPy?')
for passage in results.passages:
    print(passage)

🔗Programs — Composição de Pipelines

Um Program (ou dspy.Module) compõe vários módulos em um pipeline com lógica Python normal — condicionais, loops, transformações de dados.

Estrutura básica de um Module

import dspy

class RAGPipeline(dspy.Module):
    def __init__(self, k=3):
        # Declarar sub-modulos no __init__
        self.retrieve = dspy.Retrieve(k=k)
        self.generate = dspy.ChainOfThought('context, question -> answer')

    def forward(self, question):
        # Implementar o pipeline no forward()
        context = self.retrieve(question).passages
        pred = self.generate(
            context='\n'.join(context),
            question=question
        )
        return pred

# Usar
rag = RAGPipeline(k=5)
result = rag(question='Como funciona a fotossintese?')
print(result.answer)

Pipeline multi-stage com roteamento

class AdvancedRAG(dspy.Module):
    """RAG com geracao de query otimizada e fallback."""

    def __init__(self, k=5):
        self.generate_query = dspy.Predict('question -> search_query: str')
        self.retrieve = dspy.Retrieve(k=k)
        self.answer = dspy.ChainOfThought('context: list[str], question -> answer')
        self.classify_intent = dspy.Predict('question -> is_factual: bool')

    def forward(self, question):
        # Classificar intent para decidir a estrategia
        intent = self.classify_intent(question=question)

        if intent.is_factual:
            # Para perguntas factuais: usar RAG
            query = self.generate_query(question=question).search_query
            context = self.retrieve(query).passages
        else:
            # Para perguntas abertas: sem retrieval
            context = []

        return self.answer(context=context, question=question)

# Inspecionar componentes
rag = AdvancedRAG()
for name, predictor in rag.named_predictors():
    print(f"{name}: {predictor}")

# Inspecionar historico de chamadas ao LM
rag.inspect_history(max_lines=20)

# Salvar e carregar estado otimizado
rag.save("rag_v1.json")
rag2 = AdvancedRAG().load("rag_v1.json")

🔧Optimizers — Otimização Automática de Prompts

O grande diferencial do DSPy: compilar um program — dado um conjunto de exemplos e uma métrica, o optimizer ajusta automaticamente as instruções e os exemplos (few-shot) para maximizar a performance.

BootstrapFewShot — ponto de partida

from dspy.optimizers import BootstrapFewShot

# 1. Definir programa
student = dspy.ChainOfThought('question -> answer')

# 2. Definir metrica
def validate_answer(example, pred, trace=None):
    return example.answer.lower() in pred.answer.lower()

# 3. Criar optimizer e compilar
optimizer = BootstrapFewShot(
    metric=validate_answer,
    max_bootstrapped_demos=4,   # max. de exemplos gerados
    max_labeled_demos=16,        # max. do trainset
    metric_threshold=0.8         # so aceita demos com score >= 0.8
)

compiled = optimizer.compile(
    student,
    trainset=train_data,          # lista de dspy.Example
    teacher=dspy.ChainOfThought('question -> answer')
)

# Usar programa compilado
result = compiled(question='O que e RAG?')
print(result.answer)

MIPROv2 — otimização de instruções + exemplos

from dspy.optimizers import MIPROv2

# MIPROv2 otimiza AMBOS instrucoes e exemplos via Bayesian Optimization
optimizer = MIPROv2(
    metric=validate_answer,
    auto='medium'  # 'light' | 'medium' | 'heavy'
)

compiled = optimizer.compile(
    student,
    trainset=train_data,
    valset=dev_data,    # conjunto de validacao obrigatorio
    num_trials=30      # iteracoes do Bayesian optimizer
)

# Configuracao manual mais granular
optimizer_manual = MIPROv2(
    metric=validate_answer,
    num_bootstrapped_demos=3,
    num_candidate_instructions=5,
    num_trials=50
)

Workflow de Otimização Completo

import dspy
from dspy.evaluate import Evaluate
from dspy.optimizers import BootstrapFewShot, MIPROv2

# Separar dados - NUNCA use test no otimizador
train, dev, test = split_data(all_data, ratios=[0.7, 0.15, 0.15])

student = dspy.ChainOfThought('question -> answer')
metric = lambda ex, pred, trace=None: ex.answer.lower() in pred.answer.lower()
evaluator = Evaluate(devset=dev, metric=metric, num_threads=8)

# Fase 1: baseline
baseline = evaluator(student)
print(f"Baseline: {baseline:.2%}")

# Fase 2: BootstrapFewShot
v1 = BootstrapFewShot(metric=metric).compile(student, trainset=train)
print(f"BootstrapFewShot: {evaluator(v1):.2%}")

# Fase 3: MIPROv2 (mais poderoso)
v2 = MIPROv2(metric=metric, auto='light').compile(
    student, trainset=train, valset=dev, num_trials=20
)
print(f"MIPROv2: {evaluator(v2):.2%}")

v2.save("best_model.json")

📏Métricas e Avaliação

Uma métrica é uma função Python que recebe (example, pred, trace=None) e retorna bool, int ou float. É ela que guia todo o processo de otimização.

Tipos de métricas

# 1. Booleana simples
def exact_match(example, pred, trace=None):
    return pred.answer.lower() == example.answer.lower()

# 2. F1 score por palavras (robusta a variacoes de redacao)
def f1_score(example, pred, trace=None):
    pred_words = set(pred.answer.lower().split())
    gold_words = set(example.answer.lower().split())
    if not gold_words: return 0
    overlap = len(pred_words & gold_words)
    precision = overlap / len(pred_words) if pred_words else 0
    recall    = overlap / len(gold_words)
    return 2 * precision * recall / (precision + recall) if (precision + recall) else 0

# 3. LLM-as-judge (para outputs longos ou subjetivos)
class EvalAnswer(dspy.Signature):
    """Rate if the predicted answer is correct given the gold answer."""
    gold_answer: str = dspy.InputField()
    predicted:   str = dspy.InputField()
    score: int = dspy.OutputField(desc="0 (errado) ou 1 (correto)")

judge = dspy.Predict(EvalAnswer)

def llm_judge(example, pred, trace=None):
    r = judge(gold_answer=example.answer, predicted=pred.answer)
    return bool(int(r.score))

# 4. Multi-criterio
def quality_metric(example, pred, trace=None):
    if not pred.answer: return 0
    if len(pred.answer.split()) > 200: return 0.5  # penalizar muito longa
    return float(example.answer.lower() in pred.answer.lower())

dspy.Evaluate

from dspy.evaluate import Evaluate

evaluator = Evaluate(
    devset=dev_data,
    metric=f1_score,
    num_threads=8,          # paralelismo
    display_progress=True,
    display_table=5           # exibe as 5 primeiras previsoes
)

score = evaluator(compiled_program)
print(f"F1 medio: {score:.3f}")  # score entre 0.0 e 1.0

📚RAG com DSPy

Retrieval-Augmented Generation é o caso de uso mais comum com DSPy. O framework abstrai a busca e a geração em módulos composáveis e otimizáveis.

RAG completo com ChromaDB

import dspy
import chromadb
from dspy.retrieval import ChromadbRM
from dspy.optimizers import BootstrapFewShot

# 1. Indexar documentos no ChromaDB
client = chromadb.Client()
col = client.create_collection("knowledge_base")

for doc in documents:
    col.add(ids=[doc['id']], documents=[doc['text']])

rm = ChromadbRM(collection_name="knowledge_base", client=client)
dspy.configure(lm=lm, rm=rm)

# 2. Definir as Signatures
class GenerateAnswer(dspy.Signature):
    """Answer using retrieved context."""
    context:  list[str] = dspy.InputField(desc="relevant passages")
    question: str = dspy.InputField()
    answer:   str = dspy.OutputField()

# 3. Programa RAG com query refinada
class OptimizedRAG(dspy.Module):
    def __init__(self, k=5):
        self.refine_query = dspy.Predict('question -> search_query: str')
        self.retrieve     = dspy.Retrieve(k=k)
        self.answer       = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        query   = self.refine_query(question=question).search_query
        context = self.retrieve(query).passages
        return  self.answer(context=context, question=question)

# 4. Otimizar com dados rotulados
def rag_metric(example, pred, trace=None):
    return example.answer.lower() in pred.answer.lower()

rag = OptimizedRAG()
compiled_rag = BootstrapFewShot(metric=rag_metric).compile(
    rag, trainset=labeled_qa_pairs
)

# 5. Usar em producao
result = compiled_rag(question='Como funciona o cache do Redis?')
print(result.answer)
print(result.reasoning)

Retentativas com refinamento

class SelfRefiningRAG(dspy.Module):
    """RAG com rodada de refinamento da resposta inicial."""

    def __init__(self):
        self.retrieve = dspy.Retrieve(k=4)
        self.generate = dspy.ChainOfThought('context, question -> answer')
        self.refine   = dspy.ChainOfThought(
            'context, question, draft_answer -> refined_answer'
        )

    def forward(self, question):
        ctx     = '\n'.join(self.retrieve(question).passages)
        draft   = self.generate(context=ctx, question=question)
        refined = self.refine(
            context=ctx, question=question, draft_answer=draft.answer
        )
        return refined

✅Boas Práticas e Produção

Checklist de produção

• Separe sempre Train / Dev / Test — o conjunto de test nunca deve entrar no otimizador.
• Defina a métrica antes de escrever o programa — ela é o objetivo de negócio.
• Comece simples — dspy.Predict → ChainOfThought → Pipeline → Optimizer.
• Use cache=True no LM durante desenvolvimento — evita custos repetidos na iteração.
• Salve o programa compilado: program.save("prod_v1.json").
• Versione o JSON do programajunto com o código — é o "modelo" de produção.
• Monitore a métrica em produção — amostra de outputs reais e avalie periodicamente.
• Implemente fallback se o LM falhar — nunca deixe um erro de API chegar ao usuário.

Depuração e inspeção

import dspy

program = dspy.ChainOfThought('question -> answer')
result  = program(question='O que e entropy?')

# Ver o ultimo prompt enviado ao LM
dspy.settings.lm.inspect_history(n=1)

# Ver todos os predictores do programa
for name, pred in program.named_predictors():
    print(name, pred.signature)

# Definir LM diferente por modulo
program.generate.set_lm(
    dspy.LM('anthropic/claude-haiku-4-5-20251001')  # mais barato para etapas simples
)

Integração com observabilidade

# Langfuse - logging automatico via callback
from langfuse.dspy import LangfuseInstrumentor
LangfuseInstrumentor().instrument()  # instrumenta todas as chamadas DSPy

# Weights & Biases
import wandb
wandb.init(project="dspy-rag")

# Durante a compilacao, MIPROv2 ja loga metricas automaticamente
optimizer = MIPROv2(metric=metric, auto='medium')
compiled  = optimizer.compile(program, trainset=train, valset=dev)

Exemplos de casos de uso reais