Vector Stores 向量存储
向量存储（Vector Store）是 LangChain 中用于存储和检索向量嵌入的组件。它将文档的嵌入向量与原始内容一起存储，并提供高效的相似性搜索功能，是 RAG 系统的核心基础设施。
概述
向量存储的工作流程：
文本 → 嵌入模型 → 向量 → 存入向量存储
查询 → 嵌入模型 → 查询向量 → 相似性搜索 → 返回相似文档python
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 创建向量存储
vector_store = InMemoryVectorStore(embeddings)常用向量存储
InMemoryVectorStore（内存存储）
内存向量存储，适用于原型开发和小数据集：
python
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = InMemoryVectorStore(embeddings)
# 添加文档
vector_store.add_texts(
texts=["LangChain 是一个 LLM 框架", "文档加载器用于加载数据"],
metadatas=[{"source": "doc1"}, {"source": "doc2"}]
)
# 相似性搜索
results = vector_store.similarity_search("LangChain 是什么", k=2)
for doc in results:
print(doc.page_content, doc.metadata)Chroma
Chroma 是一个开源向量数据库，适合中小规模应用：
python
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 创建 Chroma 向量存储
vector_store = Chroma(
collection_name="my_collection",
embedding_function=embeddings,
persist_directory="./chroma_db" # 持久化目录
)
# 添加文档
vector_store.add_texts(
texts=["文档内容一", "文档内容二"],
metadatas=[{"chapter": 1}, {"chapter": 2}]
)
# 搜索
results = vector_store.similarity_search("查询内容", k=3)
# 持久化存储
vector_store.persist()FAISS
FAISS（Facebook AI Similarity Search）是 Meta 开发的高效相似性搜索库：
python
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 从文本创建
vector_store = FAISS.from_texts(
texts=["文档一", "文档二", "文档三"],
embedding=embeddings,
metadatas=[{"id": 1}, {"id": 2}, {"id": 3}]
)
# 保存到本地
vector_store.save_local("faiss_index")
# 从本地加载
loaded_store = FAISS.load_local(
"faiss_index",
embeddings,
allow_dangerous_deserialization=True
)Pinecone（云向量数据库）
适用于生产级大规模应用：
python
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = PineconeVectorStore(
index_name="my-index",
embedding=embeddings,
pinecone_api_key="YOUR_API_KEY"
)
# 添加文档
vector_store.add_texts(
texts=["云上存储的文档"],
metadatas={"project": "demo"}
)Milvus
python
from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vector_store = Milvus(
embedding_function=embeddings,
collection_name="LangChainCollection",
connection_args={"host": "localhost", "port": "19530"}
)Qdrant
python
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
vector_store = QdrantVectorStore(
url="http://localhost:6333",
collection_name="my_collection",
embedding=embeddings
)核心操作
添加文档
python
# 方式一：添加文本
vector_store.add_texts(
texts=["文档一", "文档二"],
ids=["id1", "id2"],
metadatas=[{"source": "a"}, {"source": "b"}]
)
# 方式二：从 Document 对象创建
from langchain_core.documents import Document
docs = [
Document(page_content="内容一", metadata={"id": 1}),
Document(page_content="内容二", metadata={"id": 2}),
]
vector_store.add_documents(docs)相似性搜索
python
# 基本相似性搜索
docs = vector_store.similarity_search("查询文本", k=5)
# 带分数的相似性搜索
docs_with_scores = vector_store.similarity_search_with_score(
"查询文本", k=5
)
for doc, score in docs_with_scores:
print(f"分数: {score:.4f}, 内容: {doc.page_content[:50]}")
# 向量相似性搜索（直接使用向量）
query_vector = embeddings.embed_query("查询文本")
docs = vector_store.similarity_search_by_vector(query_vector, k=5)过滤（元数据过滤）
python
# Chroma 过滤
results = vector_store.similarity_search(
"查询",
k=3,
filter={"source": "技术文档"}
)
# FAISS 过滤
results = vector_store.similarity_search(
"查询",
k=3,
filter=lambda meta: meta["chapter"] > 2
)删除文档
python
vector_store.delete(ids=["id1", "id2"])完整示例：文档索引与检索
python
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
# 1. 加载文档
loader = DirectoryLoader(
"./docs/",
glob="**/*.txt",
loader_cls=TextLoader,
show_progress=True
)
documents = loader.load()
# 2. 分割文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)
# 3. 初始化和构建向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(chunks, embeddings)
# 4. 保存索引
vector_store.save_local("docs_index")
# 5. 检索
query = "什么是 RecursiveCharacterTextSplitter"
results = vector_store.similarity_search_with_score(query, k=3)
print(f"查询: {query}\n")
for i, (doc, score) in enumerate(results):
print(f"结果 {i+1} (相似度: {score:.4f}):")
print(f" 内容: {doc.page_content[:100]}...")
print(f" 来源: {doc.metadata.get('source', 'unknown')}")
print()向量存储选择指南
| 存储 | 场景 | 特点 |
|---|---|---|
| InMemoryVectorStore | 原型/小数据集 | 无需外部服务，速度快 |
| Chroma | 中小型项目 | 开源，本地持久化 |
| FAISS | 单机高效检索 | Meta 出品，性能优秀 |
| Pinecone | 生产级大规模 | 云服务，托管运维 |
| Milvus | 企业级大规模 | 分布式，功能丰富 |
| Qdrant | 生产级部署 | 性能好，Rust 实现 |
| Weaviate | 云原生 | 支持混合搜索 |
最佳实践
- 选择合适的向量存储：根据数据规模和预算选择合适的方案
- 元数据过滤：善用元数据过滤缩小搜索范围
- 持久化：生产环境务必持久化索引，避免重复构建
- 批量操作：大批量添加文档时使用批量 API
- 异步支持：使用异步版本提高并发性能
python
# 异步操作
await vector_store.asimilarity_search("异步查询", k=5)
await vector_store.aadd_documents(docs)