向量数据库实战指南：Faiss、Milvus、Weaviate 核心原理与性能优化

1次阅读

没有评论

共计 2532 个字符，预计需要花费 7 分钟才能阅读完成。

随着 AI 和大数据技术的发展，向量数据库逐渐成为处理高维数据的核心组件。传统关系型数据库在处理非结构化数据（如图片、视频、文本）时效率低下，而向量数据库通过将数据转换为高维向量，并利用相似性搜索技术，能够高效处理这些场景。

应用场景：
推荐系统：快速找到相似用户或商品
图像搜索：基于内容的图像检索
自然语言处理：语义搜索和问答系统
异常检测：识别异常模式

Faiss：Facebook 开发的轻量级库，专注于高性能向量搜索，适合嵌入式场景
Milvus：开源分布式向量数据库，支持水平扩展和持久化存储
Weaviate：结合了向量搜索和图数据库功能，适合复杂关系场景

Faiss：支持 IVF、HNSW、PQ 等多种算法
Milvus：基于 Faiss 扩展，增加分布式支持
Weaviate：默认使用 HNSW，支持自定义算法

（以下为简略对比，实际性能需结合具体场景）

数据库	索引构建速度	查询延迟	内存占用
Faiss	快	极低	中等
Milvus	中等	低	高
Weaviate	慢	中等	高

Faiss：单机版，扩展性有限
Milvus：原生支持分布式部署
Weaviate：可通过分片扩展

聚类阶段：将所有向量通过 k -means 聚类为 n 个单元（cell）
倒排索引：建立向量到单元的映射关系
搜索阶段：先定位到最近的几个单元，再在这些单元内做精确搜索

数学公式：

最近邻搜索 = argmin ||x - y||^2

构建多层图结构，上层是下层的概要图
搜索时从上至下逐层缩小范围
结合了跳表和小世界网络的特性

优势：
– 适合高维数据
– 查询复杂度接近 O(log n)

import faiss
import numpy as np

# 生成随机数据
d = 64  # 向量维度
nb = 100000  # 数据库大小
nq = 100  # 查询数量
np.random.seed(1234)
xb = np.random.random((nb, d)).astype('float32')
xq = np.random.random((nq, d)).astype('float32')

# 构建索引
index = faiss.IndexFlatL2(d)  # 使用 L2 距离
print(index.is_trained)  # 输出 True
index.add(xb)  # 添加向量到索引
print(index.ntotal)  # 输出 100000

# 执行搜索
k = 4  # 返回最近的 4 个邻居
D, I = index.search(xb[:5], k)  # 测试前 5 个向量
print(I)  # 最近邻的索引
print(D)  # 距离值

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

# 连接服务
connections.connect("default", host="localhost", port="19530")

# 定义 schema
fields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128)
]
schema = CollectionSchema(fields, description="test collection")

# 创建集合
collection_name = "test_collection"
collection = Collection(collection_name, schema)

# 插入数据
import numpy as np
entities = [[i for i in range(10)],  # ids
    np.random.rand(10, 128)  # 10 个 128 维向量
]
collection.insert(entities)

# 创建索引
index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {"nlist": 128}
}
collection.create_index("embedding", index_params)

# 加载到内存
collection.load()

# 执行搜索
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
results = collection.search(entities[1][:1], "embedding", search_params, limit=3
)
print(results)