BM25与向量数据库的融合实践：如何优化混合检索系统的召回率

1次阅读

共计 2404 个字符，预计需要花费 7 分钟才能阅读完成。

BM25 作为经典的关键词检索算法，通过统计词频 (TF) 和逆文档频率 (IDF) 计算文档相关性得分。它的优势在于：

对精确关键词匹配效果出色
计算效率高，适合大规模文档集
无需预先训练，开箱即用

但在实际应用中我们发现：

无法处理同义词问题（如 ” 手机 ” 和 ” 智能手机 ”）
对表述差异敏感（” 如何做蛋糕 ” vs “ 蛋糕制作教程 ”）
长尾查询效果差（专业术语、新兴词汇）

维度	BM25 关键词检索	向量语义检索
准确率	精确匹配时高	语义相关时高
召回率	关键词局限导致较低	语义泛化能力较强
计算复杂度	O(1)~O(log n)	O(k log n)
内存占用	较低	较高（需存储向量）

我们的解决方案采用分层架构：

召回层：
BM25 初筛 Top 1000 结果（保证关键词匹配）
向量数据库召回相似 Top 500（补充语义结果）
融合层：
对两种结果取并集
使用线性加权分数：final_score = 0.6*bm25_score + 0.4*cosine_sim
排序层：
按融合分数重新排序
业务规则调整（如时效性加权）

from typing import List, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer
from elasticsearch import Elasticsearch
import faiss

class HybridRetriever:
    """
    混合检索器实现
    Args:
        es_client: Elasticsearch 客户端
        model_path: 句子编码模型路径
        index_dim: 向量维度
    """
    def __init__(self, es_client: Elasticsearch, model_path: str, index_dim: int=384):
        self.es = es_client
        self.model = SentenceTransformer(model_path)
        self.index = faiss.IndexFlatIP(index_dim)

    def bm25_search(self, query: str, top_k: int=1000) -> List[Tuple[str, float]]:
        """BM25 关键词检索"""
        body = {
            "query": {
                "match": {"content": query}
            },
            "size": top_k
        }
        resp = self.es.search(index="docs", body=body)
        return [(hit['_source']['id'], hit['_score']) for hit in resp['hits']['hits']]

    def vector_search(self, query: str, top_k: int=500) -> List[Tuple[str, float]]:
        """向量相似度检索"""
        query_vec = self.model.encode(query)
        D, I = self.index.search(np.array([query_vec]), top_k)
        return [(str(i), float(d)) for i, d in zip(I[0], D[0])]

    def hybrid_search(self, query: str) -> List[str]:
        """混合检索入口"""
        # 并行执行两种检索
        bm25_results = self.bm25_search(query)
        vector_results = self.vector_search(query)

        # 分数归一化
        max_bm25 = max(score for _, score in bm25_results) or 1
        max_vector = max(score for _, score in vector_results) or 1

        # 结果融合
        combined = {}
        for doc_id, score in bm25_results:
            combined[doc_id] = 0.6 * (score / max_bm25)
        for doc_id, score in vector_results:
            combined[doc_id] = combined.get(doc_id, 0) + 0.4 * (score / max_vector)

        # 按融合分数排序
        return sorted(combined.items(), key=lambda x: -x[1])