LangChainRAGretrievalvector searchPython

Building Custom Retrievers in LangChain

Pramod Dutta · Jul 2, 2026 · 11 min read

A LangChain custom retriever is a class that implements the BaseRetriever interface so any data source, not just a vector store, can plug into a chain the same way. You build one when the built-in vector store retriever cannot express your logic: combining keyword and semantic search, filtering by tenant or permission before scoring, calling an internal search API, or reranking results with a second model. This guide walks through the retriever interface, three working custom retrievers of increasing complexity, and the mistakes that quietly degrade retrieval quality.

Why the default retriever is not always enough

Most LangChain tutorials show vectorstore.as_retriever() and stop there. That works fine for a demo: embed a query, run approximate nearest neighbor search, return the top k chunks. In production the requirements usually grow past what a single vector store call can do.

Common reasons teams reach for a custom retriever:

Hybrid search: dense vector similarity misses exact-match terms like product codes, error strings, or acronyms. Combining it with a keyword or BM25 index fixes recall on those queries.
Multi-source retrieval: pulling from a vector store, a SQL database, and an internal REST API in one call, then merging results.
Access control: filtering documents by user role or tenant ID before they ever reach the LLM, not just tagging metadata after the fact.
Reranking: retrieving a wide candidate set cheaply, then reordering with a cross-encoder or an LLM call for precision.
Query rewriting: expanding an abbreviated user query into multiple search queries before hitting the index.

None of these require abandoning LangChain's ecosystem. They just require implementing the retriever contract yourself instead of relying on the default wrapper.

The BaseRetriever interface

Every retriever in LangChain, custom or built-in, is a Runnable. That means it supports .invoke(), .ainvoke(), .batch(), and streaming, and it can be dropped into an LCEL chain with the pipe operator. To build a LangChain custom retriever, subclass BaseRetriever and implement one method:

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun


class EchoRetriever(BaseRetriever):
    """Minimal retriever that returns a single hard-coded document."""

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]:
        return [Document(page_content=f"You searched for: {query}")]

That is the entire contract. _get_relevant_documents takes the query string and a callback manager, and returns a list of Document objects. LangChain wraps this in .invoke() for you, handles tracing, and lets the retriever compose with everything else in the framework.

For async code paths, implement _aget_relevant_documents as well:

class EchoRetriever(BaseRetriever):
    def _get_relevant_documents(self, query, *, run_manager):
        return [Document(page_content=f"You searched for: {query}")]

    async def _aget_relevant_documents(self, query, *, run_manager):
        return [Document(page_content=f"You searched for: {query}")]

If you skip _aget_relevant_documents, LangChain runs the sync method in a thread pool when .ainvoke() is called, which works but adds overhead under concurrent load. Implement it directly if your retriever calls an async HTTP client or an async database driver.

Any extra configuration your retriever needs (a top_k, a client object, a threshold) becomes a Pydantic field on the class, since BaseRetriever extends RunnableSerializable, which is Pydantic-based:

class ThresholdRetriever(BaseRetriever):
    vectorstore: object
    k: int = 4
    score_threshold: float = 0.75

    def _get_relevant_documents(self, query, *, run_manager):
        results = self.vectorstore.similarity_search_with_score(query, k=self.k)
        return [doc for doc, score in results if score >= self.score_threshold]

Declaring fields this way means the retriever plays nicely with LangChain's serialization and tracing, and callers can construct it with keyword arguments like any other component.

Building a hybrid search retriever

The most common reason to write a LangChain custom retriever is hybrid search: blending dense vector similarity with sparse keyword matching. Vector search is good at "documents about this topic"; keyword search is good at "documents containing this exact string." Combining them covers both.

Here is a hybrid retriever that merges a vector store with a BM25Retriever and deduplicates by document content:

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_community.retrievers import BM25Retriever


class HybridRetriever(BaseRetriever):
    vectorstore: object
    bm25_retriever: BM25Retriever
    vector_k: int = 6
    keyword_k: int = 6

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]:
        vector_docs = self.vectorstore.similarity_search(query, k=self.vector_k)
        keyword_docs = self.bm25_retriever.invoke(
            query, config={"callbacks": run_manager.get_child()}
        )[: self.keyword_k]

        merged: dict[str, Document] = {}
        for doc in vector_docs + keyword_docs:
            key = doc.page_content.strip()
            if key not in merged:
                merged[key] = doc

        return list(merged.values())

Two details matter here. First, passing run_manager.get_child() into the nested retriever call keeps the tracing tree intact, so if you inspect a run in LangSmith or another tracer, the BM25 call shows up nested under the hybrid retriever instead of as a disconnected sibling. Second, deduplication by content is a simple heuristic; if your documents carry stable IDs, dedupe on the ID field instead of the raw text, since two chunks can have identical wording but different metadata.

A stronger version of this pattern uses reciprocal rank fusion instead of naive merging, so a document that ranks well in both lists outranks one that only ranks well in one:

def reciprocal_rank_fusion(
    result_lists: list[list[Document]], k: int = 60
) -> list[Document]:
    scores: dict[str, float] = {}
    doc_lookup: dict[str, Document] = {}

    for results in result_lists:
        for rank, doc in enumerate(results):
            key = doc.page_content.strip()
            doc_lookup[key] = doc
            scores[key] = scores.get(key, 0.0) + 1.0 / (k + rank + 1)

    ranked_keys = sorted(scores, key=scores.get, reverse=True)
    return [doc_lookup[key] for key in ranked_keys]

Swap the naive merge in HybridRetriever for reciprocal_rank_fusion([vector_docs, keyword_docs]) and the retriever now rewards consensus between the two search strategies rather than just concatenating them.

Adding metadata filtering and access control

If your app serves multiple tenants or user roles from one index, filtering needs to happen inside the retriever, not after the LLM has already seen the documents. A custom retriever is the right place to enforce this, because it is the single choke point every query passes through.

class TenantScopedRetriever(BaseRetriever):
    vectorstore: object
    tenant_id: str
    k: int = 5

    def _get_relevant_documents(self, query, *, run_manager):
        return self.vectorstore.similarity_search(
            query,
            k=self.k,
            filter={"tenant_id": self.tenant_id},
        )

The filter argument's shape depends on the vector store backend; most support metadata equality filters, and some support richer boolean expressions. The key architectural point is that tenant_id lives on the retriever instance, set when the retriever is constructed for a request, not passed in as part of the free-text query where a prompt injection attempt could try to override it.

For row-level permissions that cannot be expressed as a metadata filter (say, a document is visible to a dynamic list of user IDs resolved from a separate permissions service), retrieve a wider candidate set and filter in Python after the vector search:

class PermissionFilteredRetriever(BaseRetriever):
    vectorstore: object
    permissions_client: object
    user_id: str
    k: int = 5
    fetch_multiplier: int = 4

    def _get_relevant_documents(self, query, *, run_manager):
        candidates = self.vectorstore.similarity_search(
            query, k=self.k * self.fetch_multiplier
        )
        allowed_ids = self.permissions_client.get_visible_doc_ids(self.user_id)
        filtered = [
            doc for doc in candidates if doc.metadata.get("doc_id") in allowed_ids
        ]
        return filtered[: self.k]

Over-fetching with fetch_multiplier matters here: if you only ask for k documents and then filter half of them out for permissions, you can end up returning fewer than k documents to the LLM, sometimes zero, even though relevant documents exist further down the ranking.

Adding a reranking step

Vector search is fast but approximate. A common pattern is to retrieve a wide candidate set cheaply (say 20 to 30 documents) and rerank down to a smaller set (say 4 to 6) with a more expensive but more accurate model, either a dedicated cross-encoder or an LLM prompted to score relevance.

class RerankingRetriever(BaseRetriever):
    base_retriever: BaseRetriever
    reranker: object  # exposes .score(query, documents) -> list[float]
    top_n: int = 5

    def _get_relevant_documents(self, query, *, run_manager):
        candidates = self.base_retriever.invoke(
            query, config={"callbacks": run_manager.get_child()}
        )
        if not candidates:
            return []

        scores = self.reranker.score(query, [doc.page_content for doc in candidates])
        ranked = sorted(zip(candidates, scores), key=lambda pair: pair[1], reverse=True)
        return [doc for doc, _ in ranked[: self.top_n]]

This retriever wraps another retriever rather than talking to a vector store directly, which is a useful pattern in general: build small, single-purpose retrievers and compose them, instead of writing one retriever that does hybrid search, filtering, and reranking all in one method. Composition keeps each piece testable on its own.

Wiring a custom retriever into a chain

Once a retriever implements BaseRetriever, it works anywhere LangChain expects a retriever, including inside LCEL chains with the pipe operator:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "Answer using only the context below.\n\nContext:\n{context}\n\nQuestion: {question}"
)


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


retriever = RerankingRetriever(
    base_retriever=HybridRetriever(
        vectorstore=vectorstore, bm25_retriever=bm25_retriever
    ),
    reranker=my_cross_encoder,
)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke("What error code means the payment webhook timed out?")

Nothing about the surrounding chain changes because the retriever is custom. That is the entire point of implementing BaseRetriever correctly: the rest of the LCEL graph, tracing, and streaming keep working without special-casing.

Testing a custom retriever

Because a custom retriever is a plain Python class with one method to implement, it is straightforward to unit test without spinning up a real vector store:

from unittest.mock import MagicMock


def test_tenant_scoped_retriever_applies_filter():
    mock_store = MagicMock()
    mock_store.similarity_search.return_value = [
        Document(page_content="doc a", metadata={"tenant_id": "acme"})
    ]

    retriever = TenantScopedRetriever(vectorstore=mock_store, tenant_id="acme", k=3)
    results = retriever.invoke("refund policy")

    mock_store.similarity_search.assert_called_once_with(
        "refund policy", k=3, filter={"tenant_id": "acme"}
    )
    assert len(results) == 1

For hybrid and reranking retrievers, write a test that asserts on ordering, not just presence, since the entire value of those retrievers is the ranking they produce. Feed in documents with known relevance and check that the more relevant one comes first in the returned list.

Common mistakes when building custom retrievers

Returning too many or too few documents. A retriever that always returns 20 documents regardless of relevance pushes noise into the LLM's context and increases token cost. A retriever that hard-caps at 3 documents can starve a query that genuinely needs more context to answer. Tune k per use case, and consider making it a query-time parameter rather than a fixed class default.

Not respecting the async path. If your custom retriever calls an external API and you only implement _get_relevant_documents, every async chain that uses it blocks a thread pool worker per call. Implement _aget_relevant_documents with an async HTTP client whenever the underlying call supports one.

Losing the callback chain. When a custom retriever calls another retriever or a sub-chain internally, forgetting to pass run_manager.get_child() breaks tracing. You will see the outer retriever's span in your observability tool but nothing showing what happened inside it, which makes debugging retrieval quality issues much harder.

Filtering after generation instead of before. Access control filters belong inside _get_relevant_documents, applied to the candidate set before it becomes part of the prompt. Filtering the LLM's final answer instead does not stop the model from having already read documents it should never have seen.

Silently swallowing errors. If the vector store call or reranker call fails, decide explicitly whether the retriever should raise, return an empty list, or fall back to a cheaper strategy. An unguarded exception inside _get_relevant_documents surfaces as an opaque failure several layers up the chain.

Ignoring document ordering downstream. Retrievers return documents in relevance order, but if you later pass them through a step that reorders or truncates by token count, you can undo the ranking work the retriever did. Keep the highest-relevance documents closest to the question in the final prompt, since some models weight recency in context more heavily.

FAQ

When should I write a LangChain custom retriever instead of using `as_retriever()`? Use the default wrapper when a single vector store with similarity search covers your needs. Write a custom retriever when you need to combine multiple sources, apply access control before scoring, rerank results, or call a search backend that has no LangChain integration yet.

Do I need to subclass `BaseRetriever`, or can I just write a function? A plain function wrapped in RunnableLambda can act like a retriever for simple cases and will work inside LCEL chains. Subclassing BaseRetriever is worth the extra boilerplate when you want typed configuration fields, consistent tracing spans labeled as a retriever, and compatibility with tools that specifically expect a BaseRetriever instance, such as some agent constructors.

How do I combine dense and sparse retrieval without writing my own merge logic? LangChain ships an EnsembleRetriever that takes a list of retrievers and per-retriever weights and merges results with reciprocal rank fusion internally. Reach for it first; write a fully custom hybrid retriever only when you need logic the ensemble class does not support, like custom deduplication or per-tenant weighting.

How many documents should `k` be set to? There is no universal number. Start by measuring: log the retrieved documents for a sample of real queries, check whether the answer-supporting passage shows up in the top results, and adjust k based on where it tends to land. A wider k paired with a reranking step usually beats a narrow k from a single retrieval pass.

Can a custom retriever call an LLM internally, for example to rewrite the query first? Yes. A common pattern is a query-expansion retriever that calls an LLM to generate two or three alternate phrasings of the user's question, retrieves documents for each phrasing, and merges the results before returning. Just make sure the LLM call goes through run_manager.get_child() so it shows up correctly in traces, and consider caching expansions for repeated queries to avoid redundant LLM calls.

Does a custom retriever work with LangChain's agent tools? Yes. Wrap the retriever in a tool with create_retriever_tool and an agent can call it like any other tool, deciding when retrieval is needed rather than retrieving unconditionally on every turn.