Advanced RAG Patterns with Python and ChromaDB
Most Retrieval-Augmented Generation tutorials stop at the basics: split a document, embed the chunks, store them in a vector database, and retrieve the top-k results. While that works for demos, production RAG systems demand better patterns to deliver accurate, relevant responses. Here I will cover advanced RAG techniques using Python and ChromaDB: chunking strategies, hybrid search, reranking, and metadata filtering.
Beyond naive chunking
The most common mistake in RAG pipelines is using a fixed-size text splitter without considering document structure. A 500-character chunk that splits a code block in half or separates a heading from its content will produce poor retrieval results.
Let us build a semantic chunker that respects document structure:
from dataclasses import dataclass, field
import re
from sentence_transformers import SentenceTransformer
import numpy as np
@dataclass
class Chunk:
text: str
metadata: dict = field(default_factory=dict)
embedding: list[float] | None = None
class SemanticChunker:
def __init__(
self,
model_name: str = "all-MiniLM-L6-v2",
similarity_threshold: float = 0.75,
max_chunk_size: int = 1000,
):
self.model = SentenceTransformer(model_name)
self.similarity_threshold = similarity_threshold
self.max_chunk_size = max_chunk_size
def chunk_document(
self, text: str, doc_metadata: dict | None = None
) -> list[Chunk]:
sentences = self._split_sentences(text)
embeddings = self.model.encode(sentences)
chunks = []
current_chunk_sentences = [sentences[0]]
for i in range(1, len(sentences)):
similarity = self._cosine_similarity(
embeddings[i - 1], embeddings[i]
)
combined_length = sum(
len(s) for s in current_chunk_sentences
) + len(sentences[i])
if (
similarity >= self.similarity_threshold
and combined_length <= self.max_chunk_size
):
current_chunk_sentences.append(sentences[i])
else:
chunk_text = " ".join(current_chunk_sentences)
chunks.append(
Chunk(text=chunk_text, metadata=doc_metadata or {})
)
current_chunk_sentences = [sentences[i]]
if current_chunk_sentences:
chunk_text = " ".join(current_chunk_sentences)
chunks.append(
Chunk(text=chunk_text, metadata=doc_metadata or {})
)
return chunks
def _split_sentences(self, text: str) -> list[str]:
return [
s.strip()
for s in re.split(r"(?<=[.!?])\s+", text)
if s.strip()
]
def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
This chunker groups semantically similar consecutive sentences together rather than splitting at arbitrary character boundaries. The result is chunks that contain coherent ideas, which dramatically improves retrieval quality.
Setting up ChromaDB with rich metadata
ChromaDB is an open-source embedding database that makes it easy to build RAG applications. Its metadata filtering is especially useful for advanced retrieval patterns.
import chromadb
from chromadb.config import Settings
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection(
name="knowledge_base",
metadata={"hnsw:space": "cosine"},
)
def ingest_documents(
chunker: SemanticChunker,
documents: list[dict],
) -> None:
all_texts = []
all_metadatas = []
all_ids = []
for doc in documents:
chunks = chunker.chunk_document(
text=doc["content"],
doc_metadata={
"source": doc["source"],
"category": doc["category"],
"date": doc["date"],
"doc_type": doc.get("doc_type", "general"),
},
)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc['source']}_{i}"
all_texts.append(chunk.text)
all_metadatas.append(chunk.metadata)
all_ids.append(chunk_id)
collection.add(
documents=all_texts,
metadatas=all_metadatas,
ids=all_ids,
)
print(f"Ingested {len(all_ids)} chunks from {len(documents)} documents")
Notice that we store rich metadata with each chunk: source, category, date, and document type. This metadata becomes critical when we implement filtered retrieval later.
Hybrid search: combining dense and sparse retrieval
Pure vector similarity search can miss exact keyword matches that are important for technical queries. A hybrid approach combines dense embeddings with sparse keyword matching for better coverage.
from rank_bm25 import BM25Okapi
import numpy as np
class HybridRetriever:
def __init__(
self,
collection: chromadb.Collection,
alpha: float = 0.6,
):
self.collection = collection
self.alpha = alpha # weight for dense search
self._build_bm25_index()
def _build_bm25_index(self) -> None:
results = self.collection.get(include=["documents", "metadatas"])
self.doc_ids = results["ids"]
self.documents = results["documents"]
self.metadatas = results["metadatas"]
tokenized = [doc.lower().split() for doc in self.documents]
self.bm25 = BM25Okapi(tokenized)
def search(
self,
query: str,
n_results: int = 10,
where: dict | None = None,
) -> list[dict]:
# Dense search via ChromaDB
dense_results = self.collection.query(
query_texts=[query],
n_results=n_results * 2,
where=where,
include=["documents", "metadatas", "distances"],
)
# Sparse search via BM25
tokenized_query = query.lower().split()
bm25_scores = self.bm25.get_scores(tokenized_query)
# Normalize scores
dense_scores = {}
for doc_id, distance in zip(
dense_results["ids"][0], dense_results["distances"][0]
):
dense_scores[doc_id] = 1 - distance # convert distance to score
sparse_scores = {}
max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
for doc_id, score in zip(self.doc_ids, bm25_scores):
sparse_scores[doc_id] = score / max_bm25
# Combine with weighted fusion
all_doc_ids = set(dense_scores.keys()) | set(
k for k, v in sparse_scores.items() if v > 0
)
combined = []
for doc_id in all_doc_ids:
d_score = dense_scores.get(doc_id, 0)
s_score = sparse_scores.get(doc_id, 0)
final_score = self.alpha * d_score + (1 - self.alpha) * s_score
idx = self.doc_ids.index(doc_id)
combined.append({
"id": doc_id,
"text": self.documents[idx],
"metadata": self.metadatas[idx],
"score": final_score,
})
combined.sort(key=lambda x: x["score"], reverse=True)
return combined[:n_results]
The alpha parameter controls the balance between dense and sparse retrieval. A value of 0.6 gives slightly more weight to semantic similarity while still benefiting from exact keyword matches.
Reranking for precision
Initial retrieval casts a wide net. Reranking narrows it down using a more powerful cross-encoder model that scores each query-document pair directly.
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(
self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
results: list[dict],
top_k: int = 5,
) -> list[dict]:
if not results:
return []
pairs = [(query, r["text"]) for r in results]
scores = self.model.predict(pairs)
for result, score in zip(results, scores):
result["rerank_score"] = float(score)
reranked = sorted(
results, key=lambda x: x["rerank_score"], reverse=True
)
return reranked[:top_k]
Cross-encoders are more computationally expensive than bi-encoders, which is why we use them only on the already-filtered set from the initial retrieval. This two-stage approach gives us both speed and accuracy.
Metadata filtering for scoped retrieval
One of the most practical yet underused RAG patterns is metadata filtering. Instead of searching the entire knowledge base, you filter by category, date range, or document type first.
def scoped_query(
retriever: HybridRetriever,
reranker: Reranker,
query: str,
category: str | None = None,
doc_type: str | None = None,
date_after: str | None = None,
) -> list[dict]:
where_filter = {}
if category and doc_type:
where_filter = {
"$and": [
{"category": {"$eq": category}},
{"doc_type": {"$eq": doc_type}},
]
}
elif category:
where_filter = {"category": {"$eq": category}}
elif doc_type:
where_filter = {"doc_type": {"$eq": doc_type}}
results = retriever.search(
query=query,
n_results=20,
where=where_filter if where_filter else None,
)
reranked = reranker.rerank(query=query, results=results, top_k=5)
return reranked
This is especially important for enterprise applications where you need to restrict retrieval to specific document categories, enforce access controls, or prioritize recent content.
The complete RAG pipeline
Let us tie everything together into a complete pipeline:
from openai import OpenAI
def rag_query(
query: str,
retriever: HybridRetriever,
reranker: Reranker,
category: str | None = None,
) -> str:
results = scoped_query(
retriever=retriever,
reranker=reranker,
query=query,
category=category,
)
context = "\n\n---\n\n".join(
f"[Source: {r['metadata'].get('source', 'unknown')}]\n{r['text']}"
for r in results
)
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": (
"Answer the question based on the provided context. "
"Cite sources when possible. If the context doesn't "
"contain enough information, say so."
),
},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {query}",
},
],
temperature=0.3,
)
return response.choices[0].message.content
Key takeaways
Building a production-grade RAG system requires going beyond the basics. Semantic chunking ensures your chunks contain coherent ideas. Hybrid search catches both semantic and keyword matches. Reranking boosts precision on the final result set. Metadata filtering keeps retrieval scoped and relevant.
ChromaDB handles these patterns well with its metadata filtering, persistent storage, and simple API. Combined with Python's rich ecosystem of embedding models and cross-encoders, you have everything you need to build RAG systems that actually work in production.
The next time you find your RAG pipeline returning irrelevant results, look at your chunking strategy first, add metadata filtering second, and consider reranking third. These three improvements alone will dramatically improve your retrieval quality.