Guia Completo de DSPy

🧠Conceitos e Filosofia

DSPy é um framework do Stanford NLP para programar — em vez de prompt-engineering — modelos de linguagem. A ideia central: separar o que você quer (a lógica do programa) de como o LLM é instruído (prompts, demonstrações, formatação). O framework otimiza prompts e pesos automaticamente a partir de métricas e exemplos.

Em vez de escrever strings longas com instruções frágeis e exemplos chumbados, você declara Signatures (contratos de entrada/saída) e compõe Modules (Predict, ChainOfThought, ReAct, etc.). Um Optimizer(antigo "teleprompter") depois ajusta as demonstrações e instruções para maximizar sua métrica.

# Instalacao - Python 3.10+
# pip install dspy  (pacote oficial no PyPI)

import dspy

# 1. Configure um Language Model globalmente
lm = dspy.LM('openai/gpt-4o-mini', api_key='sk-...', max_tokens=1000)
dspy.configure(lm=lm)

# 2. Declare o que voce quer (Signature inline)
#    Sintaxe: "entrada1, entrada2 -> saida1, saida2"
classify = dspy.Predict('texto -> sentimento: str')

# 3. Chame como uma funcao Python normal
resultado = classify(texto='Adorei o atendimento, muito rapido!')
print(resultado.sentimento)  # -> "positivo"

# Sob o capo, o DSPy gerou o prompt, chamou o LM, parseou a saida
# e retornou um Prediction com campos tipados.

Programação vs. Prompt Engineering

Comparação com LangChain / LlamaIndex

• LangChain: cadeias imperativas de prompts encadeados, foco em integrações. Prompts continuam sendo strings manuais.
• LlamaIndex: foco em RAG/ingestão de documentos. Ótimo para indexação; menos para composição de raciocínio.
• DSPy: compilador de programas LLM. Signatures + Modules + Optimizers. O prompt é artefato compilado, não código-fonte.

✍️Signatures — Contratos Tipados

Uma Signature descreve o quê o módulo deve fazer, não como. Pode ser inline (uma string) ou declarativa (uma classe). A docstring vira a instrução principal e os InputField/OutputField viram campos estruturados no prompt compilado.

Signatures inline

import dspy

# Forma mais curta - otima para prototipos
qa = dspy.Predict('question -> answer')
summarize = dspy.Predict('document -> summary')

# Com tipos (DSPy 2.5+ respeita anotacoes em signatures inline)
classify = dspy.Predict('text -> sentiment: str, confidence: float')

# Multiplas entradas e saidas
translate = dspy.Predict('text, target_language -> translation, quality_score: float')

out = classify(text='O produto chegou quebrado.')
print(out.sentiment)    # "negativo"
print(out.confidence)   # 0.92

Signatures declarativas (classes)

from typing import Literal
import dspy

class ClassifyEmotion(dspy.Signature):
    """Classifica a emocao dominante em um texto curto em portugues."""

    text: str = dspy.InputField(desc='Texto do usuario, 1 a 3 frases.')
    emotion: Literal['alegria', 'raiva', 'medo', 'tristeza', 'surpresa', 'nojo'] = dspy.OutputField(
        desc='Emocao dominante entre as 6 universais de Ekman.'
    )
    intensity: float = dspy.OutputField(desc='Intensidade de 0.0 a 1.0.')

classify = dspy.Predict(ClassifyEmotion)
r = classify(text='Nao acredito que ele mentiu de novo, estou fervendo.')
print(r.emotion)     # "raiva"
print(r.intensity)   # 0.87

Tipos complexos com Pydantic

from typing import List, Literal
from pydantic import BaseModel
import dspy

class Entity(BaseModel):
    name: str
    type: Literal['pessoa', 'empresa', 'local', 'valor']
    start: int
    end: int

class ExtractEntities(dspy.Signature):
    """Extrai entidades nomeadas do texto, com posicoes absolutas."""
    text: str = dspy.InputField()
    entities: List[Entity] = dspy.OutputField(desc='Lista de entidades encontradas.')

extract = dspy.Predict(ExtractEntities)
r = extract(text='Kaique trabalha na Anthropic em Sao Paulo desde 2024.')
for e in r.entities:
    print(e.name, e.type)

🧩Modules — Predict, ChainOfThought, ReAct

Módulos são blocos compostáveis que executam uma signature. Cada um implementa um padrão de raciocínio diferente e, ao ser compilado, gera o prompt apropriado para aquele padrão.

dspy.Predict — chamada direta

import dspy

answer = dspy.Predict('question -> answer')
r = answer(question='Qual a capital do Japao?')
print(r.answer)  # "Toquio"

dspy.ChainOfThought — raciocínio passo a passo

Adiciona automaticamente um campo reasoning antes da resposta final. O LM é forçado a pensar em voz alta, o que costuma melhorar tarefas que exigem lógica ou contagem.

import dspy

cot = dspy.ChainOfThought('question -> answer')
r = cot(question='Se 3 camisas secam em 2 horas ao sol, quanto tempo secam 9 camisas?')
print(r.reasoning)  # "As camisas secam em paralelo, nao em serie..."
print(r.answer)     # "2 horas"

dspy.ReAct — raciocínio + ferramentas

Implementa o padrão ReAct (Reasoning + Acting). O modelo alterna entre pensar, chamar ferramenta e observar, até ter informação suficiente para responder.

import dspy

def search_wikipedia(query: str) -> str:
    """Busca na Wikipedia e retorna o resumo do primeiro resultado."""
    import requests
    r = requests.get(f'https://en.wikipedia.org/api/rest_v1/page/summary/{query}')
    return r.json().get('extract', 'Nada encontrado.')

def calculator(expression: str) -> str:
    """Avalia uma expressao matematica Python simples."""
    return str(eval(expression, {'__builtins__': {}}, {}))

agent = dspy.ReAct(
    'question -> answer',
    tools=[search_wikipedia, calculator],
    max_iters=5,
)

r = agent(question='Qual e a raiz quadrada da populacao atual de Toquio?')
print(r.answer)

dspy.ProgramOfThought — gera e executa código

import dspy

# Gera codigo Python, executa em sandbox e usa o stdout como resposta.
pot = dspy.ProgramOfThought('question -> answer')
r = pot(question='Qual o 15o numero de Fibonacci?')
print(r.answer)  # "610"

Compondo módulos

A parte mais poderosa do DSPy: módulos viram classes que herdam de dspy.Module e se compõem como camadas em PyTorch.

import dspy

class RAGPipeline(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought('context, question -> answer')

    def forward(self, question: str):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

rag = RAGPipeline(num_passages=5)
resposta = rag(question='Como o DSPy otimiza prompts?')
print(resposta.answer)

🔄Optimizers — Compilando seu Programa

Otimizadores (antigamente "teleprompters") transformam um programa DSPy + métrica + exemplos em um programa compilado com melhores prompts e/ou demonstrações few-shot. É o "backprop" do DSPy.

Anatomia de um dataset DSPy

import dspy

# Exemplos sao dspy.Example com .with_inputs() indicando quais campos sao entrada
trainset = [
    dspy.Example(
        question='Capital do Brasil?',
        answer='Brasilia'
    ).with_inputs('question'),
    dspy.Example(
        question='Quantos planetas tem o sistema solar?',
        answer='8'
    ).with_inputs('question'),
    # ... geralmente 20-200 exemplos
]

devset = [...]  # para validacao

Métrica — o sinal de otimização

def exact_match(example, pred, trace=None) -> bool:
    """Retorna True se a resposta prevista bate com a esperada."""
    return example.answer.strip().lower() == pred.answer.strip().lower()

# Metricas podem ser qualquer funcao: regex, BLEU, similaridade semantica,
# ou ate um LLM-as-judge (outro dspy.Predict).
def llm_judge(example, pred, trace=None) -> float:
    judge = dspy.Predict('question, gold_answer, predicted_answer -> correct: bool')
    r = judge(question=example.question, gold_answer=example.answer,
              predicted_answer=pred.answer)
    return 1.0 if r.correct else 0.0

BootstrapFewShot — o começo rápido

Executa seu programa nos exemplos de treino, guarda as execuções bem-sucedidas como demonstrações e injeta as melhores no prompt final.

from dspy.teleprompt import BootstrapFewShot

optimizer = BootstrapFewShot(
    metric=exact_match,
    max_bootstrapped_demos=4,   # demos geradas pelo proprio modelo
    max_labeled_demos=4,        # demos do trainset puro
)

compiled_rag = optimizer.compile(
    student=RAGPipeline(),
    trainset=trainset,
)

# compiled_rag agora tem demos embutidas e funciona como qualquer Module
r = compiled_rag(question='...')

# Persistencia
compiled_rag.save('rag_compiled.json')

new_rag = RAGPipeline()
new_rag.load('rag_compiled.json')

MIPROv2 — o otimizador de ponta

Multiprompt Instruction Proposal Optimizer v2. Propõe novas instruções para cada Predict, testa combinações via Bayesian Optimization e retém a melhor. Costuma render ganhos significativos em benchmarks.

from dspy.teleprompt import MIPROv2

optimizer = MIPROv2(
    metric=exact_match,
    auto='medium',          # 'light' | 'medium' | 'heavy'
    num_threads=8,
)

compiled = optimizer.compile(
    student=RAGPipeline(),
    trainset=trainset,
    valset=devset,
    requires_permission_to_run=False,
)

BootstrapFinetune — otimizando pesos, não só prompts

from dspy.teleprompt import BootstrapFinetune

# Gera dados sinteticos via modelo professor e fine-tuna um modelo menor.
optimizer = BootstrapFinetune(metric=exact_match)
compiled = optimizer.compile(
    student=RAGPipeline(),   # rodara em um LM pequeno
    teacher=RAGPipeline(),   # tipicamente GPT-4o ou Claude
    trainset=trainset,
)

COPRO — otimização de instruções

from dspy.teleprompt import COPRO

optimizer = COPRO(metric=exact_match, breadth=10, depth=3)
compiled = optimizer.compile(
    student=RAGPipeline(),
    trainset=trainset,
    eval_kwargs={'num_threads': 8, 'display_progress': True},
)

📚RAG com DSPy

RAG (Retrieval-Augmented Generation) é cidadão de primeira classe no DSPy. O framework tem integrações nativas com ColBERTv2, ChromaDB, Qdrant, Weaviate, Pinecone, LanceDB, MongoDB Atlas Vector e mais.

Configurando um retriever

import dspy

# Opcao 1: ColBERTv2 hospedado (otimo para demos)
colbert = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(lm=lm, rm=colbert)

# Opcao 2: ChromaDB local
from dspy.retrieve.chromadb_rm import ChromadbRM

chroma = ChromadbRM(
    collection_name='docs',
    persist_directory='./chroma_db',
    k=5,
)
dspy.configure(lm=lm, rm=chroma)

RAG simples com ChainOfThought

class SimpleRAG(dspy.Module):
    def __init__(self, k=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=k)
        self.answer = dspy.ChainOfThought('context, question -> answer')

    def forward(self, question: str):
        passages = self.retrieve(question).passages
        return self.answer(context=passages, question=question)

rag = SimpleRAG(k=5)
r = rag(question='O que e retrieval-augmented generation?')
print(r.answer)

Multi-hop RAG — quando uma busca não basta

Perguntas como Qual a idade do CEO da empresa que comprou a Activision? exigem várias rodadas de busca. DSPy torna isso elegante.

class MultiHopRAG(dspy.Module):
    def __init__(self, num_hops=3, k=5):
        super().__init__()
        self.num_hops = num_hops
        self.retrieve = dspy.Retrieve(k=k)
        self.generate_query = dspy.ChainOfThought(
            'context, question -> search_query'
        )
        self.generate_answer = dspy.ChainOfThought(
            'context, question -> answer'
        )

    def forward(self, question: str):
        context = []
        for hop in range(self.num_hops):
            query = self.generate_query(context=context, question=question).search_query
            new_passages = self.retrieve(query).passages
            context = list(set(context + new_passages))  # dedup
        return self.generate_answer(context=context, question=question)

multihop = MultiHopRAG(num_hops=3)
r = multihop(question='Qual a nacionalidade do diretor de Parasita?')
print(r.answer)

Combinando RAG + Optimizer

from dspy.teleprompt import MIPROv2

def rag_metric(example, pred, trace=None):
    # Verifica se a resposta contem a resposta esperada
    return example.answer.lower() in pred.answer.lower()

optimizer = MIPROv2(metric=rag_metric, auto='light', num_threads=8)
compiled_rag = optimizer.compile(
    student=MultiHopRAG(),
    trainset=rag_trainset,
    valset=rag_devset,
    requires_permission_to_run=False,
)

# Agora cada Predict interno tem demos selecionadas e instrucoes otimizadas.
compiled_rag.save('multihop_compiled.json')

⚙️Produção — LMs, Cache, Async, Tracing

Na prática, rodar DSPy em produção significa: escolher o LM certo, gerenciar cache, observar execuções e lidar com concorrência.

Configurando múltiplos LMs

import dspy

# dspy.LM usa LiteLLM por baixo - suporta 100+ providers
gpt4o = dspy.LM('openai/gpt-4o', api_key='sk-...', max_tokens=2000, temperature=0.0)
gpt4o_mini = dspy.LM('openai/gpt-4o-mini', api_key='sk-...', max_tokens=1000)
claude = dspy.LM('anthropic/claude-sonnet-4-5', api_key='sk-ant-...')
ollama = dspy.LM('ollama_chat/llama3.1:8b',
                 api_base='http://localhost:11434',
                 api_key='')

# Configuracao global
dspy.configure(lm=gpt4o_mini)

# Override pontual via context manager
with dspy.context(lm=gpt4o):
    critico = dspy.ChainOfThought('argumento -> critica_rigorosa')
    r = critico(argumento='DSPy e so wrapper de string.')

Cache — nativo e transparente

# Cache automatico em disco (~/.dspy_cache/ por padrao)
# Mesmas entradas = mesma saida, sem bater no LM.

# Controle fino
lm = dspy.LM(
    'openai/gpt-4o-mini',
    cache=True,           # default
    cache_in_memory=True  # cache RAM alem do disco
)

# Desabilitar cache para uma chamada especifica
with dspy.context(lm=dspy.LM('openai/gpt-4o-mini', cache=False)):
    r = classify(text='...')

Execução assíncrona e paralela

import asyncio
import dspy

# Qualquer Module tem .acall() para chamada assincrona
async def main():
    qa = dspy.ChainOfThought('question -> answer')

    questions = ['Capital da Franca?', 'Capital do Japao?', 'Capital do Brasil?']

    # Paralelismo trivial
    tasks = [qa.acall(question=q) for q in questions]
    results = await asyncio.gather(*tasks)
    for q, r in zip(questions, results):
        print(q, '->', r.answer)

asyncio.run(main())

# Para batch sincrono com threads
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=8) as ex:
    results = list(ex.map(lambda q: qa(question=q), questions))

Observabilidade — MLflow e Phoenix

# --- Opcao 1: MLflow Tracing (nativo desde DSPy 2.5)
import mlflow

mlflow.dspy.autolog()
mlflow.set_tracking_uri('http://localhost:5000')
mlflow.set_experiment('rag-pipeline')

# Qualquer chamada DSPy agora aparece como trace no MLflow UI
r = compiled_rag(question='...')

# --- Opcao 2: Arize Phoenix (OpenTelemetry)
import phoenix as px
from openinference.instrumentation.dspy import DSPyInstrumentor

px.launch_app()
DSPyInstrumentor().instrument()

# Abra http://localhost:6006 para inspecionar spans, prompts e latencias.

Inspecionando o histórico

# Ver as ultimas N interacoes com o LM - ouro para debug
dspy.inspect_history(n=3)

# Saida inclui: prompt final gerado, resposta crua, tempo, tokens.

Retries e rate-limits

lm = dspy.LM(
    'openai/gpt-4o-mini',
    num_retries=8,   # tenta 8x com backoff
    max_tokens=2000,
)

# O LiteLLM interno trata 429s com exponential backoff.
# Para timeouts customizados:
import litellm
litellm.request_timeout = 60  # segundos

Estrutura recomendada de projeto

project/
├── dspy_app/
│   ├── __init__.py
│   ├── signatures.py       # classes dspy.Signature
│   ├── modules.py          # dspy.Module compondo Predict/CoT/ReAct
│   ├── metrics.py          # funcoes de avaliacao
│   ├── pipelines.py        # entrypoints que instanciam tudo
│   └── compiled/
│       └── rag_v1.json     # programas compilados versionados
├── data/
│   ├── trainset.jsonl
│   └── devset.jsonl
├── scripts/
│   ├── compile.py          # roda MIPROv2 e salva em compiled/
│   └── eval.py
└── tests/
    └── test_rag.py         # smoke tests com cache ligado