Dspy Qdrant

Qdrant — Vector Database Integration for DSPy

Guide the user through setting up Qdrant with DSPy using the official dspy-qdrant package, plus custom retriever patterns for Pinecone, ChromaDB, and Weaviate.

What is Qdrant

Qdrant is an open-source vector search engine written in Rust. It's the only vector database with an official DSPy integration package (dspy-qdrant). Features: hybrid search (dense + sparse), payload filtering, multi-tenancy, and horizontal scaling.

Why Qdrant for DSPy

DSPy 3.0 removed all community-contributed retriever modules (ChromadbRM, PineconeRM, WeaviateRM, QdrantRM from the main repo). The dspy-qdrant package is the official replacement — maintained separately with full DSPy compatibility.

For other vector databases, you write a short custom dspy.Retrieve subclass (~15 lines). This skill covers that pattern too.

Setup

Install

pip install dspy-qdrant

This installs both the Qdrant client and the DSPy retriever module.

Start Qdrant

Option 1: Docker (local development)

docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant

Option 2: Qdrant Cloud (managed, free tier available)

1. Sign up at cloud.qdrant.io 2. Create a cluster (free tier: 1GB, 1 node) 3. Copy your URL and API key

export QDRANT_URL="https://your-cluster.aws.cloud.qdrant.io"
export QDRANT_API_KEY="your-api-key"

Option 3: pip install (in-memory, for testing)

from qdrant_client import QdrantClient
client = QdrantClient(":memory:")  # no server needed

Using QdrantRM in DSPy

Basic setup

import dspy
from dspy_qdrant import QdrantRM

retriever = QdrantRM(
    qdrant_collection_name="my_docs",
    qdrant_client_url="http://localhost:6333",  # or your cloud URL
    qdrant_client_api_key=None,                 # set for cloud
    k=5,
)

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"), rm=retriever)

# Now dspy.Retrieve() uses Qdrant
search = dspy.Retrieve(k=5)
result = search("How do refunds work?")
print(result.passages)

QdrantRM with custom embeddings

from dspy_qdrant import QdrantRM

retriever = QdrantRM(
    qdrant_collection_name="my_docs",
    qdrant_client_url="http://localhost:6333",
    k=5,
    embedding_model="openai/text-embedding-3-small",  # LiteLLM format
    embedding_dimensions=512,
)

Using Qdrant Cloud

import os
from dspy_qdrant import QdrantRM

retriever = QdrantRM(
    qdrant_collection_name="my_docs",
    qdrant_client_url=os.environ["QDRANT_URL"],
    qdrant_client_api_key=os.environ["QDRANT_API_KEY"],
    k=5,
)

Indexing documents into Qdrant

Before you can search, you need to populate your Qdrant collection:

from qdrant_client import QdrantClient, models
import dspy

client = QdrantClient("http://localhost:6333")
embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)

# Your documents
docs = [
    {"id": 1, "text": "Refunds are processed within 5-7 business days.", "category": "billing"},
    {"id": 2, "text": "Reset your password at Settings > Security.", "category": "account"},
    {"id": 3, "text": "Enterprise plans include SSO and dedicated support.", "category": "plans"},
]

# Create collection
client.create_collection(
    collection_name="my_docs",
    vectors_config=models.VectorParams(size=512, distance=models.Distance.COSINE),
)

# Upsert with embeddings
vectors = embedder([d["text"] for d in docs])
client.upsert(
    collection_name="my_docs",
    points=[
        models.PointStruct(
            id=d["id"],
            vector=v,
            payload={"text": d["text"], "category": d["category"]},
        )
        for d, v in zip(docs, vectors)
    ],
)

RAG pipeline with Qdrant

import dspy
from dspy_qdrant import QdrantRM

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

retriever = QdrantRM(
    qdrant_collection_name="my_docs",
    qdrant_client_url="http://localhost:6333",
    k=5,
)

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

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

rag = RAG()
result = rag(question="How do refunds work?")
print(result.answer)

Hybrid search (dense + sparse)

Qdrant supports hybrid search combining dense (semantic) and sparse (keyword) vectors in the same collection. This improves recall for queries that need both semantic understanding and exact keyword matching.

from qdrant_client import QdrantClient, models

client = QdrantClient("http://localhost:6333")

# Create collection with both dense and sparse vectors
client.create_collection(
    collection_name="hybrid_docs",
    vectors_config=models.VectorParams(size=512, distance=models.Distance.COSINE),
    sparse_vectors_config={
        "keywords": models.SparseVectorParams(
            modifier=models.Modifier.IDF,
        ),
    },
)

Then query with both:

results = client.query_points(
    collection_name="hybrid_docs",
    prefetch=[
        models.Prefetch(query=dense_vector, using="", limit=20),
        models.Prefetch(query=sparse_vector, using="keywords", limit=20),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # reciprocal rank fusion
    limit=5,
)

Other vector DBs with DSPy

Since DSPy 3.0 removed built-in community retrievers, use a custom dspy.Retrieve subclass for any vector database. The pattern is always the same:

Custom retriever pattern

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

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

Pinecone custom retriever

from pinecone import Pinecone
import dspy

class PineconeRetriever(dspy.Retrieve):
    def __init__(self, index_name, embedder, k=3):
        super().__init__(k=k)
        pc = Pinecone()  # reads PINECONE_API_KEY from env
        self.index = pc.Index(index_name)
        self.embedder = embedder

    def forward(self, query, k=None):
        k = k or self.k
        vector = self.embedder(query)
        results = self.index.query(vector=vector, top_k=k, include_metadata=True)
        passages = [m["metadata"]["text"] for m in results["matches"]]
        return dspy.Prediction(passages=passages)

# Usage
embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
retriever = PineconeRetriever("my-index", embedder, k=5)

ChromaDB custom retriever

import chromadb
import dspy

class ChromaRetriever(dspy.Retrieve):
    def __init__(self, collection_name, k=3):
        super().__init__(k=k)
        client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = client.get_or_create_collection(collection_name)

    def forward(self, query, k=None):
        k = k or self.k
        results = self.collection.query(query_texts=[query], n_results=k)
        return dspy.Prediction(passages=results["documents"][0])

# Usage
retriever = ChromaRetriever("my_docs", k=5)

Weaviate custom retriever

import weaviate
import dspy

class WeaviateRetriever(dspy.Retrieve):
    def __init__(self, class_name, url="http://localhost:8080", k=3):
        super().__init__(k=k)
        self.client = weaviate.connect_to_local(host=url.replace("http://", "").split(":")[0])
        self.collection = self.client.collections.get(class_name)

    def forward(self, query, k=None):
        k = k or self.k
        results = self.collection.query.near_text(query=query, limit=k)
        passages = [obj.properties["text"] for obj in results.objects]
        return dspy.Prediction(passages=passages)

# Usage
retriever = WeaviateRetriever("MyDocs", k=5)

Vector DB comparison

Choosing a vector DB

Starting a new DSPy project?
  → Qdrant (official DSPy package, easiest setup)

Prototyping locally, smallest footprint?
  → ChromaDB (pip install, in-memory or persistent, no server)

Already using Pinecone/Weaviate in production?
  → Write a custom retriever (15 lines, shown above)

Need hybrid search (keyword + semantic)?
  → Qdrant or Weaviate

Gotchas

1. DSPy 3.0 removed community retrievers — from dspy.retrieve.chromadb_rm import ChromadbRM no longer works. Use dspy-qdrant or write a custom subclass. 2. QdrantRM expects a `text` payload field — when indexing, store the document text in a payload field named text (or configure the field name in QdrantRM). 3. Embeddings must match — the embedding model and dimensions used for indexing must match what QdrantRM uses for querying. 4. ChromaDB is great for prototyping but not production — it's single-process, no replication. Migrate to Qdrant or Pinecone for production.

Cross-references

• DSPy retrieval basics (Retrieve, ColBERTv2, Embedder, Embeddings) — /dspy-retrieval
• Building RAG pipelines end-to-end — /ai-searching-docs
• Evaluating RAG quality with decomposed metrics — /dspy-ragas
• Stopping hallucinations in RAG — /ai-stopping-hallucinations
• For worked examples, see examples.md

Qdrant Examples

Index and search with Qdrant + DSPy

import dspy
from qdrant_client import QdrantClient, models
from dspy_qdrant import QdrantRM

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

# --- Step 1: Index documents ---
client = QdrantClient("http://localhost:6333")
embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)

docs = [
    "Our refund policy allows returns within 30 days of purchase.",
    "To reset your password, visit Settings > Security > Change Password.",
    "Enterprise plans include SSO, audit logs, and a dedicated account manager.",
    "Free trials last 14 days. No credit card required.",
    "We support integration with Slack, Teams, and Discord.",
    "Data is encrypted at rest (AES-256) and in transit (TLS 1.3).",
    "API rate limits: 1000 requests/minute for Pro, 100 for Free.",
    "Billing is monthly. Switch to annual for a 20% discount.",
]

# Create collection
client.recreate_collection(
    collection_name="support_docs",
    vectors_config=models.VectorParams(size=512, distance=models.Distance.COSINE),
)

# Embed and upsert
vectors = embedder(docs)
client.upsert(
    collection_name="support_docs",
    points=[
        models.PointStruct(id=i, vector=v, payload={"text": doc})
        for i, (doc, v) in enumerate(zip(docs, vectors))
    ],
)

# --- Step 2: Set up QdrantRM ---
retriever = QdrantRM(
    qdrant_collection_name="support_docs",
    qdrant_client_url="http://localhost:6333",
    k=3,
)

# --- Step 3: Build RAG pipeline ---
class SupportRAG(dspy.Module):
    def __init__(self):
        self.retrieve = retriever
        self.answer = dspy.ChainOfThought("context, question -> answer")

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

rag = SupportRAG()
result = rag(question="What's the refund policy?")
print(f"Answer: {result.answer}")

Custom Pinecone retriever for DSPy

from pinecone import Pinecone
import dspy

class PineconeRetriever(dspy.Retrieve):
    def __init__(self, index_name, embedder, namespace=None, k=3):
        super().__init__(k=k)
        pc = Pinecone()  # reads PINECONE_API_KEY from env
        self.index = pc.Index(index_name)
        self.embedder = embedder
        self.namespace = namespace

    def forward(self, query, k=None):
        k = k or self.k
        vector = self.embedder(query)
        results = self.index.query(
            vector=vector,
            top_k=k,
            include_metadata=True,
            namespace=self.namespace,
        )
        passages = [m["metadata"]["text"] for m in results["matches"]]
        return dspy.Prediction(passages=passages)

# Usage in a DSPy pipeline
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
embedder = dspy.Embedder("openai/text-embedding-3-small", dimensions=512)
retriever = PineconeRetriever("support-index", embedder, k=5)

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

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

rag = RAG()
print(rag(question="What integrations do you support?").answer)

Custom ChromaDB retriever for prototyping

import chromadb
import dspy

class ChromaRetriever(dspy.Retrieve):
    def __init__(self, collection_name, persist_dir="./chroma_db", k=3):
        super().__init__(k=k)
        client = chromadb.PersistentClient(path=persist_dir)
        self.collection = client.get_or_create_collection(collection_name)

    def forward(self, query, k=None):
        k = k or self.k
        results = self.collection.query(query_texts=[query], n_results=k)
        return dspy.Prediction(passages=results["documents"][0])

# Quick prototype — index and search in one script
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("quick_test")
collection.add(
    documents=[
        "Python is a programming language.",
        "DSPy is a framework for programming LMs.",
        "Qdrant is a vector search engine.",
    ],
    ids=["1", "2", "3"],
)

retriever = ChromaRetriever("quick_test", k=2)
result = retriever("What is DSPy?")
print(result.passages)
# ['DSPy is a framework for programming LMs.', 'Python is a programming language.']