dspy-embedding-retrieval

DSPy Embedding Retrieval

Goal

Build semantic retrieval over an application-owned text corpus with dspy.Embedder and dspy.Embeddings.

Basic Hosted Embedder

import dspy

corpus = [
    "DSPy programs are composed from modules.",
    "MIPROv2 optimizes instructions and demonstrations.",
    "RLM explores large contexts with a sandboxed REPL.",
]

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

result = search("Which optimizer tunes prompts?")
print(result.passages)
print(result.indices)

Use in RAG

class LocalRAG(dspy.Module):
    def __init__(self, retriever):
        super().__init__()
        self.retriever = retriever
        self.answer = dspy.ChainOfThought("context: list[str], question -> answer")

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

Custom Local Embeddings

Wrap any callable that accepts list[str] and returns a 2D numeric array:

from sentence_transformers import SentenceTransformer
import dspy

model = SentenceTransformer("sentence-transformers/static-retrieval-mrl-en-v1")
embedder = dspy.Embedder(model.encode)
search = dspy.Embeddings(corpus=corpus, embedder=embedder, k=5)

Scores, FAISS, and Persistence

Use dspy.EmbeddingsWithScores when downstream logic needs similarity thresholds or reranking.

For corpora at or above the brute_force_threshold default of 20_000, DSPy builds a FAISS index. Install FAISS first:

pip install faiss-cpu

Persist the index when embedding the corpus is expensive:

search.save("./retrieval-index")
loaded = dspy.Embeddings.from_saved("./retrieval-index", embedder=embedder)

Related Skills

• Build a complete pipeline: dspy-rag-pipeline
• Design typed context fields: dspy-signature-designer
• Harden caches: dspy-production-deployment