1
# 向量数据库性能基准测试框架
2
class VectorDBBenchmark:
3
    def __init__(self, db_type: str):
4
        self.db_type = db_type
5

6
    def benchmark(self, dataset_size: int = 1000000):
7
        """QPS 与召回率基准测试"""
8
        results = {
9
            "milvus": {"qps": 5000, "recall": 0.95, "p99_latency": "15ms"},
10
            "qdrant": {"qps": 8000, "recall": 0.97, "p99_latency": "8ms"},
11
            "weaviate": {"qps": 3000, "recall": 0.93, "p99_latency": "20ms"},
12
        }
13
        return results.get(self.db_type, {})

1
# HNSW 搜索算法伪代码
2
class HNSWIndex:
3
    def __init__(self, m: int = 16, ef_construction: int = 200):
4
        self.m = m  # 每一层最多连接数
5
        self.ef_construction = ef_construction  # 构建时动态列表大小
6

7
    def search(self, query_vector: list, ef: int = 100):
8
        """
9
        搜索过程：
10
        1. 从顶层开始，找到最近邻
11
        2. 逐层向下，更新候选集
12
        3. 最终在底层返回精确结果
13
        """
14
        candidates = self._search_layer_0(query_vector, ef)
15

16
        # 层间跳转
17
        for layer in range(self.max_layer, 0, -1):
18
            candidates = self._expand_neighbors(
19
                candidates,
20
                query_vector,
21
                ef=ef_construction
22
            )
23

24
        return candidates[:ef]

1
# IVF 索引原理
2
class IVFIndex:
3
    def __init__(self, nlist: int = 1024):
4
        self.nlist = nlist  # 聚类中心数量
5

6
    def build(self, vectors: list):
7
        """1. k-means 聚类建立倒排表"""
8
        self.centers, assignments = kmeans(vectors, self.nlist)
9

10
        # 2. 建立倒排表
11
        self.inverted_index = {}
12
        for i, cluster_id in enumerate(assignments):
13
            if cluster_id not in self.inverted_index:
14
                self.inverted_index[cluster_id] = []
15
            self.inverted_index[cluster_id].append(i)
16

17
    def search(self, query: list, nprobe: int = 10):
18
        """搜索时只扫描 nprobe 个聚类中心"""
19
        # 1. 找到最近的 nprobe 个聚类中心
20
        nearest_centers = self._find_nearest_centers(
21
            query,
22
            nprobe
23
        )
24

25
        # 2. 在这些聚类中暴力搜索
26
        candidates = []
27
        for center_id in nearest_centers:
28
            candidates.extend(self.inverted_index[center_id])
29

30
        return self._brute_force_search(query, candidates)

1
# Product Quantization 原理
2
class PQIndex:
3
    def __init__(self, m: int = 8, ks: int = 256):
4
        """
5
        m: 分段数（通常 8-16）
6
        ks: 每个子空间码本大小（通常 256）
7
        """
8
        self.m = m
9
        self.ks = ks
10
        self.codebooks = []
11

12
    def train(self, vectors: list):
13
        """训练码书"""
14
        dim = len(vectors[0])
15
        sub_dim = dim // self.m
16

17
        for i in range(self.m):
18
            # 对第 i 个子空间进行 k-means
19
            sub_vectors = vectors[:, i*sub_dim:(i+1)*sub_dim]
20
            codebook = kmeans(sub_vectors, self.ks)
21
            self.codebooks.append(codebook)
22

23
    def encode(self, vector: list):
24
        """编码：每个子空间找最近码字"""
25
        codes = []
26
        for i, codebook in enumerate(self.codebooks):
27
            sub_vec = vector[i*self.sub_dim:(i+1)*self.sub_dim]
28
            code = np.argmin(np.linalg.norm(codebook - sub_vec, axis=1))
29
            codes.append(code)
30
        return codes
31

32
    def search(self, query: list, candidates: list):
33
        """距离计算使用查表而非暴力计算"""
34
        # 预计算查询向量各子空间到各码字距离
35
        distance_table = self._build_distance_table(query)
36

37
        # 查表累加获取最终距离
38
        for vector_id in candidates:
39
            total_dist = sum(
40
                distance_table[i][self.codes[vector_id][i]]
41
                for i in range(self.m)
42
            )

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

3
# 1. 连接 Milvus
4
connections.connect(host='localhost', port='19530')
5

6
# 2. 定义 Schema
7
fields = [
8
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
9
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
10
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=100),
11
]
12
schema = CollectionSchema(fields=fields, description="向量检索集合")
13

14
# 3. 创建 Collection
15
collection = Collection(name="documents", schema=schema)
16

17
# 4. 创建索引
18
index_params = {
19
    "index_type": "HNSW",
20
    "params": {"M": 16, "efConstruction": 200},
21
    "metric_type": "L2"
22
}
23
collection.create_index(field_name="embedding", index_params=index_params)
24

25
# 5. 插入数据
26
entities = [
27
    [1, 2, 3],  # id
28
    [[0.1]*768, [0.2]*768],  # embeddings
29
    ["科技", "娱乐"]  # category
30
]
31
collection.insert(entities)
32

33
# 6. 搜索
34
search_params = {"metric_type": "L2", "params": {"ef": 100}}
35
results = collection.search(
36
    data=[[0.1]*768],
37
    anns_field="embedding",
38
    param=search_params,
39
    limit=10,
40
    expr="category == '科技'"
41
)

1
# Qdrant 配置文件
2
storage:
3
  # 存储路径
4
  storage_path: /qdrant/storage
5

6
  # HNSW 参数
7
  hnsw_index:
8
    m: 16 # 连接数
9
    ef_construct: 100 # 构建时搜索深度
10
    full_scan_threshold: 10000 # 小于此规模用暴力搜索
11

12
  # 量化配置
13
  quantization:
14
    binary: false
15
    product:
16
      compression: 8 # 压缩比

1
from qdrant_client import QdrantClient
2
from qdrant_client.models import Distance, VectorParams, Filter
3

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

6
# 1. 创建 Collection
7
client.create_collection(
8
    collection_name="articles",
9
    vectors_config=VectorParams(size=768, distance=Distance.COSINE)
10
)
11

12
# 2. 插入向量
13
client.upsert(
14
    collection_name="articles",
15
    points=[
16
        {
17
            "id": 1,
18
            "vector": [0.1]*768,
19
            "payload": {"title": "Python 教程", "category": "编程"}
20
        },
21
    ]
22
)
23

24
# 3. 搜索
25
results = client.search(
26
    collection_name="articles",
27
    query_vector=[0.1]*768,
28
    query_filter=Filter(
29
        must=[
30
            {"key": "category", "match": {"value": "编程"}}
31
        ]
32
    ),
33
    limit=5
34
)
35

36
# 4. 范围搜索（按分数过滤）
37
results = client.search(
38
    collection_name="articles",
39
    query_vector=[0.1]*768,
40
    score_threshold=0.8,  # 只返回相似度 > 0.8 的结果
41
    limit=10
42
)

1
# 向量数据库性能优化清单
2
optimization_checklist = {
3
    "索引参数": {
4
        "HNSW_M": "16-64，越大越精确但越慢",
5
        "HNSW_ef": "100-500，搜索时动态调整",
6
        "HNSW_ef_construction": "200-400，构建时精度"
7
    },
8
    "量化策略": {
9
        "binary": "内存减半，速度加倍，精度略降",
10
        "product": "内存压缩 4-16 倍，精度可调",
11
        "scalar": "不影响精度，略微提升速度"
12
    },
13
    "硬件配置": {
14
        "内存": "能装下全部向量 + 索引 > 80% 命中率",
15
        "CPU": "影响索引构建速度，HNSW 构建 CPU 密集",
16
        "SSD": "向量量大时必须，内存不足时代替"
17
    }
18
}