向量数据库:AI应用的基础设施

RAG系统、语义搜索、推荐系统——这些AI应用的核心基础设施都是向量数据库。2026年的向量数据库市场已经从早期的"够用就行"进化到"精打细算"的阶段,选型直接影响系统性能和成本。

核心技术维度

索引算法

向量数据库的性能核心在于近似最近邻搜索(ANN)算法:

HNSW(Hierarchical Navigable Small World)

  • 原理:多层图结构,顶层稀疏快速导航,底层密集精确搜索
  • 优势:查询速度快,召回率高
  • 劣势:内存占用大,构建慢
  • 适合:中小规模(<1000万),高召回需求

IVF(Inverted File Index)

  • 原理:将向量空间聚类为N个桶,查询时只搜索最近的几个桶
  • 优势:内存效率好,支持大规模数据
  • 劣势:需要训练聚类模型,召回率受桶数影响
  • 适合:大规模(>1000万),召回率可接受场景

PQ(Product Quantization)

  • 原理:将高维向量分成子向量,每个子向量量化编码
  • 优势:存储压缩比高(10-100倍)
  • 劣势:精度损失
  • 适合:超大规模,成本敏感场景

组合索引:IVF+PQ或HNSW+PQ结合各自优势:

# Milvus中的组合索引配置
collection_config = {
    "index_type": "IVF_PQ",
    "params": {
        "nlist": 1024,      # IVF聚类中心数
        "m": 16,            # PQ子向量数
        "nbits": 8,         # 每个子向量的编码位数
    },
    "metric_type": "COSINE"
}

量化与压缩

class QuantizationComparison:
    """不同量化方案的效果对比"""
    
    results = {
        "FP32 (无压缩)": {
            "recall": 1.0,
            "memory": "100%",
            "speed": "基准"
        },
        "FP16": {
            "recall": 0.999,
            "memory": "50%",
            "speed": "1.2x"
        },
        "INT8 (标量量化)": {
            "recall": 0.99,
            "memory": "25%",
            "speed": "1.5x"
        },
        "PQ8 (乘积量化8bit)": {
            "recall": 0.95,
            "memory": "12.5%",
            "speed": "2.0x"
        },
        "PQ4 (乘积量化4bit)": {
            "recall": 0.88,
            "memory": "6.25%",
            "speed": "2.5x"
        }
    }

主流方案对比

Milvus

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接Milvus
connections.connect(host="localhost", port="19530")

# 创建Collection
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="metadata", dtype=DataType.JSON),
]
schema = CollectionSchema(fields, "文档向量集合")
collection = Collection("documents", schema)

# 创建索引
collection.create_index(
    field_name="embedding",
    index_params={
        "index_type": "HNSW",
        "metric_type": "COSINE",
        "params": {"M": 16, "efConstruction": 256}
    }
)

# 搜索
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"params": {"ef": 64}},
    limit=10,
    expr='department == "engineering"'  # 标量过滤
)

优势

  • 开源,支持私有化部署
  • 支持多种索引类型
  • 标量字段过滤
  • 分布式架构,支持十亿级向量
  • 生态完善

劣势

  • 部署复杂(依赖etcd、MinIO、Pulsar等)
  • 内存占用较高
  • 小规模场景"杀鸡用牛刀"

Qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
)

# 插入数据
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],
            payload={"department": "engineering", "title": "..."}
        )
    ]
)

# 搜索(带过滤)
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    query_filter={
        "must": [
            {"key": "department", "match": {"value": "engineering"}}
        ]
    },
    limit=10
)

优势

  • Rust实现,性能优异
  • 单机部署简单(单二进制)
  • 内置标量过滤
  • 支持量化(INT8、PQ)
  • API设计优雅

劣势

  • 分布式功能仍在完善
  • 社区规模小于Milvus
  • 生态工具较少

Pinecone

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")

# 创建索引
pc.create_index(
    name="documents",
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

# 使用
index = pc.Index("documents")
index.upsert(vectors=[
    {"id": "1", "values": [0.1, ...], "metadata": {"dept": "eng"}}
])

results = index.query(
    vector=[0.1, ...],
    filter={"dept": {"$eq": "eng"}},
    top_k=10
)

优势

  • 全托管,零运维
  • 自动扩缩容
  • Serverless按使用付费
  • 全球部署

劣势

  • 成本高(大规模场景)
  • 数据不在本地(合规风险)
  • 定制化能力有限

其他方案

Weaviate:内置模块化向量化,支持多模态 Chroma:轻量级,适合原型开发 pgvector:PostgreSQL扩展,适合已有PG环境 Elasticsearch:向量+全文搜索混合

性能基准

在1亿条1024维向量的数据集上测试:

方案索引类型召回率@10QPS延迟P99内存
MilvusHNSW98.5%200015ms80GB
MilvusIVF_PQ93.2%50008ms15GB
QdrantHNSW98.3%220012ms75GB
QdrantINT897.1%350010ms40GB
Pinecone-97.5%180020ms-

选型决策框架

class VectorDBSelector:
    def recommend(self, requirements):
        score = {
            "milvus": 0,
            "qdrant": 0,
            "pinecone": 0,
            "chroma": 0,
            "pgvector": 0
        }
        
        # 规模
        if requirements["scale"] > 100_000_000:
            score["milvus"] += 3
            score["pinecone"] += 2
        elif requirements["scale"] > 1_000_000:
            score["qdrant"] += 3
            score["milvus"] += 2
        else:
            score["chroma"] += 3
            score["qdrant"] += 2
        
        # 部署方式
        if requirements["deployment"] == "self-hosted":
            score["milvus"] += 2
            score["qdrant"] += 3
        elif requirements["deployment"] == "cloud":
            score["pinecone"] += 3
        
        # 运维能力
        if requirements["ops_capacity"] == "low":
            score["pinecone"] += 3
            score["chroma"] += 2
        elif requirements["ops_capacity"] == "high":
            score["milvus"] += 2
            score["qdrant"] += 2
        
        # 已有基础设施
        if requirements["existing_db"] == "postgresql":
            score["pgvector"] += 3
        
        # 成本敏感度
        if requirements["cost_sensitive"]:
            score["qdrant"] += 2
            score["milvus"] += 1
            score["pgvector"] += 2
        
        # 混合搜索需求
        if requirements["hybrid_search"]:
            score["milvus"] += 2
            score["qdrant"] += 1
        
        return sorted(score.items(), key=lambda x: x[1], reverse=True)

生产实践建议

数据导入优化

def batch_import(vector_db, documents, batch_size=1000):
    """批量导入优化"""
    total = len(documents)
    
    for i in range(0, total, batch_size):
        batch = documents[i:i+batch_size]
        
        # 并行编码
        embeddings = parallel_embed(batch, workers=8)
        
        # 批量写入
        vector_db.upsert(
            vectors=[
                {"id": doc.id, "values": emb, "metadata": doc.metadata}
                for doc, emb in zip(batch, embeddings)
            ]
        )
        
        if i % 10000 == 0:
            print(f"进度: {i}/{total}")
    
    # 构建索引(导入完成后)
    vector_db.create_index()

监控指标

class VectorDBMonitor:
    def metrics(self):
        return {
            "qps": self._query_rate(),
            "latency_p50": self._latency(50),
            "latency_p99": self._latency(99),
            "recall_rate": self._sample_recall(),  # 抽样计算召回率
            "memory_usage": self._memory(),
            "disk_usage": self._disk(),
            "index_health": self._index_status(),
        }

结语

向量数据库选型没有银弹——Milvus适合大规模企业部署,Qdrant适合追求简洁高效的团队,Pinecone适合不想运维的团队,pgvector适合已有PostgreSQL的场景。选型时应先明确自己的规模、预算、运维能力和性能需求,再做出理性决策。记住:最好的向量数据库不是性能最强的,而是最适合你的场景的。