RAG混合检索：关键词 + 向量的最佳组合

1
❌ 场景示例：
2
  用户查询 → "IPH-15-PRO-256 的价格"
3
  向量检索 → 返回"苹果手机最新款评测"（语义漂移）
4
  正确答案 → iPhone 15 Pro 256GB 产品页（精确匹配）

1
❌ 场景示例：
2
  用户查询 → "怎么让网页加载更快"
3
  BM25 检索 → 无匹配（文档中是"前端性能优化指南"）
4
  向量检索 → 准确命中（语义等价）

1
┌─────────────────────────────────────────────────────┐
2
│               检索能力对比矩阵                        │
3
├──────────────────┬──────────────────┬───────────────┤
4
│    查询类型       │   向量检索        │   BM25        │
5
├──────────────────┼──────────────────┼───────────────┤
6
│ 语义近义词        │ ✅ 优秀          │ ❌ 失效        │
7
│ 精确标识符        │ ❌ 漂移          │ ✅ 优秀        │
8
│ 领域外新词        │ ❌ 失效          │ ✅ 可命中      │
9
│ 多语言概念        │ ✅ 较好          │ ❌ 依赖词汇    │
10
│ 错别字/近似词     │ ✅ 容错          │ ❌ 严格匹配    │
11
│ 代码/函数名       │ ❌ 语义漂移      │ ✅ 精确命中    │
12
└──────────────────┴──────────────────┴───────────────┘

1
关键词检索（稀疏表示）              向量检索（稠密表示）
2
─────────────────────────────────────────────────────
3
词汇空间维度：~50,000+             嵌入空间维度：768 / 1536
4
每个文档：绝大多数维度为 0         每个文档：所有维度均有值
5
"数据库" → [0,0,1,0,0,0,0,...]    "数据库" → [0.12,-0.34,0.87,...]
6
精确词汇匹配                        近似语义相似度
7
倒排索引，毫秒级                    ANN 近似最近邻，毫秒级
8
无需 GPU                           需要 GPU 或专用推理服务

1
                        ┌──────────────┐
2
                        │  用户查询    │
3
                        └──────┬───────┘
4
                               │
5
               ┌───────────────┴───────────────┐
6
               │                               │
7
               ▼                               ▼
8
   ┌───────────────────────┐     ┌───────────────────────┐
9
   │    关键词检索          │     │    向量检索            │
10
   │   BM25 / SPLADE       │     │  Dense Embedding      │
11
   └──────────┬────────────┘     └──────────┬────────────┘
12
              │                             │
13
   ┌──────────▼────────────┐     ┌──────────▼────────────┐
14
   │   倒排索引             │     │   向量数据库           │
15
   │  Inverted Index       │     │   HNSW / IVF-PQ      │
16
   └──────────┬────────────┘     └──────────┬────────────┘
17
              │    Top-K 候选                │    Top-K 候选
18
              └───────────┬─────────────────┘
19
                          │
20
              ┌───────────▼───────────────┐
21
              │      结果融合层           │
22
              │  RRF · 加权融合 · DBSF   │
23
              └───────────┬───────────────┘
24
                          │   合并候选集
25
              ┌───────────▼───────────────┐
26
              │    重排序 Reranker        │
27
              │  Cross-encoder · ColBERT  │
28
              └───────────┬───────────────┘
29
                          │   精排结果
30
              ┌───────────▼───────────────┐
31
              │       LLM 生成           │
32
              │   基于上下文生成答案      │
33
              └───────────────────────────┘

1
阶段一（召回）：追求广度   →   宁可多召回，不要漏
2
阶段二（精排）：追求精度   →   从广泛候选中挑出最优
3
阶段三（生成）：利用上下文 →   LLM 基于精排结果作答

1
                         tf(qi, D) · (k1 + 1)
2
Score(D, Q) = Σ IDF(qi) · ─────────────────────────────────────
3
                         tf(qi, D) + k1 · (1 - b + b · |D|/avgdl)

1
┌──────────────────────────────────────────────────────────────┐
2
│                    BM25 三核心因子                            │
3
├─────────────────┬────────────────────┬───────────────────────┤
4
│   词频 TF        │  逆文档频率 IDF     │   长度归一化           │
5
├─────────────────┼────────────────────┼───────────────────────┤
6
│ 词在文档中出现   │ 罕见词权重高       │ 防止长文档仅凭          │
7
│ 越多分越高，     │ 常见词（"的""是"） │ 体积优势压制           │
8
│ 但有饱和上限     │ 权重大幅降低       │ 简短精准的文档          │
9
│                 │                    │                       │
10
│ 避免"词频刷分"  │ "数据库"比"的"    │ 500字文档 ≈ 5000字     │
11
│ 的朴素做法      │ 有更高区分度       │ 文档（按比例）          │
12
└─────────────────┴────────────────────┴───────────────────────┘

1
✅ 擅长：                          ❌ 不擅长：
2
  - 产品型号精确匹配                  - "慢查询" vs "数据库性能"
3
  - 法律条款编号定位                  - 跨语言语义匹配
4
  - 错误码 / 日志检索                 - 同义词理解
5
  - 专有名词 / 缩写                   - 意图推断
6
  - 人名 / 地名                       - 上下文理解

1
文本 → Embedding 模型 → 高维向量 → 向量空间
2

3
"如何修复慢查询"   →  [0.12, -0.34, 0.87, ...]  ─┐
4
"数据库性能优化"   →  [0.11, -0.32, 0.85, ...]  ─┤→ 余弦相似度 ≈ 0.97（高度相关）
5
"今天天气怎么样"   →  [0.63,  0.21,-0.14, ...]  ─┘→ 余弦相似度 ≈ 0.12（无关）

1
                         n
2
RRF_Score(d) = Σ  ─────────────────
3
               i  k + rank_i(d)
4

5
其中 k 通常取 60，rank_i(d) 为文档 d 在第 i 路的排名

1
# LangChain EnsembleRetriever 内置 RRF
2
ensemble = EnsembleRetriever(
3
    retrievers=[dense_retriever, sparse_retriever],
4
    weights=[0.5, 0.5]   # weights 影响 RRF 中各路的权重系数
5
)

1
Hybrid_Score(d) = α · Score_dense(d) + (1-α) · Score_sparse(d)
2

3
α = 1.0  →  纯向量检索
4
α = 0.5  →  均衡混合（默认起点）
5
α = 0.0  →  纯关键词检索

1
# 手动实现加权融合（带归一化）
2
def normalize_scores(docs_scores):
3
    """Min-Max 归一化"""
4
    scores = [s for _, s in docs_scores]
5
    min_s, max_s = min(scores), max(scores)
6
    if max_s == min_s:
7
        return [(d, 1.0) for d, _ in docs_scores]
8
    return [(d, (s - min_s) / (max_s - min_s)) for d, s in docs_scores]

1
# Qdrant 中使用 DBSF
2
from qdrant_client.models import FusionQuery, Fusion
3

4
results = client.query_points(
5
    collection_name="my_collection",
6
    prefetch=[dense_prefetch, sparse_prefetch],
7
    query=FusionQuery(fusion=Fusion.DBSF),  # 使用 DBSF
8
)

1
你的情况                              推荐策略
2
─────────────────────────────────────────────────────
3
快速上线，无时间调参              →  RRF（开箱即用）
4
有标注数据，需要最优性能          →  加权融合 + evaluate_alpha()
5
使用 Qdrant，需精细分数控制       →  DBSF
6
生产系统，ES/OpenSearch 后端      →  RRF（原生支持）

1
方案                                  nDCG 相对增益指数
2
─────────────────────────────────────────────────────
3
纯向量检索                                  62
4
纯 BM25 关键词                              55
5
BM25 + 向量（二路）                          74  ↑+19%
6
稀疏向量(SPLADE) + 向量（二路）               77  ↑+22%
7
BM25 + 向量 + 稀疏向量（三路）               86  ↑+38%
8
三路 + ColBERT 重排                          94  ↑+51%

1
┌──────────────────────────────────────────────────────────────┐
2
│                       三路检索分工                            │
3
├──────────────┬───────────────────────────────────────────────┤
4
│ BM25         │ 精确匹配标识符、法条编号、产品型号              │
5
│              │ 覆盖所有 OOD 词汇（不依赖训练集）              │
6
├──────────────┼───────────────────────────────────────────────┤
7
│ SPLADE       │ 稀疏语义向量，介于词汇与语义之间               │
8
│（稀疏向量）   │ 对近义词有一定泛化，但对新词仍有盲区           │
9
├──────────────┼───────────────────────────────────────────────┤
10
│ Dense        │ 深度语义理解，捕捉意图                         │
11
│（稠密向量）   │ 跨词汇障碍，多语言泛化                         │
12
└──────────────┴───────────────────────────────────────────────┘

1
传统流程：检索 Top-20 → 外部 Reranker（延迟 +200ms）→ Top-5
2
ColBERT：检索 Top-1000 → 库内重排（延迟 +50ms）→ Top-5
3

4
更大召回范围 + 更低延迟 = 显著质量提升

1
# ── 安装依赖 ──────────────────────────────────────
2
# pip install langchain langchain-community langchain-openai
3
# pip install rank-bm25 chromadb cohere
4

5
from langchain.retrievers import BM25Retriever, EnsembleRetriever
6
from langchain_community.vectorstores import Chroma
7
from langchain_openai import OpenAIEmbeddings
8
from langchain.text_splitter import RecursiveCharacterTextSplitter
9

10
# ── Step 1: 文档切块 ───────────────────────────────
11
def prepare_retriever(docs_path: str, dense_weight: float = 0.5):
12
    from langchain_community.document_loaders import DirectoryLoader
13
    loader = DirectoryLoader(docs_path, glob="**/*.txt")
14
    raw_docs = loader.load()
15

16
    splitter = RecursiveCharacterTextSplitter(
17
        chunk_size=512,
18
        chunk_overlap=64,
19
        separators=["\n\n", "\n", "。", ".", " "]
20
    )
21
    chunks = splitter.split_documents(raw_docs)
22

23
    # ── Step 2: 构建向量库（稠密检索器）─────────────
24
    vectorstore = Chroma.from_documents(
25
        documents=chunks,
26
        embedding=OpenAIEmbeddings(model="text-embedding-3-small")
27
    )
28
    dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
29

30
    # ── Step 3: 构建 BM25（稀疏检索器）──────────────
31
    sparse_retriever = BM25Retriever.from_documents(chunks)
32
    sparse_retriever.k = 10
33

34
    # ── Step 4: 融合，RRF 算法 ──────────────────────
35
    ensemble = EnsembleRetriever(
36
        retrievers=[dense_retriever, sparse_retriever],
37
        weights=[dense_weight, 1 - dense_weight]
38
    )
39
    return ensemble, chunks, vectorstore
40

41
# 使用
42
retriever, chunks, vs = prepare_retriever("./docs", dense_weight=0.5)
43
results = retriever.invoke("如何申请年假？")

1
权重方案：向量 0.4 / BM25 0.6
2
理由：条款编号不在 Embedding 训练集中，BM25 主导确保精确命中

1
from langchain.retrievers import ContextualCompressionRetriever
2
from langchain_cohere import CohereRerank
3

4
# 法律场景：BM25 略微主导
5
legal_ensemble = EnsembleRetriever(
6
    retrievers=[dense_retriever, sparse_retriever],
7
    weights=[0.4, 0.6]
8
)
9

10
# 强烈建议配合 Reranker 精排
11
compressor = CohereRerank(
12
    model="rerank-multilingual-v3.0",
13
    top_n=5
14
)
15
legal_retriever = ContextualCompressionRetriever(
16
    base_compressor=compressor,
17
    base_retriever=legal_ensemble
18
)
19

20
# 测试效果
21
test_queries = [
22
    "第 12 条第 2 款的数据处理义务",  # 含条款编号
23
    "数据主体有哪些权利",             # 纯语义
24
    "GDPR Article 17 删除权"         # 含英文标识符
25
]
26
for q in test_queries:
27
    results = legal_retriever.invoke(q)
28
    print(f"查询: {q}")
29
    print(f"Top-1: {results[0].page_content[:100]}\n")

1
查询类型                         纯向量      纯 BM25     混合 0.4/0.6
2
────────────────────────────────────────────────────────────────────
3
"第 12 条第 2 款的义务"          ❌语义漂移  ✅精确命中   ✅精确命中
4
"数据主体有哪些权利"             ✅语义丰富  ❌词汇依赖   ✅语义+全面
5
"GDPR Article 17 删除权"        ❌漏标识符  ✅命中       ✅命中且语义丰富

1
权重方案：向量 0.2 / BM25 0.8
2
理由：产品编号是 OOD 数据，BM25 强主导确保型号命中

1
# 电商场景：BM25 强主导 + SKU 专项子索引
2
from langchain.retrievers import BM25Retriever, EnsembleRetriever
3

4
# 对产品编号字段单独建 BM25 子索引
5
product_chunks = [
6
    doc for doc in chunks
7
    if doc.metadata.get("type") == "product"
8
]
9
sku_retriever = BM25Retriever.from_documents(product_chunks)
10
sku_retriever.k = 5
11

12
# 主 BM25（全局）
13
global_sparse = BM25Retriever.from_documents(chunks)
14
global_sparse.k = 5
15

16
# 三路融合：向量 + 全局BM25 + SKU专项BM25
17
ecommerce_retriever = EnsembleRetriever(
18
    retrievers=[dense_retriever, global_sparse, sku_retriever],
19
    weights=[0.2, 0.4, 0.4]
20
)

1
查询                              纯向量结果              混合结果
2
────────────────────────────────────────────────────────────────────
3
"IPH-15-PRO-256 多少钱"          返回错误型号            ✅ 精确命中
4
"苹果最新旗舰手机"               ✅ 语义匹配              ✅ 兼顾
5
"256G 蓝色手机推荐"              ✅ 语义理解              ✅ 规格+语义

1
import re
2
from langchain.retrievers import EnsembleRetriever
3

4
# 标识符检测规则：数字字母混合、版本号、文件编号
5
IDENTIFIER_PATTERN = re.compile(
6
    r'[A-Z]{2,}-\d+|'    # HR-2024, OA-001
7
    r'\d{4}/\d+|'         # 2024/003
8
    r'v\d+\.\d+|'         # v2.1
9
    r'[A-Z]{3,}\d{3,}'    # OKR001, SLA100
10
)
11

12
def smart_retriever(query: str, vectorstore, chunks):
13
    """根据查询类型动态切换权重"""
14
    dense_ret = vectorstore.as_retriever(search_kwargs={"k": 10})
15
    sparse_ret = BM25Retriever.from_documents(chunks)
16
    sparse_ret.k = 10
17

18
    has_identifier = bool(IDENTIFIER_PATTERN.search(query))
19

20
    if has_identifier:
21
        # 含标识符：BM25 主导
22
        weights = [0.3, 0.7]
23
        print(f"[路由] 含标识符 → BM25 主导 (0.3/0.7)")
24
    else:
25
        # 纯语义查询：向量主导
26
        weights = [0.7, 0.3]
27
        print(f"[路由] 纯语义 → 向量主导 (0.7/0.3)")
28

29
    retriever = EnsembleRetriever(
30
        retrievers=[dense_ret, sparse_ret],
31
        weights=weights
32
    )
33
    return retriever.invoke(query)
34

35
# 测试动态路由
36
test_cases = [
37
    "年假怎么申请",               # → 向量主导
38
    "HR-NORM-2024/003 政策内容",  # → BM25 主导
39
    "OA 系统在哪里提交申请",      # → BM25 主导（含系统名）
40
    "试用期转正需要哪些材料",      # → 向量主导
41
]

1
查询类型                          推荐向量权重    推荐 BM25 权重
2
────────────────────────────────────────────────────────────
3
"年假怎么申请"                      0.7           0.3
4
"HR-NORM-2024/003 政策"            0.2           0.8
5
"OA 系统请假在哪"                   0.5           0.5
6
"试用期转正流程"                    0.6           0.4

1
权重方案：向量 0.35 / BM25 0.65
2
理由：代码场景精确匹配优先，MMR 模式减少结果冗余

1
from langchain.text_splitter import RecursiveCharacterTextSplitter, Language
2

3
# 代码专用切块：保持代码块完整性
4
code_splitter = RecursiveCharacterTextSplitter.from_language(
5
    language=Language.PYTHON,
6
    chunk_size=512,
7
    chunk_overlap=64
8
)
9
code_chunks = code_splitter.split_documents(tech_docs)
10

11
# BM25 对代码特别有效：小写归一化统一大小写
12
tech_sparse = BM25Retriever.from_documents(
13
    code_chunks,
14
    preprocess_func=lambda x: x.lower()  # 统一大小写
15
)
16
tech_sparse.k = 8
17

18
# 向量使用 MMR 模式，减少重复文档
19
tech_dense = vectorstore.as_retriever(
20
    search_type="mmr",
21
    search_kwargs={
22
        "k": 8,
23
        "fetch_k": 20,
24
        "lambda_mult": 0.7  # 0.7 偏相关性，0.3 偏多样性
25
    }
26
)
27

28
tech_retriever = EnsembleRetriever(
29
    retrievers=[tech_dense, tech_sparse],
30
    weights=[0.35, 0.65]
31
)

1
查询                                       纯向量结果              混合结果
2
─────────────────────────────────────────────────────────────────────────
3
AttributeError: 'list' has no attr 'keys'  返回"错误处理最佳实践"  ✅精确返回该报错文档
4
requests.get() 用法                         返回其他 HTTP 库文档    ✅精确命中 requests 库
5
"如何优化慢 SQL"                            ✅ 语义理解好            ✅ 同样好

1
权重方案：向量 0.7 / BM25 0.3
2
理由：学术概念需要语义泛化，"自注意力机制"和"Transformer 注意力"是同一概念

1
academic_retriever = EnsembleRetriever(
2
    retrievers=[dense_retriever, sparse_retriever],
3
    weights=[0.7, 0.3]
4
)

1
权重方案：向量 0.6 / BM25 0.4 + 动态路由（检测到订单号时 BM25 加权）

1
ORDER_PATTERN = re.compile(r'[A-Z]{2}\d{10,}|\d{14,}')
2

3
def cs_retriever(query: str):
4
    has_order_id = bool(ORDER_PATTERN.search(query))
5
    weights = [0.3, 0.7] if has_order_id else [0.6, 0.4]
6
    return EnsembleRetriever(
7
        retrievers=[dense_retriever, sparse_retriever],
8
        weights=weights
9
    ).invoke(query)

1
import time
2
import re
3
import numpy as np
4
from typing import List, Optional
5
from dataclasses import dataclass, field
6
from langchain_core.documents import Document
7
from langchain.retrievers import BM25Retriever, EnsembleRetriever
8
from langchain.retrievers import ContextualCompressionRetriever
9
from langchain_cohere import CohereRerank
10

11

12
@dataclass
13
class RetrievalMetric:
14
    query: str
15
    latency_ms: float
16
    num_results: int
17
    dense_weight: float
18
    used_reranker: bool
19

20

21
class HybridRAGPipeline:
22
    """
23
    生产级混合检索 Pipeline
24
    支持：权重调节 / 自动寻优 / 延迟监控 / Reranker 集成
25
    """
26

27
    def __init__(
28
        self,
29
        vectorstore,
30
        documents: List[Document],
31
        dense_weight: float = 0.5,
32
        top_k: int = 10,
33
        rerank_top_n: int = 5,
34
        use_reranker: bool = True,
35
        dynamic_routing: bool = False,
36
    ):
37
        self.dense_weight = dense_weight
38
        self.top_k = top_k
39
        self.use_reranker = use_reranker
40
        self.dynamic_routing = dynamic_routing
41
        self.metrics: List[RetrievalMetric] = []
42

43
        # 构建两路检索器
44
        self.dense_retriever = vectorstore.as_retriever(
45
            search_kwargs={"k": top_k}
46
        )
47
        self.sparse_retriever = BM25Retriever.from_documents(documents)
48
        self.sparse_retriever.k = top_k
49

50
        # 可选：Reranker
51
        self._reranker = None
52
        if use_reranker:
53
            self._reranker = CohereRerank(
54
                model="rerank-multilingual-v3.0",
55
                top_n=rerank_top_n
56
            )
57

58
        # 标识符检测规则（动态路由用）
59
        self._id_pattern = re.compile(
60
            r'[A-Z]{2,}-\d+|\d{4}/\d+|v\d+\.\d+|[A-Z]{3,}\d{3,}'
61
        )
62

63
    def _build_ensemble(self, dense_weight: float) -> EnsembleRetriever:
64
        return EnsembleRetriever(
65
            retrievers=[self.dense_retriever, self.sparse_retriever],
66
            weights=[dense_weight, 1 - dense_weight]
67
        )
68

69
    def _detect_weight(self, query: str) -> float:
70
        """动态路由：根据查询特征自动调整权重"""
71
        if not self.dynamic_routing:
72
            return self.dense_weight
73
        has_id = bool(self._id_pattern.search(query))
74
        return 0.3 if has_id else 0.7
75

76
    def retrieve(self, query: str) -> List[Document]:
77
        start = time.time()
78

79
        # 确定权重
80
        weight = self._detect_weight(query)
81
        ensemble = self._build_ensemble(weight)
82

83
        # 加 Reranker
84
        if self._reranker:
85
            retriever = ContextualCompressionRetriever(
86
                base_compressor=self._reranker,
87
                base_retriever=ensemble
88
            )
89
        else:
90
            retriever = ensemble
91

92
        results = retriever.invoke(query)
93
        elapsed_ms = (time.time() - start) * 1000
94

95
        # 记录指标
96
        self.metrics.append(RetrievalMetric(
97
            query=query,
98
            latency_ms=round(elapsed_ms, 1),
99
            num_results=len(results),
100
            dense_weight=weight,
101
            used_reranker=self.use_reranker,
102
        ))
103

104
        return results
105

106
    def evaluate_alpha(
107
        self,
108
        test_queries: List[str],
109
        ground_truth: dict,
110
        k: int = 5,
111
        alpha_steps: float = 0.1,
112
    ) -> dict:
113
        """
114
        自动搜索最优 alpha 权重
115

116
        Args:
117
            test_queries: 测试查询列表
118
            ground_truth: {query: [relevant_doc_ids]} 标注数据
119
            k: Precision@k
120
            alpha_steps: 搜索步长
121

122
        Returns:
123
            {'best_alpha': 0.4, 'scores': {0.1: 0.62, 0.2: 0.71, ...}}
124
        """
125
        results_log = {}
126
        best_alpha, best_score = 0.5, 0.0
127

128
        for alpha in np.arange(0.1, 1.0, alpha_steps):
129
            alpha = round(float(alpha), 1)
130
            hits = 0
131
            ensemble = self._build_ensemble(alpha)
132

133
            for q in test_queries:
134
                docs = ensemble.invoke(q)
135
                retrieved_ids = [d.metadata.get("id", "") for d in docs[:k]]
136
                relevant = ground_truth.get(q, [])
137
                hits += len(set(retrieved_ids) & set(relevant))
138

139
            score = hits / (len(test_queries) * k) if test_queries else 0
140
            results_log[alpha] = round(score, 4)
141

142
            if score > best_score:
143
                best_score, best_alpha = score, alpha
144

145
        print("─" * 50)
146
        print(f"  Precision@{k} 各 alpha 得分：")
147
        for a, s in results_log.items():
148
            bar = "█" * int(s * 40)
149
            print(f"  α={a:.1f}  {bar} {s:.4f}")
150
        print(f"\n  ✅ 最优 alpha = {best_alpha:.1f}，Precision@{k} = {best_score:.4f}")
151
        print("─" * 50)
152

153
        return {"best_alpha": best_alpha, "scores": results_log}
154

155
    def print_metrics_summary(self):
156
        """打印检索性能统计"""
157
        if not self.metrics:
158
            print("暂无检索记录")
159
            return
160
        latencies = [m.latency_ms for m in self.metrics]
161
        print(f"\n检索统计（共 {len(self.metrics)} 次）：")
162
        print(f"  平均延迟：{np.mean(latencies):.1f}ms")
163
        print(f"  P95 延迟：{np.percentile(latencies, 95):.1f}ms")
164
        print(f"  最大延迟：{max(latencies):.1f}ms")

1
# ── 初始化 Pipeline ────────────────────────────────
2
pipeline = HybridRAGPipeline(
3
    vectorstore=vectorstore,
4
    documents=chunks,
5
    dense_weight=0.5,
6
    top_k=10,
7
    rerank_top_n=5,
8
    use_reranker=True,
9
    dynamic_routing=True,   # 开启动态路由
10
)
11

12
# ── 单次检索 ──────────────────────────────────────
13
results = pipeline.retrieve("如何申请年假？")
14
for i, doc in enumerate(results, 1):
15
    print(f"{i}. {doc.page_content[:80]}...")
16

17
# ── 自动寻优最优 alpha ─────────────────────────────
18
best = pipeline.evaluate_alpha(
19
    test_queries=[
20
        "年假政策是什么",
21
        "HR-NORM-2024/003 规定",
22
        "请假流程怎么走",
23
    ],
24
    ground_truth={
25
        "年假政策是什么":   ["doc_001", "doc_002"],
26
        "请假流程怎么走":   ["doc_005", "doc_006"],
27
    },
28
    k=5,
29
)
30

31
# ── 性能报告 ──────────────────────────────────────
32
pipeline.print_metrics_summary()

1
问题                   根本原因                    解决方案
2
──────────────────────────────────────────────────────────────────────
3
延迟过高               两路检索串行执行             用 asyncio.gather() 并行
4
BM25 中文分词差         默认按空格切词              接入 jieba 自定义 preprocess_func
5
向量效果退化            业务数据持续更新             定期增量重建向量索引
6
结果高度重复            两路召回来自相同文档          向量侧改用 MMR 检索
7
新产品/新词召回差        OOD 数据问题                提升 BM25 权重，补充领域词典
8
Reranker 延迟高         每次调用外部 API            改用本地 ColBERT 或批量调用
9
内存占用过大            HNSW 全量加载               改用 IVF-PQ 量化压缩索引

1
import jieba
2

3
def chinese_tokenizer(text: str) -> list:
4
    """为 BM25 提供中文精准分词"""
5
    return list(jieba.cut(text))
6

7
# 初始化时注入分词器
8
zh_sparse_retriever = BM25Retriever.from_documents(
9
    chunks,
10
    preprocess_func=chinese_tokenizer
11
)
12

13
# 可选：加载领域词典提升专业词精准度
14
jieba.load_userdict("domain_dict.txt")
15
# domain_dict.txt 格式：每行一个词，如：
16
# 混合检索 5 n
17
# EnsembleRetriever 10 n
18
# BM25 10 n

1
import asyncio
2
from langchain_core.documents import Document
3

4
async def async_hybrid_retrieve(
5
    query: str,
6
    dense_retriever,
7
    sparse_retriever,
8
    weights=(0.5, 0.5)
9
) -> List[Document]:
10
    """并行执行两路检索，显著降低延迟"""
11
    dense_task = asyncio.create_task(
12
        asyncio.to_thread(dense_retriever.invoke, query)
13
    )
14
    sparse_task = asyncio.create_task(
15
        asyncio.to_thread(sparse_retriever.invoke, query)
16
    )
17
    dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)
18

19
    # RRF 融合
20
    return rrf_fusion(dense_results, sparse_results, weights)

1
┌─────────────────────────────────────────────────────────────┐
2
│                    混合检索技术选型                          │
3
├───────────────┬─────────────────────────────────────────────┤
4
│ 向量数据库    │ Qdrant（三路混合原生支持）                    │
5
│               │ Chroma（轻量本地，适合快速验证）              │
6
│               │ Weaviate（大规模生产）                       │
7
├───────────────┼─────────────────────────────────────────────┤
8
│ 关键词引擎    │ Elasticsearch 8.9+（RRF 原生支持）           │
9
│               │ BM25Retriever（LangChain 内置，快速起步）    │
10
├───────────────┼─────────────────────────────────────────────┤
11
│ 重排序器      │ Cohere Rerank（云端，效果最优）              │
12
│               │ ColBERT（本地部署，延迟低）                   │
13
│               │ BGE-Reranker（中文场景推荐）                 │
14
├───────────────┼─────────────────────────────────────────────┤
15
│ 框架          │ LangChain EnsembleRetriever（快速上线）      │
16
│               │ LlamaIndex（更细粒度控制）                   │
17
└───────────────┴─────────────────────────────────────────────┘

1
阶段 1（1–2 天）快速验证
2
├── EnsembleRetriever(weights=[0.5, 0.5])
3
├── 本地 Chroma + BM25Retriever
4
└── 验证混合检索 vs 单路的基础效果差异
5

6
阶段 2（1 周）调优
7
├── 收集 20–50 条标注查询
8
├── evaluate_alpha() 自动寻优
9
├── 针对业务场景微调权重
10
└── 引入 Cohere Rerank 精排
11

12
阶段 3（2–4 周）生产化
13
├── 迁移到 Elasticsearch 或 Qdrant
14
├── 实现动态路由（标识符检测）
15
├── 接入延迟监控和质量报警
16
└── 建立周期性向量索引刷新机制
17

18
阶段 4（持续）闭环优化
19
├── 收集用户反馈，扩充标注集
20
├── 监控 precision@5、recall@10
21
├── 考虑引入三路混合（BM25 + SPLADE + Dense）
22
└── 探索 ColBERT 本地重排降低延迟