向量数据库

向量数据库是现代 AI 应用特别是 RAG 系统的核心组件，专门用于存储文档的向量表示并支持高效的语义检索。本章将深入介绍向量数据库的原理、选择策略和高级用法。

什么是向量数据库？

向量数据库 是一种专门设计用于存储、索引和查询高维向量的数据库系统。与传统数据库基于精确匹配不同，向量数据库通过计算向量间的相似度来检索数据，这使得语义层面的搜索成为可能。

为什么需要向量数据库？

考虑一个场景：用户问"如何处理数据库连接超时"，传统关键词搜索会尝试匹配"数据库"、"连接"、"超时"这些词，但如果文档中写的是"DB连接失败"或"connection timeout"，关键词搜索就会漏掉这些相关内容。而向量数据库通过语义理解，能够找到所有与"数据库连接问题"相关的文档，无论具体用词如何。

核心优势包括：

• 语义检索：基于内容含义而非字面匹配，用户用不同表述方式提问也能找到相关内容
• 高效相似度计算：使用专门的索引结构（如 HNSW、IVF）加速向量检索，支持亿级向量毫秒级响应
• 多模态支持：不仅能处理文本，还能处理图像、音频等的向量表示

向量相似度计算

向量数据库通过计算向量间的距离或相似度来衡量文档的相关性。常用的度量方法有：

余弦相似度（Cosine Similarity）

最常用的度量方式，计算两个向量夹角的余弦值，取值范围 [-1, 1]，值越大表示越相似：

similarity = cos(θ) = (A · B) / (||A|| × ||B||)

余弦相似度只关注向量方向，不受向量长度影响，适合文本嵌入场景。

欧氏距离（Euclidean Distance）

计算向量间的直线距离，值越小表示越相似：

distance = ||A - B|| = sqrt(Σ(a_i - b_i)²)

点积（Dot Product）

计算向量的内积，在某些模型中与相似度正相关：

dot_product = A · B = Σ(a_i × b_i)

选择度量方式时，应与嵌入模型的训练方式保持一致。OpenAI 的 text-embedding 系列使用余弦相似度。

常用向量数据库选型

选择向量数据库需要综合考虑部署方式、性能、功能和生态支持。以下是主流选择：

1. Chroma

定位：轻量级本地向量数据库，适合开发测试和小规模应用

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# 从文档创建
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="my_collection"
)

# 持久化到磁盘
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 指定存储路径
)

# 从磁盘加载
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_collection"
)

# 检索
results = vectorstore.similarity_search("查询内容", k=3)

Chroma 特点：

• 零配置启动，无需额外服务
• 支持元数据过滤
• 内置嵌入计算（可选）
• 适合开发原型和小规模生产

2. FAISS

定位：Facebook 开源的高性能向量检索库，适合本地高性能场景

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

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

# 保存到磁盘
vectorstore.save_local("faiss_index")

# 加载（需要 allow_dangerous_deserialization=True）
vectorstore = FAISS.load_local(
    "faiss_index",
    embeddings,
    allow_dangerous_deserialization=True
)

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

FAISS 特点：

• 极致的检索性能
• 支持多种索引类型（Flat、IVF、HNSW）
• 纯内存操作，重启后需重新加载
• 适合对延迟敏感的场景

3. Pinecone

定位：全托管云向量数据库，适合生产环境大规模部署

from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings
from pinecone import Pinecone

# 初始化 Pinecone 客户端
pc = Pinecone(api_key="your-api-key")

# 创建向量存储
vectorstore = PineconeVectorStore.from_documents(
    documents=documents,
    index_name="my-index",
    embedding=embeddings
)

# 或连接现有索引
vectorstore = PineconeVectorStore(
    index_name="my-index",
    embedding=embeddings
)

Pinecone 特点：

• 完全托管，无需运维
• 自动扩展，支持十亿级向量
• 内置监控和告警
• 按使用量计费

4. Qdrant

定位：高性能开源向量数据库，支持云部署和自托管

from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
from qdrant_client import QdrantClient

# 内存模式（开发测试）
client = QdrantClient(":memory:")

# 连接远程服务
client = QdrantClient(url="http://localhost:6333")

# 创建向量存储
vectorstore = QdrantVectorStore.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_name="my_collection",
    client=client
)

Qdrant 特点：

• Rust 实现，性能优异
• 支持复杂的元数据过滤
• 提供丰富的过滤表达式
• 支持量化压缩降低存储成本

5. Milvus

定位：国产开源向量数据库，企业级功能完善

from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings

# 连接 Milvus
vectorstore = Milvus.from_documents(
    documents=documents,
    embedding=embeddings,
    connection_args={
        "host": "localhost",
        "port": "19530"
    },
    collection_name="my_collection"
)

Milvus 特点：

• 支持多种索引类型
• 分布式架构，水平扩展
• 企业级稳定性和可靠性
• 国产化项目首选

选型建议

场景	推荐	原因
开发测试	Chroma / FAISS	零配置、轻量级
小规模生产	Chroma + 持久化	部署简单、成本低
大规模生产	Pinecone / Qdrant / Milvus	高可用、可扩展
国产化要求	Milvus	完整国内支持
极致性能	FAISS	最快的检索速度

嵌入模型

向量数据库的效果很大程度上取决于嵌入模型的质量。嵌入模型将文本转换为固定维度的向量，语义相似的文本其向量也相近。

OpenAI Embeddings

OpenAI 提供的嵌入模型，性能稳定、易用：

from langchain_openai import OpenAIEmbeddings

# 使用最新模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 大模型（更高维度，更好效果）
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 嵌入单个文本
vector = embeddings.embed_query("Python 是一种编程语言")
print(f"维度: {len(vector)}")  # 1536 或 3072

# 批量嵌入
vectors = embeddings.embed_documents([
    "第一段文本",
    "第二段文本"
])

模型对比：

模型	维度	价格（/1M tokens）	特点
text-embedding-3-small	1536	$0.02	性价比高，适合大多数场景
text-embedding-3-large	3072	$0.13	更高精度，适合对召回率要求高的场景

本地嵌入模型

使用 HuggingFace 的开源模型，无需 API 调用：

from langchain_huggingface import HuggingFaceEmbeddings

# 使用多语言模型
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# 使用中文优化模型
embeddings = HuggingFaceEmbeddings(
    model_name="shibing624/text2vec-base-chinese"
)

本地模型的优点是完全离线、无 API 成本，缺点是需要本地计算资源。

其他嵌入模型

# Google Generative AI
from langchain_google_genai import GoogleGenerativeAIEmbeddings
embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")

# Cohere
from langchain_cohere import CohereEmbeddings
embeddings = CohereEmbeddings(model="embed-english-v3.0")

# Azure OpenAI
from langchain_openai import AzureOpenAIEmbeddings
embeddings = AzureOpenAIEmbeddings(
    azure_deployment="my-deployment",
    azure_endpoint="https://xxx.openai.azure.com/"
)

检索策略详解

1. 基础相似度检索

最简单的检索方式，返回与查询最相似的 k 个文档：

# 返回最相似的 3 个文档
results = vectorstore.similarity_search("查询内容", k=3)

for doc in results:
    print(f"内容: {doc.page_content[:100]}")
    print(f"元数据: {doc.metadata}")
    print("---")

2. 带分数的相似度检索

获取相似度分数，用于排序或过滤：

# 返回文档和相似度分数
results = vectorstore.similarity_search_with_score("查询内容", k=3)

for doc, score in results:
    print(f"分数: {score:.4f}")  # 分数越高越相似
    print(f"内容: {doc.page_content[:100]}")
    print("---")

3. 阈值检索

只返回相似度超过阈值的文档：

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 10,              # 最多返回 10 个
        "score_threshold": 0.8  # 相似度阈值
    }
)

# 如果没有文档超过阈值，返回空列表
docs = retriever.invoke("查询内容")

阈值检索适合对答案质量要求高的场景，避免返回不相关内容。

4. MMR 多样性检索

最大边际相关性（MMR）在相似性和多样性之间取得平衡，避免返回内容高度重复的文档：

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,              # 最终返回数量
        "fetch_k": 20,       # 候选文档数量
        "lambda_mult": 0.5   # 多样性权重，0 最多样，1 最相似
    }
)

docs = retriever.invoke("查询内容")

MMR 原理：

传统检索只考虑文档与查询的相似度，可能导致返回的文档内容高度重复。MMR 在选择文档时同时考虑：

1. 与查询的相似度
2. 与已选文档的差异度

通过 lambda_mult 参数控制两者的平衡。值为 0.5 时，同等重视相似性和多样性。

5. 元数据过滤

基于文档元数据进行过滤：

# 在检索时过滤
results = vectorstore.similarity_search(
    "查询内容",
    k=3,
    filter={"source": "product_manual"}  # 只检索特定来源
)

# 使用检索器配置
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {
            "category": "技术文档",
            "year": {"$gte": 2023}
        }
    }
)

元数据过滤让检索更加精确，适合多租户、多类别的场景。

检索器详解

检索器（Retriever）是 LangChain 中检索文档的统一接口。所有向量存储都可以转换为检索器：

# 创建检索器
retriever = vectorstore.as_retriever()

# 检索器是一个 Runnable，可以在 LCEL 链中使用
docs = retriever.invoke("查询内容")

检索器的优势

检索器相比直接调用向量存储的方法有几个好处：

1. 统一的 Runnable 接口：可以在 LCEL 链中组合使用
2. 配置化：通过参数配置检索行为，代码更清晰
3. 可替换性：不同检索器实现相同接口，易于切换

自定义检索器

可以实现自定义检索器，结合多种检索策略：

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List

class EnsembleRetriever(BaseRetriever):
    """组合多个检索器的结果"""
    
    def __init__(self, retrievers, weights=None):
        super().__init__()
        self.retrievers = retrievers
        self.weights = weights or [1/len(retrievers)] * len(retrievers)
    
    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 从每个检索器获取结果
        all_docs = []
        for retriever, weight in zip(self.retrievers, self.weights):
            docs = retriever.invoke(query)
            for doc in docs:
                # 加权处理
                doc.metadata["score"] = doc.metadata.get("score", 1.0) * weight
                all_docs.append(doc)
        
        # 去重并排序
        seen = set()
        unique_docs = []
        for doc in sorted(all_docs, key=lambda x: x.metadata["score"], reverse=True):
            if doc.page_content not in seen:
                seen.add(doc.page_content)
                unique_docs.append(doc)
        
        return unique_docs[:5]

# 使用
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

高级功能

1. 混合检索

结合关键词检索和向量检索的优势：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 创建 BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 创建向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 组合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 向量检索权重更高
)

docs = ensemble_retriever.invoke("查询内容")

为什么要混合检索？

向量检索擅长语义理解，但可能漏掉精确匹配的内容。比如查询"Python 3.11 新特性"，向量检索可能找到所有 Python 相关文档，而关键词检索能精确定位"3.11"版本的内容。混合检索取长补短，提高召回率。

2. 上下文压缩

使用 LLM 对检索结果进行压缩，只保留与查询相关的部分：

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")

# 创建压缩器
compressor = LLMChainExtractor.from_llm(model)

# 创建压缩检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever()
)

# 检索结果会被压缩，只保留相关内容
docs = compression_retriever.invoke("查询内容")

压缩后的文档内容更精简，减少噪音信息，提高 LLM 回答质量。

3. 重排序

先粗检索大量候选，再用精排模型重新排序：

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 加载重排序模型
reranker = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")

# 创建重排序压缩器
compressor = CrossEncoderReranker(
    model=reranker,
    top_n=5  # 最终返回数量
)

# 创建重排序检索器
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})  # 先召回更多
)

重排序模型专门训练用于判断文档与查询的相关性，比向量相似度更准确，但计算成本更高。典型做法是先向量检索召回 50-100 个候选，再用重排序模型选出 top 5。

4. 嵌入缓存

避免重复计算嵌入，节省成本和时间：

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import InMemoryByteStore

# 创建缓存层
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=embeddings,
    document_embedding_cache=InMemoryByteStore(),
    namespace="my_namespace"  # 不同命名空间独立缓存
)

# 相同文本只会计算一次嵌入
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=cached_embeddings
)

实践建议

文档分块策略

分块质量直接影响检索效果：

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 块大小：根据嵌入模型限制设置
    chunk_overlap=50,    # 重叠：保证上下文连续性
    length_function=len,
    separators=["\n\n", "\n", "。", ".", " ", ""]  # 分割优先级
)

# 块太大：检索不够精确
# 块太小：上下文不完整
# 建议范围：300-1000 字符

监控检索质量

在生产环境中监控检索效果：

def retrieve_with_metrics(query: str, retriever, k: int = 5):
    """带指标监控的检索"""
    import time
    
    start = time.time()
    docs = retriever.invoke(query)
    latency = time.time() - start
    
    # 记录指标
    metrics = {
        "query": query,
        "latency_ms": latency * 1000,
        "num_results": len(docs),
        "avg_doc_length": sum(len(d.page_content) for d in docs) / len(docs) if docs else 0
    }
    
    # 发送到监控系统
    # log_metrics(metrics)
    
    return docs, metrics

处理空结果

优雅处理检索无结果的情况：

def safe_retrieve(query: str, retriever, min_score: float = 0.7):
    """安全的检索，处理无结果情况"""
    try:
        docs = retriever.invoke(query)
        
        if not docs:
            return None, "未找到相关文档"
        
        # 如果有分数，过滤低分结果
        if hasattr(docs[0], "metadata") and "score" in docs[0].metadata:
            docs = [d for d in docs if d.metadata["score"] >= min_score]
            
            if not docs:
                return None, "相关文档的置信度太低"
        
        return docs, None
    
    except Exception as e:
        return None, f"检索出错: {str(e)}"

常见问题

1. 检索结果不相关

可能原因和解决方案：

• 嵌入模型不匹配：确保使用与文档语言匹配的嵌入模型
• 分块太大：减小 chunk_size，提高检索粒度
• 查询太短：使用查询扩展或重写技术

2. 检索速度慢

• 使用索引（HNSW、IVF）加速
• 减少检索数量 k
• 使用量化压缩向量
• 考虑分布式部署

3. 内存占用高

• 使用磁盘存储的向量库（Chroma、Milvus）
• 启用向量量化
• 分批处理大规模文档

下一步

现在你已经掌握了向量数据库的核心概念和高级用法，接下来学习：

• RAG 检索增强 - 构建完整的检索增强生成系统
• Tools 工具 - 为代理添加工具能力
• 回调与监控 - LangSmith 调试和监控