LangChain 进阶:Vector Stores 向量存储详解
01 介绍
在构建基于大语言模型(LLM)的应用时,Vector Stores(向量存储)是连接语言模型与实际应用数据的关键桥梁。它为构建智能化、高效的语言处理应用提供了强大的基础设施支持,特别是在检索增强生成(RAG)架构中扮演着核心角色。
理解向量存储前需了解向量嵌入(Embeddings)。文本向量化是将非结构化的文本转换成数值向量的过程,这些向量能够在多维空间中捕捉词语或文档之间的语义相似性。常见的嵌入模型如 Word2Vec、BERT、Sentence Transformers 等,可以被用来生成这样的向量。
Vector Stores 的核心用途
- 高效检索:Vector Stores 主要用于存储这些嵌入向量,并支持高效的相似性搜索,使得用户能够根据输入的查询(也是一个向量)快速找到最相关的文档、段落或信息。
- 语义搜索:由于向量能够表达语义关系,Vector Stores 使得基于内容而非关键词的搜索成为可能,大大提高了搜索的准确性和相关性,解决了传统关键词匹配无法理解上下文的问题。
Vector Stores 实现方式
LangChain 支持多种向量存储后端,开发者可以根据项目需求选择本地部署或云服务:
- FAISS:Facebook AI Similarity Search,一个高效的相似性搜索库,特别适合大规模的向量数据集,常用于本地开发测试。
- Pinecone、Qdrant:云原生的向量数据库服务,提供了 API 接口,便于管理和检索向量数据,适合生产环境的高可用需求。
- Weaviate:一个语义搜索引擎,支持向量搜索和知识图谱管理,具备较强的扩展性。
- Chroma:一个开源的向量数据库,专为机器学习和 NLP 应用设计,轻量级且易于集成。
Vector Stores 功能特性
- 索引构建:可以为文档集合创建索引,这个过程涉及将文档转换为向量并存储起来,以便后续快速检索。
- 更新与删除:支持对向量数据的动态管理,包括源文档更新时的向量重计算以及删除不再需要的向量。
- 检索优化:通过近似最近邻(Approximate Nearest Neighbor, ANN)算法,在保证较高精度的同时,实现了对大规模数据集的高效检索。
Vector Stores 应用场景
- 问答系统:快速从大量文档中找到与问题最相关的答案,减少模型幻觉。
- 个性化推荐:基于用户历史行为和偏好生成的向量,来推荐相似或相关的内容。
- 知识图谱增强:结合向量搜索提高知识图谱节点间链接的发现和查询效率。
- 文档检索系统:企业内部文档、网页内容的快速语义搜索。
LangChain 提供了统一的 API 接口来与不同的 Vector Stores 交互,使得开发者无需深入了解每个后端的具体实现细节,即可轻松集成和切换向量存储解决方案,提升了开发效率和灵活性。
02 Vector Store 使用指南
LangChain 中向量数据库的使用基本遵循四个标准步骤:加载文档、切分文本、生成嵌入、构建索引。
1. 相似性搜索 (Similarity Search)
如果我们使用的 OpenAI 相关的模型,我们可以这么使用:
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
embeddings = OpenAIEmbeddings(openai_api_key='xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx')
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=embeddings)
query = '生活就像巧克力'
docs = db.similarity_search(query)
或者我们也可替换成本地的词嵌入模型,这里我经常用(项目用)的词嵌入模型是 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='第一行')
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)
db2.merge_from(db1)
5. 异步操作 (Asynchronous Operations)
向量存储通常作为单独的服务运行,需要一些 IO 操作,因此它们可能会被异步调用。这带来了性能优势,因为您不必浪费时间等待外部服务的响应。如果您使用异步框架(如 FastAPI),这可能也很重要。
LangChain 支持对向量存储进行异步操作。所有方法都可以使用其异步对应项进行调用,前缀为 a,意思是 async。
Qdrant 是一个在线向量数据库,它支持所有异步操作,同时提供 API 和管理端,方便操作。
import asyncio
async def main():
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
6. 最大边际相关性搜索 (MMR)
最大边际相关性搜索(Maximum Marginal Relevance)是一种在信息检索和推荐系统中用于挑选结果的策略,旨在平衡查询项与已有结果集的相关性以及结果之间的多样性。
简单来说,MMR 不仅仅考虑单个结果与查询的相似度,还力求所选集合中的文档彼此不那么相似,以此来提高信息覆盖的全面性和减少重复信息。
MMR 通过结合两个指标来选择下一个最佳文档:
- 相关性(Relevance):衡量文档与查询的直接相关程度,通常通过相似度分数表示。
- 多样性和新颖性(Marginality):衡量新文档相对于已选择集合的新颖程度,鼓励选择能够增加信息多样性的文档。
query = "What did the president say about Ketanji Brown Jackson"
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
docs = db.similarity_search(
query="金融政策",
filter={"source": "./policy.txt"}
)
3. 向量库选型对比
| 向量库 | 类型 | 优点 | 缺点 | 适用场景 |
|---|
| FAISS | 本地库 | 速度快,内存占用低,免费 | 不支持持久化存储,无服务端 | 原型开发,本地测试 |
| Qdrant | 云服务/本地 | 支持过滤,高性能,REST API | 需要部署维护 | 生产环境,复杂查询 |
| Chroma | 本地/云 | 易用,Python 友好,轻量 | 大规模并发能力较弱 | 小型项目,快速验证 |
| Pinecone | 云服务 | 托管服务,免运维,高可用 | 成本较高,闭源 | 企业级应用,高 SLA 要求 |
4. 安全性与序列化
在使用 load_local 时,allow_dangerous_deserialization=True 允许加载任意 Python 对象,存在安全风险。在生产环境中,应确保加载的文件来源可信,或采用更安全的数据持久化方案。
05 总结
为什么我们在大模型应用开发中需要 Vector Store?这里主要是 2 点原因:
- 效率优化:我们不能每次处理源数据的时候都去做一次 embedding 获取向量,这样不合理且效率不高。通过预构建索引,可以实现毫秒级检索。
- 准确性提升:我们通过向量数据库做相似度搜索,将相关数据给到 LLM,让它做输出会更加精确,有效减少幻觉,实现基于私有知识的问答。
掌握 Vector Stores 的使用是构建高质量 RAG 应用的基础。开发者应根据业务规模、延迟要求和预算,选择合适的向量存储后端,并结合文本切分、元数据过滤等技巧优化检索效果。