RAG05：向量检索
理解向量检索的核心机制，包括TopK检索、召回率优化、相似度阈值控制和噪声过滤，构建高质量的检索系统。
向量检索是 RAG 的核心环节，它决定了模型能否找到相关的知识片段。理解向量检索的机制，才能构建高效、准确的检索系统。
向量检索的基本原理
什么是向量检索
向量检索是在向量空间中找到与查询向量最相似的向量的过程。它与传统关键词检索的核心区别在于：
| 维度 | 关键词检索 | 向量检索 |
|---|---|---|
| 匹配方式 | 精确匹配或模糊匹配 | 语义相似度匹配 |
| 理解能力 | 字面含义 | 语义含义 |
| 同义词 | 需要手动配置 | 自动理解 |
| 上下文 | 不考虑 | 考虑 |
相似度度量
相似度度量决定了两个向量之间的”距离”：
| 度量方法 | 公式 | 特点 | 适用场景 |
|---|---|---|---|
| 余弦相似度 | cos(θ) = A·B / ( | A | · |
| 欧氏距离 | √Σ(Ai-Bi)² | 衡量空间距离 | 图像检索 |
| 内积 | A·B | 同时考虑方向和长度 | 需要考虑重要性时 |
| 曼哈顿距离 | Σ|Ai-Bi| | L1 距离，计算简单 | 稀疏数据 |
余弦相似度 vs 内积：
import numpy as np
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def inner_product(a, b):
return np.dot(a, b)
# 向量长度不同时的差异
a = np.array([1, 2, 3])
b = np.array([2, 4, 6])
print(f"余弦相似度: {cosine_similarity(a, b):.4f}")
print(f"内积: {inner_product(a, b)}")TopK 检索策略
TopK 基本概念
TopK 检索是返回与查询向量最相似的 K 个结果。K 值的选择直接影响检索效果：
| K 值 | 效果 | 适用场景 |
|---|---|---|
| 小 K（5-10） | 结果更精准，但可能遗漏相关内容 | 需要高精准度的场景 |
| 中 K（10-30） | 平衡精准度和召回率 | 大多数场景 |
| 大 K（30-100） | 召回率高，但噪声多 | 需要高召回率的场景 |
TopK 的实现方式
| 方法 | 说明 | 时间复杂度 | 适用数据量 |
|---|---|---|---|
| 暴力搜索 | 计算所有向量的相似度 | O(n) | 小规模数据（<1万） |
| 树索引 | 使用 KD-Tree、Ball Tree | O(log n) | 中等规模（1万-100万） |
| 向量数据库 | 使用 HNSW、IVF 等索引 | O(log n)~O(ε log n) | 大规模（>100万） |
TopK 的优化策略
# 简单的 TopK 实现
def topk_search(query_vector, vectors, k=10, similarity_fn=cosine_similarity):
scores = [(i, similarity_fn(query_vector, vec)) for i, vec in enumerate(vectors)]
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:k]
# 使用 FAISS 进行高效检索
import faiss
def faiss_search(query_vector, index, k=10):
distances, indices = index.search(query_vector.reshape(1, -1), k)
return list(zip(indices[0], distances[0]))召回率与准确率的权衡
召回率与准确率的定义
| 指标 | 定义 | 关注点 |
|---|---|---|
| 召回率 | 正确检索到的相关文档数 / 所有相关文档数 | 不漏掉相关内容 |
| 准确率 | 正确检索到的相关文档数 / 检索到的文档总数 | 检索结果精准 |
如何平衡召回率与准确率
| 策略 | 方法 | 效果 |
|---|---|---|
| 提高召回率 | 增大 K 值、降低相似度阈值 | 可能引入噪声 |
| 提高准确率 | 减小 K 值、提高相似度阈值 | 可能遗漏相关内容 |
| 两阶段检索 | 先用大 K 召回，再用精排过滤 | 平衡两者 |
| 语义扩展 | 对查询进行扩展，生成多个相关查询 | 提高召回率 |
召回率优化技巧
- 查询扩展：对原始查询进行语义扩展
- 多路召回：使用不同的检索策略并行召回
- 跨语言检索：支持多语言的语义匹配
- 时间衰减：考虑文档的时效性
相似度阈值与噪声控制
相似度阈值的作用
相似度阈值用于过滤不相关的检索结果：
| 阈值设置 | 效果 | 适用场景 |
|---|---|---|
| 高阈值（>0.8） | 只返回高度相关的结果 | 需要精准回答 |
| 中阈值（0.5-0.8） | 返回相关度适中的结果 | 大多数场景 |
| 低阈值（<0.5） | 返回更多结果，包括弱相关 | 需要全面覆盖 |
动态阈值策略
def dynamic_threshold(query_vector, results, base_threshold=0.6):
"""根据查询和结果动态调整阈值"""
if len(results) == 0:
return []
avg_score = sum(score for _, score in results) / len(results)
if avg_score > 0.8:
threshold = 0.7
elif avg_score < 0.5:
threshold = 0.4
else:
threshold = base_threshold
return [(idx, score) for idx, score in results if score >= threshold]噪声过滤方法
| 方法 | 说明 | 示例 |
|---|---|---|
| 阈值过滤 | 根据相似度分数过滤 | 过滤分数 < 0.5 的结果 |
| 重复检测 | 检测内容重复的片段 | 合并相似的段落 |
| 长度过滤 | 过滤过短或过长的片段 | 过滤 < 10 词或 > 500 词的内容 |
| 质量评分 | 根据内容质量评分过滤 | 过滤低质量的摘要 |
向量数据库选择
常见向量数据库对比
| 数据库 | 核心特点 | 适用场景 | 部署复杂度 |
|---|---|---|---|
| FAISS | 高性能，Facebook 开源 | 单机、小规模 | 低 |
| Milvus | 分布式，支持多种索引 | 大规模、高并发 | 中 |
| Pinecone | 托管服务，易用 | 快速上线、不想运维 | 低（托管） |
| Chroma | 轻量级，易用 | 开发测试、小型项目 | 低 |
| Weaviate | 支持语义搜索和 GraphQL | 需要高级查询功能 | 中 |
索引类型选择
| 索引类型 | 原理 | 特点 | 适用场景 |
|---|---|---|---|
| HNSW | 层次化导航小世界图 | 速度快、召回率高 | 大多数场景 |
| IVF | 倒排文件索引 | 适合大规模数据 | 超大规模数据 |
| FLAT | 暴力搜索 | 召回率100% | 小规模数据 |
| PQ | 乘积量化 | 内存占用小 | 内存受限场景 |
向量检索的常见问题
问题1：召回率低
表现：检索不到相关的知识片段
解决方案：
- 增大 K 值
- 降低相似度阈值
- 优化 Embedding 模型
- 使用查询扩展
问题2：噪声过多
表现：检索结果包含大量不相关内容
解决方案：
- 提高相似度阈值
- 使用更严格的过滤策略
- 增加重排序环节
问题3：检索速度慢
表现：检索延迟高，无法满足实时需求
解决方案：
- 使用更高效的索引类型（如 HNSW）
- 减少向量维度
- 使用向量数据库的缓存机制
问题4：语义漂移
表现：检索结果与查询语义不一致
解决方案：
- 使用更好的 Embedding 模型
- 优化文档切分策略
- 添加业务规则过滤
向量检索的最佳实践
检索流程
1. 查询预处理
- 文本清洗
- 查询扩展（可选）
2. 向量转换
- 使用 Embedding 模型将查询转换为向量
3. 向量检索
- 使用 TopK 检索获取候选结果
- 根据相似度阈值过滤
4. 结果后处理
- 去重
- 排序
- 格式化输出评估指标
1. 召回率（Recall）：> 90%
2. 准确率（Precision）：> 80%
3. 检索延迟：< 100ms（P99）
4. TopK 命中率：> 95%优化策略
# 完整的向量检索流程
class VectorRetriever:
def __init__(self, index, embedding_model, k=10, threshold=0.5):
self.index = index
self.embedding_model = embedding_model
self.k = k
self.threshold = threshold
def retrieve(self, query):
# 1. 向量化
query_vector = self.embedding_model.encode(query)
# 2. TopK 检索
distances, indices = self.index.search(query_vector.reshape(1, -1), self.k)
# 3. 阈值过滤
results = []
for idx, dist in zip(indices[0], distances[0]):
similarity = 1 - dist # 转换为相似度
if similarity >= self.threshold:
results.append((idx, similarity))
return results项目判断清单
- 数据量小（<1万条）→ 使用 FAISS 或 Chroma
- 数据量大（>100万条）→ 使用 Milvus 或 Pinecone
- 需要分布式部署 → 使用 Milvus
- 需要快速上线 → 使用 Pinecone 或 Chroma
- 召回率低 → 增大 K 值或优化 Embedding 模型
- 噪声过多 → 提高阈值或增加重排序
- 检索慢 → 使用 HNSW 索引或减少向量维度
- 需要支持多语言 → 使用多语言 Embedding 模型