为什么需要混合检索？

上周我接到一个线上问题：用户反馈知识库问答的准确率突然下降了15%。排查后发现，单纯依赖向量检索在某些场景下确实存在明显短板。

举个真实案例：用户问"如何申请测试机"，向量检索返回了"设备管理制度"相关文档，但压根没提到申请流程。原因是"测试机"和"设备"的语义相似度不够高，导致召回失败。

这时我意识到，单一路召回注定有盲区。向量检索擅长语义匹配，但在精确关键词、专有名词、ID匹配上表现很差。要想做到企业级准确率，必须上混合检索。

我的判断：2025年做企业知识库还只用向量检索，就是在偷懒。别跟我说"简单够用"，用户会因为15%的准确率下降直接弃用你的产品。

混合检索的核心思路

简单说就是：多路召回 + Reranker重排。

打个比方：多路召回就像HR初筛简历，向量检索、关键词检索、知识图谱检索三路并行，各自按自己的标准筛选候选人（文档），各自返回Top-K结果。这一步的目标是宁可错杀一千，不可放过一个——召回率优先，准确率靠后。

而Reranker就是最终面试官，HR初筛通过的候选人（候选文档）都来到终面，Reranker逐一面试（计算Query-Document相关性分数），最后决定谁被录用（返回Top-N给LLM）。

• 第一层：多路召回（HR初筛） —— 向量检索、关键词检索、知识图谱检索同时跑，各自返回Top-K结果

• 第二层：结果融合 —— 用Reranker模型（最终面试官）对所有候选文档重新打分排序

• 第三层：最终返回 —— 取重排后的Top-N送给LLM生成答案

这样做的好处是：各路召回互补短板，Reranker保证最终质量。

技术架构设计

在我们的企业知识库项目中，混合检索架构如下：

用户Query（求职者）

↓

HR初筛1 → 向量检索（Embedding + FAISS）"语义理解派"


HR初筛2 → 关键词检索（BM25 + Elasticsearch）"精确匹配派"


HR初筛3 → 知识图谱检索（Neo4j + Cypher）"关系挖掘派"

↓

合并去重（所有HR初筛通过的候选人集合）

↓

最终面试官（Reranker重排，Cross-Encoder）

↓

录用Top-N（送入LLM）

第一路：向量检索实战（语义理解派HR）

向量检索就像HR初筛时的"感觉派"——不看精确匹配，看整体语义是否合适。我们用的是 sentence-transformers + FAISS。

先看完整代码：

from sentence_transformers import SentenceModel
import faiss
import numpy as np

# 加载Embedding模型（把文字转成向量）
model = SentenceModel(
    'BAAI/bge-large-zh-v1.5'
)

# 向量化查询（用户问题变成512维向量）
query_embedding = model.encode(
    [query_text]
)[0]

# FAISS检索Top-50（在向量空间找最近的50个文档）
index = faiss.read_index(
    'knowledge_base.index'
)
scores, indices = index.search(
    query_embedding.reshape(1, -1),
    k=50
)

代码解读：

• SentenceModel 负责把文字转成数字向量（就像把简历转成能力画像）

• faiss.read_index 加载预先构建好的向量库

• index.search 在向量空间找最相似的50个文档

这里有个坑：召回数量要设大一点（50-100），因为后面Reranker会重新排序，前期宁可多召回一些，也别漏掉相关文档。

第二路：关键词检索（精确匹配派HR）

BM25就像HR初筛时的"硬核派"——只看关键词是否精确匹配，不care语义。对于专有名词、产品型号、精确匹配的查询效果拔群。我们用Elasticsearch实现：

完整代码：

from elasticsearch import Elasticsearch

# 连接Elasticsearch（我们的文档搜索引擎）
es = Elasticsearch(
    ['localhost:9200']
)

# BM25检索（关键词匹配）
query_body = {
    "query": {
        "multi_match": {
            "query": query_text,  # 用户问题
            "fields": [
                "title^2",  # 标题匹配权重x2
                "content"    # 正文匹配
            ]
        }
    },
    "size": 50  # 返回Top-50
}

# 执行搜索
results = es.search(
    index='knowledge_base',  # 索引名
    body=query_body
)

代码解读：

• multi_match 同时在title和content两个字段搜索

• title^2 表示标题匹配权重是正文的两倍（标题更重要）

• size: 50 返回Top-50结果

实战技巧：Title字段加权（^2），因为标题匹配通常比正文匹配更重要。

第三路：知识图谱检索（关系挖掘派HR）

知识图谱就像HR初筛时的"背景调查派"——不仅看简历本身，还看候选人之间的关系（比如"这个人和那个项目什么关系？“）。对于实体关系类问题（比如"张三负责哪些项目”），知识图谱有天然优势。

我们的图谱结构：

实体类型	关系类型
员工、部门、项目	隶属、负责、参与
文档、知识条目	包含、引用、相关

检索示例（Cypher查询语言）：

// 查询"测试机申请"相关实体
MATCH (d:Document)-[r:CONTAINS]->
     (k:Knowledge)
WHERE d.title CONTAINS '测试机' 
      OR k.content CONTAINS 
      '申请流程'
RETURN d, r, k
LIMIT 50

代码解读：

• MATCH 匹配图模式（文档-包含->知识条目）

• WHERE 过滤条件（标题或内容包含关键词）

• RETURN 返回匹配的文档、关系、知识条目

结果融合：去重与归一化（三个HR的打分尺度不一样）

三路召回的结果需要合并，但各路的分数尺度不同（向量相似度是0-1，BM25是任意正数，图谱是匹配数量），必须归一化。

类比：三个HR给候选人打分，HR1打60-90分，HR2打70-80分，HR3打1-5分。直接平均没意义，得先归一化到同一把尺子（0-1）上。

归一化 + 合并代码：

# 第一步：归一化分数到[0,1]
def normalize_scores(results):
    """Min-Max归一化"""
    scores = [r['score'] for r in results]
    min_s, max_s = min(scores), max(scores)
    
    # 如果所有分数一样，统一设为1.0
    if max_s == min_s:
        return [{'doc': r['doc'], 
                 'score': 1.0} 
                for r in results]
    
    # Min-Max归一化
    for r in results:
        r['score'] = (
            (r['score'] - min_s) / 
            (max_s - min_s)
        )
    
    return results

# 第二步：合并三路结果（去重 + 平均）
def merge_results(vec_results, 
                   bm25_results, 
                   kg_results):
    # 先归一化各路分数
    vec_norm = normalize_scores(vec_results)
    bm25_norm = normalize_scores(bm25_results)
    kg_norm = normalize_scores(kg_results)
    
    # 按doc_id合并（去重）
    merged = {}
    for r in vec_norm + bm25_norm + kg_norm:
        doc_id = r['doc']['id']
        if doc_id not in merged:
            merged[doc_id] = {
                'doc': r['doc'],
                'scores': []
            }
        merged[doc_id]['scores'].append(
            r['score']
        )
    
    # 取平均分作为初步分数
    for doc_id in merged:
        scores = merged[doc_id]['scores']
        merged[doc_id]['avg_score'] = (
            sum(scores) / len(scores)
        )
    
    # 按平均分排序（降序）
    return sorted(
        merged.values(),
        key=lambda x: x['avg_score'],
        reverse=True
    )

代码解读：

• normalize_scores 把各路分数映射到0-1（三个HR的尺子统一）

• merge_results 合并三路结果，按doc_id去重，取平均分

• 返回按平均分排序的候选文档列表（等待最终面试官Reranker拍板）

Reranker重排模型实战（最终面试官）

这是整个混合检索的核心。Reranker用Cross-Encoder架构，直接计算Query-Document的相关性分数，比Bi-Encoder（向量检索）准确得多。

我的观点：Reranker不是可选项，是必选项。80ms的延迟换15%的准确率提升，任何ToB场景都应该上。别拿"性能优化"当借口，用户要的是准确结果，不是快但错的答案。

我们用的是 BAAI/bge-reranker-large，用法非常简单：

from sentence_transformers import CrossEncoder

# 加载Reranker模型（最终面试官）
reranker = CrossEncoder(
    'BAAI/bge-reranker-large'
)

# 准备候选文档（多路召回合并后取Top-100）
candidates = merge_results(
    vec_results,
    bm25_results,
    kg_results
)[:100]

# 构造Query-Document对（给最终面试官的考题）
pairs = [[query_text, doc['content']] 
         for doc in candidates]

# Reranker打分（最终面试官逐一面试）
scores = reranker.predict(pairs)

# 按新分数排序（录用排名）
reranked = sorted(
    zip(candidates, scores),
    key=lambda x: x[1],
    reverse=True
)

# 最终返回Top-5（录用5人）
final_results = [doc for doc, score 
                in reranked[:5]]

代码解读：

• CrossEncoder 加载Reranker模型（最终面试官上岗）

• pairs 构造Query-Document对（考题）

• reranker.predict 计算相关性分数（面试官打分）

• 返回Top-5（录用前5名送给LLM生成答案）

性能优化技巧

问题：Reranker模型通常比较大（1B+参数），预测速度慢。如果每次查询都要对100个候选文档做Cross-Encoder，延迟会飙到500ms+。

解决方案：异步批量推理 + 模型量化

import torch
from torch.quantization import quantize_dynamic

# 1. 模型量化（INT8，把模型压小压快）
reranker.model = quantize_dynamic(
    reranker.model,
    {torch.nn.Linear},
    dtype=torch.qint8
)

# 2. 批量推理（一次面试32个候选人，提高效率）
def batch_rerank(pairs, batch_size=32):
    all_scores = []
    for i in range(0, len(pairs), batch_size):
        batch = pairs[i:i+batch_size]
        scores = reranker.predict(batch)
        all_scores.extend(scores)
    return all_scores

# 3. 异步执行（面试官加班，不阻塞招聘流程）
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

async def async_rerank(pairs):
    loop = asyncio.get_event_loop()
    scores = await loop.run_in_executor(
        executor,
        batch_rerank,
        pairs
    )
    return scores

代码解读：

• quantize_dynamic 把模型从FP32量化到INT8（模型变小，推理变快）

• batch_rerank 批量推理（一次处理32个，不是逐个处理）

• async_rerank 异步执行（Reranker在后台工作，不阻塞主流程）

经过优化，Reranker延迟从500ms降到80ms，满足生产要求。

实战效果对比

我们在1000条真实用户查询上做了A/B测试：

方案	Recall@5	MRR	延迟
仅向量检索	68.2%	0.72	45ms
仅BM25	61.5%	0.68	30ms
混合检索（无Reranker）	75.3%	0.76	80ms
混合检索+Reranker	83.7%	0.84	160ms

结论：混合检索+Reranker把Recall@5从68.2%提升到83.7%，MRR从0.72提升到0.84。虽然延迟增加了115ms，但在企业知识库场景下，这个trade-off是完全值得的。

我的态度：别跟我说"延迟增加了115ms，用户会不耐烦"。用户要的是正确答案，不是快但错的答案。如果你做的产品是"快>准"，那可能不适合上Reranker；但企业知识库这种场景，准确率优先。

踩坑记录

坑1：Reranker输入长度截断

bge-reranker-large的最大输入长度是512个token。如果文档内容超长，直接截断会导致关键信息丢失。

解决方案：滑动窗口切分长文档，每个窗口512token，Reranker对每个窗口打分，取最高分作为该文档的分数。

坑2：多路召回的重复文档

三路召回可能返回同一文档的多个片段（比如向量检索返回段落3，BM25返回段落5），合并时容易重复。

解决方案：按doc_id + chunk_id去重，而不是仅按doc_id。

坑3：Reranker模型占用显存

如果把Reranker部署在GPU上，1B+参数模型要占4GB+显存。如果并发请求多，容易OOM。

解决方案：用ONNX Runtime量化模型，转成INT8，显存占用降到1GB以下。或者直接用CPU推理（我们的做法），虽然慢点但成本低。

经验总结
混合检索不是银弹，要根据业务场景权衡。如果你们的知识库主要是技术文档（术语多、精确匹配需求强），BM25权重可以调高；如果是通用问答，向量检索权重可以调高。最重要的是：用真实用户查询做评估，别拍脑袋调参数。

下一步计划

下一篇我们会讲知识库系统的缓存策略：如何用Redis + 语义缓存（Semantic Cache）把频繁查询的响应时间从160ms降到5ms。这个项目越来越有意思了。

预告：缓存策略是另一个"必选项"。160ms对于ToB场景还能接受，但如果是ToC场景（比如客服机器人），必须降到50ms以内。语义缓存就是为此而生。

系列进度：8/20 | 下一篇：缓存策略——Redis + 语义缓存实战