向量数据库

向量数据库是 RAG 系统的核心组件，用于高效存储和检索高维向量。

为什么需要向量数据库？

传统数据库（MySQL、PostgreSQL）不支持向量相似度搜索，而向量数据库专门优化了：

• 高维向量索引：百万级向量毫秒级检索
• 近似最近邻搜索（ANN）：牺牲少量精度换取速度
• 分布式扩展：支持 PB 级数据
• Filters（过滤）：结合元数据查询

核心概念

向量（Embedding）

文本/图像通过模型转换为浮点数数组：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
text = "AI Cloud Hook 文档"
vector = model.encode(text)  # shape: (384,)

距离度量

• 余弦相似度（Cosine）：最常用，范围 [-1, 1]
• 欧氏距离（L2）：空间距离
• 内积（IP）：快速计算，对归一化向量等价于余弦

from sklearn.metrics.pairwise import cosine_similarity

similarity = cosine_similarity([query_vector], [doc_vector])[0][0]

主流向量数据库对比

数据库	开源	性能	功能	适用场景
Milvus	✅	极高	丰富	大规模生产
Pinecone	❌	高	全托管	快速上线
Qdrant	✅	高	过滤强	复杂查询
Weaviate	✅	中	模块化	多种嵌入
Chroma	✅	中	简单	快速原型
Pinecone	❌	高	全托管	快速上线
pgvector	✅	中	PostgreSQL 扩展	已有 PG 生态

实战：Milvus

安装

docker-compose -f docker-compose.yml up -d

连接与操作

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

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

# 2. 定义集合（Collection）
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=500),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=384),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=50)
]

schema = CollectionSchema(fields, description="文档集合")
collection = Collection(name="documents", schema=schema)

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

# 4. 插入数据
texts = ["文档1内容", "文档2内容", ...]
embeddings = model.encode(texts).tolist()
entities = [
    ["id1", "文档1", embeddings[0], "技术"],
    ["id2", "文档2", embeddings[1], "产品"]
]
collection.insert(entities)

# 5. 搜索
collection.load()  # 加载到内存

search_params = {"metric_type": "COSINE", "params": {"ef": 100}}
results = collection.search(
    data=[query_embedding],
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr="category == '技术'"  # 过滤条件
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance}, 文本: {hit.entity.get('text')}")

实战：Qdrant

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=384, distance=Distance.COSINE)
)

# 插入（带 payload）
points = [
    PointStruct(
        id=1,
        vector=embedding.tolist(),
        payload={"text": "文档内容", "category": "tech", "source": "wiki"}
    )
]
client.upsert(collection_name="documents", points=points)

# 搜索 + 过滤
results = client.search(
    collection_name="documents",
    query_vector=embedding.tolist(),
    limit=5,
    filter={
        "must": [
            {"key": "category", "match": {"value": "tech"}}
        ]
    }
)

Annoy / FAISS（轻量级）

如果数据量不大（< 100 万），可以用单机库：

Annoy（Spotify）

from annoy import AnnoyIndex

f = 384  # 向量维度
t = AnnoyIndex(f, 'angular')
for i, vec in enumerate(embeddings):
    t.add_item(i, vec)
t.build(10)  # 10 棵树
t.save('index.ann')

# 搜索
indices, distances = t.get_nns_by_vector(query_vec, 5, include_distances=True)

FAISS（Meta）

import faiss

d = 384
index = faiss.IndexFlatIP(d)  # 内积相似度
index.add(embeddings_np)

D, I = index.search(query_np, k=5)  # D: 距离, I: 索引

生产建议

1. 数据分片

当数据超过单机内存时：

• 按时间分片（daily/hourly）
• 按类别分片
• 使用分布式数据库（Milvus Cluster）

2. 索引优化参数

• HNSW：M (16-64), efConstruction (100-500), efSearch (50-200)
• IVF：nlist (聚类数), nprobe (搜索聚类数)

平衡：efSearch 越高精度越高，但速度越慢。

3. 数据更新策略

• 实时插入：小批量实时写入
• 批量重建：定期全量重建索引
• TTL（生存时间）：自动过期旧数据

4. 监控指标

• QPS：每秒查询数
• P99 Latency：99% 分位数延迟
• 召回率：Top-K 的命中率
• 索引大小：存储成本

5. 成本优化

• 压缩量化：FP16 → INT8 节省 50% 存储
• 冷热分离：热数据放内存，冷数据放磁盘
• 缓存层：Redis 缓存热门查询

常见问题

Q: 召回率低怎么办？

A: 增大 ef 或 nprobe 参数，或使用更好的 Embedding 模型。

Q: 插入太慢？

A: 使用批量插入，或调整 bulk_insert 参数；关闭自动索引，定时重建。

Q: 内存不够？

A: 使用量化（ quantization=INT8），或选择支持磁盘索引的数据库（如 pgvector）。

Q: 过滤查询慢？

A: 建立复合索引（向量+标量字段），或预过滤数据集。

快速选型建议

• 学习/原型：Chroma（超级简单）
• 中小规模生产：Qdrant（过滤强，易运维）
• 大规模生产：Milvus（性能最高，生态丰富）
• 已有 PostgreSQL：pgvector（零运维成本）

掌握向量数据库是构建现代 RAG 系统的必备技能！