Dspy Retrieval

Retrieval Modules in DSPy

Guide the user through DSPy's retrieval modules for searching documents, computing embeddings, and building RAG (retrieval-augmented generation) pipelines.

Step 1: Gather context

Before building retrieval into a DSPy program, clarify:

1. What are you searching over? Your own documents, a knowledge base, an external corpus like Wikipedia? 2. How large is the corpus? A few hundred docs (in-memory FAISS works) vs. millions (need a dedicated vector store like Pinecone, Qdrant, or Chroma)? 3. Do you already have a search backend? If you have Elasticsearch, Pinecone, or another store, subclass dspy.Retrieve to wrap it. If not, use dspy.retrievers.Embeddings for a local solution. 4. Single-hop or multi-hop? Simple questions need one retrieval step. Compositional questions (e.g., "Where was the designer of the Eiffel Tower born?") need chained retrieval.

What retrieval modules are

DSPy provides retrieval modules that fetch relevant documents or passages given a query. These modules plug into DSPy programs just like dspy.Predict or dspy.ChainOfThought -- declare them in __init__, call them in forward(), and optimizers handle the rest.

There are four key components:

dspy.Retrieve

The base class for all retrievers. Use it directly with a configured retrieval model (rm), or subclass it to wrap your own search backend.

Using with a configured RM

import dspy

# Configure a retrieval model globally
colbert = dspy.ColBERTv2(url="http://your-server:8893/api/search")
lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm, rm=colbert)

# Use dspy.Retrieve -- it delegates to the configured rm
retriever = dspy.Retrieve(k=5)
result = retriever("What is retrieval-augmented generation?")
print(result.passages)  # list[str] of top-k passages

Key parameters

• `k` (int) -- number of passages to retrieve. Can be set at init time or overridden per call.

Return value

dspy.Retrieve returns a dspy.Prediction with a .passages attribute -- a list[str] of the top-k retrieved passages.

Subclassing for custom backends

Wrap any search system by subclassing dspy.Retrieve and implementing forward():

class MyRetriever(dspy.Retrieve):
    def __init__(self, search_client, k=3):
        super().__init__(k=k)
        self.client = search_client

    def forward(self, query, k=None):
        k = k or self.k
        results = self.client.search(query, top_k=k)
        return dspy.Prediction(passages=[r["text"] for r in results])

The forward() method must: 1. Accept query (str) and optional k (int) 2. Return a dspy.Prediction with a passages field (list of strings)

dspy.ColBERTv2

A retrieval client that queries a hosted ColBERTv2 server. ColBERTv2 is a neural retrieval model that provides high-quality passage retrieval.

Constructor

colbert = dspy.ColBERTv2(url="http://your-server:8893/api/search")

Parameters:

• `url` (str) -- URL of the ColBERTv2 server endpoint

Usage

# Direct call
results = colbert("What is DSPy?", k=3)
# Returns list of dicts with 'text', 'score', etc.

# As a configured retrieval model
dspy.configure(lm=lm, rm=colbert)
retriever = dspy.Retrieve(k=5)
passages = retriever("search query").passages

Setting up a ColBERTv2 server

Stanford hosts a public ColBERTv2 server for Wikipedia that you can use for testing:

colbert = dspy.ColBERTv2(url="http://20.102.90.50:2017/wiki17_abstracts")
dspy.configure(lm=lm, rm=colbert)

For your own data, you need to run a ColBERTv2 server. See the ColBERT repository for setup instructions.

dspy.Embedder

Computes embeddings for text using any LiteLLM-supported embedding model. This is not a retriever itself -- it turns text into vectors that you can use with dspy.retrievers.Embeddings or your own vector store.

Constructor

embedder = dspy.Embedder(
    "openai/text-embedding-3-small",  # model identifier (LiteLLM format)
    dimensions=512,                    # optional: output dimensions
)

Parameters:

• model (str | Callable) -- embedding model in LiteLLM format (e.g., "openai/text-embedding-3-small", "cohere/embed-english-v3.0"), or a callable for custom embedding functions
• batch_size (int, default 200) -- batch size for embedding multiple texts
• caching (bool, default True) -- whether to cache embedding responses for hosted models
• *\\kwargs* -- additional model-specific arguments (e.g., dimensions=512 for models that support it)

Usage

# Embed a single text
vector = embedder("What is DSPy?")
# Returns a 1D numpy array

# Embed multiple texts
vectors = embedder(["text one", "text two", "text three"])
# Returns a 2D numpy array (shape: num_texts x embedding_dim)

Supported providers

Any embedding model supported by LiteLLM works:

# OpenAI
embedder = dspy.Embedder("openai/text-embedding-3-small")

# Cohere
embedder = dspy.Embedder("cohere/embed-english-v3.0")

# Local via Ollama
embedder = dspy.Embedder("ollama/nomic-embed-text")

dspy.retrievers.Embeddings

A local vector search retriever that uses FAISS under the hood. Give it an Embedder and a corpus, and it builds an in-memory index for fast similarity search.

Constructor

import dspy

embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
search = dspy.retrievers.Embeddings(
    corpus=corpus,      # list[str] of documents
    embedder=embedder,
    k=5,                # number of results to return (default 5)
)

Parameters:

• `corpus` (list[str]) -- the documents to index and search over
• `embedder` -- a dspy.Embedder instance
• `k` (int, default 5) -- default number of results to return
• `brute_force_threshold` (int, default 20000) -- corpus size above which FAISS indexing kicks in (below this, brute-force search)
• `normalize` (bool, default True) -- whether to normalize embeddings

Saving and loading embeddings

Avoid re-embedding large corpora on every run:

# Save after initial indexing
search.save("./my_embeddings")

# Load later without re-computing
search = dspy.retrievers.Embeddings.from_saved("./my_embeddings", embedder=embedder)

Usage

# Search
result = search("How do I reset my password?")
print(result.passages)  # list[str] of top-k matching documents

# Use in a module
class QA(dspy.Module):
    def __init__(self, search):
        self.search = search
        self.answer = dspy.ChainOfThought("context, question -> answer")

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

When to use Embeddings vs. ColBERTv2

Building RAG pipelines

RAG is the most common use of retrieval in DSPy. The pattern: retrieve relevant passages, then generate an answer grounded in them.

Basic RAG

import dspy

class RAG(dspy.Module):
    def __init__(self, retriever, k=3):
        self.retrieve = retriever
        self.generate = dspy.ChainOfThought("context, question -> answer")

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

# With Embeddings retriever
embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=my_docs, k=5)
rag = RAG(retriever=search)
result = rag(question="How do refunds work?")
print(result.answer)

RAG with source grounding

Use dspy.Refine to enforce that answers stay grounded in the retrieved context:

class GroundedRAG(dspy.Module):
    def __init__(self, retriever):
        self.retrieve = retriever
        self.generate = dspy.ChainOfThought(
            "context, question -> answer, cited_sources: list[int]"
        )

    def forward(self, question):
        passages = self.retrieve(question).passages
        result = self.generate(context=passages, question=question)
        return dspy.Prediction(
            answer=result.answer,
            cited_sources=result.cited_sources,
            passages=passages,
        )

def grounding_reward(args, pred):
    score = 1.0
    if not pred.cited_sources or len(pred.cited_sources) == 0:
        score -= 0.3  # soft penalty for missing citations
    return score

grounded_rag = dspy.Refine(module=GroundedRAG(retriever=search), N=3, reward_fn=grounding_reward, threshold=0.8)

Multi-hop RAG

When a question needs information from multiple documents, chain retrieval steps:

class MultiHopRAG(dspy.Module):
    def __init__(self, retriever, hops=2):
        self.retrieve = retriever
        self.generate_query = [
            dspy.ChainOfThought("context, question -> search_query")
            for _ in range(hops)
        ]
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = []
        for hop in self.generate_query:
            query = hop(context=context, question=question).search_query
            new_passages = self.retrieve(query).passages
            context = list(dict.fromkeys(context + new_passages))  # deduplicate

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

Configuring retrievers

There are two ways to wire up a retriever:

Option 1: Global configuration with dspy.configure

colbert = dspy.ColBERTv2(url="http://your-server:8893/api/search")
dspy.configure(lm=lm, rm=colbert)

# dspy.Retrieve() now uses colbert automatically
retriever = dspy.Retrieve(k=5)

Option 2: Pass the retriever directly

embedder = dspy.Embedder("openai/text-embedding-3-small")
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=docs, k=5)

class MyRAG(dspy.Module):
    def __init__(self):
        self.search = search  # use directly, no global config needed
        self.answer = dspy.ChainOfThought("context, question -> answer")

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

Option 2 is more explicit and avoids global state. Prefer it when your program uses a single retriever.

The k parameter

The k parameter controls how many passages to retrieve. It can be set at multiple levels:

# At init time
retriever = dspy.Retrieve(k=5)

# Override per call
result = retriever("query", k=10)

# In Embeddings constructor
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=docs, k=3)

Choosing k:

• Start with k=3 to k=5 for most tasks
• Increase k for questions that need broader context
• Decrease k for faster inference and lower token costs
• More passages means more context for the LM, but also more noise and higher cost
• Use evaluation to find the optimal k for your specific task

Grounded generation with Citations

dspy.experimental.Citations is a type (not a module) that you use as an OutputField to get structured source references from the LM. It works with Anthropic models that support native citations, or falls back to LM-generated citation extraction.

from dspy.experimental import Citations, Document

class AnswerWithSources(dspy.Signature):
    """Answer the question and cite the source documents."""
    documents: list[Document] = dspy.InputField()
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()
    citations: Citations = dspy.OutputField()

lm = dspy.LM("anthropic/claude-sonnet-4-5-20250929")  # or "openai/gpt-4o", etc.
predictor = dspy.Predict(AnswerWithSources)
result = predictor(documents=docs, question="What is the refund policy?")
# result.citations contains structured Citation objects with cited_text, document_index, etc.

When to use: RAG pipelines where claims need to trace back to source documents with exact quoted text and document indices.

Note: This is in dspy.experimental — the API may change. For broader anti-hallucination patterns, see /ai-stopping-hallucinations.

Gotchas

• Using `dspy.Retrieve` without configuring `rm`. Claude often writes dspy.Retrieve(k=5) without setting dspy.configure(rm=...) first. Without a configured retrieval model, calling Retrieve raises a confusing error. Either configure rm globally or pass a concrete retriever (like dspy.retrievers.Embeddings) directly to your module.
• Re-embedding the corpus on every run. Claude builds dspy.retrievers.Embeddings(corpus=docs, embedder=embedder) in scripts without saving. For corpora over a few hundred docs, this wastes time and API calls. Use search.save("./embeddings") after initial indexing and Embeddings.from_saved("./embeddings", embedder=embedder) on subsequent runs.
• Forgetting `.with_inputs()` on RAG examples. When building training data for RAG optimization, Claude creates dspy.Example(question=q, answer=a) without calling .with_inputs("question"). The optimizer silently treats all fields as inputs. Always chain .with_inputs() to mark which fields are inputs vs. expected outputs.
• Returning raw dicts instead of `dspy.Prediction` from custom retrievers. When subclassing dspy.Retrieve, the forward() method must return dspy.Prediction(passages=[...]) — not a list or dict. Returning the wrong type causes downstream modules to fail when they access .passages.
• Setting k too high for the context window. Claude defaults to k=10 or higher, which can stuff too many passages into the generation prompt and exceed the LM context or degrade answer quality. Start with k=3 to k=5 and increase based on evaluation results.

Additional resources

• dspy.Retrieve API docs
• dspy.ColBERTv2 API docs
• dspy.Embedder API docs
• For API details, see reference.md
• For worked examples, see examples.md

Cross-references

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• Building custom modules to wrap retrieval logic -- see /dspy-modules
• Vector database setup (Qdrant, Pinecone, ChromaDB, Weaviate) -- see /dspy-qdrant
• End-to-end document search with vector stores and chunking -- see /ai-searching-docs
• Keeping answers grounded and avoiding hallucination -- see /ai-stopping-hallucinations
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

dspy-retrieval -- Worked Examples

Example 1: RAG pipeline with ColBERTv2

A complete RAG pipeline that uses a hosted ColBERTv2 server to retrieve Wikipedia passages and answer questions with citations.

import dspy
from typing import Literal


class AnswerWithConfidence(dspy.Signature):
    """Answer the question using only the provided context. Say 'insufficient information' if the context doesn't contain the answer."""
    context: list[str] = dspy.InputField(desc="Retrieved passages from the knowledge base")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Answer grounded in the context")
    confidence: Literal["high", "medium", "low"] = dspy.OutputField()


class ColBERTRAG(dspy.Module):
    """RAG pipeline backed by ColBERTv2 retrieval."""

    def __init__(self, k=5):
        self.retrieve = dspy.Retrieve(k=k)
        self.generate = dspy.ChainOfThought(AnswerWithConfidence)

    def forward(self, question):
        # Retrieve relevant passages
        passages = self.retrieve(question).passages

        if not passages:
            return dspy.Prediction(
                answer="No relevant passages found.",
                confidence="low",
                passages=[],
            )

        # Generate a grounded answer
        result = self.generate(context=passages, question=question)

        return dspy.Prediction(
            answer=result.answer,
            confidence=result.confidence,
            passages=passages,
        )


def rag_confidence_reward(args, pred):
    """Prefer high- or medium-confidence answers; penalize low confidence."""
    if pred.confidence == "low":
        return 0.5
    return 1.0


# --- Setup ---

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
colbert = dspy.ColBERTv2(url="http://20.102.90.50:2017/wiki17_abstracts")
dspy.configure(lm=lm, rm=colbert)

# --- Usage ---

rag = dspy.Refine(module=ColBERTRAG(k=5), N=3, reward_fn=rag_confidence_reward, threshold=1.0)
result = rag(question="What is the capital of France?")
print(f"Answer: {result.answer}")
print(f"Confidence: {result.confidence}")
print(f"Retrieved {len(result.passages)} passages")

# --- Evaluation ---

devset = [
    dspy.Example(question="What is the capital of France?", answer="Paris").with_inputs("question"),
    dspy.Example(question="Who wrote Hamlet?", answer="William Shakespeare").with_inputs("question"),
]


def rag_metric(example, prediction, trace=None):
    answer_correct = example.answer.lower() in prediction.answer.lower()
    high_confidence = prediction.confidence in ("high", "medium")
    return answer_correct + 0.3 * high_confidence


from dspy.evaluate import Evaluate

evaluator = Evaluate(devset=devset, metric=rag_metric, num_threads=2)
score = evaluator(rag)
print(f"Score: {score}")

# --- Optimization ---

optimizer = dspy.BootstrapFewShot(metric=rag_metric, max_bootstrapped_demos=4)
optimized_rag = optimizer.compile(rag, trainset=devset)
optimized_rag.save("colbert_rag_optimized.json")

Key points:

• dspy.ColBERTv2 is set as the global retrieval model via dspy.configure(rm=colbert)
• dspy.Retrieve(k=5) delegates to ColBERTv2 automatically
• The module handles the empty-results edge case before calling the LM
• dspy.Refine retries generation when confidence is low, nudging the model toward higher-quality answers
• Optimization tunes the answer generation prompt while leaving retrieval unchanged

Example 2: Custom retriever with embeddings

Build a local retriever using dspy.Embedder and dspy.retrievers.Embeddings over your own document corpus. No external server needed.

import dspy


# --- Prepare corpus ---

documents = [
    "DSPy is a framework for programming language models instead of prompting them.",
    "Retrieval-augmented generation (RAG) combines search with language model generation.",
    "ColBERTv2 is a neural retrieval model that uses late interaction for efficient passage ranking.",
    "FAISS is a library for efficient similarity search and clustering of dense vectors.",
    "Few-shot learning uses a small number of examples to teach a model a new task.",
    "Chain-of-thought prompting improves reasoning by generating intermediate steps.",
    "Vector databases store embeddings for fast nearest-neighbor search.",
    "DSPy optimizers tune prompts automatically using training examples and a metric.",
    "Embeddings map text to dense vectors where similar texts are close together.",
    "Multi-hop retrieval chains multiple search steps to answer complex questions.",
]


# --- Build retriever ---

embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=documents, k=3)


# --- Build RAG module ---

class EmbeddingsRAG(dspy.Module):
    """RAG using local embeddings-based retrieval."""

    def __init__(self, retriever):
        self.retriever = retriever
        self.answer = dspy.ChainOfThought("context, question -> answer")

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


# --- Usage ---

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

rag = EmbeddingsRAG(retriever=search)
result = rag(question="What is DSPy?")
print(result.answer)
print(result.reasoning)

# --- Inspect what was retrieved ---

retrieved = search("What is DSPy?")
for i, passage in enumerate(retrieved.passages):
    print(f"  [{i}] {passage}")

Key points:

• dspy.Embedder handles embedding computation via LiteLLM -- works with OpenAI, Cohere, Ollama, etc.
• dspy.retrievers.Embeddings builds a FAISS index in memory from the corpus
• No global rm configuration needed -- the retriever is passed directly to the module
• The corpus is a plain list[str] -- load your documents however you like
• For larger corpora, consider using a dedicated vector store (Chroma, Pinecone) instead

Example 3: Multi-hop retrieval pattern

Answer complex questions that require combining information from multiple documents. Each hop generates a new search query based on what has been found so far.

import dspy


class GenerateSearchQuery(dspy.Signature):
    """Generate a search query to find information that is still missing."""
    context: list[str] = dspy.InputField(desc="Information gathered so far")
    question: str = dspy.InputField(desc="The original question to answer")
    search_query: str = dspy.OutputField(desc="A focused search query for the next retrieval step")


class AnswerFromContext(dspy.Signature):
    """Answer the question using all gathered context. Cite specific facts from the passages."""
    context: list[str] = dspy.InputField(desc="All retrieved passages across search steps")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Comprehensive answer grounded in the context")


class MultiHopRAG(dspy.Module):
    """Multi-hop retrieval: iteratively search, gather context, then answer."""

    def __init__(self, retriever, hops=2, passages_per_hop=3):
        self.retriever = retriever
        self.generate_query = [
            dspy.ChainOfThought(GenerateSearchQuery) for _ in range(hops)
        ]
        self.answer = dspy.ChainOfThought(AnswerFromContext)

    def forward(self, question):
        context = []

        for hop in self.generate_query:
            # Generate a search query based on what we know so far
            search_query = hop(context=context, question=question).search_query

            # Retrieve new passages
            new_passages = self.retriever(search_query).passages

            # Deduplicate and accumulate context
            context = list(dict.fromkeys(context + new_passages))

        # Generate final answer from all gathered context
        result = self.answer(context=context, question=question)

        return dspy.Prediction(
            answer=result.answer,
            reasoning=result.reasoning,
            context=context,
            num_hops=len(self.generate_query),
        )


# --- Setup with Embeddings retriever ---

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

corpus = [
    "The Eiffel Tower is located in Paris, France. It was built in 1889.",
    "Paris is the capital and most populous city of France.",
    "The Eiffel Tower was designed by Gustave Eiffel's engineering company.",
    "Gustave Eiffel was born on December 15, 1832 in Dijon, France.",
    "The Eiffel Tower stands 330 meters tall and was the tallest structure in the world until 1930.",
    "France is a country in Western Europe with a population of about 67 million.",
    "The Chrysler Building surpassed the Eiffel Tower as the tallest structure in 1930.",
    "Dijon is a city in eastern France, known as the capital of the Burgundy region.",
]

embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
search = dspy.retrievers.Embeddings(embedder=embedder, corpus=corpus, k=3)


def multihop_context_reward(args, pred):
    """Reward multi-hop results that gathered passages from multiple search steps."""
    if len(pred.context) < 2:
        return 0.5
    return 1.0


# --- Usage ---

multihop = dspy.Refine(
    module=MultiHopRAG(retriever=search, hops=2, passages_per_hop=3),
    N=3,
    reward_fn=multihop_context_reward,
    threshold=1.0,
)
result = multihop(question="Where was the designer of the Eiffel Tower born?")

print(f"Answer: {result.answer}")
print(f"Reasoning: {result.reasoning}")
print(f"Hops: {result.num_hops}")
print(f"Context gathered ({len(result.context)} passages):")
for i, passage in enumerate(result.context):
    print(f"  [{i}] {passage}")


# --- Evaluation ---

devset = [
    dspy.Example(
        question="Where was the designer of the Eiffel Tower born?",
        answer="Dijon, France",
    ).with_inputs("question"),
    dspy.Example(
        question="What surpassed the Eiffel Tower as the tallest structure?",
        answer="The Chrysler Building",
    ).with_inputs("question"),
]


def multihop_metric(example, prediction, trace=None):
    answer_correct = example.answer.lower() in prediction.answer.lower()
    used_multiple_passages = len(prediction.context) > 3
    return answer_correct + 0.2 * used_multiple_passages


# --- Optimization ---

# optimizer = dspy.BootstrapFewShot(metric=multihop_metric, max_bootstrapped_demos=3)
# optimized = optimizer.compile(multihop, trainset=devset)
# optimized.save("multihop_rag_optimized.json")

Key points:

• Each hop generates a new search query using ChainOfThought, which reasons about what information is still missing
• The generate_query list creates separate ChainOfThought instances per hop -- each gets its own optimizable prompts
• dict.fromkeys() deduplicates passages while preserving order
• Multi-hop is essential for compositional questions like "Where was the designer of X born?" where no single passage has the full answer
• The optimizer tunes both query generation and answer generation together, improving end-to-end accuracy

Retrieval API Reference

Condensed from dspy.ai/api. Verify against upstream for latest.

dspy.Retrieve

dspy.Retrieve(k=3)

Returns: dspy.Prediction with .passages: list[str]

Requires dspy.configure(rm=...) or a subclass implementing forward().

dspy.ColBERTv2

dspy.ColBERTv2(url="http://0.0.0.0", port=None, post_requests=False)

dspy.Embedder

dspy.Embedder(model, batch_size=200, caching=True, **kwargs)

Returns: 1D numpy array (single text) or 2D array (multiple texts).

dspy.retrievers.Embeddings

dspy.retrievers.Embeddings(corpus, embedder, k=5, brute_force_threshold=20000, normalize=True)

Key methods:

• save(path) -- persist embeddings to disk
• Embeddings.from_saved(path, embedder=embedder) -- load without re-embedding