Retrievers in depth

Similarity search, MMR, multi-query and contextual compression, hybrid search, reranking, and writing your own retriever.

Retrieval modes

from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnableLambda

store = FAISS.from_documents(chunks, local_emb)

# 1. plain similarity: top-k nearest
basic = store.as_retriever(search_kwargs={"k": 6})

# 2. MMR: relevant but mutually diverse, good for summaries over a broad question
mmr = store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 6, "fetch_k": 30, "lambda_mult": 0.5},
)

# 3. score threshold: abstain rather than return irrelevant passages
thresholded = store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.25, "k": 6},
)

for name, retriever in [("basic", basic), ("mmr", mmr), ("threshold", thresholded)]:
    docs = retriever.invoke("how do refunds work for annual plans?")
    print(name, len(docs), [d.metadata.get("h2") for d in docs])

• MMR's fetch_k must be larger than k. The point is to fetch a wide candidate set and then choose a diverse subset from it.
• A score threshold turns a bad retrieval into an empty one. That is usually better than feeding an irrelevant passage to a model that will answer from it anyway.
• Parent-document retrieval is the practical fix for the chunk-size trade-off: embed small chunks for matching, return the parent section for context.
• Retrieval parameters are the highest-leverage knobs in a RAG system. Tune k, the search type and the threshold before touching the prompt.

Multi-query, compression and hybrid

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.retrievers import ContextualCompressionRetriever, EnsembleRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.retrievers import BM25Retriever

# multi-query: the model rewrites the question several ways, results are unioned
multi = MultiQueryRetriever.from_llm(retriever=basic, llm=model)

# contextual compression: keep only the sentences that bear on the question
compressor = LLMChainExtractor.from_llm(model)
compressed = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=basic)

# hybrid: lexical BM25 alongside the dense retriever, fused by rank
bm25 = BM25Retriever.from_documents(chunks)
bm25.k = 6
hybrid = EnsembleRetriever(retrievers=[bm25, basic], weights=[0.4, 0.6])

import logging
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

for name, r in [("multi", multi), ("compressed", compressed), ("hybrid", hybrid)]:
    docs = r.invoke("what happens if I exceed the API rate limit?")
    print(name, len(docs), docs[0].page_content[:70])

• Multi-query costs one extra model call per query and helps most when the user's words do not match the document's words.
• Contextual compression costs one model call per retrieved document. It shrinks the prompt but can be slower than simply passing more context, so measure both.
• Hybrid retrieval needs score normalisation before fusion; rank-based fusion (reciprocal rank fusion) is more robust than weighted score averaging.
• Every extra retriever stage adds latency. For an interactive product, budget the total: retrieval, reranking and generation should fit the response-time target together.

A custom retriever and its evaluation

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List

class SectionAwareRetriever(BaseRetriever):
    """Prefer chunks whose heading matches a section hinted at in the question."""

    store: object
    k: int = 6
    boost: float = 1.4

    def _get_relevant_documents(self, query: str, *, run_manager=None) -> List[Document]:
        candidates = self.store.similarity_search_with_score(query, k=self.k * 4)
        scored = []
        for doc, score in candidates:
            heading = (doc.metadata.get("h2") or "").lower()
            bonus = self.boost if heading and heading in query.lower() else 1.0
            scored.append((score * bonus, doc))
        scored.sort(key=lambda pair: pair[0], reverse=True)
        return [doc for _, doc in scored[: self.k]]

retriever = SectionAwareRetriever(store=store)
print(len(retriever.invoke("billing refund policy")))

# evaluate retrieval separately from generation
def recall_at_k(retriever, cases, k=6):
    hits = 0
    for case in cases:
        docs = retriever.invoke(case["question"])[:k]
        ids = {d.metadata.get("chunk_id") for d in docs}
        hits += case["expected_chunk_id"] in ids
    return hits / len(cases)

cases = [
    {"question": "how do I request a refund?", "expected_chunk_id": "billing-03"},
    {"question": "what is the SLA for enterprise?", "expected_chunk_id": "sla-11"},
]
print(round(recall_at_k(basic, cases), 3))

FAQ

Related

Retrieval-augmented generation Embeddings and vector stores

Last refreshed 2026-09-18.