How do you implement a custom retriever in LangChain?

What you need to know

A retriever's contract is simple: string in, list of Document out. A custom one is how you connect LangChain to anything that is not a built-in vector store — an internal search API, Elasticsearch with custom scoring, a SQL query, or a mix.

1from langchain_core.retrievers import BaseRetriever2from langchain_core.callbacks import CallbackManagerForRetrieverRun3from langchain_core.documents import Document45class TicketSearchRetriever(BaseRetriever):6    """Searches the support-ticket API and returns tickets as Documents."""7    client: object          # any object with .search(query, limit)8    k: int = 59    min_score: float = 0.31011    def _get_relevant_documents(12        self, query: str, *, run_manager: CallbackManagerForRetrieverRun13    ) -> list[Document]:14        hits = self.client.search(query, limit=self.k)15        return [16            Document(page_content=h["text"],17                     metadata={"id": h["id"], "score": h["score"], "source": "tickets"})18            for h in hits if h["score"] >= self.min_score19        ]2021retriever = TicketSearchRetriever(client=ticket_api, k=5)22docs = retriever.invoke("refund not received after UPI payment")

Why subclass instead of writing a plain function

• Runnable interface — invoke, batch, stream, ainvoke, with_retry, with_fallbacks all work.
• Tracing — the call appears in LangSmith as a retriever step, with the query and returned documents.
• Drop-in — anything that accepts a retriever (an LCEL chain, EnsembleRetriever, a retrieval tool) accepts yours.

Details that matter

• Pydantic fields — BaseRetriever is a Pydantic model. Declare client: object, k: int = 5 as fields. A custom __init__ usually breaks validation.
• Async — without _aget_relevant_documents, ainvoke runs the sync method in a thread pool. That works but wastes a thread per call; add a real async version if the client supports it.
• Metadata — put the source id and score in metadata. Citations, dedupe and evaluation all depend on it.
• run_manager — lets you emit custom callback events; you can ignore it for simple cases.

The lighter option

1from langchain_core.runnables import chain23@chain4def ticket_retriever(query: str) -> list[Document]:5    return vector_store.similarity_search(query, k=3)

@chain turns a function into a runnable. Use it for quick wrappers; use BaseRetriever when you want configuration fields and a reusable class.

A real-life example

A fintech's support bot must answer "Why did my UPI payment fail?" from two sources: product docs in a vector store and the company's existing Elasticsearch index of resolved tickets, which has years of tuned scoring. Re-indexing tickets into a vector store would lose that tuning.

The engineer writes TicketSearchRetriever around the Elasticsearch client, filtering out hits below a score of 0.3 and tagging each document source="tickets". It joins the vector-store retriever in an EnsembleRetriever. In LangSmith traces, the team can now see which answers came from past tickets versus docs, and they find that ticket hits resolve 30% more "payment failed" questions in their test set.

Follow-up questions to expect

• "How do you test it?" — Pass a fake client with fixed hits and assert on the returned documents and metadata; no network needed.
• "How would you add per-user filtering?" — Take the user or tenant id as a field set when the retriever is created per request, and pass it to the backend query — never trust the query text for it.
• "get_relevant_documents or _get_relevant_documents?" — Implement the underscore method; callers use invoke. The public get_relevant_documents is the deprecated caller-side name.