4.1 相似度计算算法

4.1 相似度计算算法 — RAG 知识库实战核心检索技术

本节导读：深入理解 RAG 系统中的相似度计算原理，掌握从基础距离度量到高级语义匹配算法的完整技术栈。学完本节，你将能够根据业务场景选择最优的相似度策略，并实现高效的向量相似度计算。

学习目标

• 掌握余弦相似度、欧氏距离、点积等基础距离度量的数学原理与适用场景
• 理解最大内积搜索（MIPS）在 RAG 检索中的核心地位
• 学会用 Python 实现多种相似度计算算法并做性能对比
• 了解批次计算、NumPy 向量化、近似搜索等性能优化手段
• 能够根据嵌入模型特性和业务需求做出合理的相似度策略选型

核心概念

相似度计算是 RAG 系统检索引擎的"度量尺"——它决定了当用户提出一个问题时，系统如何从数以万计的文档向量中找到"最相关"的那些。选错度量方式，检索质量会大打折扣；选对了，后续的重排和生成才有好的输入基础。

从"距离"到"相似度"

在向量空间中，我们用两种互补的视角看待向量间的关系：

• 距离（Distance）：值越小表示越接近，如欧氏距离、曼哈顿距离
• 相似度（Similarity）：值越大表示越接近，如余弦相似度、皮尔逊相关系数

RAG 系统通常使用相似度而非距离，因为"得分越高越相关"这个直觉更符合排序逻辑。但距离和相似度之间可以相互转换——比如对欧氏距离取负数或用核函数映射。

为什么相似度选择如此重要？

一个常被忽略的事实：不同的嵌入模型对不同的相似度度量敏感程度不同。

• 归一化嵌入（如 OpenAI text-embedding-3-small、BGE 系列）：输出向量的 L2 范数已归一化到 1，此时余弦相似度等价于点积，计算更高效
• 未归一化嵌入（部分老模型）：向量长度不一致，必须用余弦相似度消除长度影响
• 对比学习模型（如 SimCSE）：训练时就用余弦相似度做对比，推理时也应保持一致

核心结论：90% 的 RAG 场景用余弦相似度就够了，但"够用"和"最优"之间可能差 5-10% 的检索准确率。本节会帮你判断你的场景是否属于那需要优化的 10%。

环境准备

# 核心依赖
numpy>=1.24.0
scipy>=1.10.0
scikit-learn>=1.3.0

# 可选：用于大规模近似搜索的测试
faiss-cpu>=1.7.4    # CPU 版
# faiss-gpu>=1.7.4  # GPU 版，需要 CUDA
​

前置知识：线性代数基础（向量点积、范数）、Python NumPy 基本操作。

分步实战

步骤 1：实现四大基础相似度算法

先从最基础的四种度量方式入手，理解各自的数学含义和代码实现。

1. 余弦相似度（Cosine Similarity）—— RAG 的默认选择

余弦相似度衡量两个向量方向上的一致性，忽略向量的绝对大小。公式如下：

取值范围 [-1, 1]，RAG 中通常为 [0, 1]（因为嵌入模型的输出多在正半空间）。

import numpy as np

def cosine_similarity(query: np.ndarray, docs: np.ndarray) -> np.ndarray:
    """
    计算查询向量与文档向量矩阵的余弦相似度。
    
    Args:
        query: 形状 (dim,) 的查询向量
        docs: 形状 (n_docs, dim) 的文档向量矩阵
    
    Returns:
        形状 (n_docs,) 的相似度得分数组
    """
    # 分子：query 与每个 doc 的点积
    numerator = docs @ query
    # 分母：各自的 L2 范数
    query_norm = np.linalg.norm(query)
    doc_norms = np.linalg.norm(docs, axis=1)
    # 防止除零
    denominator = query_norm * doc_norms
    denominator = np.maximum(denominator, 1e-10)
    return numerator / denominator
​

2. 点积（Dot Product / Inner Product）—— 最快的度量

当向量已归一化时，点积等价于余弦相似度，但省去了范数计算，速度更快。

def dot_product_similarity(query: np.ndarray, docs: np.ndarray) -> np.ndarray:
    """
    点积相似度。适用于已归一化的嵌入向量。
    当向量 L2 范数为 1 时，结果与余弦相似度完全一致。
    """
    return docs @ query
​

3. 欧氏距离（Euclidean Distance）—— 考虑绝对差异

欧氏距离衡量两个向量在空间中的直线距离，值越小越相似。

def euclidean_similarity(query: np.ndarray, docs: np.ndarray) -> np.ndarray:
    """
    将欧氏距离转换为相似度得分（使用高斯核映射）。
    原始距离越小，转换后相似度越高。
    """
    distances = np.linalg.norm(docs - query, axis=1)
    # 高斯核转换：σ 控制衰减速度
    sigma = 1.0
    return np.exp(-distances**2 / (2 * sigma**2))
​

4. 曼哈顿距离（Manhattan / L1 Distance）—— 对异常值更鲁棒

def manhattan_similarity(query: np.ndarray, docs: np.ndarray) -> np.ndarray:
    """
    曼哈顿距离转相似度。对高维空间中的异常值维度不敏感，
    在某些稀疏向量场景下表现优于欧氏距离。
    """
    distances = np.sum(np.abs(docs - query), axis=1)
    max_dist = np.max(distances) if np.max(distances) > 0 else 1.0
    return 1.0 - distances / max_dist
​

步骤 2：对比测试——哪种度量最适合你的数据？

光看公式不够，用真实嵌入数据跑一下对比。这里用 scikit-learn 的 TF-IDF 生成向量做演示（实际 RAG 中应替换为你的嵌入模型输出）。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import normalize

# 模拟 RAG 文档库
documents = [
    "RAG 系统通过检索外部知识增强大语言模型的回答质量",
    "向量数据库是存储和检索高维向量的专用数据库系统",
    "文档分割策略直接影响 RAG 系统的检索精度和上下文质量",
    "嵌入模型将文本转换为高维向量表示以捕获语义信息",
    "重排序模型对初步检索结果进行精细排序以提升相关性",
    "提示工程是优化大语言模型输出的关键技术手段",
    "混合检索结合稀疏检索和稠密检索的优点提升召回率",
    "知识图谱可以与 RAG 系统结合实现结构化知识推理",
]

query = "如何提升 RAG 系统的检索准确性"

# 生成 TF-IDF 向量（实际项目替换为嵌入模型）
vectorizer = TfidfVectorizer()
doc_vectors = vectorizer.fit_transform(documents).toarray().astype(np.float32)
doc_vectors = normalize(doc_vectors)  # L2 归一化
query_vector = vectorizer.transform([query]).toarray().astype(np.float32)
query_vector = normalize(query_vector)

# 计算四种相似度
cos_scores = cosine_similarity(query_vector.flatten(), doc_vectors)
dot_scores = dot_product_similarity(query_vector.flatten(), doc_vectors)
euc_scores = euclidean_similarity(query_vector.flatten(), doc_vectors)
man_scores = manhattan_similarity(query_vector.flatten(), doc_vectors)

# 输出 Top-3 对比
print(f"{'文档':<40} {'余弦':>6} {'点积':>6} {'欧氏':>6} {'曼哈顿':>6}")
print("-" * 70)
for i in np.argsort(-cos_scores)[:3]:
    print(f"{documents[i][:38]:<40} {cos_scores[i]:>6.4f} {dot_scores[i]:>6.4f} "
          f"{euc_scores[i]:>6.4f} {man_scores[i]:>6.4f}")
​

运行后你会发现：在归一化向量上，余弦相似度和点积的排序结果完全一致，但点积的计算速度快约 30%（省去了范数计算）。这就是为什么生产环境中，如果确认嵌入模型输出已归一化，优先使用点积。

步骤 3：最大内积搜索（MIPS）—— 大规模 RAG 的底层引擎

当文档数量从几百扩展到百万级时，逐一计算相似度变得不可接受。MIPS（Maximum Inner Product Search）是 RAG 检索的核心计算问题。

MIPS 的挑战：暴力搜索的时间复杂度是 O(n × d)（n=文档数，d=向量维度），百万级 768 维向量意味着每秒数亿次浮点运算。

解决方案演进：

步骤 4：性能优化实战——从 2 秒到 20 毫秒

import time

def benchmark_similarity(n_docs=100000, dim=768, top_k=10):
    """对比不同实现方式的查询延迟"""
    np.random.seed(42)
    docs = np.random.randn(n_docs, dim).astype(np.float32)
    docs = docs / np.linalg.norm(docs, axis=1, keepdims=True)  # 归一化
    query = np.random.randn(dim).astype(np.float32)
    query = query / np.linalg.norm(query)
    
    # 方法 1：纯 Python 循环（最慢，仅作对比基准）
    start = time.time()
    scores_py = []
    for i in range(min(n_docs, 1000)):
        score = np.dot(query, docs[i])
        scores_py.append(score)
    time_py = time.time() - start
    
    # 方法 2：NumPy 向量化点积
    start = time.time()
    scores_np = docs @ query  # 矩阵乘法
    topk_idx = np.argpartition(-scores_np, top_k)[:top_k]
    time_np = time.time() - start
    
    # 方法 3：FAISS 索引搜索
    import faiss
    index = faiss.IndexFlatIP(dim)  # 内积索引
    index.add(docs)
    start = time.time()
    scores_faiss, idx_faiss = index.search(query.reshape(1, -1), top_k)
    time_faiss = time.time() - start
    
    print(f"{'方法':<25} {'耗时':>10} {'加速比':>10}")
    print("-" * 50)
    print(f"{'Python 循环(1K)':<25} {time_py*1000:>8.1f}ms {'1x':>10}")
    print(f"{'NumPy 向量化(100K)':<25} {time_np*1000:>8.1f}ms {'~1000x':>10}")
    print(f"{'FAISS 索引(100K)':<25} {time_faiss*1000:>8.1f}ms {'~10000x':>10}")
    
    return idx_faiss.flatten()

# 运行基准测试
top_results = benchmark_similarity()
print(f"\nTop-10 文档 ID: {top_results}")
​

在典型服务器上，100K × 768 维的场景下：NumPy 向量化约 2ms，FAISS 约 0.2ms。关键优化点：确保使用 float32 而非 float64（速度提升约 2 倍），确保查询向量形状为 (1, dim) 以利用 FAISS 的批量优化。

完整示例：生产级相似度检索器

以下是一个可直接用于 RAG 项目的相似度检索器封装，支持多种度量方式和 FAISS 加速：

import numpy as np
import faiss
from typing import List, Tuple, Optional

class SimilarityRetriever:
    """
    生产级向量相似度检索器。
    
    支持：余弦相似度、点积、欧氏距离
    后端：FAISS 索引加速（自动回退到 NumPy）
    """
    
    METRICS = {
        'cosine': faiss.METRIC_INNER_PRODUCT,   # 归一化后等价
        'dot_product': faiss.METRIC_INNER_PRODUCT,
        'l2': faiss.METRIC_L2,
    }
    
    def __init__(
        self,
        dim: int = 768,
        metric: str = 'cosine',
        use_faiss: bool = True,
        normalize: bool = True,
    ):
        self.dim = dim
        self.metric = metric
        self.normalize = normalize
        self.use_faiss = use_faiss
        self._index = None
        self._doc_ids: List[int] = []
        self._is_normalized = False
    
    def build_index(self, vectors: np.ndarray, doc_ids: Optional[List[int]] = None):
        """
        构建检索索引。
        
        Args:
            vectors: 形状 (n, dim) 的文档向量矩阵
            doc_ids: 对应的文档 ID 列表
        """
        assert vectors.shape[1] == self.dim
        n = vectors.shape[0]
        self._doc_ids = doc_ids or list(range(n))
        
        # 根据度量方式选择是否归一化
        if self.metric in ('cosine', 'dot_product') and self.normalize:
            vectors = vectors / np.linalg.norm(
                vectors, axis=1, keepdims=True
            ).clip(min=1e-10)
            self._is_normalized = True
        
        if self.use_faiss:
            faiss_metric = self.METRICS[self.metric]
            self._index = faiss.IndexFlatIP(self.dim) if faiss_metric == faiss.METRIC_INNER_PRODUCT else faiss.IndexFlatL2(self.dim)
            self._index.add(vectors.astype(np.float32))
        else:
            self._index = vectors  # 直接存 NumPy 数组
    
    def search(
        self, query: np.ndarray, top_k: int = 5
    ) -> List[Tuple[int, float]]:
        """
        检索最相似的文档。
        
        Returns:
            [(doc_id, score), ...] 按相似度降序排列
        """
        if self.metric in ('cosine', 'dot_product') and self.normalize:
            query = query / np.linalg.norm(query).clip(min=1e-10)
        
        if self.use_faiss:
            scores, indices = self._index.search(
                query.reshape(1, -1).astype(np.float32), top_k
            )
            results = []
            for score, idx in zip(scores[0], indices[0]):
                if idx >= 0:  # FAISS 用 -1 表示无效索引
                    results.append((self._doc_ids[idx], float(score)))
            return results
        else:
            scores = self._index @ query
            top_indices = np.argpartition(-scores, top_k)[:top_k]
            top_indices = top_indices[np.argsort(-scores[top_indices])]
            return [(self._doc_ids[i], float(scores[i])) for i in top_indices]
    
    @property
    def doc_count(self) -> int:
        return len(self._doc_ids)


# 使用示例
if __name__ == '__main__':
    # 模拟 5 万篇文档的嵌入向量（768 维，已归一化）
    n_docs = 50000
    dim = 768
    np.random.seed(42)
    doc_vectors = np.random.randn(n_docs, dim).astype(np.float32)
    doc_vectors /= np.linalg.norm(doc_vectors, axis=1, keepdims=True)
    
    # 构建检索器
    retriever = SimilarityRetriever(dim=dim, metric='cosine')
    retriever.build_index(doc_vectors)
    
    # 模拟用户查询
    query = np.random.randn(dim).astype(np.float32)
    
    # 检索 Top-5
    results = retriever.search(query, top_k=5)
    for rank, (doc_id, score) in enumerate(results, 1):
        print(f"  #{rank} 文档ID={doc_id}, 相似度={score:.4f}")
​

步骤 5：理解相似度分数的"校准"问题

在实际项目中，你很快会遇到一个令人困惑的现象：同样的查询和文档，用不同嵌入模型得到的相似度分数范围完全不同。OpenAI 的 text-embedding-3-small 输出的最高分可能只有 0.65，而 BGE-large-zh 的最高分可能达到 0.92。这会让设置相似度阈值变得困难。

根本原因：不同模型训练时使用的对比损失函数不同，输出向量的分布也不同。有些模型训练时就把正样本对的余弦相似度推向 0.7 左右（温度系数的影响），有些则推向 0.95。

实用对策：不要依赖绝对阈值做过滤，而是用以下两种策略之一：

# 策略 1：基于排名比例的动态阈值
def dynamic_threshold(scores: np.ndarray, ratio: float = 0.7) -> float:
    """
    取 Top-1 分数的 ratio 倍作为阈值。
    例如 Top-1=0.65, ratio=0.7, 则阈值为 0.455
    """
    return float(np.max(scores) * ratio)

# 策略 2：基于分数分布的统计阈值
def statistical_threshold(scores: np.ndarray, z_score: float = -1.5) -> float:
    """
    取均值减去 z_score 个标准差作为阈值。
    适用于分数分布相对稳定的情况。
    """
    mean = np.mean(scores)
    std = np.std(scores)
    return float(mean + z_score * std)

# 使用示例
threshold = dynamic_threshold(scores_array)
filtered = [(did, s) for did, s in results if s >= threshold]
​

策略 1 在实践中更稳健——它不依赖分数的绝对分布，只依赖"最相关文档"的分数作为锚点。当你的检索 Top-1 质量可靠时，这个策略能有效过滤掉不相关的噪声文档。在实际部署中，我建议先用策略 1 跑一批测试查询，观察过滤后的 Precision@K 变化，再决定是否需要更精细的策略 2。

常见问题 FAQ

Q1：RAG 系统中余弦相似度和点积有什么区别？该用哪个？

A：当嵌入向量的 L2 范数已归一化为 1 时（绝大多数现代嵌入模型默认如此），余弦相似度和点积在数学上完全等价，排序结果一模一样。区别仅在计算效率：点积省去了分母的范数运算，速度快约 20-30%。实际建议：如果你的嵌入模型确定输出归一化向量（如 OpenAI text-embedding 系列、BGE 系列、E5 系列），直接用点积；如果不确定，先用余弦相似度兜底。

Q2：FAISS 和纯 NumPy 向量搜索差距有多大？

A：差距取决于数据规模。1 万条以下差异不明显（都在毫秒级）；10 万到 100 万条时，FAISS 的 HNSW 索引比 NumPy 暴力搜索快 50-500 倍。但 FAISS 的优势不仅在于速度——它还提供了 GPU 加速、量化压缩（PQ/SQ）和多索引融合等高级功能，这些都是生产环境必需的。如果文档量超过 5 万，强烈建议使用 FAISS。

Q3：相似度分数很低（比如最高才 0.3）是否正常？

A：完全正常，不需要担心。相似度的绝对值没有跨模型的统一标准——OpenAI 的嵌入通常在 0.2-0.8 之间，BGE 模型可能在 0.5-0.95 之间。重要的是相对排序而非绝对值。只要最相关的文档排在前面，分数低不是问题。如果排序质量不好，应该排查嵌入模型是否适合你的语料，而不是去调相似度的阈值。

Q4：能否混合使用多种相似度度量？

A：可以，这就是"混合检索"的思想（详见本教程 4.4 节混合检索方法）。典型做法是将稠密向量的余弦相似度与稀疏向量（如 BM25 得分）通过 RRF（Reciprocal Rank Fusion）融合。但注意：不要同时使用余弦相似度和欧氏距离对同一组稠密向量做双重计算再融合——在归一化向量上它们提供的排序信息是冗余的，融合不会带来提升。

最佳实践与避坑

• 确认嵌入模型是否归一化：写一段代码打印 10 个随机文档向量的 L2 范数，如果都接近 1.0，则可以直接用点积替代余弦相似度
• 不要忽略 float32：确保向量和查询都是 float32 类型。float64 不仅慢一倍，FAISS 部分索引也不支持
• Top-K 不要设太大：RAG 检索中 Top-5 到 Top-10 通常是最佳范围。超过 20 个文档送入 LLM 不仅增加成本，还会因为噪声文档干扰生成质量
• 批次查询优于单条查询：如果需要同时查询多个问题，拼成批次 (batch_size, dim) 一次性传入 FAISS，利用矩阵运算的并行性
• 监控相似度分布：定期统计 Top-1 分数的均值和 P95。如果均值持续下降，可能意味着新入库的文档与查询语义空间不匹配，需要重新训练嵌入模型或调整分割策略

本节小结

本节从数学原理出发，系统讲解了 RAG 系统中四种核心相似度计算算法——余弦相似度、点积、欧氏距离和曼哈顿距离，并通过代码实现了完整的对比测试。我们进一步探讨了大规模场景下的 MIPS 问题和 FAISS 加速方案，最终封装了一个可直接用于生产环境的 SimilarityRetriever 类。

核心要点：大多数 RAG 场景用余弦相似度（或等价的点积）就足够了，但理解底层原理能帮你在遇到检索质量瓶颈时做出正确的优化决策。下一节 4.2 将在这个基础上，讲解如何设计更复杂的检索策略——包括多路召回、查询改写和元数据过滤，进一步提升检索系统的整体表现。

延伸阅读

• 官方文档：FAISS 官方文档（Facebook AI Research），包含索引类型详解和 GPU 使用指南
• 相关论文：Johnson 等人 2019 年发表的 FAISS 论文，系统介绍了高效相似度搜索的工程实现
• 相关章节：本教程 3.4 节向量索引与检索优化（详解 HNSW、IVF 等索引结构），4.4 节混合检索方法（结合稠密与稀疏检索的融合策略）

关键词：RAG 知识库实战, 相似度计算, 余弦相似度, 点积, FAISS, 向量检索, MIPS, 检索优化
难度：入门到进阶
预计阅读：15 分钟