3.1 向量相似度搜索原理

本节导读：深入理解向量相似度搜索的数学基础和算法原理，掌握距离度量的选择依据，学习HNSW算法的优化机制，为高效向量检索提供理论支撑。

学习目标

• 理解向量相似度的数学定义和计算方法
• 掌握各种距离度量的原理和适用场景
• 学习HNSW算法的工作原理和优化机制
• 了解向量检索的性能评估指标

核心概念

向量相似度搜索是Qdrant的核心功能，通过数学方法计算向量间的距离，找到最相似的数据点。这一过程基于距离度量和空间索引算法，实现高效的高维数据检索。

![向量相似度搜索示意图：展示高维向量空间中距离计算的几何意义，以及相似度排序的概念图]

向量相似度基础

3.1.1 距离度量类型

向量相似度搜索的核心是距离度量，Qdrant支持多种距离计算方法：

余弦相似度（Cosine）

余弦相似度计算两个向量间的夹角余弦值，范围在[-1,1]之间，其中1表示完全相似，-1表示完全相反，0表示正交。

import numpy as np

def cosine_similarity(vec1, vec2):
    """计算余弦相似度"""
    dot_product = np.dot(vec1, vec2)
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)
    return dot_product / (norm1 * norm2)

# 示例
class VectorSimilarity:
    @staticmethod
    def cosine_similarity(vec1, vec2):
        """计算余弦相似度"""
        dot_product = np.dot(vec1, vec2)
        norm1 = np.linalg.norm(vec1)
        norm2 = np.linalg.norm(vec2)
        similarity = dot_product / (norm1 * norm2)
        return similarity
    
    @staticmethod
    def euclidean_distance(vec1, vec2):
        """计算欧几里得距离"""
        return np.sqrt(np.sum((vec1 - vec2) ** 2))
    
    @staticmethod
    def dot_product(vec1, vec2):
        """计算内积"""
        return np.dot(vec1, vec2)
​

优势：

• 不受向量长度影响，适合文本语义相似度
• 计算相对简单，适合高维数据
• 对于归一化向量，等价于内积计算

欧几里得距离（Euclidean）

欧几里得距离是最直观的距离度量，表示两点之间的直线距离。

def euclidean_distance(vec1, vec2):
    """计算欧几里得距离"""
    return np.sqrt(np.sum((vec1 - vec2) ** 2))

# 数学公式
d = sqrt(∑(x_i - y_i)²)
​

应用场景：

• 图像识别和特征匹配
• 数值型数据的距离计算
• 需要考虑向量长度的情况

内积（Dot Product）

内积是向量的基本运算，在某些情况下可作为相似度度量。

def dot_product(vec1, vec2):
    """计算内积"""
    return np.dot(vec1, vec2)

# 数学公式
vec1 · vec2 = ∑(x_i * y_i)
​

特性：

• 计算效率高
• 归一化后等同于余弦相似度
• 在推荐系统中广泛应用

3.1.2 距离特性分析

不同的距离度量具有不同的特性，适用于不同的应用场景：

HNSW算法详解

3.1.3 HNSW算法原理

HNSW（Hierarchical Navigable Small World）是Qdrant使用的核心索引算法，专为高维向量搜索设计：

层次化结构

HNSW构建多层图结构，每层都有不同的连接密度：

class HNSWNode:
    def __init__(self, vector_id, vector, level):
        self.id = vector_id
        self.vector = vector
        self.level = level
        self.connections = {}  # level -> [neighbor_ids]
        
    def add_connection(self, level, neighbor_id):
        if level not in self.connections:
            self.connections[level] = []
        self.connections[level].append(neighbor_id)
        
    def get_connections(self, level):
        return self.connections.get(level, [])
​

![HNSW层次化结构图：展示多层图的构建过程，每层节点连接密度递减，顶层连接稀疏，底层连接密集]

贪心搜索算法

HNSW采用贪心算法进行快速搜索：

import heapq
import numpy as np

class HNSWSearch:
    def __init__(self, entry_point, max_connections=16, ef_search=50):
        self.entry_point = entry_point
        self.max_connections = max_connections
        self.ef_search = ef_search
        
    def search(self, query_vector, ef=50):
        """执行HNSW搜索"""
        # 1. 从顶层开始粗略搜索
        candidates = set()
        visited = set()
        
        # 从顶层开始搜索
        current_level = self.get_top_level()
        current_node = self.entry_point
        
        while current_level >= 0:
            # 贪心搜索：找到最接近的邻居
            neighbors = self.get_neighbors(current_node, current_level)
            
            best_neighbor = None
            best_distance = float('inf')
            
            for neighbor_id in neighbors:
                if neighbor_id in visited:
                    continue
                    
                neighbor_vector = self.get_vector(neighbor_id)
                distance = self.calculate_distance(query_vector, neighbor_vector)
                
                if distance < best_distance:
                    best_distance = distance
                    best_neighbor = neighbor_id
            
            if best_neighbor is not None:
                visited.add(best_neighbor)
                candidates.add((best_distance, best_neighbor))
                current_node = best_neighbor
            else:
                break
                
            # 如果当前层搜索到更好的结果，可以下降到下一层
            if self.should_descent(current_level, best_distance):
                current_level -= 1
                
        # 2. 在底层进行精确搜索
        results = []
        for distance, node_id in candidates:
            heapq.heappush(results, (distance, node_id))
            
        # 取top-k结果
        final_results = []
        for _ in range(min(ef, len(results))):
            distance, node_id = heapq.heappop(results)
            final_results.append((node_id, distance))
            
        return final_results
​