langchain-core retrieval reference

calendarSep 6, 2026 Β· 13 min read Β· ai langchain  Β·
Share on: linkedincopy

Pinned version: langchain-core==1.6.2 (released 2026-09-04) β€” the same package as the langchain-core reference ; that page covers the chat and tool types, this one covers the retrieval types. Two extras are needed for the examples: numpy (resolved to 2.5.2), which InMemoryVectorStore and the fake embeddings import lazily and do not declare β€” without it the first search fails with NameError: name 'np' is not defined β€” and langchain-text-splitters==1.1.2 for the one load_and_split line. Everything below was derived by inspecting the installed packages, and every example was executed against them on Python 3.13. Each heading links to reference.langchain.com, which is unversioned and always serves the latest 1.x. Links open in a new tab.

Part of a five-page set β€” see how the LangChain stack fits together for how langchain-core, langgraph, langchain and langchain-text-splitters fit together.

What the retrieval half of langchain-core is

Retrieval-augmented generation is a pipeline of four steps, and langchain-core defines the type that flows between them and the base class for each step:

1   load                  split                     embed + store                    retrieve
2BaseLoader.lazy_load  β†’  TextSplitter.split_documents  β†’  VectorStore.add_documents  β†’  BaseRetriever.invoke(query)
3   list[Document]         list[Document]              Embeddings.embed_documents        list[Document]
ModuleWhat it defines
langchain_core.documentsDocument (text + metadata + id), BaseDocumentTransformer (what a text splitter is), BaseDocumentCompressor
langchain_core.document_loadersBaseLoader (yields Documents), Blob/BlobLoader/BaseBlobParser for raw bytes
langchain_core.embeddingsEmbeddings (text β†’ vector), plus FakeEmbeddings and DeterministicFakeEmbedding for tests
langchain_core.vectorstoresVectorStore (the store interface), InMemoryVectorStore (the one concrete store), VectorStoreRetriever
langchain_core.retrieversBaseRetriever β€” a Runnable[str, list[Document]]

What it does not ship: loaders for real formats (PDF, HTML, S3, … live in langchain-community and per-source packages), a real embedding model (langchain-openai, langchain-huggingface, langchain-ollama, … subclass Embeddings; langchain-anthropic 1.7.1 has no embeddings class at all), a production vector store (langchain-chroma, langchain-postgres, langchain-pinecone, … subclass VectorStore), or the splitters themselves β€” those are the separate langchain-text-splitters package, which depends on langchain-core alone.

Two facts that shape everything below:

  • A retriever is a Runnable. BaseRetriever subclasses RunnableSerializable[str, list[Document]], so it takes its place in a | chain, a RunnableParallel, or a langgraph node exactly like a prompt or a model β€” the last example on this page prints the graph of a complete RAG chain.
  • Document is not a message. It is the unit of retrieval; what you send to a model is still a HumanMessage/SystemMessage, usually built by formatting the retrieved documents into a prompt.

The examples use a six-line Embeddings subclass (BagOfWords, counts of four words) instead of a provider so that similarity scores are exact and the outputs are reproducible offline.

Index

NameWhat it defines / doesMost-used methods & fields
DocumentPydantic model: page_content (the text), metadata (a dict β€” source, page, headers), optional id (the key vector stores use)page_content, metadata, id
BaseLoaderAbstract source of documents: a subclass implements lazy_load() (a generator); load() collects it, load_and_split() chunks itlazy_load, load, load_and_split
EmbeddingsAbstract text→vector interface: embed_documents(texts) and embed_query(text); providers subclass itembed_documents, embed_query
VectorStoreAbstract store: a subclass implements similarity_search and from_texts; it inherits add_documents, from_documents, get_by_ids, delete, as_retrieverfrom_documents, add_documents, similarity_search, as_retriever
InMemoryVectorStoreThe concrete store in core: a dict of {id, vector, text, metadata} searched by cosine similarity in numpy; dump/load to JSONsimilarity_search_with_score, filter=, max_marginal_relevance_search, dump, load
BaseRetrieverAbstract Runnable[str, list[Document]]: a subclass implements _get_relevant_documents(query); it gets invoke/batch/| and callbacksinvoke, batch
VectorStoreRetrieverWhat as_retriever() returns: a BaseRetriever that runs similarity_search/max_marginal_relevance_search on its store with fixed search_kwargssearch_type, search_kwargs, invoke

Document

The unit everything on this page produces or consumes: a piece of text plus what you know about it. Its fields:

  • page_content: str β€” the text; the only positional argument.
  • metadata: dict β€” anything a filter, a citation or a prompt might need: source, page, the section headers a splitter recorded, a start_index. Defaults to {}.
  • id: str | None β€” the key a vector store files it under; add_documents assigns one when it is missing, and get_by_ids/delete take these.
  • type β€” fixed to "Document", for serialisation.

It is a Pydantic model (BaseMedia β†’ Serializable β†’ BaseModel), so it validates on construction and repr shows the fields; str() gives the page_content=… metadata=… form you see when a list of results is printed. Loaders create them, splitters copy metadata onto every chunk, stores keep the text beside the vector, and retrievers hand them back.

1from langchain_core.documents import Document
2
3doc = Document("Cats sleep sixteen hours a day.", metadata={"source": "cats.txt", "page": 1})
4print(doc.page_content)               # Cats sleep sixteen hours a day.
5print(doc.metadata)                   # {'source': 'cats.txt', 'page': 1}
6print(doc.id, "|", doc.type)          # None | Document
7print(repr(Document("hi", id="d1")))  # Document(id='d1', metadata={}, page_content='hi')
8print(str(doc))                       # page_content='Cats sleep sixteen hours a day.' metadata={'source': 'cats.txt', 'page': 1}

↩ back to index

BaseLoader

The interface for "where documents come from". A subclass implements one method, lazy_load(self) -> Iterator[Document], as a generator that yields one Document per file, page, row or record with the provenance in metadata. In return it gets:

  • load() β€” list(self.lazy_load()), for when the whole corpus fits in memory. (Older loaders implemented load instead; the base lazy_load then wraps it, and raises NotImplementedError if neither is written.)
  • alazy_load() / aload() β€” async twins; the defaults run the sync versions in a thread.
  • load_and_split(text_splitter=None) β€” load() followed by text_splitter.split_documents. With no splitter it constructs RecursiveCharacterTextSplitter() from langchain-text-splitters, and raises ImportError if that package is absent. Its docstring marks it "do not override, consider it deprecated"; calling it is fine.

Core ships no concrete loader for any file format β€” those live in langchain-community and per-source packages β€” and a loader for raw bytes is split into BlobLoader (finds Blobs) plus BaseBlobParser (turns one into documents).

 1from collections.abc import Iterator
 2
 3from langchain_core.document_loaders import BaseLoader
 4from langchain_core.documents import Document
 5
 6FILES = {"cats.txt": "Cats sleep a lot.\n\nCats purr.", "dogs.txt": "Dogs fetch."}
 7
 8
 9class DictLoader(BaseLoader):
10    """One Document per entry; lazy_load is the only method a loader must write."""
11
12    def __init__(self, files: dict[str, str]):
13        self.files = files
14
15    def lazy_load(self) -> Iterator[Document]:
16        for name, text in self.files.items():
17            yield Document(text, metadata={"source": name})
18
19
20loader = DictLoader(FILES)
21print(type(loader.lazy_load()).__name__)                  # generator
22for doc in loader.load():                                 # load() = list(lazy_load())
23    print(doc.metadata["source"], "|", repr(doc.page_content))
24# cats.txt | 'Cats sleep a lot.\n\nCats purr.'
25# dogs.txt | 'Dogs fetch.'
26
27chunks = loader.load_and_split()                          # RecursiveCharacterTextSplitter() by default
28print(len(chunks), [c.page_content for c in chunks])      # 2 ['Cats sleep a lot.\n\nCats purr.', 'Dogs fetch.']
29
30from langchain_text_splitters import CharacterTextSplitter
31
32chunks = loader.load_and_split(CharacterTextSplitter(chunk_size=20, chunk_overlap=0))
33for c in chunks:
34    print(c.metadata["source"], "|", repr(c.page_content))
35# cats.txt | 'Cats sleep a lot.'
36# cats.txt | 'Cats purr.'
37# dogs.txt | 'Dogs fetch.'

↩ back to index

Embeddings

The text-to-vector interface. It is a plain ABC (not a Runnable) with two abstract methods β€” embed_documents(texts: list[str]) -> list[list[float]] for what goes into the store and embed_query(text: str) -> list[float] for what is searched for β€” because some providers embed the two sides differently. The async aembed_documents/aembed_query default to running the sync methods in a thread. A vector store holds one Embeddings and calls embed_documents on add_documents and embed_query on every search, so the same object must be used for both, or the vectors will not be comparable.

Core ships no provider implementation, only two test doubles (both need numpy): FakeEmbeddings(size=) returns random vectors, and DeterministicFakeEmbedding(size=) seeds the random generator with a hash of the text, so the same text always gets the same vector β€” but two different texts get unrelated vectors, so nearest-neighbour search over it only finds exact matches. That is why the examples on this page use the small subclass below instead: its vectors mean something, and cosine scores come out exact.

 1import re
 2
 3from langchain_core.embeddings import DeterministicFakeEmbedding, Embeddings
 4
 5VOCAB = ["cat", "dog", "kernel", "python"]
 6
 7
 8class BagOfWords(Embeddings):
 9    """The two members Embeddings leaves abstract, on a four-word vocabulary."""
10
11    def embed_documents(self, texts: list[str]) -> list[list[float]]:
12        return [self.embed_query(t) for t in texts]
13
14    def embed_query(self, text: str) -> list[float]:
15        words = re.findall(r"[a-z]+", text.lower())
16        return [float(words.count(w)) for w in VOCAB]
17
18
19emb = BagOfWords()
20print(emb.embed_query("cat cat dog"))                    # [2.0, 1.0, 0.0, 0.0]
21print(emb.embed_documents(["python kernel", "dog"]))     # [[0.0, 0.0, 1.0, 1.0], [0.0, 1.0, 0.0, 0.0]]
22
23fake = DeterministicFakeEmbedding(size=4)                # core's own test double: seeded by hash(text)
24a, b = fake.embed_query("cat"), fake.embed_query("cat")
25print(len(a), a == b)                                    # 4 True   <- same text, same vector
26print(fake.embed_query("cat") == fake.embed_query("dog"))  # False

↩ back to index

VectorStore

The abstract store: documents in, nearest documents out. A backend implements two members β€” similarity_search(query, k=4, **kwargs) -> list[Document] and the classmethod from_texts(texts, embedding, metadatas=None, ids=None) β€” plus one of add_texts/add_documents (each is defined in terms of the other). Everything else is inherited:

  • Loading. from_documents(docs, embedding) unpacks the documents and calls from_texts; add_documents(docs, ids=None) returns the ids used (a document's own id when it has one).
  • Searching. similarity_search (documents only), similarity_search_with_score (plus the backend's raw score), similarity_search_with_relevance_scores (normalised to 0–1, which needs the backend to define a relevance function), similarity_search_by_vector (skip the query embedding), max_marginal_relevance_search(query, k, fetch_k, lambda_mult) (fetch fetch_k, then pick k that are relevant and mutually different). Each has an a-prefixed async twin.
  • Housekeeping. get_by_ids(ids), delete(ids), and the embeddings property.
  • as_retriever(search_type=…, search_kwargs=…) β€” wraps the store in a VectorStoreRetriever so it can sit in a chain.

Provider packages (langchain-chroma, langchain-postgres, langchain-pinecone, …) subclass it; core's own implementation is InMemoryVectorStore , used below.

 1import re
 2
 3from langchain_core.documents import Document
 4from langchain_core.embeddings import Embeddings
 5from langchain_core.vectorstores import InMemoryVectorStore, VectorStore
 6
 7VOCAB = ["cat", "dog", "kernel", "python"]
 8
 9
10class BagOfWords(Embeddings):
11    def embed_documents(self, texts):
12        return [self.embed_query(t) for t in texts]
13
14    def embed_query(self, text):
15        words = re.findall(r"[a-z]+", text.lower())
16        return [float(words.count(w)) for w in VOCAB]
17
18
19print(sorted(VectorStore.__abstractmethods__))          # ['from_texts', 'similarity_search']
20
21docs = [
22    Document("The cat sat on the mat.", metadata={"topic": "pets"}, id="d1"),
23    Document("A dog chased the cat.", metadata={"topic": "pets"}, id="d2"),
24    Document("The kernel schedules the python process.", metadata={"topic": "os"}, id="d3"),
25]
26store = InMemoryVectorStore.from_documents(docs, BagOfWords())   # inherited: embeds + add_documents
27print(isinstance(store, VectorStore), store.embeddings.__class__.__name__)   # True BagOfWords
28
29print([d.id for d in store.similarity_search("cat", k=2)])       # ['d1', 'd2']
30print(store.add_documents([Document("Another cat.", id="d4")]))  # ['d4']
31print([d.id for d in store.get_by_ids(["d4", "d1"])])            # ['d4', 'd1']
32print(store.delete(["d4"]), len(store.store))                    # None 3
33
34retriever = store.as_retriever(search_kwargs={"k": 1})
35print(type(retriever).__name__, retriever.invoke("kernel")[0].id)   # VectorStoreRetriever d3

↩ back to index

InMemoryVectorStore

The one concrete VectorStore in core, and the right one for tests, notebooks and anything that fits in RAM. InMemoryVectorStore(embedding) keeps a plain dict, store, of id β†’ {"id", "vector", "text", "metadata"}, embeds on add_documents, and searches by computing the cosine similarity between the query vector and every stored vector in numpy. What it adds to the interface:

  • similarity_search_with_score(query, k=4, filter=None) returns (Document, score) pairs where the score is the cosine similarity β€” 1.0 for a document pointing the same way as the query, 0.0 for an orthogonal one. filter is a callable Document -> bool applied before ranking, so it can use any metadata.
  • max_marginal_relevance_search(query, k=4, fetch_k=20, lambda_mult=0.5) is implemented, so search_type="mmr" works on a retriever built from it. similarity_search_with_relevance_scores β€” and therefore search_type="similarity_score_threshold" β€” is not: this store defines no relevance function, and the call raises NotImplementedError.
  • dump(path) / load(path, embedding) β€” JSON on disk. The vectors are saved, the Embeddings object is not, so load is handed the same kind of embedder again.
  • get_by_ids, delete(ids), and the _by_vector search variants.
 1import os
 2import re
 3import tempfile
 4
 5from langchain_core.documents import Document
 6from langchain_core.embeddings import Embeddings
 7from langchain_core.vectorstores import InMemoryVectorStore
 8
 9VOCAB = ["cat", "dog", "kernel", "python"]
10
11
12class BagOfWords(Embeddings):
13    def embed_documents(self, texts):
14        return [self.embed_query(t) for t in texts]
15
16    def embed_query(self, text):
17        words = re.findall(r"[a-z]+", text.lower())
18        return [float(words.count(w)) for w in VOCAB]
19
20
21store = InMemoryVectorStore(BagOfWords())
22store.add_documents([
23    Document("cat cat", metadata={"lang": "en"}, id="d1"),
24    Document("cat dog", metadata={"lang": "en"}, id="d2"),
25    Document("python kernel", metadata={"lang": "code"}, id="d3"),
26])
27print(store.store["d1"])
28# {'id': 'd1', 'vector': [2.0, 0.0, 0.0, 0.0], 'text': 'cat cat', 'metadata': {'lang': 'en'}}
29
30for doc, score in store.similarity_search_with_score("cat", k=3):
31    print(f"{score:.3f}", doc.id, doc.page_content)
32# 1.000 d1 cat cat        <- cosine similarity: same direction as the query
33# 0.707 d2 cat dog
34# 0.000 d3 python kernel
35
36only_code = store.similarity_search("kernel cat", k=3, filter=lambda d: d.metadata["lang"] == "code")
37print([d.id for d in only_code])                          # ['d3']
38
39path = os.path.join(tempfile.mkdtemp(), "store.json")
40store.dump(path)
41reloaded = InMemoryVectorStore.load(path, BagOfWords())   # vectors are stored; the embedder is not
42print(len(reloaded.store), reloaded.similarity_search("dog", k=1)[0].id)   # 3 d2

↩ back to index

BaseRetriever

The abstract "string in, documents out" component, and the reason retrieval plugs into the rest of the ecosystem: it subclasses RunnableSerializable[str, list[Document]]. A subclass declares its configuration as Pydantic fields and implements one method, _get_relevant_documents(self, query: str, *, run_manager) -> list[Document] (the run_manager parameter may be omitted from the signature; _aget_relevant_documents is the optional async native version). In return it gets:

  • The Runnable surface. invoke(query), batch(queries), ainvoke, astream, and | composition β€” retriever | format_docs | prompt | model is the RAG chain. Input and output schemas are generated and named after the class.
  • Callbacks. invoke wraps the call in on_retriever_start/on_retriever_end/on_retriever_error, so a retriever shows up as its own span in LangSmith traces with the query and the documents.
  • Config. tags, metadata, and everything with_config accepts.

A vector store is not a retriever, but as_retriever() makes one; write your own subclass when the source is not a vector store at all β€” a keyword index, an HTTP API, a database query.

 1from langchain_core.callbacks import CallbackManagerForRetrieverRun
 2from langchain_core.documents import Document
 3from langchain_core.retrievers import BaseRetriever
 4from langchain_core.runnables import Runnable, RunnableLambda
 5
 6
 7class KeywordRetriever(BaseRetriever):
 8    """Every document containing the query word, most mentions first."""
 9
10    docs: list[Document]
11    k: int = 2
12
13    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> list[Document]:
14        scored = [(d.page_content.lower().split().count(query.lower()), d) for d in self.docs]
15        return [d for n, d in sorted(scored, key=lambda t: -t[0]) if n > 0][: self.k]
16
17
18retriever = KeywordRetriever(docs=[
19    Document("cat cat cat", id="d1"),
20    Document("dog", id="d2"),
21    Document("cat and dog", id="d3"),
22])
23print(isinstance(retriever, Runnable))                          # True
24print([d.id for d in retriever.invoke("cat")])                  # ['d1', 'd3']
25print([[d.id for d in r] for r in retriever.batch(["dog", "python"])])   # [['d2', 'd3'], []]
26
27count = retriever | RunnableLambda(lambda docs: f"{len(docs)} hit(s)")
28print(count.invoke("dog"))                                      # 2 hit(s)
29print(retriever.get_input_schema().__name__, "->", retriever.get_output_schema().__name__)
30# KeywordRetrieverInput -> KeywordRetrieverOutput

↩ back to index

VectorStoreRetriever

What VectorStore.as_retriever() returns: a BaseRetriever whose _get_relevant_documents calls the store. Its fields fix how the store is searched, so the chain that holds it only ever passes a query string:

  • search_type β€” "similarity" (default) calls similarity_search; "mmr" calls max_marginal_relevance_search; "similarity_score_threshold" calls similarity_search_with_relevance_scores and keeps what scores above search_kwargs["score_threshold"] (unsupported by InMemoryVectorStore, see above). Anything else is rejected at construction.
  • search_kwargs β€” forwarded to that call: k, filter, fetch_k and lambda_mult for MMR. Keyword arguments passed to invoke are merged on top for a single call.
  • vectorstore, tags (the store's class name and its embedder's, for tracing), metadata.

Because it is a Runnable, a whole RAG chain is Runnable composition and can print its own graph, which is what the example ends with.

 1import re
 2
 3from langchain_core.documents import Document
 4from langchain_core.embeddings import Embeddings
 5from langchain_core.language_models import GenericFakeChatModel
 6from langchain_core.messages import AIMessage
 7from langchain_core.output_parsers import StrOutputParser
 8from langchain_core.prompts import ChatPromptTemplate
 9from langchain_core.runnables import RunnableParallel, RunnablePassthrough
10from langchain_core.vectorstores import InMemoryVectorStore
11
12VOCAB = ["cat", "dog", "kernel", "python"]
13
14
15class BagOfWords(Embeddings):
16    def embed_documents(self, texts):
17        return [self.embed_query(t) for t in texts]
18
19    def embed_query(self, text):
20        words = re.findall(r"[a-z]+", text.lower())
21        return [float(words.count(w)) for w in VOCAB]
22
23
24store = InMemoryVectorStore.from_documents([
25    Document("A cat sleeps sixteen hours a day.", id="cat"),
26    Document("A dog needs a walk every day.", id="dog"),
27    Document("The kernel schedules every python thread.", id="os"),
28], BagOfWords())
29
30retriever = store.as_retriever(search_kwargs={"k": 1})
31print(retriever.search_type, retriever.search_kwargs)          # similarity {'k': 1}
32print([d.id for d in retriever.invoke("How long does a cat sleep?")])   # ['cat']
33
34diverse = store.as_retriever(search_type="mmr", search_kwargs={"k": 2, "fetch_k": 3})
35print([d.id for d in diverse.invoke("cat cat dog")])           # ['cat', 'dog']
36
37# The retriever is a Runnable, so it is the first stage of a RAG chain.
38prompt = ChatPromptTemplate.from_messages([
39    ("system", "Answer from the context only.\n\n{context}"),
40    ("human", "{question}"),
41])
42model = GenericFakeChatModel(messages=iter([AIMessage("About sixteen hours.")]))
43
44
45def format_docs(docs: list[Document]) -> str:
46    return "\n".join(d.page_content for d in docs)
47
48
49gather = RunnableParallel(context=retriever | format_docs, question=RunnablePassthrough())
50gather.name = "gather"                                        # names the diagram's fork/join nodes
51chain = gather | prompt | model | StrOutputParser()
52
53print(chain.invoke("How long does a cat sleep?"))             # About sixteen hours.
54print(chain.get_graph().draw_mermaid(with_styles=False))

Diagram β€” generated by the code above:

gather_input to VectorStoreRetriever to format_docs to gather_output; gather_input to Passthrough to gather_output; gather_output to ChatPromptTemplate to GenericFakeChatModel to StrOutputParser to StrOutputParserOutput.

The fork after gather_input is the RunnableParallel: the question goes down one branch to the retriever and, unchanged, down the other; gather_output is the dict {"context", "question"} the prompt template fills. Without the gather.name line the two nodes are called Parallel<context,question>Input/Output.

↩ back to index