生产环境中的向量数据库：选择与优化指南

向量数据库为 RAG 系统的核心——相似性搜索——提供动力。选错数据库或配置不当，会直接影响 LLM 应用的质量和成本。

向量搜索的工作原理

传统数据库通过相等或范围运算符比较值。向量搜索则通过计算高维嵌入空间中的距离来寻找 K 近邻：

import numpy as np

def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 精确 KNN 每次查询复杂度为 O(n*d)
# 100 万向量 * 1536 维 = 每次查询 15 亿次操作
# ANN 算法以微小精度损失换取巨大速度提升

ANN 索引类型

HNSW（分层可导航小世界）：适用于大多数场景

• 查询：O(log n)，高召回率（>95%）
• 内存：约 O(n * d * 4 字节) —— 昂贵

IVF（倒排文件索引）：适用于内存受限的大数据集

• 将向量量化为簇
• 查询：O(sqrt(n))，较低召回率（70-90%）
• 内存：比 HNSW 少 8-16 倍

PQ（乘积量化）：压缩

• 将 1536 个 float32（6KB）压缩至 96 字节（64 倍压缩）
• IVF+PQ 组合是十亿级规模的理想选择

Pinecone（托管云服务）

最适合：希望零基础设施开销的团队。

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key=os.environ['PINECONE_API_KEY'])
pc.create_index(
    name='documents',
    dimension=1536,
    metric='cosine',
    spec=ServerlessSpec(cloud='aws', region='us-east-1')
)
index = pc.Index('documents')

# 带元数据的 Upsert
index.upsert(vectors=[
    ('doc-001', embedding_vector, {
        'text': 'chunk text here',
        'source': 'handbook.pdf',
        'category': 'hr-policy'
    })
], namespace='production')

# 带元数据过滤的查询
results = index.query(
    vector=query_embedding,
    top_k=10,
    filter={'category': {'$in': ['hr-policy', 'benefits']}},
    include_metadata=True,
    namespace='production'
)

Qdrant（自托管 / 托管）

最适合：大规模下的成本效益，高级过滤。

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct, HnswConfigDiff,
    Filter, FieldCondition, MatchValue, Range
)

client = QdrantClient(url='http://localhost:6333')

client.create_collection(
    collection_name='documents',
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    hnsw_config=HnswConfigDiff(
        m=16,           # 每个节点的连接数
        ef_construct=200  # 构建质量
    )
)

# 批量 Upsert
client.upsert(
    collection_name='documents',
    points=[PointStruct(id=i, vector=emb, payload={'text': txt})
            for i, (emb, txt) in enumerate(data)],
    wait=True
)

# 复杂过滤
results = client.search(
    collection_name='documents',
    query_vector=query_embedding,
    limit=10,
    query_filter=Filter(
        must=[FieldCondition(key='category', match=MatchValue(value='technical'))],
        must_not=[FieldCondition(key='archived', match=MatchValue(value=True))]
    ),
    score_threshold=0.75
)

Chroma（本地 / 简单 RAG）

最适合：开发、原型设计、小规模（<100 万向量）。

import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path='./chroma_db')
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.environ['OPENAI_API_KEY'],
    model_name='text-embedding-3-small'
)
collection = client.get_or_create_collection(
    name='documents', embedding_function=openai_ef
)

# 添加文档（自动嵌入）
collection.add(
    documents=['The refund policy allows returns within 30 days...'],
    metadatas=[{'source': 'policy.pdf', 'category': 'support'}],
    ids=['doc-001']
)

# 自然语言查询（自动嵌入）
results = collection.query(
    query_texts=['What is the return policy?'],
    n_results=5,
    where={'category': 'support'}
)

混合搜索：稠密 + 稀疏

将向量搜索与 BM25 结合以获得更好的召回率：

def hybrid_search(query, collection, bm25_index, alpha=0.7):
    dense = collection.query(query_texts=[query], n_results=20)
    sparse = bm25_index.get_scores(query.split())
    combined = {}
    for rank, doc_id in enumerate(dense['ids'][0]):
        combined[doc_id] = alpha / (rank + 60)
    for rank, (doc_id, _) in enumerate(
        sorted(sparse.items(), key=lambda x: x[1], reverse=True)[:20]
    ):
        combined[doc_id] = combined.get(doc_id, 0) + (1-alpha) / (rank + 60)
    return sorted(combined.items(), key=lambda x: x[1], reverse=True)[:10]

选择指南

选择 Pinecone 用于快速生产部署。选择 Qdrant 用于成本控制和高级过滤。使用 Chroma 进行开发。