混合检索策略：语义搜索与关键词搜索的最佳融合

引言

纯向量检索并非万能。当用户查询包含专有名词、缩写或精确匹配需求时（如”RAG-2024-001”、“GPT-4o”），语义检索可能找不到精确结果。混合检索（Hybrid Search） 将向量语义搜索与关键词精确搜索相结合，弥补各自的不足。

向量检索（语义）                   关键词检索（精确）
    │                                    │
    │   "最便宜的向量数据库"              │   "Milvus 2.4 配置"
    │                                    │
    ▼                                    ▼
┌─────────────────┐           ┌─────────────────┐
│ cosine(语义向量) │           │   BM25(词频)    │
│ "向量数据库优惠" │           │ "Milvus 2.4"   │
└────────┬────────┘           └────────┬────────┘
         │                             │
         └──────────┬──────────────────┘
                    ▼
         ┌─────────────────────┐
         │   RRF 合并排序      │
         │   Reciprocal Rank   │
         │   Fusion            │
         └─────────┬───────────┘
                   ▼
         最终检索结果（兼顾语义+精确）

一、为什么需要混合检索？

1.1 纯向量检索的盲区

1.2 两种检索模式的优势互补

二、混合检索的实现方式

2.1 RRF（Reciprocal Rank Fusion）

最经典的混合检索算法。核心思想：将两个排序结果的排名倒数相加，排名高的结果获得更高权重。

def rrf_merge(
    vector_results: list[dict],
    bm25_results: list[dict],
    k: int = 60
) -> list[dict]:
    """
    RRF 合并排序
    k: 常数，通常 60（Elasticsearch 默认值）
    """
    scores = {}

    # 向量检索结果
    for rank, doc in enumerate(vector_results):
        doc_id = doc['id']
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # BM25 检索结果
    for rank, doc in enumerate(bm25_results):
        doc_id = doc['id']
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # 按总分排序
    ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    return [
        {"id": doc_id, "score": total_score}
        for doc_id, total_score in ranked
    ]

2.2 加权融合

给两种检索结果分配权重，适合对某种检索方式有偏好时使用：

def weighted_fusion(
    vector_results: list[dict],
    bm25_results: list[dict],
    vector_weight: float = 0.7,
    bm25_weight: float = 0.3,
    top_k: int = 10
) -> list[dict]:
    """
    加权融合两种检索结果
    vector_weight + bm25_weight 不一定等于1
    """
    scores = {}

    for rank, doc in enumerate(vector_results):
        doc_id = doc['id']
        scores[doc_id] = scores.get(doc_id, 0) + \
            vector_weight / (rank + 1)

    for rank, doc in enumerate(bm25_results):
        doc_id = doc['id']
        scores[doc_id] = scores.get(doc_id, 0) + \
            bm25_weight / (rank + 1)

    ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return ranked[:top_k]

三、实践：Elasticsearch + 向量检索混合

3.1 Elasticsearch 原生混合检索

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")

# 创建索引，同时支持 BM25 和向量检索
mapping = {
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "ik_smart"  # 中文分词
            },
            "content": {
                "type": "text",
                "analyzer": "ik_smart"
            },
            "embedding": {
                "type": "dense_vector",
                "dims": 768,
                "index": True,
                "similarity": "cosine"
            }
        }
    }
}

es.indices.create(index="hybrid_docs", body=mapping)

# 混合检索查询
def hybrid_search_es(
    es_client,
    index: str,
    query: str,
    query_vector: list[float],
    alpha: float = 0.5
):
    """在 Elasticsearch 中执行混合检索"""
    body = {
        "size": 10,
        "query": {
            "bool": {
                "should": [
                    # BM25 关键词检索
                    {
                        "multi_match": {
                            "query": query,
                            "fields": ["title^2", "content"],
                            "type": "best_fields"
                        }
                    },
                    # 向量检索
                    {
                        "script_score": {
                            "query": {"match_all": {}},
                            "script": {
                                "source": f"{alpha} * cosineSimilarity(params.query_vector, 'embedding') + {(1-alpha)} * _score",
                                "params": {"query_vector": query_vector}
                            }
                        }
                    }
                ]
            }
        }
    }
    return es_client.search(index=index, body=body)

3.2 LangChain 混合检索

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# BM25 检索器
bm25_retriever = BM25Retriever.from_documents(
    documents,  # 需要原始文本
    k=10
)

# 向量检索器
vector_retriever = Chroma.from_documents(
    documents=documents,
    embedding=OpenAIEmbeddings()
).as_retriever(search_kwargs={"k": 10})

# 混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # BM25 权重 0.3，向量权重 0.7
)

# 使用混合检索
results = ensemble_retriever.invoke("Milvus 2.4 版本的部署配置")

四、实践：LlamaIndex 混合检索

from llama_index.core import (
    VectorStoreIndex, SimpleDirectoryReader
)
from llama_index.core.retrievers import (
    VectorIndexRetriever,
    KeywordTableRetriever
)
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import (
    SentenceTransformerRerank
)
from llama_index.core.retrievers import (
    CustomRetriever
)

class HybridRetriever(CustomRetriever):
    """自定义混合检索器"""
    def __init__(
        self,
        vector_retriever,
        keyword_retriever,
        mode: str = "RRF",
        top_k: int = 10,
        vector_weight: float = 0.7
    ):
        self.vector_retriever = vector_retriever
        self.keyword_retriever = keyword_retriever
        self.mode = mode
        self.top_k = top_k
        self.vector_weight = vector_weight

    def _retrieve(self, query):
        vector_nodes = self.vector_retriever.retrieve(query)
        keyword_nodes = self.keyword_retriever.retrieve(query)

        if self.mode == "RRF":
            return self._rrf_merge(vector_nodes, keyword_nodes)
        elif self.mode == "weighted":
            return self._weighted_merge(
                vector_nodes, keyword_nodes
            )

# 使用混合检索
hybrid_retriever = HybridRetriever(
    vector_retriever=VectorIndexRetriever(
        index=index, similarity_top_k=10
    ),
    keyword_retriever=KeywordTableRetriever(
        index=index, similarity_top_k=10
    ),
    mode="RRF"
)

query_engine = RetrieverQueryEngine.from_args(
    retriever=hybrid_retriever,
    node_postprocessors=[
        SentenceTransformerRerank(top_k=5)
    ]
)

五、混合检索参数调优

六、何时使用混合检索

你的查询类型？
    │
    ├── 开放式问题（"什么是...？"）
    │   └── 纯向量检索即可
    │
    ├── 混合查询（"Milvus的性能指标"）
    │   └── 混合检索（向量 0.6 + BM25 0.4）
    │
    ├── 精确匹配（"版本 2.4.0 的配置"）
    │   └── 混合检索（向量 0.3 + BM25 0.7）
    │
    └── 编号/代码查询（"DOC-2024-0123"）
        └── 纯关键词检索

七、总结

混合检索的最佳实践：

1. 不是所有场景都需要 — 开放式问答纯向量检索就够用
2. 根据查询类型动态调整权重 — 检测到专有名词时增加 BM25 权重
3. 配合重排序效果更好 — 混合检索取 TOP 50，再用 Cross-encoder 重排序
4. RRF 比加权融合更稳健 — 不受分数尺度不一致的影响
5. 关注延迟 — 混合检索执行两个查询，延迟增加一倍

相关资源：

• Elasticsearch 混合搜索
• LangChain Ensemble Retriever
• RRF 论文