Retrieval is where RAG quality is won or lost. Tune top-k, add keyword search for exact terms, and rerank the candidates so the best passages reach the model.
Why: you embed the query with the same model and pull the k nearest chunks — k trades recall (more context) against noise and token cost. When: start around k=5 and adjust by spot-checking real queries. Where: return the content to answer from and the metadata to cite.
def retrieve(query, k=5):
q = model.encode([query], normalize_embeddings=True).astype("float32")
scores, idx = index.search(q, k)
return [
{"content": texts[i], "meta": meta[i], "score": float(s)}
for i, s in zip(idx[0], scores[0])
]
for hit in retrieve("how long do refunds take?"):
print(round(hit["score"], 3), hit["meta"], hit["content"][:60])Why: vectors match meaning but miss exact tokens — product codes, error numbers, rare names — so combining keyword (BM25) search with vector search catches both. When: add hybrid retrieval whenever queries contain identifiers or jargon that must match literally. Where: fuse the two ranked lists (e.g. reciprocal rank fusion) into one candidate set.
# pip install rank-bm25
from rank_bm25 import BM25Okapi
bm25 = BM25Okapi([t.split() for t in texts])
def hybrid(query, k=5):
kw = bm25.get_top_n(query.split(), texts, n=k) # exact-term matches
vec = [h["content"] for h in retrieve(query, k)] # meaning matches
# Reciprocal rank fusion: reward items ranked high in either list.
scores = {}
for lst in (kw, vec):
for rank, doc in enumerate(lst):
scores[doc] = scores.get(doc, 0) + 1 / (60 + rank)
return sorted(scores, key=scores.get, reverse=True)[:k]Why: a cross-encoder reads the query and each passage together and scores true relevance far more accurately than vector distance — so retrieve a wide net, then rerank down to the few best. When: add reranking when top-k contains relevant passages but in the wrong order. Where: it is slower per pair, so only rerank the ~20 candidates from retrieval, not the whole corpus.
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank(query, candidates, top=3):
pairs = [(query, c) for c in candidates]
scored = sorted(zip(candidates, reranker.predict(pairs)),
key=lambda x: x[1], reverse=True)
return [c for c, _ in scored[:top]]
wide = [h["content"] for h in retrieve("refund on a download", k=20)]
best = rerank("refund on a download", wide) # the 3 most relevant passages