dspy-rag

Building Retrieval-Augmented Generation pipelines with DSPy and vector databases

DSPy RAG

Scope: RAG patterns, retrieval modules, vector databases, context optimization Lines: ~420 Last Updated: 2025-10-25

When to Use This Skill

Activate this skill when:

• Building RAG (Retrieval-Augmented Generation) systems with DSPy
• Integrating vector databases (ChromaDB, Weaviate, Qdrant) with DSPy
• Optimizing retrieval quality and context selection
• Implementing multi-hop reasoning over documents
• Creating question-answering systems over custom knowledge bases
• Reducing hallucinations with grounded generation

Core Concepts

What is RAG?

Retrieval-Augmented Generation: Combine retrieval + generation

1. Retrieve: Find relevant documents from knowledge base
2. Augment: Add retrieved context to prompt
3. Generate: LM produces answer based on context

Benefits:

• Reduce hallucinations (grounded in facts)
• Access up-to-date information
• Domain-specific knowledge
• Transparent (can cite sources)

DSPy RAG Components

dspy.Retrieve: Built-in retrieval module

• Queries vector database or search API
• Returns top-k relevant passages
• Integrates with ChromaDB, Weaviate, Qdrant, etc.

Custom RAG modules: Combine retrieval + generation

• Control retrieval strategy
• Format context
• Cite sources
• Handle multi-hop reasoning

Vector Database Integration

Supported:

• ChromaDB (local, simple)
• Weaviate (production, scalable)
• Qdrant (high performance)
• Pinecone (managed cloud)
• Custom retrievers (API-based)

---

Patterns

Pattern 1: Simple RAG with ChromaDB

import dspy
import chromadb

# Set up ChromaDB
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("knowledge_base")

# Add documents
documents = [
    "DSPy is a framework for programming language models.",
    "Paris is the capital of France.",
    "Python is a popular programming language.",
]

collection.add(
    documents=documents,
    ids=[str(i) for i in range(len(documents))],
)

# Configure DSPy with retriever
retriever = dspy.Retrieve(k=3)
rm = dspy.chromadb_rm.ChromadbRM(
    collection_name="knowledge_base",
    persist_directory="./chroma_db"
)
dspy.settings.configure(rm=rm)

# Simple RAG module
class SimpleRAG(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)
        self.generate = dspy.ChainOfThought("context, question -> answer")

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

        # Generate answer
        return self.generate(context=context, question=question)

# Use RAG
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

rag = SimpleRAG()
result = rag(question="What is DSPy?")
print(result.answer)

When to use:

• Small knowledge bases (<10K documents)
• Local development
• Quick prototyping

Pattern 2: RAG with Citations

import dspy

class CitedRAG(dspy.Module):
    """RAG system that cites sources."""

    def __init__(self, k=5):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=k)

        # Signature includes citations
        self.generate = dspy.ChainOfThought(
            "context, question -> answer, citations: list[int]"
        )

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

        # Format context with passage numbers
        context_parts = []
        for i, passage in enumerate(passages):
            context_parts.append(f"[{i+1}] {passage}")

        context = "\n\n".join(context_parts)

        # Generate with citations
        result = self.generate(context=context, question=question)

        # Attach retrieved passages for reference
        result.retrieved_passages = passages

        return result

# Use RAG with citations
rag = CitedRAG(k=5)
result = rag(question="What is DSPy?")

print("Answer:", result.answer)
print("Citations:", result.citations)
print("\nSources:")
for citation_idx in result.citations:
    print(f"[{citation_idx}] {result.retrieved_passages[citation_idx-1][:100]}...")

Benefits:

• Verifiable answers
• Transparent reasoning
• User can check sources

Pattern 3: Multi-Hop RAG

import dspy

class MultiHopRAG(dspy.Module):
    """RAG with multiple retrieval rounds for complex questions."""

    def __init__(self, max_hops=2):
        super().__init__()
        self.max_hops = max_hops
        self.retrieve = dspy.Retrieve(k=3)

        # Generate follow-up queries
        self.gen_query = dspy.ChainOfThought(
            "question, context -> needs_more_info: bool, follow_up_query"
        )

        # Final answer generation
        self.generate_answer = dspy.ChainOfThought(
            "question, all_context -> answer"
        )

    def forward(self, question):
        all_passages = []
        current_question = question

        # Multi-hop retrieval
        for hop in range(self.max_hops):
            # Retrieve for current query
            passages = self.retrieve(current_question).passages
            all_passages.extend(passages)

            # Check if we need more information
            context = "\n".join(all_passages)
            query_result = self.gen_query(question=question, context=context)

            if not query_result.needs_more_info:
                break

            # Generate follow-up query
            current_question = query_result.follow_up_query

        # Generate final answer with all context
        all_context = "\n\n".join(all_passages)
        result = self.generate_answer(question=question, all_context=all_context)

        result.num_hops = hop + 1
        result.all_passages = all_passages

        return result

# Use multi-hop RAG
rag = MultiHopRAG(max_hops=3)
result = rag(question="What programming languages are used in DSPy development?")

print(f"Answer: {result.answer}")
print(f"Required {result.num_hops} retrieval hops")

When to use:

• Complex questions requiring multiple pieces of information
• Questions with implicit dependencies
• Deep knowledge base exploration

Pattern 4: RAG with Weaviate

import dspy
import weaviate

# Connect to Weaviate
weaviate_client = weaviate.Client("http://localhost:8080")

# Create schema
schema = {
    "class": "Document",
    "properties": [
        {"name": "content", "dataType": ["text"]},
        {"name": "title", "dataType": ["string"]},
    ],
}

weaviate_client.schema.create_class(schema)

# Add documents
documents = [
    {"content": "DSPy is a framework...", "title": "DSPy Overview"},
    # ... more documents
]

for doc in documents:
    weaviate_client.data_object.create(doc, "Document")

# Configure DSPy with Weaviate
rm = dspy.WeaviateRM("Document", weaviate_client=weaviate_client)
dspy.settings.configure(rm=rm)

# Use retriever
class WeaviateRAG(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=5)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        passages = self.retrieve(question).passages
        context = "\n\n".join(passages)
        return self.generate(context=context, question=question)

# Use RAG
rag = WeaviateRAG()
result = rag(question="What is DSPy?")

When to use:

• Production RAG systems
• Large knowledge bases (>100K documents)
• Need for scalability and performance

Pattern 5: RAG with Reranking

import dspy

class RerankedRAG(dspy.Module):
    """RAG with passage reranking for better quality."""

    def __init__(self, retrieve_k=10, use_k=3):
        super().__init__()
        self.retrieve_k = retrieve_k
        self.use_k = use_k

        self.retrieve = dspy.Retrieve(k=retrieve_k)

        # Reranker: score each passage for relevance
        self.rerank = dspy.Predict(
            "question, passage -> relevance_score: float"
        )

        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Initial retrieval (over-fetch)
        passages = self.retrieve(question).passages

        # Rerank passages
        scored_passages = []
        for passage in passages:
            score_result = self.rerank(question=question, passage=passage)
            try:
                score = float(score_result.relevance_score)
            except:
                score = 0.5

            scored_passages.append((score, passage))

        # Sort by score and take top-k
        scored_passages.sort(reverse=True, key=lambda x: x[0])
        top_passages = [p for _, p in scored_passages[:self.use_k]]

        # Generate answer with reranked context
        context = "\n\n".join(top_passages)
        return self.generate(context=context, question=question)

# Use reranked RAG
rag = RerankedRAG(retrieve_k=10, use_k=3)
result = rag(question="What is DSPy?")

Benefits:

• Better context quality
• Reduce noise from irrelevant passages
• Improved answer accuracy

Trade-off: More LM calls (one per passage for reranking)

Pattern 6: RAG Optimization with BootstrapFewShot

import dspy

# Define RAG module
class SimpleRAG(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)
        self.generate = dspy.ChainOfThought("context, question -> answer")

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

# Prepare training data
trainset = [
    dspy.Example(
        question="What is DSPy?",
        answer="A framework for programming language models"
    ).with_inputs("question"),
    # ... more examples
]

# Define metric
def rag_metric(example, pred, trace=None):
    """Check if answer is correct."""
    return example.answer.lower() in pred.answer.lower()

# Optimize RAG pipeline
optimizer = dspy.BootstrapFewShot(metric=rag_metric)

rag = SimpleRAG()
optimized_rag = optimizer.compile(student=rag, trainset=trainset)

# Optimized RAG now has better prompts for generation
result = optimized_rag(question="What is DSPy?")

What gets optimized:

• Generation module prompts
• Few-shot examples for generation
• Context formatting (implicitly)

Pattern 7: RAG with Modal-Hosted Models

import dspy
import modal

# Deploy retrieval-optimized model on Modal
app = modal.App("dspy-rag")

@app.function(
    image=modal.Image.debian_slim().pip_install("vllm", "sentence-transformers"),
    gpu="L40S",
)
@modal.web_endpoint(method="POST")
def embed_documents(request: dict):
    """Embedding endpoint for RAG retrieval."""
    from sentence_transformers import SentenceTransformer

    model = SentenceTransformer("all-MiniLM-L6-v2")
    texts = request.get("texts", [])
    embeddings = model.encode(texts)

    return {"embeddings": embeddings.tolist()}

# Configure DSPy with Modal-hosted LM for generation
lm = dspy.LM(
    "openai/meta-llama/Meta-Llama-3.1-8B-Instruct",
    api_base="https://your-app--serve.modal.run/v1",
    api_key="EMPTY",
)

dspy.configure(lm=lm)

# Use with RAG
class ModalRAG(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=5)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        passages = self.retrieve(question).passages
        context = "\n\n".join(passages)
        return self.generate(context=context, question=question)

# Use Modal-powered RAG
rag = ModalRAG()
result = rag(question="What is DSPy?")

Benefits:

• Cost-effective GPU inference
• Serverless scaling
• Custom model deployment

Pattern 8: Production RAG Pipeline

import dspy
from typing import List, Optional

class ProductionRAG(dspy.Module):
    """Production-ready RAG with error handling and logging."""

    def __init__(self, k=5, min_relevance=0.3):
        super().__init__()
        self.k = k
        self.min_relevance = min_relevance

        self.retrieve = dspy.Retrieve(k=k)
        self.generate = dspy.ChainOfThought(
            "context, question -> answer, confidence: float"
        )

    def forward(self, question: str) -> dspy.Prediction:
        try:
            # Retrieve passages
            retrieval_result = self.retrieve(question)
            passages = retrieval_result.passages

            if not passages:
                # Fallback: no retrieval results
                return dspy.Prediction(
                    answer="I don't have enough information to answer this question.",
                    confidence=0.0,
                    error="No passages retrieved"
                )

            # Format context
            context = "\n\n".join(passages)

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

            # Validate confidence
            try:
                confidence = float(result.confidence)
            except:
                confidence = 0.5

            # Low confidence warning
            if confidence < self.min_relevance:
                result.warning = "Low confidence answer"

            result.num_passages = len(passages)
            return result

        except Exception as e:
            # Error handling
            return dspy.Prediction(
                answer="An error occurred while processing your question.",
                confidence=0.0,
                error=str(e)
            )

# Use production RAG
rag = ProductionRAG(k=5, min_relevance=0.6)
result = rag(question="What is DSPy?")

print(f"Answer: {result.answer}")
print(f"Confidence: {result.confidence}")
if hasattr(result, 'warning'):
    print(f"Warning: {result.warning}")

Production features:

• Error handling
• Confidence scoring
• Validation
• Fallback responses
• Logging/monitoring hooks

---

Quick Reference

Basic RAG Pattern

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

    def forward(self, question):
        passages = self.retrieve(question).passages
        context = "\n\n".join(passages)
        return self.generate(context=context, question=question)

Vector DB Setup

# ChromaDB (local)
import chromadb
client = chromadb.Client()
collection = client.create_collection("docs")
rm = dspy.chromadb_rm.ChromadbRM("docs")

# Weaviate (production)
import weaviate
client = weaviate.Client("http://localhost:8080")
rm = dspy.WeaviateRM("Document", weaviate_client=client)

# Configure DSPy
dspy.settings.configure(rm=rm)

RAG Best Practices

✅ DO: Use reranking for better quality
✅ DO: Include citations for transparency
✅ DO: Optimize retrieval (k value, chunking strategy)
✅ DO: Handle empty retrieval results
✅ DO: Monitor retrieval quality

❌ DON'T: Retrieve too many passages (context overflow)
❌ DON'T: Skip passage deduplication
❌ DON'T: Ignore retrieval failures
❌ DON'T: Use RAG when LM already has knowledge

---

Anti-Patterns

❌ Too many retrieved passages: Context overflow, slow generation

# Bad
self.retrieve = dspy.Retrieve(k=50)  # Too many!

✅ Use reasonable k and rerank:

# Good
self.retrieve = dspy.Retrieve(k=10)
# Rerank to top 3

❌ No error handling: Crashes on retrieval failure

# Bad
passages = self.retrieve(question).passages  # May fail!
context = "\n".join(passages)

✅ Handle errors:

# Good
try:
    passages = self.retrieve(question).passages
    if not passages:
        return fallback_response()
except Exception as e:
    log_error(e)
    return error_response()

❌ Ignoring retrieval quality: Using irrelevant passages

# Bad - use all retrieved passages blindly
passages = self.retrieve(question).passages
context = "\n".join(passages)  # May include irrelevant content

✅ Filter or rerank:

# Good - rerank or filter
passages = self.retrieve(question).passages
relevant = [p for p in passages if is_relevant(question, p)]
context = "\n".join(relevant)

---

Related Skills

• dspy-modules.md - Building RAG modules
• dspy-optimizers.md - Optimizing RAG pipelines
• dspy-evaluation.md - Evaluating RAG quality
• database/mongodb-basics.md - Document storage
• database/postgres-query-optimization.md - Metadata storage

---

Last Updated: 2025-10-25 Format Version: 1.0 (Atomic)