RAG06：混合检索 | 旋大爷的博客

混合检索是将多种检索策略结合起来，取长补短，提升检索效果。单一的检索方式有其局限性，混合检索能更全面地覆盖用户需求。

混合检索的必要性

单一检索方式的局限性

检索方式	优点	缺点	典型问题
关键词检索	精确、高效、可控	无法理解语义、同义词处理困难	”苹果”无法同时匹配水果和公司
向量检索	语义理解、同义词自动匹配	可能遗漏精确匹配、噪声较多	精确术语可能被语义相似内容淹没
结构化检索	精确过滤、条件明确	无法处理非结构化内容	无法检索自由文本

混合检索的价值

混合检索通过多种方式互补：

关键词检索 → 精确匹配专业术语
向量检索   → 语义理解和同义词扩展
结构化检索 → 按类别、时间等条件过滤
业务规则   → 按业务逻辑进行优先级调整

混合检索的策略

策略1：并行检索 + 结果融合

同时使用多种检索方式，然后融合结果：

步骤	说明	示例
并行检索	同时进行关键词、向量、结构化检索	分别获取3组结果
结果去重	去除重复的文档	相同文档只保留一次
分数融合	合并不同检索方式的分数	加权求和或归一化
最终排序	根据融合分数排序	返回 TopK 结果

分数融合示例：

def fuse_results(keyword_results, vector_results, struct_results, weights=[0.3, 0.5, 0.2]):
    """融合多种检索结果"""
    result_map = {}
    
    # 关键词检索结果
    for doc_id, score in keyword_results:
        result_map[doc_id] = result_map.get(doc_id, 0) + score * weights[0]
    
    # 向量检索结果
    for doc_id, score in vector_results:
        result_map[doc_id] = result_map.get(doc_id, 0) + score * weights[1]
    
    # 结构化检索结果（匹配为1，不匹配为0）
    for doc_id, match in struct_results:
        result_map[doc_id] = result_map.get(doc_id, 0) + match * weights[2]
    
    # 排序返回
    sorted_results = sorted(result_map.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

策略2：串联检索

按顺序使用多种检索方式，逐步缩小范围：

步骤	说明	目的
结构化过滤	先按类别、时间等条件过滤	缩小检索范围
关键词检索	在过滤后的结果中进行关键词检索	精确匹配术语
向量检索	在剩余结果中进行语义检索	补充语义相关内容

串联检索示例：

def cascade_retrieval(query, docs, category_filter=None, time_range=None):
    """串联检索：先过滤，再关键词，最后向量"""
    # 1. 结构化过滤
    filtered_docs = docs
    if category_filter:
        filtered_docs = [d for d in filtered_docs if d.category == category_filter]
    if time_range:
        filtered_docs = [d for d in filtered_docs if time_range[0] <= d.date <= time_range[1]]
    
    # 2. 关键词检索
    keyword_results = keyword_search(query, filtered_docs)
    
    # 3. 向量检索（补充关键词没找到的）
    keyword_ids = {r[0] for r in keyword_results}
    remaining_docs = [d for d in filtered_docs if d.id not in keyword_ids]
    vector_results = vector_search(query, remaining_docs)
    
    return keyword_results + vector_results

策略3：增强检索

使用一种检索方式作为主检索，其他方式作为增强：

主检索	增强方式	适用场景
向量检索	关键词过滤	需要精确术语匹配的场景
关键词检索	向量扩展	需要语义扩展的场景
结构化检索	向量排序	需要精确过滤+语义排序的场景

关键词检索与向量检索的结合

BM25 + 向量检索

BM25 是经典的关键词检索算法，与向量检索结合效果很好：

组件	作用	优势
BM25	精确匹配关键词	专业术语、代码、人名等精确匹配
向量检索	语义理解	同义词、语义相似内容

结合策略：

from rank_bm25 import BM25Okapi

def bm25_vector_hybrid(query, docs, vector_index, embedding_model, k=10):
    """BM25 + 向量检索混合"""
    # BM25 检索
    tokenized_docs = [doc.text.split() for doc in docs]
    bm25 = BM25Okapi(tokenized_docs)
    tokenized_query = query.split()
    bm25_scores = bm25.get_scores(tokenized_query)
    
    # 向量检索
    query_vector = embedding_model.encode(query)
    _, vector_indices = vector_index.search(query_vector.reshape(1, -1), k*2)
    
    # 融合分数
    results = {}
    for i, score in enumerate(bm25_scores):
        results[i] = results.get(i, 0) + score * 0.3
    
    for idx in vector_indices[0]:
        results[idx] = results.get(idx, 0) + (1 - _[0][list(vector_indices[0]).index(idx)]) * 0.7
    
    return sorted(results.items(), key=lambda x: x[1], reverse=True)[:k]

Elasticsearch + 向量检索

Elasticsearch 支持关键词检索，同时可以集成向量检索：

方案	说明	适用场景
ES 内置向量	使用 ES 的 dense_vector 类型	小规模向量检索
ES + FAISS	ES 做关键词过滤，FAISS 做向量检索	需要精确过滤+向量检索
ES + 外部向量库	ES 存储元数据，外部库存储向量	大规模数据

结构化条件检索

结构化字段类型

字段类型	示例	过滤方式
类别	技术文档、产品文档、API 文档	精确匹配
时间	创建时间、更新时间	范围查询
作者	张三、李四	精确匹配
标签	Java、Python、AI	多值匹配
权限	公开、内部、机密	权限过滤

结构化过滤示例

def structured_filter(docs, filters):
    """根据结构化条件过滤文档"""
    filtered = docs
    
    if 'category' in filters:
        filtered = [d for d in filtered if d.category == filters['category']]
    
    if 'time_range' in filters:
        start, end = filters['time_range']
        filtered = [d for d in filtered if start <= d.date <= end]
    
    if 'tags' in filters:
        required_tags = set(filters['tags'])
        filtered = [d for d in filtered if required_tags.issubset(set(d.tags))]
    
    if 'permission' in filters:
        filtered = [d for d in filtered if d.permission <= filters['permission']]
    
    return filtered

业务规则过滤

业务规则类型

规则类型	说明	示例
权限规则	根据用户权限过滤	普通用户看不到机密文档
时效性规则	根据文档时效性过滤	优先展示最近更新的文档
相关性规则	根据业务相关性过滤	特定业务线只看相关文档
质量规则	根据文档质量过滤	过滤低质量或草稿文档

业务规则实现

def apply_business_rules(results, user_context):
    """应用业务规则过滤"""
    filtered = []
    
    for doc_id, score in results:
        doc = get_document(doc_id)
        
        # 权限检查
        if doc.permission > user_context['permission_level']:
            continue
        
        # 时效性调整
        time_decay = calculate_time_decay(doc.update_time)
        score *= time_decay
        
        # 业务相关性检查
        if doc.business_line not in user_context['allowed_lines']:
            score *= 0.5
        
        filtered.append((doc_id, score))
    
    return sorted(filtered, key=lambda x: x[1], reverse=True)

def calculate_time_decay(update_time, half_life_days=30):
    """计算时间衰减因子"""
    days_since_update = (datetime.now() - update_time).days
    decay = 0.5 ** (days_since_update / half_life_days)
    return max(decay, 0.1)  # 最小保留 10% 权重

混合检索的常见问题

问题1：分数融合困难

表现：不同检索方式的分数范围不同，难以直接融合

解决方案：

使用归一化（如 Min-Max、Z-score）
使用相对排名替代绝对分数
使用学习排序（Learning to Rank）

问题2：检索延迟高

表现：多种检索方式并行导致延迟增加

解决方案：

使用异步并行检索
优化检索顺序，先快速过滤再精确检索
使用缓存机制

问题3：结果重复

表现：不同检索方式返回相同的文档

解决方案：

添加去重逻辑
在融合阶段合并相同文档的分数

问题4：规则冲突

表现：业务规则与检索结果产生冲突

解决方案：

明确规则优先级
使用可配置的规则引擎
添加人工干预机制

混合检索的最佳实践

检索架构设计

用户查询
    ↓
┌─────────────────────────────┐
│  查询分析与预处理           │
│  - 意图识别                 │
│  - 查询扩展                 │
│  - 条件提取                 │
└─────────────────────────────┘
    ↓
┌─────────────────────────────┐
│  并行检索层                 │
│  - 关键词检索（BM25/ES）    │
│  - 向量检索（FAISS/Milvus） │
│  - 结构化检索（数据库）      │
└─────────────────────────────┘
    ↓
┌─────────────────────────────┐
│  结果融合层                 │
│  - 去重                     │
│  - 分数归一化               │
│  - 加权融合                 │
└─────────────────────────────┘
    ↓
┌─────────────────────────────┐
│  业务规则层                 │
│  - 权限过滤                 │
│  - 时效性调整               │
│  - 质量过滤                 │
└─────────────────────────────┘
    ↓
最终检索结果

评估方法

指标	定义	目标值
召回率	正确检索到的相关文档比例	> 95%
准确率	检索结果中相关文档比例	> 85%
延迟	端到端检索时间	< 200ms
用户满意度	用户对结果的满意度评分	> 4.5/5

权重调整策略

def adaptive_weights(query, context):
    """根据查询类型自适应调整权重"""
    query_type = analyze_query_type(query)
    
    if query_type == 'technical':
        # 技术查询：更依赖关键词
        return {'keyword': 0.5, 'vector': 0.3, 'struct': 0.2}
    elif query_type == 'semantic':
        # 语义查询：更依赖向量
        return {'keyword': 0.2, 'vector': 0.6, 'struct': 0.2}
    elif query_type == 'fact':
        # 事实查询：更依赖结构化
        return {'keyword': 0.3, 'vector': 0.3, 'struct': 0.4}
    else:
        return {'keyword': 0.3, 'vector': 0.5, 'struct': 0.2}

项目判断清单

需要精确术语匹配 → 增加关键词检索（BM25）
需要语义理解 → 增加向量检索
需要按条件过滤 → 增加结构化检索
结果质量差 → 检查融合策略和权重
检索延迟高 → 使用异步并行或优化顺序
需要权限控制 → 添加业务规则层
文档有时效性 → 添加时间衰减因子
需要持续优化 → 建立评估指标和反馈机制