人工智能知识库

混合检索技术

16分钟阅读

混合检索技术：别把鸡蛋放一个篮子里

这篇文章解决什么问题

为什么你的RAG系统有时候能找到正确答案，有时候又查不到？很可能是只用了单一检索方式的问题。混合检索就是把多种检索方法组合起来，取长补短，大幅提升检索效果。

前言：我踩过的坑

先说个真实故事。

我之前做了一个法律知识库问答系统，用的是纯向量检索。开始效果还不错，直到有一天用户问：

“合同法第二十三条是什么？”

向量检索返回了一堆”合同法解读”、“合同法案例分析”，愣是没找到第二十三条的具体内容。

你猜怎么着？因为”第二十三条”这几个字在法律文本里太稀有了，Embedding模型根本没法准确表示它。

后来加了BM25关键词检索，问题解决了。

这就是为什么需要混合检索——单一检索方式总有盲区，组合起来才能COVER全场。

一、为什么需要混合检索

1.1 单一检索的局限性

检索方式优点缺点
纯向量检索语义理解强、同义词友好精确术语弱、数值检索差
纯关键词检索精确匹配强、术语友好语义理解差、同义词不识别
知识图谱检索关系推理强结构化要求高、覆盖有限

1.2 打个比方

向量检索像一个热情的图书管理员：

  • 你说”找本讲解决问题的书”
  • 他给你推荐了《高效能人士的七个习惯》《问题分析与解决》《麦肯锡方法》
  • 但你真正想要的可能只是《毛泽东选集》里讲矛盾论的那篇

关键词检索像一个死板的索引员：

  • 你说”找第二十三条”
  • 他精准定位到合同法第二十三条
  • 但你说”找违约相关的条款”，他完全不知道你在说什么

混合检索就是两个都叫上，取长补短。

1.3 混合检索的价值

  • 召回率提升：不同检索方式覆盖不同内容
  • 鲁棒性增强：一种方式拉胯，另一种顶上
  • 适用场景更广：简单问题和复杂问题都能搞定

二、混合检索的核心技术

2.1 BM25：关键词检索的老将

BM25是关键词检索领域的老牌算法，比TF-IDF更聪明。

原理：基于词频和文档长度的概率检索模型

import math
from collections import Counter
 
def bm25_score(query, document, corpus_stats, k1=1.5, b=0.75):
    """
    计算BM25分数
    
    参数：
    - query: 查询词列表
    - document: 文档词列表
    - corpus_stats: 语料库统计信息
    - k1: 词频饱和参数
    - b: 长度归一化参数
    """
    
    doc_len = len(document)
    avg_len = corpus_stats["avg_doc_len"]
    doc_freq = corpus_stats["doc_freq"]  # 词出现的文档数
    
    score = 0.0
    
    for term in query:
        if term not in document:
            continue
        
        # 计算词频
        tf = document.count(term)
        
        # IDF：稀有词权重更高
        idf = math.log((corpus_stats["num_docs"] - doc_freq.get(term, 0) + 0.5) / 
                       (doc_freq.get(term, 0) + 0.5) + 1)
        
        # BM25公式
        term_score = idf * (tf * (k1 + 1)) / (tf + k1 * (1 - b + b * doc_len / avg_len))
        score += term_score
    
    return score

BM25的特点：

  • 对罕见词友好（IDF高）
  • 词频有上限（不会因为词多就无限加分）
  • 对文档长度敏感（短的文档更容易匹配）

2.2 Dense Retrieval：向量检索的主力

向量检索通过Embedding模型将文本转为向量，在向量空间中找相似内容。

import numpy as np
 
def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
 
def dense_search(query_vector, document_vectors, top_k=10):
    """
    密集检索：在向量空间中找最相似的结果
    """
    scores = []
    
    for doc_id, doc_vector in enumerate(document_vectors):
        score = cosine_similarity(query_vector, doc_vector)
        scores.append((doc_id, score))
    
    # 按分数排序，取top_k
    scores.sort(key=lambda x: x[1], reverse=True)
    
    return scores[:top_k]

2.3 RRF：融合结果的利器

拿到多个检索方式的结果后，需要把它们融合起来。RRF（Reciprocal Rank Fusion）是目前最流行的融合算法。

原理：排名越靠前的结果，最终得分越高

def rrf_fusion(results_lists, k=60):
    """
    Reciprocal Rank Fusion
    
    参数：
    - results_lists: 多个检索方式的结果列表，每个列表是 [(doc_id, score), ...] 按分数降序
    - k: 调节参数，k越大，不同排名的结果差距越小
    """
    
    scores = {}  # doc_id -> 最终分数
    
    for results in results_lists:
        for rank, (doc_id, original_score) in enumerate(results):
            # RRF分数：1/(k + rank)
            rrf_score = 1 / (k + rank + 1)
            
            if doc_id not in scores:
                scores[doc_id] = 0
            scores[doc_id] += rrf_score
    
    # 按RRF分数排序
    final_ranking = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    
    return final_ranking

为什么RRF好用：

  • 不依赖原始分数的绝对值
  • 只看相对排名，避免不同检索方式分数范围不一致的问题
  • k参数可以调节平滑程度

2.4 实际使用RRF

async def hybrid_search_with_rrf(query, vector_retriever, bm25_retriever, k=60):
    """
    使用RRF融合的混合检索
    """
    
    # 1. 向量检索
    vector_results = await vector_retriever.search(query, top_k=20)
    
    # 2. BM25检索
    bm25_results = await bm25_retriever.search(query, top_k=20)
    
    # 3. RRF融合
    fused_results = rrf_fusion([vector_results, bm25_results], k=k)
    
    # 4. 返回最终结果
    return fused_results

三、混合检索的实操

3.1 使用LangChain实现

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain.schema import Document
 
# 假设你有一些文档
documents = [
    Document(page_content="合同法第二十三条：当事人对合同条款有争议的，应当按照合同的目的、交易习惯确定。", metadata={"source": "合同法"}),
    Document(page_content="侵权责任法第二十三条：因紧急避险造成损害的，由引起险情发生的人承担民事责任。", metadata={"source": "侵权责任法"}),
    # ... 更多文档
]
 
# 1. 创建向量检索器
vectorstore = FAISS.from_documents(documents, embedding=your_embedding)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
 
# 2. 创建BM25检索器
bm25_retriever = BM25Retriever.from_documents(documents)
 
# 3. 创建混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]  # BM25和向量各占50%
)
 
# 4. 使用
results = ensemble_retriever.get_relevant_documents("合同法第二十三条")

3.2 使用RAG-Fusion增强效果

RAG-Fusion是一种高级混合检索策略，它会：

  1. 原始查询检索
  2. LLM生成多个相关查询
  3. 每个查询分别检索
  4. 用RRF融合结果
async def rag_fusion(query, llm, retriever, num_queries=5):
    """
    RAG-Fusion检索
    """
    
    # 1. LLM生成多个查询
    query_prompt = f"""
基于以下查询，生成{num_queries}个不同的表达方式：
 
原始查询：{query}
 
要求：
- 每个查询要表达相同的意思，但用词不同
- 涵盖不同的角度和侧重点
"""
    
    new_queries = await llm.generate(query_prompt)
    new_queries = new_queries.split("\n")[:num_queries]
    
    # 2. 所有查询都检索一遍
    all_results = []
    for new_query in new_queries:
        results = await retriever.search(new_query, top_k=10)
        all_results.append(results)
    
    # 3. RRF融合
    fused_results = rrf_fusion(all_results, k=60)
    
    return fused_results

3.3 权重调优

混合检索的权重设置很关键：

# 不同的权重组合
 
# 方案1：平均权重
weights_equal = [0.5, 0.5]
 
# 方案2：偏向关键词
weights_keyword = [0.7, 0.3]
 
# 方案3：偏向语义
weights_semantic = [0.3, 0.7]
 
# 方案4：动态权重
def dynamic_weights(query):
    """
    根据查询类型动态调整权重
    """
    # 检测是否是精确查询
    is_precise = any(indicator in query for indicator in [
        "第", "条", "条款", "规定", "第X章"
    ])
    
    if is_precise:
        return [0.8, 0.2]  # 偏向关键词
    else:
        return [0.3, 0.7]  # 偏向语义

四、实战案例：法律知识库

4.1 场景描述

做一个法律知识库问答系统，需要检索：

  • 法律条文（精确匹配）
  • 法律解读（语义理解）
  • 相关案例（混合理解）

4.2 实现方案

from dataclasses import dataclass
from typing import List, Tuple
import numpy as np
 
@dataclass
class LawQueryResult:
    doc_id: str
    content: str
    source: str
    final_score: float
 
class LawHybridRetriever:
    """法律知识库混合检索器"""
    
    def __init__(self, vector_db, bm25_index):
        self.vector_db = vector_db
        self.bm25_index = bm25_index
        self.vector_weight = 0.4
        self.bm25_weight = 0.6  # 法律场景更看重精确匹配
    
    async def search(self, query: str, top_k: int = 10) -> List[LawQueryResult]:
        # 1. 判断查询类型
        query_type = self._classify_query(query)
        
        # 2. 调整权重
        weights = self._adjust_weights(query_type)
        
        # 3. 并行检索
        vector_results = await self.vector_db.search(query, top_k=top_k * 2)
        bm25_results = self.bm25_index.search(query, top_k=top_k * 2)
        
        # 4. 分数归一化
        vector_scores = self._normalize_scores(vector_results)
        bm25_scores = self._normalize_scores(bm25_results)
        
        # 5. 加权融合
        all_doc_ids = set(vector_scores.keys()) | set(bm25_scores.keys())
        final_scores = {}
        
        for doc_id in all_doc_ids:
            v_score = vector_scores.get(doc_id, 0)
            b_score = bm25_scores.get(doc_id, 0)
            final_scores[doc_id] = (
                weights["vector"] * v_score + 
                weights["bm25"] * b_score
            )
        
        # 6. 返回top_k
        sorted_results = sorted(final_scores.items(), key=lambda x: x[1], reverse=True)
        
        return [
            LawQueryResult(
                doc_id=doc_id,
                content=self._get_content(doc_id),
                source=self._get_source(doc_id),
                final_score=score
            )
            for doc_id, score in sorted_results[:top_k]
        ]
    
    def _classify_query(self, query: str) -> str:
        """分类查询类型"""
        precise_indicators = ["第", "条", "条款", "规定", "第X章", "法"]
        semantic_indicators = ["解释", "理解", "含义", "为什么", "如何"]
        
        if any(ind in query for ind in precise_indicators):
            return "precise"
        elif any(ind in query for ind in semantic_indicators):
            return "semantic"
        else:
            return "mixed"
    
    def _adjust_weights(self, query_type: str) -> dict:
        """调整检索权重"""
        if query_type == "precise":
            return {"vector": 0.2, "bm25": 0.8}
        elif query_type == "semantic":
            return {"vector": 0.8, "bm25": 0.2}
        else:
            return {"vector": 0.4, "bm25": 0.6}
    
    def _normalize_scores(self, results: List[Tuple[str, float]]) -> dict:
        """分数归一化到[0, 1]"""
        if not results:
            return {}
        
        scores = [r[1] for r in results]
        min_s, max_s = min(scores), max(scores)
        
        if max_s == min_s:
            return {r[0]: 1.0 for r in results}
        
        return {
            r[0]: (r[1] - min_s) / (max_s - min_s) 
            for r in results
        }
    
    def _get_content(self, doc_id: str) -> str:
        # 实际从数据库获取
        return ""
    
    def _get_source(self, doc_id: str) -> str:
        # 实际从数据库获取
        return ""

4.3 测试效果

# 测试用例
test_cases = [
    "合同法第二十三条是什么？",  # 精确查询
    "什么是合同的善意履行？",      # 语义查询
    "定金和违约金的区别是什么？",  # 混合查询
]
 
# 运行测试
retriever = LawHybridRetriever(vector_db, bm25_index)
 
for query in test_cases:
    results = await retriever.search(query, top_k=5)
    print(f"\n查询：{query}")
    print(f"找到 {len(results)} 条结果")
    for i, r in enumerate(results[:3], 1):
        print(f"  {i}. [{r.final_score:.3f}] {r.source}: {r.content[:50]}...")

五、HyDE：假设性文档嵌入

5.1 什么是HyDE

HyDE（Hypothetical Document Embeddings）是一种高级检索策略：

  1. 让LLM根据查询生成一个”假答案”
  2. 用这个假答案去检索
  3. 返回真实的相关文档

5.2 为什么HyDE有效

因为”假答案”包含了：

  • 查询的语义信息
  • 答案的结构信息
  • 相关的概念和术语

这些信息比单纯的用户查询更丰富，能更好地匹配目标文档。

5.3 HyDE实现

async defhyde_search(query, llm, retriever):
    """
    HyDE检索
    """
    
    # 1. 生成假答案
    hyde_prompt = f"""
基于以下问题，生成一个假设性的答案。
这个答案不需要准确，但要用到相关的术语和概念。
 
问题：{query}
 
请用中文回答，生成一段50-100字的假设性答案：
"""
    
    hypothetical_doc = await llm.generate(hyde_prompt)
    
    # 2. 用假答案检索
    results = await retriever.search(hypothetical_doc, top_k=10)
    
    return results

5.4 混合使用HyDE

async def hybrid_hyde_search(query, llm, retriever, use_hyde=True):
    """
    结合HyDE的混合检索
    """
    
    # 1. 传统混合检索
    normal_results = await hybrid_search(query, retriever)
    
    # 2. HyDE检索（可选）
    if use_hyde:
        hyde_results = await hyde_search(query, llm, retriever)
        
        # 3. RRF融合
        fused = rrf_fusion([normal_results, hyde_results])
        return fused
    
    return normal_results

六、常见问题

6.1 什么时候用混合检索

适合用：

  • 检索质量不稳定
  • 文档类型多样（术语+解释+案例）
  • 用户查询方式多样（精确+模糊）
  • 对召回率要求高

不需要用：

  • 文档类型单一
  • 性能要求极高
  • 资源有限

6.2 权重怎么调

没有标准答案，需要实验：

# 建议的调参流程
weights_to_test = [
    [0.5, 0.5],  # 平均
    [0.3, 0.7],  # 偏向向量
    [0.7, 0.3],  # 偏向BM25
    [0.2, 0.8],  # 极度偏向BM25
    [0.8, 0.2],  # 极度偏向向量
]
 
for weights in weights_to_test:
    retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, vector_retriever],
        weights=weights
    )
    
    metrics = evaluate(retriever, test_queries)
    print(f"Weights {weights}: Recall@{10} = {metrics['recall@10']:.3f}")

6.3 检索速度变慢怎么办

混合检索确实比单一检索慢，可以优化：

# 优化方案1：并行检索
async def parallel_hybrid_search(query, retriever1, retriever2):
    """并行执行两个检索"""
    results1, results2 = await asyncio.gather(
        retriever1.search(query),
        retriever2.search(query)
    )
    return rrf_fusion([results1, results2])
 
# 优化方案2：减少检索数量
top_k = 10  # 而不是50
 
# 优化方案3：用ANN近似搜索
vector_retriever.search_method = "ann"  # 而不是精确搜索

七、总结

混合检索的核心：

  • 组合多种检索方式，取长补短
  • RRF是最常用的结果融合算法
  • 根据场景调整权重

实战建议：

  • 先用简单的平均权重
  • 根据查询类型动态调整
  • 用HyDE进一步增强效果

记住：没有最好的检索方式，只有最适合的组合。

相关主题

更新记录

  • 2026-04-24：改写完成，语言风格优化
  • 增加HyDE和RAG-Fusion高级技巧

关系图谱

反向链接