HoRain云--向量数据库

📅 2026/7/13 14:45:40
HoRain云--向量数据库
在 AI 时代我们处理的数据形态发生了根本变化。传统数据库存储的是结构化数据数字、字符串、日期。AI 应用处理的是语义一段文本的含义、一张图片的内容、一段语音的意思这些语义信息被表示成向量——一串浮点数比如 [0.123, -0.456, 0.789, ...]。向量数据库就是专门用来存储、索引、查询这些向量的数据库没有它RAG检索增强生成做不了语义搜索实现不了推荐系统也无法高效运行。向量数据库的核心价值在百万级甚至十亿级向量中毫秒级找出与查询向量最相似的前 K 个结果。向量相似度搜索的挑战先理解问题的本质才能理解为什么需要专门的算法。暴力搜索的复杂度问题最简单的思路是暴力搜索计算查询向量与库中每个向量的相似度排序取前 K 个。实例# # 暴力搜索演示简单但低效# import numpy as npfrom typing import List, Tupledef l2_distance(v1: np.ndarray, v2: np.ndarray) - float:计算 L2 欧氏距离越小越相似return np.sqrt(np.sum((v1 - v2) ** 2))def brute_force_search(query: np.ndarray,vectors: List[np.ndarray],top_k: int 5) - List[Tuple[int, float]]:暴力搜索计算与所有向量的距离排序返回 Top K优点简单准确100% 召回缺点数据量大时极慢# 计算与每个向量的距离distances []for idx, vec in enumerate(vectors):dist l2_distance(query, vec)distances.append((idx, dist))# 按距离排序取前 K 个distances.sort(keylambda x: x[1])return distances[:top_k]# 生成 10000 个 128 维的随机向量模拟向量库np.random.seed(42)vector_count 10000dimension 128vectors [np.random.randn(dimension) for _ in range(vector_count)]# 生成一个查询向量query_vector np.random.randn(dimension)# 执行暴力搜索print(f在 {vector_count} 个向量中搜索 Top 5...)results brute_force_search(query_vector, vectors, top_k5)print(搜索结果)for idx, dist in results:print(f 向量 {idx:4d}, 距离 {dist:.4f})# 输出类似# 在 10000 个向量中搜索 Top 5...# 搜索结果# 向量 8236, 距离 13.3986# 向量 8946, 距离 13.5604# 向量 9891, 距离 13.5688# 向量 5172, 距离 13.5987# 向量 5223, 距离 13.6081暴力搜索的问题很明显每次查询都要遍历所有向量。如果有 100 万个向量每次查询要算 100 万次距离。如果有 10 亿个向量每次查询要算 10 亿次——这在生产环境完全不可接受。高维空间的维度诅咒雪上加霜的是向量通常是高维的。OpenAI 的 text-embedding-3-small 是 1536 维large 是 3072 维。在高维空间里传统的空间索引方法比如 KD-Tree效率会急剧下降甚至比暴力搜索还慢。数据规模暴力搜索耗时估算能否接受1 万~1 毫秒可以接受100 万~100 毫秒有点慢1000 万~1 秒太慢了1 亿~10 秒完全不可接受10 亿~100 秒无法使用这就是为什么需要近似最近邻ANN算法放弃 100% 的准确率换取 100 倍甚至 1000 倍的速度提升。近似最近邻ANN算法ANN 算法的核心思想构建索引结构让查询时不用遍历所有向量。主流算法有三个HNSW、IVF、PQ。HNSW分层小世界图HNSWHierarchical Navigable Small World是目前最流行的算法之一它的灵感来自六度分隔理论——世界上任意两个人最多通过 6 个中间人就能联系上。HNSW 通过多层稀疏小世界图从顶层稀疏节点快速粗定位逐层下降到底层全量数据层精细搜索用分层跳跃大幅减少距离计算次数实现高速高召回的近似最近邻检索。分层结构底层 Layer0 存全部向量层数越高节点越少充当高速通道同个向量跨层用虚线垂直关联。检索流程蓝线路径顶层入口粗搜找到本层离查询向量最近点。垂直下落到中层再次粗定位。落到 Layer0 全量层精细遍历输出最近邻。核心原理高层快速缩小搜索范围底层精确匹配把暴力查找优化为对数复杂度速度快、召回高是主流向量检索算法。实例# # HNSW 原理简化演示非生产实现# import numpy as npimport randomfrom typing import List, Dict, Set, Tupleclass SimpleHNSW:简化版 HNSW仅用于演示原理def __init__(self, dim: int, max_level: int 3, ef_construction: int 5):self.dim dimself.max_level max_level # 最大层数self.ef_construction ef_construction # 建图时考虑的候选数self.levels: List[Dict[int, np.ndarray]] [{} for _ in range(max_level)]self.edges: List[Dict[int, Set[int]]] [{} for _ in range(max_level)]self.next_id 0self.enter_point None # 入口点顶层某个点def _random_level(self) - int:随机选择一个层级越高概率越小level 0while random.random() 0.5 and level self.max_level - 1:level 1return leveldef _distance(self, v1: np.ndarray, v2: np.ndarray) - float:L2 距离return np.sqrt(np.sum((v1 - v2) ** 2))def _search_layer(self,query: np.ndarray,level: int,entry: int,ef: int) - List[Tuple[float, int]]:在某一层搜索从 entry 开始贪心找到 ef 个最近点if entry not in self.levels[level]:return []visited set([entry])candidates [(self._distance(query, self.levels[level][entry]), entry)]result list(candidates)while candidates:# 取出当前最近的候选candidates.sort()dist, current candidates.pop(0)# 遍历邻居if current not in self.edges[level]:continuefor neighbor in self.edges[level][current]:if neighbor in visited:continuevisited.add(neighbor)neighbor_dist self._distance(query, self.levels[level][neighbor])# 加入候选和结果集candidates.append((neighbor_dist, neighbor))result.append((neighbor_dist, neighbor))# 保持结果集大小不超过 efresult.sort()result result[:ef]return resultdef add(self, vector: np.ndarray) - int:添加一个向量vec_id self.next_idself.next_id 1# 随机确定这个向量出现在哪些层max_level_for_vec self._random_level()# 在所有层存储向量实际实现中通常只在底层存完整向量for level in range(max_level_for_vec 1):self.levels[level][vec_id] vectorself.edges[level][vec_id] set()# 如果是第一个点设为入口if self.enter_point is None:self.enter_point vec_idreturn vec_id# 从顶层向下搜索找到每一层的插入位置current_entry self.enter_point# 先在高于 max_level_for_vec 的层搜索更新入口for level in range(self.max_level - 1, max_level_for_vec, -1):if current_entry in self.levels[level]:layer_result self._search_layer(vector, level, current_entry, 1)if layer_result:current_entry layer_result[0][1]# 在 max_level_for_vec 及以下层建立连接for level in range(max_level_for_vec, -1, -1):# 在这一层搜索最近的 ef_construction 个点layer_result self._search_layer(vector, level, current_entry, self.ef_construction)# 连接到这些点简化双向连接for dist, neighbor_id in layer_result:self.edges[level][vec_id].add(neighbor_id)self.edges[level][neighbor_id].add(vec_id)# 更新下一层的入口if layer_result:current_entry layer_result[0][1]# 更新顶层入口if max_level_for_vec self._random_level(): # 简化的判断self.enter_point vec_idreturn vec_iddef search(self, query: np.ndarray, top_k: int 5, ef: int 10) - List[Tuple[int, float]]:搜索 Top K 最近邻if self.enter_point is None:return []current_entry self.enter_point# 从顶层向下直到第一层for level in range(self.max_level - 1, 0, -1):if current_entry in self.levels[level]:layer_result self._search_layer(query, level, current_entry, 1)if layer_result:current_entry layer_result[0][1]# 在底层搜索 ef 个点然后返回 Top Klayer_result self._search_layer(query, 0, current_entry, ef)layer_result.sort()return [(vec_id, dist) for dist, vec_id in layer_result[:top_k]]# 测试np.random.seed(42)dim 128hnsw SimpleHNSW(dimdim, max_level3, ef_construction5)# 添加 100 个向量print(正在构建 HNSW 索引添加 100 个向量...)for i in range(100):vec np.random.randn(dim)hnsw.add(vec)print(f索引构建完成入口点 {hnsw.enter_point})# 生成查询向量query np.random.randn(dim)# 搜索print(\n执行 HNSW 搜索...)results hnsw.search(query, top_k5, ef10)print(搜索结果)for vec_id, dist in results:print(f 向量 {vec_id:3d}, 距离 {dist:.4f})# 输出类似# 正在构建 HNSW 索引添加 100 个向量...# 索引构建完成入口点 54## 执行 HNSW 搜索...# 搜索结果# 向量 86, 距离 13.7515# 向量 30, 距离 13.8651# 向量 28, 距离 14.0754# 向量 56, 距离 14.1106# 向量 34, 距离 14.2131HNSW 的优点1.查询快通常只需访问少量节点就能找到最近邻。2.构建快可以增量添加向量不需要重建整个索引。3.参数简单主要调整 ef_construction建图质量和 ef搜索质量。IVF倒排文件索引IVFInverted File的思路是分而治之。步骤1.聚类把向量空间分成 K 个桶用 K-Means 聚类。2.索引每个向量分到最近的桶里。3.查询先找查询向量最近的 N 个桶只在这 N 个桶里搜索。实例# # IVF 原理简化演示# import numpy as npfrom typing import List, Tupleclass SimpleIVF:简化版 IVF仅用于演示原理def __init__(self, dim: int, nlist: int 10):self.dim dimself.nlist nlist # 聚类中心数量桶的数量self.centroids: np.ndarray None # 聚类中心self.inverted_lists: List[List[Tuple[int, np.ndarray]]] [[] for _ in range(nlist)]self.next_id 0def _l2_distance(self, v1: np.ndarray, v2: np.ndarray) - float:L2 距离return np.sqrt(np.sum((v1 - v2) ** 2))def _find_nearest_centroid(self, vector: np.ndarray) - int:找到最近的聚类中心distances [self._l2_distance(vector, c) for c in self.centroids]return int(np.argmin(distances))def fit(self, vectors: List[np.ndarray]):用 K-Means 初始化聚类中心简化版随机选# 简化随机选 nlist 个向量作为初始中心# 实际实现中应该用 K-Means 迭代indices np.random.choice(len(vectors), self.nlist, replaceFalse)self.centroids np.array([vectors[i] for i in indices])def add(self, vector: np.ndarray) - int:添加向量vec_id self.next_idself.next_id 1# 找到最近的桶centroid_idx self._find_nearest_centroid(vector)# 放进这个桶里self.inverted_lists[centroid_idx].append((vec_id, vector))return vec_iddef search(self, query: np.ndarray, top_k: int 5, nprobe: int 3) - List[Tuple[int, float]]:搜索 Top Knprobe: 搜索多少个桶越多越准确但越慢# 找到最近的 nprobe 个桶centroid_distances [(self._l2_distance(query, c), i)for i, c in enumerate(self.centroids)]centroid_distances.sort()nearest_centroids [i for dist, i in centroid_distances[:nprobe]]# 只在这 nprobe 个桶里搜索candidates []for centroid_idx in nearest_centroids:for vec_id, vec in self.inverted_lists[centroid_idx]:dist self._l2_distance(query, vec)candidates.append((dist, vec_id))# 排序返回 Top Kcandidates.sort()return [(vec_id, dist) for dist, vec_id in candidates[:top_k]]# 测试np.random.seed(42)dim 128# 生成 1000 个向量vectors [np.random.randn(dim) for _ in range(1000)]# 构建 IVF 索引ivf SimpleIVF(dimdim, nlist20)ivf.fit(vectors)# 添加向量for vec in vectors:ivf.add(vec)print(fIVF 索引构建完成桶数 {ivf.nlist})for i in range(ivf.nlist):print(f 桶 {i:2d}: {len(ivf.inverted_lists[i])} 个向量)# 查询query np.random.randn(dim)print(\n执行 IVF 搜索nprobe3...)results ivf.search(query, top_k5, nprobe3)print(搜索结果)for vec_id, dist in results:print(f 向量 {vec_id:3d}, 距离 {dist:.4f})# 输出类似# IVF 索引构建完成桶数 20# 桶 0: 63 个向量# 桶 1: 56 个向量# ...## 执行 IVF 搜索nprobe3...# 搜索结果# 向量 863, 距离 13.5409# 向量 189, 距离 13.7083# 向量 491, 距离 13.8628# 向量 264, 距离 13.9812# 向量 781, 距离 14.0021IVF 的优点是内存占用小可以动态调整查询时的 nprobe 参数来平衡速度和准确率。PQ乘积量化PQProduct Quantization是一种压缩算法。核心思路1.切分把高维向量切分成多个小段subvectors。2.量化对每个小段单独聚类用聚类中心的 ID 代替原始向量。3.查表查询时用查表法快速计算近似距离。比如 128 维向量切分成 8 段每段 16 维每段用 256 个中心量化结果是每个向量只需要 8 字节存储原来需要 128 * 4 512 字节。IVF-PQ 组合生产环境中通常把 IVF 和 PQ 结合起来用IVF 负责分桶减少搜索范围PQ 负责压缩减少内存和计算量。算法速度准确率内存适用场景暴力搜索慢100%高数据量小需要 100% 准确HNSW极快高中查询密集需要低延迟IVF快中低大规模数据可接受少量精度损失IVF-PQ很快中极低超大规模数据内存受限