6-5 混合检索 · Hermes教程

纯向量检索的局限

向量检索（语义搜索）很强大，但有一个明显的弱点：对精确关键词匹配不敏感。例如：

用户问："BGE-M3模型的最大输入长度是多少？" — 向量检索可能返回一堆关于"Embedding模型"的泛泛介绍，而不是那个包含"BGE-M3"和"输入长度"的具体文档。
技术文档中的型号、版本号、专有名词（如"Hermes v2.3"、"函数名calc_similarity"）向量检索很难精确命中。
产品代码、SKU编号这类完全无语义的字符串，向量检索束手无策。

BM25关键词检索原理

BM25是传统搜索引擎的核心算法，是TF-IDF的改进版。它的核心思想：

TF（词频）：关键词在文档中出现越多，相关性越高，但有饱和上限（不是线性增长）。
IDF（逆文档频率）：一个词在越少文档中出现，区分度越强，权重越高（"的/是"这类停用词IDF极低）。
文档长度归一化：短文档中的词频比长文档更有意义。

BM25不需要GPU，计算极快，对精确关键词命中非常可靠。缺点是完全不理解语义（"汽车"和"轿车"对BM25来说是两个不同的词）。

混合检索：向量 + BM25双路

混合检索的思路：分别用向量检索和BM25检索各取Top-K结果，再用融合算法合并排序。两路检索互补：向量召回语义相关内容，BM25召回精确关键词命中内容。

RRF（倒数排名融合）算法

RRF是最常用的融合方法，简单且效果好：

RRF_score(d) = Σ 1 / (k + rank_i(d))

其中：
- k = 60（常数，防止高排名差距过大，通常固定为60）
- rank_i(d) = 文档d在第i路检索结果中的排名
- 对每路检索结果求和

排名越靠前的文档得分越高，多路都靠前的文档综合得分最高。

何时必须使用混合检索？

文档中含有大量专有名词、型号、代码、SKU
用户经常使用精确的技术术语进行查询
知识库包含结构化文本（API文档、技术规格表）
纯向量检索的Recall@5指标低于70%

混合检索完整实现代码

pip install rank-bm25 chromadb openai

import numpy as np
from rank_bm25 import BM25Okapi
import chromadb
from openai import OpenAI

# ===== 准备数据 =====
documents = [
    "BGE-M3模型支持最长8192个token的输入",
    "RAG通过检索增强生成质量，减少幻觉",
    "向量数据库使用HNSW索引加速ANN检索",
    "BM25是传统搜索引擎的核心排序算法",
    "混合检索结合向量搜索和关键词搜索的优势",
]
doc_ids = [f"doc_{i}" for i in range(len(documents))]

client = OpenAI()

def get_embedding(text):
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return resp.data[0].embedding

# ===== 构建BM25索引 =====
# 中文需要先分词，这里用简单的字符级分词
tokenized_docs = [list(doc) for doc in documents]
bm25 = BM25Okapi(tokenized_docs)

# ===== 构建向量索引（Chroma）=====
chroma_client = chromadb.Client()
collection = chroma_client.create_collection("hybrid_search")

embeddings = [get_embedding(doc) for doc in documents]
collection.add(
    ids=doc_ids,
    documents=documents,
    embeddings=embeddings
)

# ===== RRF融合函数 =====
def rrf_fusion(rankings_list, k=60):
    """
    rankings_list: 每路检索返回的 [(doc_id, rank), ...] 列表
    返回：按RRF得分排序的 [(doc_id, score), ...]
    """
    scores = {}
    for rankings in rankings_list:
        for doc_id, rank in rankings:
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

# ===== 混合检索主函数 =====
def hybrid_search(query, top_k=3):
    # --- 路1：BM25关键词检索 ---
    tokenized_query = list(query)
    bm25_scores = bm25.get_scores(tokenized_query)
    bm25_ranking = sorted(enumerate(bm25_scores), key=lambda x: x[1], reverse=True)
    bm25_results = [(doc_ids[idx], rank+1) for rank, (idx, _) in enumerate(bm25_ranking[:top_k*2])]

    # --- 路2：向量语义检索 ---
    query_embedding = get_embedding(query)
    vec_results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k * 2
    )
    vector_results = [(doc_ids[int(id_.split("_")[1])], rank+1)
                     for rank, id_ in enumerate(vec_results["ids"][0])]

    # --- RRF融合 ---
    fused = rrf_fusion([bm25_results, vector_results])

    # 返回Top-K结果
    final_results = []
    for doc_id, score in fused[:top_k]:
        idx = int(doc_id.split("_")[1])
        final_results.append({
            "id": doc_id,
            "content": documents[idx],
            "rrf_score": score
        })
    return final_results

# ===== 测试 =====
query = "BGE-M3最大输入token数"
results = hybrid_search(query, top_k=3)
for r in results:
    print(f"[{r['rrf_score']:.4f}] {r['content']}")

使用LangChain的EnsembleRetriever（简化版）

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 构建BM25 Retriever
bm25_retriever = BM25Retriever.from_texts(documents)
bm25_retriever.k = 5

# 构建向量 Retriever
vectorstore = Chroma.from_texts(documents, OpenAIEmbeddings())
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 混合检索（权重：向量0.6，BM25 0.4）
ensemble = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

results = ensemble.invoke("什么是BM25算法？")
for doc in results:
    print(doc.page_content)