向量存储（Vector Stores） #

向量存储是 RAG 应用的核心组件，负责存储文档的向量表示并支持高效的相似度检索。LangChain 支持多种向量数据库，满足不同场景的需求。

向量存储概述 #

text 
┌─────────────────────────────────────────────────────────────┐
│                    向量存储的作用                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  存储：                                                     │
│  文档 ───> 向量化 ───> 存储到向量数据库                      │
│                                                             │
│  检索：                                                     │
│  查询 ───> 向量化 ───> 相似度搜索 ───> 返回相关文档          │
│                                                             │
│  核心功能：                                                  │
│  ✅ 高效存储大量向量                                        │
│  ✅ 快速相似度搜索                                          │
│  ✅ 支持元数据过滤                                          │
│  ✅ 支持增删改查                                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

基本使用 #

创建向量存储 #

python 
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# 创建嵌入模型
embeddings = OpenAIEmbeddings()

# 方式一：从文档列表创建
documents = [
    Document(page_content="机器学习是AI的一个分支", metadata={"source": "doc1"}),
    Document(page_content="深度学习使用神经网络", metadata={"source": "doc2"}),
]

vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 方式二：从文本列表创建
texts = ["机器学习是AI的一个分支", "深度学习使用神经网络"]
metadatas = [{"source": "doc1"}, {"source": "doc2"}]

vectorstore = Chroma.from_texts(
    texts=texts,
    embedding=embeddings,
    metadatas=metadatas,
    persist_directory="./chroma_db"
)

相似度搜索 #

python 
# 基础搜索
results = vectorstore.similarity_search(
    query="什么是AI技术",
    k=4  # 返回最相似的 4 个文档
)

for doc in results:
    print(doc.page_content)
    print(doc.metadata)
    print("-" * 50)

# 带分数的搜索
results_with_scores = vectorstore.similarity_search_with_score(
    query="什么是AI技术",
    k=4
)

for doc, score in results_with_scores:
    print(f"Score: {score:.4f}")
    print(doc.page_content)
    print("-" * 50)

添加和删除文档 #

python 
# 添加文档
vectorstore.add_documents([
    Document(page_content="新文档内容", metadata={"source": "new"})
])

# 添加文本
vectorstore.add_texts(
    texts=["另一个新文档"],
    metadatas=[{"source": "another"}]
)

# 删除文档
vectorstore.delete(ids=["doc-id-1", "doc-id-2"])

常用向量数据库 #

FAISS（本地） #

Facebook 开源的高效向量搜索库：

python 
from langchain_community.vectorstores import FAISS

# 创建
vectorstore = FAISS.from_documents(
    documents=documents,
    embedding=embeddings
)

# 保存到本地
vectorstore.save_local("./faiss_index")

# 从本地加载
vectorstore = FAISS.load_local(
    "./faiss_index",
    embeddings=embeddings,
    allow_dangerous_deserialization=True
)

# 合并多个索引
vectorstore1.merge_from(vectorstore2)

Chroma（本地） #

功能丰富的本地向量数据库：

python 
from langchain_community.vectorstores import Chroma

# 创建（持久化）
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="my_collection"
)

# 加载已有数据库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_collection"
)

# 内存模式
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings
)

Pinecone（云端） #

托管式向量数据库：

python 
from langchain_community.vectorstores import Pinecone
import pinecone

# 初始化 Pinecone
pinecone.init(
    api_key="your-api-key",
    environment="your-environment"
)

# 创建索引
index_name = "my-index"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine"
    )

# 创建向量存储
vectorstore = Pinecone.from_documents(
    documents=documents,
    embedding=embeddings,
    index_name=index_name
)

# 连接已有索引
vectorstore = Pinecone.from_existing_index(
    index_name=index_name,
    embedding=embeddings
)

Weaviate #

开源向量搜索引擎：

python 
from langchain_community.vectorstores import Weaviate
import weaviate

# 连接 Weaviate
client = weaviate.Client("http://localhost:8080")

# 创建向量存储
vectorstore = Weaviate.from_documents(
    documents=documents,
    embedding=embeddings,
    client=client,
    index_name="MyCollection"
)

# 连接已有集合
vectorstore = Weaviate(
    client=client,
    index_name="MyCollection",
    text_key="text",
    embedding=embeddings
)

Milvus #

高性能分布式向量数据库：

python 
from langchain_community.vectorstores import Milvus

# 创建向量存储
vectorstore = Milvus.from_documents(
    documents=documents,
    embedding=embeddings,
    connection_args={
        "host": "localhost",
        "port": "19530"
    },
    collection_name="my_collection"
)

Qdrant #

高性能向量数据库：

python 
from langchain_community.vectorstores import Qdrant
from qdrant_client import QdrantClient

# 创建客户端
client = QdrantClient(url="http://localhost:6333")

# 创建向量存储
vectorstore = Qdrant.from_documents(
    documents=documents,
    embedding=embeddings,
    url="http://localhost:6333",
    collection_name="my_collection"
)
text 
┌─────────────────────────────────────────────────────────────┐
│                    向量数据库对比                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  数据库     │ 类型   │ 特点           │ 适用场景            │
│  ─────────────────────────────────────────────────────────  │
│  FAISS      │ 本地   │ 快速、轻量     │ 开发测试、小规模    │
│  Chroma     │ 本地   │ 功能丰富       │ 本地应用、开发      │
│  Pinecone   │ 云端   │ 托管、易用     │ 生产环境、无运维    │
│  Weaviate   │ 开源   │ 功能全面       │ 自托管、企业级      │
│  Milvus     │ 开源   │ 高性能、分布式 │ 大规模、高并发      │
│  Qdrant     │ 开源   │ 高效、Rust     │ 高性能需求          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

高级功能 #

元数据过滤 #

python 
# Chroma 元数据过滤
results = vectorstore.similarity_search(
    query="机器学习",
    k=4,
    filter={
        "source": "doc1"
    }
)

# 复杂过滤
results = vectorstore.similarity_search(
    query="机器学习",
    k=4,
    filter={
        "$and": [
            {"source": {"$eq": "doc1"}},
            {"year": {"$gte": 2023}}
        ]
    }
)

# Pinecone 元数据过滤
results = vectorstore.similarity_search(
    query="机器学习",
    k=4,
    filter={
        "category": "tech",
        "year": {"$gte": 2023}
    }
)

MMR 搜索 #

最大边际相关性搜索，提高结果多样性：

python 
# MMR 搜索
results = vectorstore.max_marginal_relevance_search(
    query="机器学习",
    k=4,          # 返回结果数
    fetch_k=20,   # 候选结果数
    lambda_mult=0.5  # 多样性权重
)

# 带分数的 MMR 搜索
results = vectorstore.max_marginal_relevance_search_with_score(
    query="机器学习",
    k=4,
    fetch_k=20
)
text 
┌─────────────────────────────────────────────────────────────┐
│                    MMR vs 普通搜索                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  普通相似度搜索：                                            │
│  查询: "机器学习"                                           │
│  结果:                                                      │
│  1. 机器学习是AI分支 (0.95)                                 │
│  2. 机器学习算法介绍 (0.94)  ← 内容相似                     │
│  3. 机器学习应用场景 (0.93)  ← 内容相似                     │
│                                                             │
│  MMR 搜索：                                                 │
│  查询: "机器学习"                                           │
│  结果:                                                      │
│  1. 机器学习是AI分支 (0.95)                                 │
│  2. 深度学习神经网络 (0.82)  ← 内容多样                     │
│  3. 自然语言处理技术 (0.78)  ← 内容多样                     │
│                                                             │
│  MMR 在保持相关性的同时增加结果多样性                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

创建检索器 #

python 
# 基础检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# MMR 检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20}
)

# 分数阈值检索器
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8, "k": 4}
)

# 使用检索器
docs = retriever.invoke("什么是机器学习？")

管理操作 #

查看存储信息 #

python 
# Chroma
collection = vectorstore._collection
print(f"文档数量: {collection.count()}")

# FAISS
print(f"索引大小: {vectorstore.index.ntotal}")

# Pinecone
index = pinecone.Index("my-index")
stats = index.describe_index_stats()
print(f"向量数量: {stats.total_vector_count}")

批量操作 #

python 
# 批量添加
batch_documents = [
    Document(page_content=f"文档 {i}", metadata={"id": i})
    for i in range(1000)
]

vectorstore.add_documents(batch_documents)

# 分批处理大数据集
def batch_add_documents(documents, batch_size=100):
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        vectorstore.add_documents(batch)

更新文档 #

python 
# 删除旧文档
vectorstore.delete(ids=["old-doc-id"])

# 添加新文档
vectorstore.add_documents([
    Document(
        page_content="更新后的内容",
        metadata={"id": "new-doc-id"}
    )
])

完整示例 #

文档知识库 #

python 
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

class KnowledgeBase:
    """文档知识库"""
    
    def __init__(self, persist_directory: str):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma(
            persist_directory=persist_directory,
            embedding_function=self.embeddings
        )
    
    def add_documents(self, documents: list):
        """添加文档"""
        # 分割文档
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        splits = text_splitter.split_documents(documents)
        
        # 添加到向量存储
        self.vectorstore.add_documents(splits)
    
    def search(self, query: str, k: int = 4) -> list:
        """搜索文档"""
        return self.vectorstore.similarity_search(query, k=k)
    
    def search_with_scores(self, query: str, k: int = 4) -> list:
        """带分数的搜索"""
        return self.vectorstore.similarity_search_with_score(query, k=k)
    
    def as_retriever(self, **kwargs):
        """获取检索器"""
        return self.vectorstore.as_retriever(**kwargs)

# 使用
kb = KnowledgeBase("./knowledge_base")

# 添加文档
loader = DirectoryLoader("./documents", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()
kb.add_documents(documents)

# 搜索
results = kb.search("什么是机器学习？")
for doc in results:
    print(doc.page_content[:100])
    print("-" * 50)

性能优化 #

1. 选择合适的索引 #

python 
# FAISS - 选择索引类型
import faiss

# 平面索引（精确搜索，适合小数据集）
index = faiss.IndexFlatL2(1536)

# IVF 索引（近似搜索，适合大数据集）
nlist = 100  # 聚类中心数
quantizer = faiss.IndexFlatL2(1536)
index = faiss.IndexIVFFlat(quantizer, 1536, nlist)

2. 批量处理 #

python 
# 批量添加文档
def batch_add(vectorstore, documents, batch_size=100):
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        vectorstore.add_documents(batch)

3. 缓存嵌入 #

python 
from langchain.embeddings import CacheBackedEmbeddings
from langchain_community.storage import LocalFileStore

# 使用缓存的嵌入
store = LocalFileStore("./embedding_cache")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    OpenAIEmbeddings(),
    store
)

vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=cached_embeddings
)

最佳实践 #

1. 选择合适的向量数据库 #

python 
# 开发阶段 - 使用本地数据库
vectorstore = Chroma.from_documents(documents, embeddings)

# 生产环境 - 使用云端或分布式数据库
vectorstore = Pinecone.from_documents(documents, embeddings)

2. 合理设置参数 #

python 
# 分割参数
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,     # 根据模型上下文调整
    chunk_overlap=50    # 适当的重叠
)

# 检索参数
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 4}  # 根据需求调整
)

3. 持久化存储 #

python 
# 定期持久化
vectorstore.persist()

# 备份向量数据库
import shutil
shutil.copytree("./chroma_db", "./chroma_db_backup")

下一步 #

最后更新：2026-03-30