LangChain 进阶：Vector Stores 向量存储详解

或者我们也可替换成本地的词嵌入模型，这里我经常用（项目用）的词嵌入模型是 m3e 系列：

from langchain_community.vectorstores.utils import DistanceStrategy
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
from transformers.utils import is_torch_cuda_available, is_torch_mps_available
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS

# 词嵌入模型设备选择
EMBEDDING_DEVICE = "cuda" if is_torch_cuda_available() else "mps" if is_torch_mps_available() else "cpu"
embedding = HuggingFaceEmbeddings(
    model_name='D:\models\m3e-base', 
    model_kwargs={'device': EMBEDDING_DEVICE}
)

text_loader = TextLoader('./index.txt', encoding='utf-8', autodetect_encoding=True)
raw_documents = text_loader.load()

text_splitter = CharacterTextSplitter(chunk_size=150, chunk_overlap=80)
documents = text_splitter.split_documents(raw_documents)

# 指定距离策略为余弦相似度
db = FAISS.from_documents(
    documents=documents, 
    embedding=embedding, 
    distance_strategy=DistanceStrategy.COSINE
)

db.save_local(folder_path='./vector/FAISS.db', index_name='cpm-index')

query = '生活就像巧克力'
docs = db.similarity_search(query)

2. 带分数值的相似性搜索

前面我们通过 similarity_search 做向量相似度查询，也可以通过 similarity_search_with_score 查询结果并返回相似度'分数'，'分数'越低相似度越高（取决于距离策略）。

docs = db.similarity_search_with_score(query='第一行')

# 返回格式为 [(Document, score), ...]
for doc, score in docs:
    print(f"Score: {score}, Content: {doc.page_content[:50]}...")

也可以使用 similarity_search_by_vector 搜索与给定嵌入向量类似的文档，它接受嵌入向量作为参数而不是字符串。这在需要将外部向量直接传入时使用。

embedding_vector = OpenAIEmbeddings().embed_query(query)
docs = db.similarity_search_by_vector(embedding_vector)

提示：similarity_search、similarity_search_with_score 的本质还是去调用 similarity_search_by_vector，内部会自动完成 query 到 vector 的转换。

3. 本地保存与加载 (Save & Load)

如果我们项目使用的是本地向量数据库，那我们就可以指定向量数据库的保存位置，将 FAISS 索引、docstore 和 index_to_docstore_id 保存到本地磁盘。

这样做的好处就是我们可以提前对源数据做向量化处理，后面需要用到的时候直接加载就行，无需再次向量化，节省计算资源和时间。通过 save_local 实现：

db = FAISS.from_documents(documents=documents, embedding=embedding)
db.save_local(folder_path='./vector/FAISS.db', index_name='cpm-index')

加载并使用，通过 load_local 实现：

db = FAISS.load_local(
    folder_path='./vector/FAISS.db', 
    embeddings=embedding, 
    index_name='cpm-index',
    allow_dangerous_deserialization=True  # 生产环境请谨慎使用此参数
)
docs = db.similarity_search(query='狮子王', k=1)

4. 向量库合并 (Vector Merge)

有时候我们对元数据做向量化的时候并不是一蹴而就，可能是有源数据的时候就向量化一次，这就可能会导致我们会保存很多 .faiss 文件和 .pkl 文件。这个时候，我们更希望将多个文件进行合并，通过 merge_from 实现：

db1 = FAISS.from_documents(documents=documents, embedding=embedding)
db2 = FAISS.from_documents(documents=documents, embedding=embedding)

# 将 db1 的数据合并到 db2 中
db2.merge_from(db1)

5. 异步操作 (Asynchronous Operations)

向量存储通常作为单独的服务运行，需要一些 IO 操作，因此它们可能会被异步调用。这带来了性能优势，因为您不必浪费时间等待外部服务的响应。如果您使用异步框架（如 FastAPI），这可能也很重要。

LangChain 支持对向量存储进行异步操作。所有方法都可以使用其异步对应项进行调用，前缀为 a，意思是 async。

Qdrant 是一个在线向量数据库，它支持所有异步操作，同时提供 API 和管理端，方便操作。

import asyncio

async def main():
    # 异步初始化 Qdrant
    db = await Qdrant.afrom_documents(
        documents, 
        embeddings, 
        "http://localhost:6333"
    )
    query = "What did the president say about Ketanji Brown Jackson"
    
    # 异步搜索
    docs = await db.asimilarity_search(query)
    
    # 或者使用向量搜索
    embedding_vector = embeddings.embed_query(query)
    docs = await db.asimilarity_search_by_vector(embedding_vector)
    return docs

# asyncio.run(main())

6. 最大边际相关性搜索 (MMR)

最大边际相关性搜索（Maximum Marginal Relevance）是一种在信息检索和推荐系统中用于挑选结果的策略，旨在平衡查询项与已有结果集的相关性以及结果之间的多样性。

简单来说，MMR 不仅仅考虑单个结果与查询的相似度，还力求所选集合中的文档彼此不那么相似，以此来提高信息覆盖的全面性和减少重复信息。

MMR 通过结合两个指标来选择下一个最佳文档：

• 相关性（Relevance）：衡量文档与查询的直接相关程度，通常通过相似度分数表示。
• 多样性和新颖性（Marginality）：衡量新文档相对于已选择集合的新颖程度，鼓励选择能够增加信息多样性的文档。

query = "What did the president say about Ketanji Brown Jackson"
# k: 返回数量，fetch_k: 初始检索数量，lambda_mult: 多样性权重 (0~1)
found_docs = await qdrant.amax_marginal_relevance_search(query, k=2, fetch_k=10, lambda_mult=0.5)

03 相似度计算策略

LangChain 中有一个 DistanceStrategy 枚举类，里面定义了开发者可以选择的相似度计算方式。选择合适的距离度量对于检索效果至关重要。

from langchain_community.vectorstores.utils import DistanceStrategy

class DistanceStrategy(str, Enum):
    EUCLIDEAN_DISTANCE = "EUCLIDEAN_DISTANCE"
    MAX_INNER_PRODUCT = "MAX_INNER_PRODUCT"
    DOT_PRODUCT = "DOT_PRODUCT"
    JACCARD = "JACCARD"
    COSINE = "COSINE"

下面是每个参数的简要说明及适用场景：

• EUCLIDEAN_DISTANCE（欧氏距离）: 衡量两个点之间直线距离的方法。适用于连续数值特征，如图像识别、语音识别、聚类算法等。
• MAX_INNER_PRODUCT（最大内积）: 指两个向量的元素对应相乘后的结果的最大值。在某些情境下，这可以被看作是一种相似度度量，尤其是在二进制特征向量中寻找匹配项时。适用于推荐系统中的用户 - 物品匹配。
• DOT_PRODUCT（点积）: 也称为标量积，是两个向量的对应元素相乘后求和的结果。点积的大小可以反映两个向量方向上的相关性。适用于计算向量之间的相似度、机器学习中的权重更新规则等。
• JACCARD（杰卡德相似系数）: 用于比较有限样本集之间的相似性和差异性，定义为两个集合交集的元素个数除以并集的元素个数。适用于文本分类、推荐系统的用户兴趣相似度计算、生物信息学中的基因序列比较等。
• COSINE（余弦相似度）: 通过计算两个非零向量的夹角余弦值来评估它们的方向性相似度，不受向量长度的影响，仅关注方向。这是 NLP 任务中最常用的指标，适用于文档分类、文本挖掘、搜索引擎中的文档排名等。

这些参数的选择取决于具体的应用场景和需求，例如计算相似度的目的、数据的特性（如稀疏性、维度）等。如下所示配置：

# 省略部分代码...
db = FAISS.from_documents(
    documents=documents, 
    embedding=embedding, 
    distance_strategy=DistanceStrategy.COSINE
)

04 最佳实践与选型建议

在实际工程中，除了基础用法，还需要注意以下最佳实践：

1. 文本切分策略

默认的 CharacterTextSplitter 按字符数切分，但在处理代码或结构化文本时可能破坏语义。建议使用 RecursiveCharacterTextSplitter，它可以按不同层级的分隔符（如段落、句子、单词）递归切分，更好地保留上下文。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)
documents = splitter.split_documents(raw_documents)

2. 元数据过滤

在检索时，可以通过元数据过滤缩小范围。例如，只搜索特定年份或类型的文档。

from langchain.schema import Document

# 假设文档带有 metadata 字段
docs = db.similarity_search(
    query="金融政策",
    filter={"source": "./policy.txt"}  # 仅搜索该来源的文档
)

3. 向量库选型对比

4. 安全性与序列化

在使用 load_local 时，allow_dangerous_deserialization=True 允许加载任意 Python 对象，存在安全风险。在生产环境中，应确保加载的文件来源可信，或采用更安全的数据持久化方案。

05 总结

为什么我们在大模型应用开发中需要 Vector Store？这里主要是 2 点原因：

1. 效率优化：我们不能每次处理源数据的时候都去做一次 embedding 获取向量，这样不合理且效率不高。通过预构建索引，可以实现毫秒级检索。
2. 准确性提升：我们通过向量数据库做相似度搜索，将相关数据给到 LLM，让它做输出会更加精确，有效减少幻觉，实现基于私有知识的问答。

掌握 Vector Stores 的使用是构建高质量 RAG 应用的基础。开发者应根据业务规模、延迟要求和预算，选择合适的向量存储后端，并结合文本切分、元数据过滤等技巧优化检索效果。