Bài 23: Retrievers — abstract layer trên vector DB

Sau bài này bạn sẽ:

• Hiểu retriever là Runnable với signature .invoke(query: str) -> list[Document].
• Biết cách tạo retriever từ vector store với các search type khác nhau (similarity, mmr, similarity_score_threshold).
• Kết hợp nhiều retriever bằng EnsembleRetriever (hybrid BM25 + vector).
• Rerank kết quả bằng ContextualCompressionRetriever + Cohere Rerank.
• Dùng MultiQueryRetriever, ParentDocumentRetriever, SelfQueryRetriever cho các use case phức tạp.
• Tích hợp retriever vào LCEL chain.

Retriever là abstraction của LangChain cho bài toán: nhận vào một query string, trả về một danh sách Document. Về mặt kỹ thuật, retriever là Runnable với signature cụ thể:

retriever.invoke("FastAPI lifespan là gì?")
# → [Document(page_content="...", metadata={...}), ...]

Nguồn dữ liệu bên dưới có thể là:

• Vector store (ChromaDB, Pinecone, Qdrant, ...)
• BM25 (keyword search truyền thống)
• Web search (Tavily, Google)
• SQL database (query theo ngôn ngữ tự nhiên)
• API nội bộ bất kỳ

Điểm mấu chốt: RAG chain không quan tâm nguồn là gì — nó chỉ gọi retriever.invoke(query). Muốn đổi từ ChromaDB sang Pinecone hay từ vector search sang hybrid, chỉ cần thay object retriever, không phải viết lại chain.

Khác biệt với vector store client thẳng

Các bài 14–18 trong Module 3 dạy cách dùng client trực tiếp của từng vector DB (ChromaDB, Pinecone, Qdrant). Cách đó phù hợp khi bạn cần kiểm soát chi tiết (batch upsert, xóa collection, quản lý index). Retriever của LangChain là lớp trên, phù hợp khi bạn đã xây xong index và cần một interface đồng nhất để query trong RAG chain.

Mọi vector store trong LangChain đều có method .as_retriever() để tạo retriever wrapper:

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    collection_name="docs",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4},
)

docs = retriever.invoke("FastAPI lifespan là gì?")
for doc in docs:
    print(doc.page_content[:120])
    print(doc.metadata)
    print("---")

search_kwargs là dict truyền thẳng vào method search của vector store. Ngoài k, một số store hỗ trợ filter để lọc theo metadata:

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 6,
        "filter": {"source": "fastapi-docs.pdf"},
    },
)

Filter này hoạt động ở tầng vector store — tương đương where clause trong Module 3, nhưng bây giờ nằm trong config của retriever thay vì gọi API client thẳng.

.as_retriever() hỗ trợ ba giá trị cho search_type:

similarity (mặc định)

Trả top-k document gần nhất theo cosine distance (hoặc metric mà store đang dùng). Không lọc thêm điều kiện gì.

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4},
)

mmr

Maximal Marginal Relevance — cân bằng giữa relevance và diversity. Xem chi tiết ở mục 5.

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

similarity_score_threshold

Chỉ trả document có similarity score vượt ngưỡng. Dùng khi muốn tránh trả doc không liên quan khi query quá xa tất cả tài liệu.

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7},
    # không set k → trả tất cả doc vượt ngưỡng (có thể nhiều hoặc ít)
)

Lưu ý: nếu không có doc nào vượt ngưỡng, invoke() trả về list rỗng []. Chain phía sau cần xử lý trường hợp này.

Với similarity thuần, nếu một đoạn nội dung xuất hiện nhiều lần trong tài liệu (ví dụ intro được lặp lại ở nhiều chapter), top-K có thể trả về 4–5 đoạn gần như giống hệt nhau. Context gửi lên LLM khi đó trùng lặp, lãng phí token window.

MMR giải quyết vấn đề này bằng thuật toán lặp:

1. Lấy fetch_k document gần nhất với query (pool ban đầu).
2. Trong mỗi vòng lặp, chọn document có điểm kết hợp cao nhất: lambda_mult * relevance - (1 - lambda_mult) * max_similarity_to_selected.
3. Lặp đến khi đủ k document.

Ý nghĩa các tham số:

• k: số document trả về cuối cùng.
• fetch_k: kích thước pool ban đầu để MMR chọn từ đó. Nên đặt fetch_k >= 3 * k để MMR có đủ lựa chọn. Mặc định là 20.
• lambda_mult: số thực trong [0, 1]. Càng gần 0 → ưu tiên diversity; càng gần 1 → ưu tiên relevance. Mặc định 0.5.

# Tìm "best practices Python" với MMR
# Không muốn 5 đoạn cùng nói về type hint
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,
        "fetch_k": 25,
        "lambda_mult": 0.4,  # nghiêng về diversity
    },
)
docs = retriever.invoke("best practices Python")
# → 5 đoạn từ 5 chủ đề khác nhau (type hint, error handling, docstring, ...)

Khi nào không cần MMR: query rất cụ thể (ví dụ "trang 3 của report.pdf mục 2.1"), similarity thuần đã đủ chính xác vì tất cả top-K doc đều liên quan chặt đến một đoạn duy nhất.

LangChain cung cấp nhiều retriever không phụ thuộc vector DB, tất cả đều dùng cùng interface .invoke(query).

BM25Retriever — keyword search cổ điển

BM25 (BM = "Best Match", từ Okapi BM25, Robertson et al. 1994) là thuật toán ranking dựa trên tần suất từ và độ dài document. Hoạt động tốt với query chứa thuật ngữ kỹ thuật cụ thể, tên riêng, mã sản phẩm — những gì embedding hay bị "mờ" vì vector hóa.

from langchain_community.retrievers import BM25Retriever

# docs là list[Document] đã load sẵn
bm25_retriever = BM25Retriever.from_documents(docs, k=4)
results = bm25_retriever.invoke("FastAPI UploadFile")

Cần cài thêm: pip install rank_bm25.

BM25Retriever lưu index trong bộ nhớ — không persist. Mỗi lần khởi động phải build lại từ document list.

TavilySearchAPIRetriever — web search

from langchain_community.retrievers import TavilySearchAPIRetriever

# Cần TAVILY_API_KEY trong env
tavily_retriever = TavilySearchAPIRetriever(k=5)
docs = tavily_retriever.invoke("LangChain 0.3 migration guide")

Kết quả là Document với page_content là snippet và metadata["source"] là URL. Phù hợp khi cần thông tin real-time không có trong corpus nội bộ.

WikipediaRetriever

from langchain_community.retrievers import WikipediaRetriever

wiki_retriever = WikipediaRetriever(top_k_results=3, doc_content_chars_max=2000)
docs = wiki_retriever.invoke("transformer neural network")

Cần: pip install wikipedia.

Lựa chọn retriever theo tính chất query

EnsembleRetriever gọi tất cả retriever trong list song song, sau đó merge kết quả bằng Reciprocal Rank Fusion (RRF). RRF tính score cho mỗi document dựa trên rank của nó trong từng danh sách: score = Σ weight_i / (rank_i + 60). Hằng số 60 là giá trị mặc định từ paper gốc (Cormack et al., 2009).

from langchain_community.retrievers import BM25Retriever
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever

# Build hai retriever
bm25 = BM25Retriever.from_documents(docs, k=4)

vectorstore = Chroma(
    collection_name="docs",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# Kết hợp
ensemble = EnsembleRetriever(
    retrievers=[bm25, vector_retriever],
    weights=[0.4, 0.6],
)

docs = ensemble.invoke("FastAPI UploadFile async")

Weights phải tổng bằng 1.0. Ví dụ trên: vector retriever (semantic) có trọng số 0.6, BM25 (keyword) có 0.4. Điều chỉnh theo tính chất corpus:

• Corpus nhiều code, tên API, mã lỗi → tăng BM25 weight.
• Corpus văn bản tự nhiên, câu hỏi mở → tăng vector weight.

EnsembleRetriever có thể kết hợp hơn hai retriever. Số lượng phần tử trong retrievers và weights phải bằng nhau.

Vector retrieval (recall@K) trả nhiều document để không bỏ sót, nhưng precision thấp — một phần trong K document không thực sự liên quan. ContextualCompressionRetriever giải quyết bằng cách thêm một bước lọc/rerank sau khi retrieve.

Cơ chế: gọi base_retriever.invoke(query) lấy K doc → đưa qua base_compressor để lọc / sắp xếp lại → trả top-N chất lượng.

Cohere Rerank

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma(
    collection_name="docs",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./chroma_db",
)
# Base retriever: retrieve rộng (top-20)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

# Compressor: rerank top-20 → top-3
# Cần COHERE_API_KEY trong env
compressor = CohereRerank(top_n=3, model="rerank-english-v3.0")

reranked_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever,
)

docs = reranked_retriever.invoke("FastAPI lifespan event")
# → tối đa 3 document, đã được rerank bởi Cohere

Cần: pip install cohere langchain-cohere.

LLMChainFilter — dùng LLM làm compressor

Khi không có Cohere API, có thể dùng LLM để filter. Cách này chậm hơn vì gọi LLM cho mỗi document, nhưng không cần service ngoài:

from langchain.retrievers.document_compressors import LLMChainFilter
from langchain_openai import ChatOpenAI

llm_filter = LLMChainFilter.from_llm(ChatOpenAI(model="gpt-4o-mini"))

filtered_retriever = ContextualCompressionRetriever(
    base_compressor=llm_filter,
    base_retriever=base_retriever,
)

Khi nào dùng rerank

Rerank thêm latency (gọi Cohere API hoặc LLM) và cost. Phù hợp khi:

• Precision của kết quả ảnh hưởng trực tiếp đến chất lượng câu trả lời (hệ thống Q&A nội bộ, chatbot hỗ trợ kỹ thuật).
• Token window của LLM giới hạn — cần lọc chặt thay vì dùng nhiều doc.

Không cần rerank khi độ chính xác retrieval đã cao (corpus nhỏ, query cụ thể) hoặc khi latency là ưu tiên số 1.

Vấn đề: user dùng từ khác với từ trong tài liệu (ví dụ user hỏi "cách giữ state giữa các request" nhưng tài liệu dùng "session management"). Vector embedding giảm thiểu vấn đề này nhưng không giải quyết hoàn toàn.

MultiQueryRetriever dùng LLM để tạo N biến thể khác nhau của query gốc, thực hiện retrieval cho từng biến thể, sau đó hợp nhất (union) toàn bộ document trả về (loại trùng theo page_content).

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

base_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

mq_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=ChatOpenAI(model="gpt-4o-mini"),
)

# Bật logging để xem query được generate
import logging
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

docs = mq_retriever.invoke("cách giữ state giữa các request trong FastAPI")
# LLM generate thêm 2-3 query → retrieve → union → deduplicate

Đánh đổi: MultiQueryRetriever gọi LLM thêm một lần để generate query variants, tốn thêm latency và cost. Phù hợp khi query từ người dùng cuối không có kiểm soát, corpus dùng ngôn ngữ kỹ thuật hoặc tiếng Anh trong khi user hỏi bằng tiếng Việt.

Có hai mục tiêu mâu thuẫn khi chunking:

• Retrieval chính xác: chunk nhỏ → embedding của chunk đại diện tốt hơn cho một ý cụ thể → cosine similarity với query cao hơn.
• Ngữ cảnh đủ rộng: chunk lớn → LLM nhận được đủ context để trả lời trọn vẹn.

ParentDocumentRetriever giải quyết mâu thuẫn này bằng cách:

1. Index chunk nhỏ vào vector store (dùng để retrieve).
2. Lưu parent document lớn vào docstore.
3. Khi retrieve: tìm chunk nhỏ → lookup parent → trả parent.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# Splitter cho chunk nhỏ (dùng để index)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# Splitter cho parent (lưu trong docstore)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

vectorstore = Chroma(
    collection_name="child_chunks",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
)
docstore = InMemoryStore()  # hoặc RedisStore cho production

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# Add documents (cần làm một lần khi build index)
retriever.add_documents(raw_docs)

# Retrieve: trả về parent (2000 chars), không phải child (400 chars)
docs = retriever.invoke("FastAPI lifespan event")

Lưu ý: InMemoryStore mất dữ liệu khi restart. Trong production, dùng RedisStore (cần pip install langchain-redis) hoặc tự implement BaseStore.

SelfQueryRetriever dùng LLM để phân tích query tự nhiên và tự động tạo cả vector query lẫn metadata filter. Phù hợp khi corpus có metadata có cấu trúc (tên file, ngày, tác giả, tag, ...) và user muốn query kết hợp nội dung + điều kiện.

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_openai import ChatOpenAI

# Mô tả metadata schema cho LLM
metadata_field_info = [
    AttributeInfo(
        name="source",
        description="tên file PDF chứa đoạn text",
        type="string",
    ),
    AttributeInfo(
        name="page",
        description="số trang trong file PDF",
        type="integer",
    ),
    AttributeInfo(
        name="author",
        description="tác giả tài liệu",
        type="string",
    ),
]

sq_retriever = SelfQueryRetriever.from_llm(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    vectorstore=vectorstore,
    document_contents="đoạn text từ tài liệu kỹ thuật về AI và backend development",
    metadata_field_info=metadata_field_info,
    verbose=True,  # in ra structured query LLM generate
)

# LLM parse query → vector query "lifespan" + filter: source=report.pdf AND page=3
docs = sq_retriever.invoke("Tìm phần về lifespan ở trang 3 của report.pdf")

Khi verbose=True, output log hiện structured query dạng:

{
  "query": "lifespan",
  "filter": "and(eq('source', 'report.pdf'), eq('page', 3))"
}

Giới hạn: LLM đôi khi parse sai filter nếu query mơ hồ hoặc metadata schema không được mô tả đủ rõ trong description. Luôn bật verbose=True trong giai đoạn development để kiểm tra filter thực tế. Ngoài ra, không phải vector store nào cũng hỗ trợ tất cả operator — kiểm tra docs của store đang dùng.

Vì retriever là Runnable, nó có thể dùng trực tiếp trong pipe |:

Pattern đơn giản nhất

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

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "Trả lời dựa trên context sau.\n"
        "Context:\n{context}\n\n"
        "Nếu không đủ thông tin, nói 'Tôi không có thông tin về vấn đề này'."
    )),
    ("user", "{question}"),
])

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

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

answer = chain.invoke("FastAPI là gì?")

Pattern dùng RunnablePassthrough.assign

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

chain = (
    RunnablePassthrough.assign(
        context=lambda x: format_docs(retriever.invoke(x["question"]))
    )
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke({"question": "FastAPI lifespan là gì?"})

Phân biệt hai pattern:

• Pattern 1: input vào chain là string (query thẳng), retriever nhận string.
• Pattern 2: input là dict {"question": "..."}, linh hoạt hơn khi cần truyền thêm params (ví dụ session_id cho memory ở bài sau).

Streaming

for chunk in chain.stream({"question": "FastAPI là gì?"}):
    print(chunk, end="", flush=True)

Retrieval chạy blocking trước, sau đó LLM stream token. Người dùng thấy câu trả lời xuất hiện dần khi LLM generate, không đợi toàn bộ completion.

Khi cần integrate nguồn dữ liệu không có sẵn trong LangChain (search engine nội bộ, GraphQL API, database chuyên dụng), kế thừa BaseRetriever và override _get_relevant_documents:

from typing import List
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from langchain_core.callbacks.manager import CallbackManagerForRetrieverRun

class InternalSearchRetriever(BaseRetriever):
    """Retriever gọi internal search API."""

    api_url: str
    top_k: int = 5

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> List[Document]:
        import httpx

        response = httpx.get(
            f"{self.api_url}/search",
            params={"q": query, "limit": self.top_k},
            timeout=10.0,
        )
        response.raise_for_status()

        results = response.json()["results"]
        return [
            Document(
                page_content=r["text"],
                metadata={"id": r["id"], "score": r["score"]},
            )
            for r in results
        ]

Dùng giống bất kỳ retriever nào khác:

retriever = InternalSearchRetriever(api_url="https://search.internal.company.com", top_k=4)
docs = retriever.invoke("FastAPI lifespan")

Nếu cần async (để không block event loop trong FastAPI), override thêm _aget_relevant_documents:

    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> List[Document]:
        import httpx

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.api_url}/search",
                params={"q": query, "limit": self.top_k},
            )
        response.raise_for_status()
        # ... parse và return

1. Quên set k, dùng default 4

Default k=4 đôi khi quá ít (corpus lớn, câu hỏi rộng) hoặc quá nhiều (token window LLM giới hạn, mỗi doc dài). Xác định k dựa trên độ dài trung bình của chunk và context window của LLM đang dùng.

2. similarity_score_threshold quá cao → trả empty

# Ngưỡng 0.9 quá nghiêm → gần như không bao giờ có doc nào qua
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.9},
)
docs = retriever.invoke("...")  # → []

Kiểm tra bằng cách dùng similarity trước, xem score thực tế, rồi mới chọn ngưỡng phù hợp với corpus.

3. EnsembleRetriever weights không cân đối

Đặt BM25 weight quá thấp (ví dụ [0.05, 0.95]) → BM25 gần như không ảnh hưởng → mất đi lợi ích hybrid. Bắt đầu với [0.3, 0.7] rồi điều chỉnh dựa trên evaluation.

4. SelfQueryRetriever parse sai filter

LLM đôi khi generate filter không chính xác, đặc biệt khi query mơ hồ hoặc trường metadata có tên gần giống nhau. Luôn dùng verbose=True trong development để kiểm tra structured query được generate. Nếu parse sai thường xuyên, thêm ví dụ vào description của AttributeInfo.

5. Rerank tốn cost và latency

Cohere Rerank tính phí per request. Với base retriever k=20 và rerank top-3, mỗi query tốn ít nhất một lần gọi Cohere. Ở scale lớn, cân nhắc chỉ rerank khi precision quan trọng hơn cost, hoặc dùng cross-encoder local (sentence-transformers) thay vì Cohere.

6. ParentDocumentRetriever mất docstore khi restart

InMemoryStore không persist. Nếu dùng trong ứng dụng thực tế, cần dùng persistent store (Redis, SQLite, ...) và cơ chế reload document khi khởi động.

• Retriever là Runnable với interface duy nhất: .invoke(query) -> list[Document]. Swap nguồn dữ liệu không cần viết lại RAG chain.
• Vector store retriever: 3 search type — similarity (top-k), mmr (diverse top-k), similarity_score_threshold (filter theo score).
• MMR dùng khi muốn tránh kết quả trùng lặp; cần đặt fetch_k đủ lớn (thường ≥ 3 × k).
• BM25Retriever bổ sung keyword matching cho những gì vector embedding hay bỏ sót (tên riêng, mã kỹ thuật).
• EnsembleRetriever merge nhiều retriever bằng RRF — hybrid BM25 + vector là pattern phổ biến nhất.
• ContextualCompressionRetriever + Cohere Rerank: retrieve rộng (top-20) rồi rerank về top-3; dùng khi precision quan trọng hơn latency/cost.
• MultiQueryRetriever: LLM paraphrase query → retrieve nhiều variant → union. Giúp khi user dùng từ khác với tài liệu.
• ParentDocumentRetriever: index chunk nhỏ, trả parent lớn — giải quyết mâu thuẫn precision vs context.
• SelfQueryRetriever: LLM parse query tự nhiên → vector query + metadata filter. Luôn bật verbose=True khi debug.
• Custom retriever: kế thừa BaseRetriever, override _get_relevant_documents.

Bài 24: Memory — lưu lịch sử conversation