混合检索

混合检索是一种结合稀疏检索和密集检索优势的检索策略，它通过融合两种检索方法的结果，提高了RAG系统的整体检索效果。混合检索既保留了稀疏检索的高效率和可解释性，又获得了密集检索的语义理解能力。

1. 基本原理

核心思想

• 互补优势：稀疏检索擅长关键词匹配，密集检索擅长语义理解
• 结果融合：将两种检索方法的结果按照一定权重融合
• 平衡性能：在效率和准确性之间取得平衡

工作流程

1. 并行检索：同时执行稀疏检索和密集检索
2. 结果排序：对两种检索结果分别排序
3. 权重融合：根据预设权重融合排序结果
4. 最终排序：生成最终的检索结果

2. 融合策略

线性融合

最简单的融合方法是线性加权，根据两种检索方法的置信度分配权重。

def linear_fusion(sparse_scores, dense_scores, sparse_weight=0.3, dense_weight=0.7):
    """线性融合稀疏和密集检索结果"""
    fused_scores = {}
    
    # 处理稀疏检索结果
    for doc_id, score in sparse_scores.items():
        fused_scores[doc_id] = score * sparse_weight
    
    # 处理密集检索结果
    for doc_id, score in dense_scores.items():
        if doc_id in fused_scores:
            fused_scores[doc_id] += score * dense_weight
        else:
            fused_scores[doc_id] = score * dense_weight
    
    # 按分数排序
    sorted_docs = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_id for doc_id, score in sorted_docs]

reciprocal rank fusion (RRF)

RRF是一种更复杂的融合方法，它基于文档在不同检索结果中的排名。

def reciprocal_rank_fusion(sparse_results, dense_results, k=60):
    """
    RRF融合方法
    k: 常数，防止高排名文档的分数过于集中
    """
    fused_scores = {}
    
    # 处理稀疏检索结果
    for rank, doc_id in enumerate(sparse_results):
        fused_scores[doc_id] = fused_scores.get(doc_id, 0) + 1 / (k + rank + 1)
    
    # 处理密集检索结果
    for rank, doc_id in enumerate(dense_results):
        fused_scores[doc_id] = fused_scores.get(doc_id, 0) + 1 / (k + rank + 1)
    
    # 按分数排序
    sorted_docs = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_id for doc_id, score in sorted_docs]

加权融合

根据不同查询类型动态调整权重：

def weighted_fusion(query, sparse_results, dense_results):
    """根据查询特征动态调整权重"""
    # 判断查询类型
    if is_keyword_query(query):  # 关键词查询
        sparse_weight, dense_weight = 0.7, 0.3
    elif is_semantic_query(query):  # 语义查询
        sparse_weight, dense_weight = 0.2, 0.8
    else:  # 混合查询
        sparse_weight, dense_weight = 0.4, 0.6
    
    return linear_fusion(sparse_results, dense_results, sparse_weight, dense_weight)

3. 实现示例

完整混合检索系统

from langchain.embeddings import HuggingFaceEmbeddings
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import jieba

class HybridRetriever:
    def __init__(self, documents, sparse_weight=0.3, dense_weight=0.7):
        self.documents = documents
        self.sparse_weight = sparse_weight
        self.dense_weight = dense_weight
        
        # 初始化密集检索
        self.embeddings = HuggingFaceEmbeddings()
        self.doc_vectors = [self.embeddings.embed_query(doc) for doc in documents]
        
        # 初始化稀疏检索
        self.tokenized_docs = [' '.join(jieba.lcut(doc)) for doc in documents]
        self.vectorizer = TfidfVectorizer()
        self.tfidf_matrix = self.vectorizer.fit_transform(self.tokenized_docs)
    
    def sparse_search(self, query, top_k=5):
        """稀疏检索"""
        tokenized_query = ' '.join(jieba.lcut(query))
        query_vector = self.vectorizer.transform([tokenized_query])
        similarities = cosine_similarity(query_vector, self.tfidf_matrix)[0]
        
        top_indices = np.argsort(similarities)[::-1][:top_k]
        return {idx: similarities[idx] for idx in top_indices}
    
    def dense_search(self, query, top_k=5):
        """密集检索"""
        query_vector = self.embeddings.embed_query(query)
        similarities = [cosine_similarity([query_vector], [doc_vec])[0][0] 
                       for doc_vec in self.doc_vectors]
        
        top_indices = np.argsort(similarities)[::-1][:top_k]
        return {idx: similarities[idx] for idx in top_indices}
    
    def search(self, query, top_k=5):
        """混合检索"""
        sparse_results = self.sparse_search(query, top_k)
        dense_results = self.dense_search(query, top_k)
        
        # 融合结果
        fused_scores = {}
        for idx, score in sparse_results.items():
            fused_scores[idx] = score * self.sparse_weight
        
        for idx, score in dense_results.items():
            if idx in fused_scores:
                fused_scores[idx] += score * self.dense_weight
            else:
                fused_scores[idx] = score * self.dense_weight
        
        # 排序并返回文档
        sorted_indices = sorted(fused_scores.keys(), 
                               key=lambda x: fused_scores[x], reverse=True)
        return [self.documents[idx] for idx in sorted_indices[:top_k]]

# 使用示例
documents = [
    "RAG是一种结合检索和生成的技术",
    "TF-IDF是一种传统的信息检索算法",
    "向量数据库用于存储和检索嵌入向量",
    "机器学习是人工智能的一个分支",
    "深度学习是机器学习的一种方法"
]

retriever = HybridRetriever(documents)
results = retriever.search("什么是检索技术？", top_k=3)
for doc in results:
    print(doc)

4. 权重调优

经验法则

• 关键词查询为主：sparse_weight=0.6, dense_weight=0.4
• 语义查询为主：sparse_weight=0.2, dense_weight=0.8
• 平衡方案：sparse_weight=0.3, dense_weight=0.7

自动调优

def tune_weights(validation_queries, ground_truth):
    """通过验证集自动调优权重"""
    best_weights = (0.5, 0.5)
    best_score = 0
    
    for sparse_w in np.arange(0, 1.1, 0.1):
        dense_w = 1 - sparse_w
        score = evaluate_weights(sparse_w, dense_w, validation_queries, ground_truth)
        if score > best_score:
            best_score = score
            best_weights = (sparse_w, dense_w)
    
    return best_weights

5. 优缺点分析

优点

• 综合优势：结合两种检索方法的优点
• 鲁棒性强：对不同类型的查询都有较好的表现
• 可配置性：可以根据场景调整权重

缺点

• 计算成本：需要执行两次检索
• 复杂度增加：系统复杂度提高
• 权重调优：需要调优权重参数

6. 最佳实践

查询分类

根据查询类型自动选择检索策略：

def classify_query(query):
    """分类查询类型"""
    # 关键词特征：包含特定术语、代码、名称
    # 语义特征：自然语言问题、描述性查询
    if contains_specific_terms(query):
        return "keyword"
    elif is_natural_language(query):
        return "semantic"
    else:
        return "hybrid"

缓存优化

from functools import lru_cache

class CachedHybridRetriever(HybridRetriever):
    @lru_cache(maxsize=1000)
    def search(self, query, top_k=5):
        return super().search(query, top_k)