稀疏检索

稀疏检索是RAG系统中最传统的检索方法，基于关键词匹配和统计分析。虽然不如密集检索那样能捕捉语义关系，但它具有计算效率高、可解释性强的特点，仍然是RAG系统中的重要组成部分。

1. 基本原理

核心概念

• 词频（TF）：词语在文档中出现的频率
• 逆文档频率（IDF）：词语在整个语料库中的稀有程度
• 文档长度归一化：考虑文档长度对词频的影响

工作流程

1. 文本预处理：分词、去停用词、词干提取
2. 特征提取：计算每个词的权重
3. 相似度计算：计算查询与文档的相似度
4. 排序：按相似度排序返回结果

2. 常用算法

TF-IDF

TF-IDF（Term Frequency-Inverse Document Frequency）是最经典的稀疏检索算法。

计算公式

TF(t, d) = 词t在文档d中的出现次数 / 文档d的总词数
IDF(t) = log(语料库中文档总数 / 包含词t的文档数)
TF-IDF(t, d) = TF(t, d) × IDF(t)

实现示例

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# 文档集合
documents = [
    "RAG是一种结合检索和生成的技术",
    "TF-IDF是一种传统的信息检索算法",
    "向量数据库用于存储和检索嵌入向量"
]

# 创建TF-IDF向量化器
vectorizer = TfidfVectorizer()
doc_vectors = vectorizer.fit_transform(documents)

# 查询
query = "什么是RAG技术"
query_vector = vectorizer.transform([query])

# 计算相似度
similarities = cosine_similarity(query_vector, doc_vectors)
print(f"相似度: {similarities}")

BM25

BM25是TF-IDF的改进版本，在信息检索领域表现更好。

特点

• 考虑了文档长度归一化
• 对词频进行饱和处理
• 参数可调，适应不同场景

实现示例

from rank_bm25 import BM25Okapi

# 文档集合（已分词）
documents = [
    ["RAG", "是", "一种", "结合", "检索", "和", "生成", "的", "技术"],
    ["TF-IDF", "是", "一种", "传统", "的", "信息", "检索", "算法"],
    ["向量", "数据库", "用于", "存储", "和", "检索", "嵌入", "向量"]
]

# 创建BM25模型
bm25 = BM25Okapi(documents)

# 查询（已分词）
query = ["RAG", "技术"]
scores = bm25.get_scores(query)
print(f"BM25分数: {scores}")

3. 中文处理

分词

中文需要专门的分词工具：

import jieba

# 分词
text = "RAG是一种结合检索和生成的技术"
words = jieba.lcut(text)
print(words)
# 输出: ['RAG', '是', '一种', '结合', '检索', '和', '生成', '的', '技术']

停用词处理

# 加载停用词
with open('stopwords.txt', 'r', encoding='utf-8') as f:
    stopwords = set(f.read().splitlines())

# 去除停用词
words = [w for w in words if w not in stopwords]

4. 优缺点分析

优点

• 计算效率高：基于稀疏矩阵，计算速度快
• 可解释性强：可以清楚地看到匹配的关键词
• 无需训练：不需要预训练模型
• 适合精确匹配：对特定术语和名称匹配效果好

缺点

• 语义理解弱：无法理解同义词和语义关系
• 对变形敏感：无法处理词形变化
• 需要良好的分词：中文分词质量影响检索效果

5. 应用场景

• 精确检索：查找特定术语、名称、代码
• 大规模数据：数据量巨大时的快速筛选
• 混合检索：与密集检索结合使用
• 关键词搜索：传统的关键词匹配场景

6. 与密集检索的结合

稀疏检索和密集检索可以互补，通常采用混合策略：

def hybrid_search(query, sparse_weight=0.3, dense_weight=0.7):
    # 稀疏检索
    sparse_scores = sparse_retriever.search(query)
    
    # 密集检索
    dense_scores = dense_retriever.search(query)
    
    # 融合结果
    final_scores = {}
    for doc_id in set(sparse_scores.keys()) | set(dense_scores.keys()):
        sparse_score = sparse_scores.get(doc_id, 0)
        dense_score = dense_scores.get(doc_id, 0)
        final_scores[doc_id] = sparse_weight * sparse_score + dense_weight * dense_score
    
    # 排序返回
    return sorted(final_scores.items(), key=lambda x: x[1], reverse=True)