LangChain 设计原理分析 | 向量数据库与 Retriever 机制

本文介绍了向量检索的核心概念与实践方法，重点解析VectorStore与Retriever在LangChain中的分工，比较FAISS和Chroma的适用场景，并演示如何通过as_retriever()将向量库接入RAG流程。主要内容包括：概念区分（VectorStore负责存储检索，Retriever负责业务封装）；工具选择建议（FAISS适合大规模场景，Chroma适合快速原型）；实践示例展示

码农Q！

991人浏览 · 2025-08-21 11:16:13

码农Q！ · 2025-08-21 11:16:13 发布

本文旨在理解向量检索的实现机制、VectorStore 与 Retriever 在 LangChain 中的职责划分，掌握 FAISS 与 Chroma 的实践用法，以及如何把向量库“作为 retriever” 接入检索增强生成（RAG）流水线。

一、先把概念理清楚（最重要的点）

向量数据库 / VectorStore：把文本或片段映射为向量并存储、索引，用于快速近邻搜索（ANN）。在 LangChain 中常见实现包括 FAISS、Chroma、Milvus、Pinecone 等。
Embedding：把文本变为向量的模型（OpenAIEmbeddings / HuggingFaceEmbeddings / 本地 xinference 等）。Embedding 的语义质量直接决定检索效果。
Retriever（检索器）：一个抽象层（BaseRetriever），负责“给我最相关的 documents”。VectorStore 通常提供 .as_retriever() 把自己包装成 Retriever。LangChain 的 Retriever 遵循 Runnable 接口（支持 .invoke() / .ainvoke() 等）。

关键区别：VectorStore 负责存与查；Retriever 负责把“查询”映射成文档列表并做业务侧包装（例如加过滤、返回 score、metadata 过滤等）。

二、FAISS 与 Chroma 的快速比较（实践视角）

FAISS（Facebook AI Similarity Search）
- 优点：极高的性能与灵活的索引（IVF、PQ、HNSW 等），适合大规模离线索引与高吞吐检索。实现上常用于内存/本地服务化场景。
- 典型场景：批量构建索引后离线部署、需要自定义索引参数与性能调优时。
Chroma
- 优点：开源、易用、开发者友好，提供嵌入 + 向量存储的一体化体验，适合中小规模快速迭代。LangChain 提供了 Chroma 的一键集成。
- 典型场景：快速原型、POC、轻量级应用或与 metadata 结合做过滤。

选择建议：如果你要生产级、海量检索（亿级向量），优先 FAISS / Milvus / ANN 专有云；若只是开发迭代或中小数据，Chroma 更省心。

三、`as_retriever()` 的作用与行为

as_retriever() 是 VectorStore 提供的便捷方法，把 VectorStore 包装成 BaseRetriever（或可直接当作 Runnable 使用）。调用后你得到的 retriever 常带有 search_kwargs（比如 k），并可被 LangChain 的 create_retrieval_chain / RetrievalQA 直接消费。官方 How-to 有示例。

要点：

vectorstore.as_retriever(search_kwargs={"k":4}) → 返回 top-k 文档（含 metadata、score）。
Retriever 遵循 Runnable 接口，可直接 .invoke({"input": "..."}) 或在 Chain 中被自动调用。

四、实战示例（两版：FAISS 与 Chroma，基于最新 `create_retrieval_chain`）

说明：下面示例基于 LangChain 最新 API（create_retrieval_chain(retriever, combine_docs_chain)），并展示如何用 as_retriever()。你可以把 embedding 换成本地模型（如 Xinference / HuggingFace）。

依赖（示例环境）

pip install -U langchain langchain_community faiss-cpu chromadb sentence-transformers

示例 A：FAISS + Xinference Embeddings（小数据集演示）

import os

from langchain_community.document_loaders import DirectoryLoader
from langchain_community.embeddings import XinferenceEmbeddings
from langchain_community.vectorstores.faiss import FAISS

from langchain.text_splitter import CharacterTextSplitter
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 1. 加载文本并切片
loader = DirectoryLoader("docs", glob="**/*.txt")  # 你的文档目录
docs = loader.load()
splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=30)
chunks = splitter.split_documents(docs)

# 2. Embedding
embedding = XinferenceEmbeddings(
    server_url="http://127.0.0.1:9997",
    model_uid="bge-large-zh-v1.5"  # 这里填你在 Xinference 加载的 embedding 模型的 uid
)

# 3. 建索引（FAISS）
vectorstore = FAISS.from_documents(chunks, embedding)
vectorstore.save_local('GPT5')

# 4. 构造 retriever（as_retriever）
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 5. LLM + combine chain（stuff）
llm = ChatOpenAI(
    temperature=0,
    model="glm-4.5",
    openai_api_key=os.getenv("ZAI_API_KEY"),
    openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)  # 或本地 LLM 封装
prompt_template = "根据下面文档回答问题：\n\n{context}\n\n问题：{input}\n"
prompt = ChatPromptTemplate.from_template(prompt_template)
combine_chain = create_stuff_documents_chain(llm, prompt)

# 6. 组合成 retrieval chain
retrieval_chain = create_retrieval_chain(retriever=retriever,
                                         combine_docs_chain=combine_chain)

# 7. 调用
resp = retrieval_chain.invoke({"input": "GPT-5有什么特点？"})
print(resp["answer"])

说明/要点：

FAISS.from_documents 会调用 Embeddings，把 chunks 向量化并建立索引（可保存/load）。
retriever = vectorstore.as_retriever(...) 把向量存储包装成 Retriever，可直接给 create_retrieval_chain 使用。

输出：

示例 B：Chroma 快速原型

import os

from langchain_community.document_loaders import DirectoryLoader
from langchain_community.embeddings import XinferenceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

loader = DirectoryLoader("docs", glob="**/*.txt")
docs = loader.load()
splitter = CharacterTextSplitter(chunk_size=400, chunk_overlap=40)
chunks = splitter.split_documents(docs)

emb = XinferenceEmbeddings(
    server_url="http://127.0.0.1:9997",
    model_uid="bge-large-zh-v1.5"  # 这里填你在 Xinference 加载的 embedding 模型的 uid
)

db = Chroma.from_documents(chunks, embedding=emb)
retriever = db.as_retriever(search_kwargs={"k": 1})

llm = ChatOpenAI(
    temperature=0,
    model="glm-4.5",
    openai_api_key=os.getenv("ZAI_API_KEY"),
    openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)
prompt = ChatPromptTemplate.from_template(
    "请基于下面上下文回答问题：\n\n{context}\n\n问题：{input}\n")
combine_chain = create_stuff_documents_chain(llm, prompt)
chain = create_retrieval_chain(retriever=retriever, combine_docs_chain=combine_chain)

print(chain.invoke({"input": "一句话总结GPT5的特点！"})["answer"])

说明：

Chroma 做为一体化 DB，使用体验更简单，适合迭代。

输出：

五、实现细节与调优要点（工程级）

文本切分（chunking）
- 切分策略影响召回 vs 上下文连贯性。常见做法：chunk_size=300~800，overlap=50~200。
- 先把长文档分段并摘要，再存入向量库，这样能减少 Token 消耗。
向量归一化（normalize）
- 当使用余弦相似度时，常将向量 L2 归一化，检索时只需计算 dot product。某些 VectorStore 提供 normalize_L2 参数（FAISS 支持）。
距离度量
- FAISS 支持欧氏（L2）/内积(dot)等；Chroma 默认用余弦/内积。根据 embedding 决定（embedding 是否已归一化）。
索引类型与性能（FAISS）
- 常见索引：IndexFlatL2（精确）；IVF + PQ（大规模压缩）；HNSW（快速近邻）。选择取决于吞吐、内存、精度需求。可在创建时通过 FAISS API 参数配置。
Metadata 过滤
- 如果需要基于 metadata 做筛选（例如按 source、date），vectorstores / retrievers 常支持 filter 参数；在调用 retriever.get_relevant_documents 时可传 filters。检查具体 VectorStore API（Chroma/FAISS wrapper 支持程度不同）。
持久化 / 重建索引
- FAISS 索引与 docstore 需分开保存（FAISS.save_local / FAISS.load_local）；Chroma 提供内置持久化方式。务必做好 embedding 版本与索引版本的同步策略。
语义质量监测
- 使用小集的查询做 A/B（不同 embedding / chunking / index）评测。监控召回率、平均相似度分布、用户反馈。

六、如何自定义 Retriever（高级用法）

如果内置 retriever 不满足你的需求（例如要做多阶段检索、语境压缩、检索融合多个 index），可以继承 BaseRetriever：

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

class MyCustomRetriever(BaseRetriever):
    vector_store: VectorStore
    k: int

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun = None
    ) -> list[Document]:
        # 可以先做 query expansion / rerank / metadata filter
        docs = self.vector_store.similarity_search(query, k=self.k)
        # 做二次排序或裁剪
        docs = docs[:2]
        return docs


retriever = MyCustomRetriever(vector_store=db, k=4)

输出

自定义 retriever 的好处：可以把 检索、排序和压缩 一步完成，直接把处理干净的文档交给 Chain。官方有自定义检索器的教程。

七、实操常见问题与排错

检索到的是无关片段：检查 embedding model 是否适配你的语言域、chunking 是否过碎或过大。
向量大小不匹配 / 报错：确认索引创建时的 embedding 维度与检索时一致（避免模型升级而没重建索引）。
metadata 过滤不起作用：不同 VectorStore 对 filter 支持度不同（Chroma 支持较好），务必查看具体实现文档。

八、快速参考代码片段（保存/加载 FAISS）

from langchain_community.vectorstores.faiss import FAISS

# 3. 建索引（FAISS）
if os.path.exists("GPT5"):
    vectorstore = FAISS.load_local("GPT5", embeddings=embedding,
                                   allow_dangerous_deserialization=True)
else:
    vectorstore = FAISS.from_documents(chunks, embedding)
    vectorstore.save_local('GPT5')