RAG 全链路优化

从"能用"到"好用"。

第一章：RAG常见问题

做过 RAG 的同学大概都踩过这三个坑：

病灶	症状	根因
切片粗暴	检索回来的内容支离破碎，上下文断裂	按固定字符数硬切，把一句话切成两半
检索不精准	召回的内容跟问题关系不大	只用向量相似度，忽略了关键词匹配
没有大局观	找到了局部细节，但遗漏了全局信息	单次检索只能看到"一小块"，缺乏多视角探索

这些问题不是孤立的，它们会层层叠加：切片质量差 → 检索命中率低 → 大模型拿不到有效上下文 → 生成的答案牛头不对马嘴。

第二章：Milvus 向量数据库实战

2.1 为什么选 Milvus

Milvus 是目前国内生态最成熟的开源向量数据库之一，支持亿级向量检索，社区活跃，文档齐全。对比 FAISS（纯内存、无持久化）和 Chroma（轻量但功能有限），Milvus 在生产环境中更有优势。

官方文档：https://milvus.io/docs/zh

2.2 安装与连接

docker-compose.yml

使用docker安装
docker compose up -d

# 安装 Python SDK
uv pip install pymilvus -i https://mirrors.aliyun.com/pypi/simple

from pymilvus import MilvusClient, DataType

# 连接 Milvus 服务（默认端口 19530）
client = MilvusClient(uri="http://localhost:19530")

2.3 创建集合（Collection）

Milvus 的"集合"类似于关系型数据库的"表"。我们需要定义字段结构，然后创建集合和索引。

dimension = 1024  # 向量维度，需要和 Embedding 模型的输出维度一致
collection_name = "demo_collection"
metric_type = "COSINE"  # 余弦相似度，适合文本语义匹配

# ---- 第一步：定义 Schema ----
# Schema 描述了集合中每条数据的结构
schema = client.create_schema()
# 主键字段，INT64 类型，自动递增
schema.add_field(field_name="id", is_primary=True, auto_id=True, datatype=DataType.INT64)
# 向量字段，存储文本的 Embedding 表示
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=dimension)
# 文本字段，存储原始文本内容
schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=2000)

# ---- 第二步：创建集合 ----
# 如果已存在同名集合，先删除（开发阶段用，生产环境慎用）
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

client.create_collection(
    collection_name=collection_name,
    schema=schema,
    metric_type=metric_type,
)

# ---- 第三步：创建索引 ----
# 索引决定了向量检索的速度和精度平衡
index_params = MilvusClient.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="AUTOINDEX",  # 自动选择最优索引，适合大多数场景
)
client.create_index(
    collection_name=collection_name,
    index_params=index_params
)

索引类型选择小贴士：AUTOINDEX 是最省心的选择，Milvus 会根据数据量自动决策。如果你对性能有极致要求，可以研究 IVF_FLAT、HNSW 等具体索引类型。

2.4 插入数据

插入数据的流程：文本 → Embedding 模型转成向量 → 存入 Milvus。

from langchain_community.embeddings import DashScopeEmbeddings

# 初始化 Embedding 模型（这里用阿里通义的 text-embedding-v3）
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"  # 替换为你的真实 API Key
)

# 准备一些测试文本
docs = [
    "Milvus 就像是一个专门存放‘向量’的超级仓库，能在几毫秒内从千万级数据中找出最相似的片段。",
    "RAG 技术的本质，就是给大模型配上一个能随时查阅的‘外部外挂大脑’（知识库）。",
    "大模型之所以聪明，是因为它通过多层神经网络，学会了文字背后的隐藏含义和逻辑关系。",
    "为了应对不同的数据量，向量数据库提供了多种‘索引’策略，选对了索引，检索速度能快十倍。",
    "Embedding 就像是一座桥梁，把人类看得懂的文字，翻译成了计算机能算得清楚的空间坐标。"
]

# 将文本转为向量并组装成插入数据
data = []
for doc in docs:
    vector = embedding_model.embed_query(doc)  # 文本 → 向量
    data.append({"vector": vector, "text": doc})

# 批量插入
res = client.insert(
    collection_name=collection_name,
    data=data
)
print(f"成功插入 {len(data)} 条数据")

2.5 向量检索

检索时，用户的问题同样需要先转成向量，然后在 Milvus 中做近似最近邻（ANN）搜索。

# 加载集合到内存（检索前必须执行）
if client.has_collection(collection_name):
    client.load_collection(collection_name)

    # 将查询文本转为向量
    query_vector = embedding_model.embed_query("Milvus是什么？")

    # 执行向量检索
    res = client.search(
        collection_name=collection_name,
        data=[query_vector],       # 查询向量（支持批量）
        anns_field="vector",       # 在哪个向量字段上搜索
        limit=2,                   # 返回最相似的 Top-K 条
        output_fields=["id", "text"]  # 需要返回哪些字段
    )

    # 打印结果
    for hit in res[0]:
        print(f"相似度: {hit['distance']:.4f}  内容: {hit['entity']['text']}")
else:
    print(f"集合 '{collection_name}' 不存在，请先创建")

2.6 与 LangChain 集成

实际项目中，我们通常不会直接手写 Milvus 的增删改查，而是通过 LangChain 的封装来简化流程。

# langchain_community要使用0.4.1版本
from langchain_community.vectorstores import Milvus
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings

# 初始化 Embedding 模型
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"
)

# 第一步：加载文档（以 PDF 为例）
loader = PyMuPDFLoader("car_info.pdf")
docs = loader.load()

# 第二步：切片
# chunk_size: 每个切片的最大字符数
# chunk_overlap: 相邻切片的重叠字符数，防止上下文断裂
text_splitter = RecursiveCharacterTextSplitter(
    # 分隔符优先级：段落 → 换行 → 句号 → 空格 → 硬切
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],
    
    # 每个块最大 50 字符
    chunk_size=50,
    
    # 相邻块重叠 10 字符（chunk_size 的 20%）
    chunk_overlap=10,
    
    # 长度计算函数
    length_function=len
)
split_docs = text_splitter.split_documents(docs)

# 过滤空文档
valid_docs = [doc for doc in split_docs if doc.page_content and doc.page_content.strip()]

# 第三步：存入 Milvus（LangChain 会自动处理 Embedding 转换）
vectorstore = Milvus.from_documents(
    documents=valid_docs,
    embedding=embedding_model,
    connection_args={"uri": "http://localhost:19530"},
    collection_name="car_info_collection"
)
print(f"已创建集合，包含 {len(valid_docs)} 个文档切片")

在浏览器访问 http://{你的Milvus IP}:9091/webui 打开Milvus 内置的图形化界面查看数据库和集合 (Collection) 的列表及详细信息。

检索就更简单了：

# 创建检索器，k=1 表示返回最相似的 1 条
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})

# 执行检索
query = "在多少速度范围内，HDC才会激活？"
relevant_docs = retriever.invoke(query)

for doc in relevant_docs:
    print(f"内容: {doc.page_content}")
    print(f"来源: {doc.metadata}")

2.7 检索工具与 Agent 集成

把检索器变成工具

在 LangChain 的 Agent 架构中，检索器可以被封装成一个"工具"，让 Agent 自主决定何时调用。

from langchain_core.tools.retriever import create_retriever_tool
from langchain_community.chat_models import ChatTongyi
from langchain.agents import create_agent

# 初始化大模型
llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# 将向量存储转为检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 把检索器封装为工具
# name: 工具名称，Agent 用它来引用这个工具
# description: 工具描述，Agent 根据描述决定何时使用——写得好不好直接影响 Agent 的决策质量
retriever_tool = create_retriever_tool(
    retriever,
    name="product_info_retriever",
    description="当用户询问产品相关信息时使用此工具，包括保修政策、技术参数、使用方法等。",
)

# 创建 Agent，传入工具列表
tools = [retriever_tool]
agent = create_agent(llm, tools)

# Agent 会根据问题自动判断是否需要调用检索工具
query = "在多少速度范围内，HDC才会激活？"
res = agent.invoke({"messages": [("human", query)]})
print(res['messages'][-1])

关键点：description 是 Agent 决策的依据。写得太模糊（如"搜索信息"），Agent 可能在不该用的时候乱用；写得太具体（如"只搜索保修信息"），Agent 可能在该用的时候不用。建议写清楚工具的适用范围。

第三章：Embedding 模型选型

3.1 为什么 Embedding 选型很重要

Embedding 模型是 RAG 系统的"眼睛"——它决定了系统如何"理解"文本。选错了模型，后续所有的切片、检索、精排优化都是事倍功半。

Embedding 模型的核心任务：把文本映射到高维向量空间，使得语义相近的文本在向量空间中距离也近。

文本: "Milvus 是一个高性能向量数据库"
       ↓ Embedding 模型
向量: [0.23, -0.15, 0.87, 0.04, ..., -0.42]  (如 1024 维)

选择 Embedding 模型时，需要同时考虑效果、速度、成本、部署方式四个维度。

3.2 主流 Embedding 模型对比

MTEB 基准

MTEB（Massive Text Embedding Benchmark）是目前最权威的 Embedding 模型评测基准，涵盖分类、聚类、配对、重排、检索、语义相似度等 8 大类任务。

注意：MTEB 分数高不代表在你的数据上效果好。选型时一定要用自己的真实数据做二次评估。

主流中文模型

模型	维度	最大长度	特点	适用场景
BGE-M3	1024	8192	多语言、稀疏+稠密双表示、Matryoshka 自适应维度	通用中文 RAG、多语言场景
BGE-Large-ZH-v1.5	1024	512	中文 SOTA、轻量	纯中文短文本 RAG
text2vec-large-chinese	1024	512	中文社区广泛使用	中文语义匹配
stella-m3	1024	512	蒸馏模型、推理速度快	对延迟敏感的中文场景

主流多语言/API 模型

模型	维度	最大长度	特点	适用场景
text-embedding-v3	1024	8192	阿里通义出品、中文优化、API 调用	中文场景快速接入
text-embedding-3-large	256-3072	8191	OpenAI 出品、Matryoshka 自适应维度	快速原型、多语言
Cohere embed-v3	1024	512	英文效果好	英文场景
GTE-Qwen2	维度可变	8192	阿里通义出品、长文本支持好	长文档场景
E5-mistral-7b-instruct	4096	32768	基于 Mistral-7B、指令感知	需任务指令的复杂场景
Jina-embeddings-v3	1024	8192	多语言、任务特定 LoRA	多任务灵活切换

3.3 关键选择维度

维度一：向量维度（Dimension）

向量维度决定了存储成本和检索精度的平衡。

维度	优势	劣势	推荐场景
384-512	存储小、检索快	语义表达力有限	简单 FAQ、小规模库
768-1024	性价比最优	—	大多数 RAG 场景（推荐）
1536-4096	语义表达力强	存储大、检索慢	高精度要求、大规模库

经验值：1024 维是当前工业界最主流的选择，效果和成本的甜点位置。

维度二：最大文本长度

Embedding 模型处理文本时有长度上限，超出部分会被截断。

支持长度	适合的切片策略	代表模型
512 token	小块切片（200-500 字）	BGE-ZH-v1.5、text2vec
8192 token	大块切片 / Late Chunking	BGE-M3、text-embedding-v3、GTE-Qwen2

选择策略：

如果你的切片是 500 字符的小块 → 512 token 的模型足够
如果你想做"大块检索"或"全文 Embedding" → 需要 8192 token 的模型
BGE-M3 同时支持长文本和 Matryoshka，是当前中文场景的"万能选手"

维度三：Matryoshka 自适应维度

部分模型（text-embedding-3、BGE-M3 等）支持 Matryoshka 表示学习。核心思想：高维向量天然包含低维信息。

维度 1024 的向量
  ↓ 取前 512 维（丢弃后 512 维）
维度 512 的向量（精度损失 < 2%）
  ↓ 取前 256 维
维度 256 的向量（精度损失 < 5%）

这意味着：一套 Embedding，多套索引。

粗排用 256 维，速度快
精排用 1024 维，精度高
存储 1024 维，检索时按需截断

# BGE-M3 Matryoshka 用法示意
from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)

sentences = ["什么是向量数据库？"]
dense_vecs = model.encode(sentences, batch_size=1)['dense_vecs']

# 需要 256 维？直接取前 256 列，无需重新 Embedding
dense_256 = dense_vecs[:, :256]
# 在 Milvus 中创建 256 维的 Collection 来存储，成本降低 75%

维度四：API 服务 vs 本地部署

方式	代表	优势	劣势
API 服务	DashScope、OpenAI、Cohere	免运维、弹性伸缩、接入快	网络依赖、长期成本较高、数据需出域
本地部署	BGE-M3、GTE、E5	延迟可控、数据不出域、长期成本低	需要 GPU、需要运维

选型建议：原型验证阶段用 API 服务（跑通链路），生产上线阶段优先考虑本地部署（成本与延迟可控）。混合方案也常见——Embedding 走 API，Reranker 部署在本地 GPU。

3.4 如何在自己的数据上评估

MTEB 排名不能代替你的真实场景。一个轻量的评估流程：

# 评估框架示意（简化版）
import numpy as np

def evaluate_hit_rate(embedding_model, test_pairs, vectorstore, k=5):
    """
    简单评估 Top-K 命中率
    test_pairs: [(问题, 正确答案文本), ...]
    """
    hits = 0
    total = len(test_pairs)

    for question, ground_truth in test_pairs:
        # 检索 Top-K
        retrieved = vectorstore.similarity_search(question, k=k)
        retrieved_texts = [doc.page_content for doc in retrieved]

        # 检查正确答案是否在 Top-K 中
        if any(ground_truth in text for text in retrieved_texts):
            hits += 1

    hit_rate = hits / total
    print(f"Top-{k} 命中率: {hit_rate:.2%}")
    return hit_rate


# 使用示例
test_pairs = [
    ("产品保修期是多久？", "三年或十万公里"),
    ("如何更换备胎？", "停车并拉手刹"),
    ("发动机型号是什么？", "2.0T 涡轮增压"),
    # ... 建议准备 50-100 对
]

# 分别用不同的 Embedding 模型测试
models_to_test = {
    "DashScope text-embedding-v3": embedding_model_dashscope,
    "BGE-M3": embedding_model_bge,
    "text2vec": embedding_model_text2vec,
}

for name, model in models_to_test.items():
    print(f"\n--- {name} ---")
    evaluate_hit_rate(model, test_pairs, vectorstore, k=5)

关键评估指标：

指标	含义	计算方式	参考基准
Top-1 命中率	答案出现在第 1 位的概率	命中次数 / 总问题数	> 60% 为可用
Top-3 命中率	答案出现在前 3 位的概率	同上	> 80% 为良好
Top-5 命中率	答案出现在前 5 位的概率	同上	> 90% 为优秀
MRR	第一个正确结果排名的倒数的均值	`mean(1 / rank_of_first_hit)`	越接近 1 越好

3.5 模型推荐速查

你的场景	推荐模型	理由
中文、数据不出域	`BGE-M3`（本地部署）	多语言、长文本、Matryoshka、中文 SOTA 级效果
中文、可以走 API	`DashScope text-embedding-v3`	通义系列、中文优化、接入简单
中英混合	`BGE-M3` 或 `GTE-Qwen2`	多语言效果好
英文为主	`Cohere embed-v3`	英文 MTEB 排名靠前
快速原型验证	`text-embedding-3-small`	便宜、快、够用
短文本（< 512 token）	`BGE-Large-ZH-v1.5`	轻量、效果好
长文档（> 4096 token）	`GTE-Qwen2` 或 `BGE-M3`	支持长上下文 Embedding

一句话总结：中文场景默认选 BGE-M3（本地）或 text-embedding-v3（API），然后用自己的真实数据跑一轮命中率对比，再做最终决策。

第四章：文档智能切片策略

4.1 固定切片的局限

最常用的 RecursiveCharacterTextSplitter 本质上是按字符数硬切——不管你句子说到哪，到了 500 字就一刀下去。这在大多数场景下够用，但遇到以下情况就会翻车：

一段完整的技术说明被切成两半，语义残缺
相邻的两个切片讲的是完全不相关的话题
文档结构复杂（标题、段落、表格混排），硬切会破坏结构

4.2 语义切块（SemanticChunker）

SemanticChunker 的思路完全不同：先按句子拆分，再按语义合并。

核心原理：

将文档拆成句子级别的片段
用 Embedding 模型计算每对相邻句子的语义相似度
在语义发生"突变"的地方切开——相似度低说明话题转了

#安装依赖：uv pip install langchain_experimental==0.4.1 -i https://mirrors.aliyun.com/pypi/simple
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_community.embeddings import DashScopeEmbeddings

# 初始化 Embedding 模型
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"
)

# 创建语义分块器
# breakpoint_threshold_type="percentile" 表示用百分位数法确定切分阈值
# breakpoint_threshold_amount=95 表示只有相似度排名后 5% 的位置才会被切开
text_splitter = SemanticChunker(
    embeddings=embedding_model,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95  # 值越大，切出来的块越少（越粗）
)

loader = PyMuPDFLoader("your_document.pdf")
docs = text_splitter.split_documents(loader.load())

print(f"共切分为 {len(docs)} 个语义块")
print(f"第一个块: {docs[0].page_content[:100]}...")

4.3 百分位数法原理详解

百分位数法是语义切块的核心算法，理解它对调参至关重要。

通俗解释：从"差生划线"说起

假设全班 100 个学生考了一次试（成绩单上的 100 个分数，代表 100 对相邻句子的相似度）：

固定阈值法（传统方式）：设定一条固定分数线，比如 60 分，低于 60 的就是"差生"（分割点）。

如果题目特别难，全班第一名才 55 分 → 按 60 分标准，全班都是"差生"，全切开 → 碎片化
如果题目特别简单，全班最低也有 80 分 → 按 60 分标准，一个"差生"都找不到 → 全不切

固定阈值对"题目难度"（文档整体相似度分布）高度敏感，换个文档就失效了。

百分位数法：不设固定分数线，而是按排名划——"找出排名倒数 5% 的学生"。

题目难、大家分都低？没关系，仍是最差的那 5% 被划出来
题目简单、大家分都高？同样是最差的那 5% 被划出来

不管这次考试难度如何，始终把表现最差的 5% 的边界切出来。这样无论文档的整体相似度是高是低，都能动态、自适应地找到话题转变最剧烈的地方。

什么是百分位数

在一组按从小到大排列的数据中，第p百分位数表示：有p%的数据小于或等于这个数值，同时有(100-p)%的数据大于或等于这个数值。百分位数描述的是"数据分布中的相对位置"，用考试分数来举例最直观：

25 百分位数 = 70  →  25% 的学生分数 ≤ 70 分
50 百分位数 = 80  →  50% 的学生分数 ≤ 80 分（中位数）
75 百分位数 = 90  →  75% 的学生分数 ≤ 90 分
90 百分位数 = 97  →  90% 的学生分数 ≤ 97 分

在语义切块中，breakpoint_threshold_amount = 90 的含义是：取第 90 百分位数的相似度值作为阈值。只有相似度排在倒数 10% 的位置才会被切开——这些位置是整篇文档中话题转变最剧烈的地方。

算法步骤

假设一篇文档被拆成了 10 个句子：[S1, S2, S3, ..., S10]

Step 1：计算相邻句子的余弦相似度

cos(S1, S2) = 0.95   ← 话题连贯
cos(S2, S3) = 0.92   ← 话题连贯
cos(S3, S4) = 0.30   ← 话题突变！
cos(S4, S5) = 0.88
cos(S5, S6) = 0.90
cos(S6, S7) = 0.85
cos(S7, S8) = 0.40   ← 话题突变！
cos(S8, S9) = 0.91
cos(S9, S10) = 0.89

得到相似度数组：[0.95, 0.92, 0.30, 0.88, 0.90, 0.85, 0.40, 0.91, 0.89]

Step 2：计算百分位数阈值

设置 breakpoint_threshold_amount = 90（第 90 百分位数）。

将数组从小到大排序：[0.30, 0.40, 0.85, 0.88, 0.89, 0.90, 0.91, 0.92, 0.95]

9 个元素，第 90 百分位数大约排在第 8 位（9 × 0.9 = 8.1），阈值约等于 0.91。

Step 3：基础方法——用阈值标记候选断点

用阈值 0.91 去比对原始的相邻相似度：

0.95 > 0.91 → 相似，不切
0.92 > 0.91 → 相似，不切
0.30 < 0.91 → ✂️ S3 和 S4 之间切开（话题确实变了）
0.88 < 0.91 → ✂️ S4 和 S5 之间也会被切（问题：这里话题并没有明显转变！）
0.90 < 0.91 → ✂️ S5 和 S6 之间也会被切
0.85 < 0.91 → ✂️ ...
0.40 < 0.91 → ✂️ S7 和 S8 之间切开
0.91 = 0.91 → 恰好踩线，看实现，通常不切
0.89 < 0.91 → ✂️ S9 和 S10 之间被切

⚠️ 重要问题：如果只是简单地"一刀切"，0.88、0.90、0.85、0.89 这些中等相似度的位置都会被切开，导致过度分割——在话题过渡区域产生大量碎片。这显然不是我们想要的。

LangChain 的优化：局部最小值策略

为了避免上述过度分割问题，LangChain 的实际实现不只用单一阈值，而是分两步走：

粗筛：用百分位数阈值找出所有"可疑"的候选断点（相似度低于阈值的位置）
精确定位：在每个候选断点附近（前后若干句子），找到相似度的局部最小值（真正的"谷底"）作为最终切分位置

相似度曲线示意：

  1.0 │  ██
      │  ██  ██     ██  ██         ██  ██  ██
  0.5 │  ██  ██  ██  ██  ██  ██     ██  ██  ██
      │  ██  ██  ██  ██  ██  ██  ██  ██  ██  ██
  0.0 ├─────────────────────────────────────────
      S1—S2 S3—S4 S5—S6 S7—S8 S9—S10
                ▾           ▾
            真正谷底      真正谷底

在 S3-S4 到 S6-S7 这一段中，0.30 就是局部最小值 → 只在这里切一刀
在 S7-S8 到 S9-S10 这一段中，0.40 就是局部最小值 → 只在这里切一刀
结果：9 个候选断点缩减为 2 个真正的切分点，避免了碎片化

参数调优

breakpoint_threshold_amount 控制切分的"敏感度"：

值	含义	效果	适用场景
95	只有后 5% 的相似度被标记	切得少，每块大	长文档、连贯性强的技术文档
90	后 10% 被标记	适中（推荐默认值）	大多数场景
80	后 20% 被标记	切得多，每块小	短文档、话题跳跃频繁的 QA 文档
50	后 50% 被标记	切得非常碎	需要极高检索粒度的场景

调参经验：breakpoint_threshold_amount 值越大，阈值越高，切出来的块越少（越粗）。建议从 90 开始，观察检索效果——如果检索结果经常包含多个不相关话题，说明块太大，降低到 85–80；如果检索结果信息不全，说明切得太碎，提高到 92–95。

第五章：多路召回与混合检索(重点)

5.1 为什么需要多路召回

单一检索方式各有盲区：

检索方式	优势	劣势
向量检索	理解语义，"苹果公司"能匹配到"Apple Inc."	对精确关键词不敏感，(例如:搜：`特斯拉 Model Y 2024款后轮驱动版报价` 向量检索召回：Model 3报价、Model Y 2023款、蔚来ES6报价)
关键词检索（BM25）	精确匹配能力强，速度快	无法理解同义词和语义相似性(例如:"感冒"和"风寒")

多路召回的核心思想：两条腿走路，取长补短。

5.2 BM25 原理：图书馆管理员的打分规则

BM25 是经典的关键词检索算法，理解它有助于调优。用一个图书馆的例子来解释：

你在图书馆找一本"怎么做蛋糕"的书，管理员（搜索引擎）用三条规则打分：

规则一：关键词出现越多，越可能是好书

书 A 提了 1 次"蛋糕"，书 B 提了 10 次 → 书 B 更相关。

但不是越多越好！如果一本书写了 100 次"蛋糕蛋糕蛋糕"全是废话，BM25 会"封顶"——到一定程度就不加分了。这叫词频饱和。

规则二：短书里提到一次，比长书里提到一次更有价值

书 C 只有 5 页，认真讲了"戚风蛋糕做法"；书 D 有 500 页，在第 387 页顺带提了一句"蛋糕也是甜点"。BM25 会惩罚又长又水的书，给短小精悍的书更高分。这叫文档长度归一化。

规则三：越少见的词，越重要

查询"低糖无麸质蛋糕"——"蛋糕"很常见，很多书都有，重要性一般；"无麸质"很少见，只有几本书提到，那这些书很可能就是你要的。这叫逆文档频率（IDF）：越稀有，越珍贵。

一句话总结：BM25 喜欢内容相关、简洁不啰嗦、用词精准的文档，讨厌又长又水的"注水文"。

5.3 混合检索（EnsembleRetriever）

EnsembleRetriever 是 LangChain 中混合检索的核心组件，它做的事情很简单：并行调用多个检索器，然后加权融合结果，解决单一检索器的局限性，提升检索的精准度、召回率和鲁棒性。

#安装依赖:uv pip install rank_bm25 -i https://mirrors.aliyun.com/pypi/simple
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Milvus
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_classic.retrievers import EnsembleRetriever

# 初始化
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"
)

# ---- 创建 BM25 检索器 ----
# BM25 不需要向量，直接基于文本的关键词匹配
bm25_retriever = BM25Retriever.from_documents(valid_docs)
bm25_retriever.k = 10  # 返回 Top-10

# ---- 创建向量检索器 ----
vectorstore = Milvus(
    embedding_function=embedding_model,
    connection_args={"uri": "http://localhost:19530"},
    collection_name="car_info_collection"
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# ---- 创建混合检索器 ----
# weights 控制各检索器的权重，总和建议为 1
# 向量检索权重更高（0.6），因为语义理解通常更重要
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],
    normalize_scores=True  # 将不同检索器的分数归一化到 [0,1]，避免偏差
)

# 执行混合检索
results = ensemble_retriever.invoke("在多少速度范围内，HDC才会激活？")
print(f"混合检索返回 {len(results)} 条结果")
print(results[0].page_content)

5.4 多查询检索（MultiQueryRetriever）

向量检索有个常见问题：用户的提问方式和文档的表述方式存在"词汇鸿沟"。比如用户问"怎么修车"，但文档里写的是"车辆故障排除指南"——语义一样，措辞不同，向量检索可能匹配不上。

MultiQueryRetriever 的解决思路很巧妙：让大模型把一个问题改写成多个不同角度的问题，分别检索，合并去重。

相当于从多个"视角"去探索知识库，大大增加命中率。

from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models import ChatTongyi
from langchain_classic.retrievers.multi_query import MultiQueryRetriever

# 初始化大模型
llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# 自定义改写提示词（可选，不写则用默认的）
CUSTOM_PROMPT = PromptTemplate(
    input_variables=["question"],
    template="""你是一个专业的问题改写助手。请为下面的问题生成 4 个不同的改写版本，
每个版本应该：
- 从不同角度表达相同的意图
- 使用不同的关键词和表达方式
- 保持问题的核心含义

每个问题单独一行，不要编号。

原始问题: {question}

改写后的问题:"""
)

# 创建多查询检索器
# 底层检索器用的是上面的混合检索器，这样每个改写问题都会走混合检索
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=ensemble_retriever,
    llm=llm,
    prompt=CUSTOM_PROMPT
)

# 执行检索
# 假设生成了 4 个改写问题，每个检索 10 条，去重后可能得到 20多 条结果
results = multi_query_retriever.invoke("在多少速度范围内，HDC才会激活？")
print(f"多查询检索返回 {len(results)} 条结果")
print(results[0].page_content)

性能提醒：多查询检索相当于一次请求做了 5 次检索（1 次原始 + 4 次改写），延迟和成本都会增加。适合对召回率要求高的场景，不适合实时性要求极高的在线服务。

5.5 完整的多路召回系统

把上面的组件封装成一个可切换的检索系统：

class MultiRetrieverSystem:
    """
    多路召回检索系统
    支持 BM25、向量检索、混合检索、多查询检索四种模式
    """

    def __init__(self, documents, embedding_model, llm):
        self.documents = documents
        self.embedding_model = embedding_model
        self.llm = llm
        self.setup_retrievers()

    def setup_retrievers(self):
        """初始化所有检索器"""
        # BM25 检索器
        self.bm25 = BM25Retriever.from_documents(self.documents)
        self.bm25.k = 10

        # 向量检索器
        self.vectorstore = Milvus(
            embedding_function=embedding_model,
            connection_args={"uri": "http://localhost:19530"},
            collection_name="car_info_collection"
        )
        self.vector = self.vectorstore.as_retriever(search_kwargs={"k": 10})

        # 混合检索器
        self.ensemble = EnsembleRetriever(
            retrievers=[self.bm25, self.vector], # 待融合的基础检索器列表
            weights=[0.4, 0.6], # BM25权重0.4，向量检索权重0.6 总和建议为1，向量检索权重更高
            normalize_scores=True # 是否将不同检索器的分数归一化到[0,1]（避免因分数范围差异导致融合偏差）
        )

        # 多查询检索器
        self.multi_query = MultiQueryRetriever.from_llm(
            retriever=self.ensemble,
            llm=self.llm,
            prompt=CUSTOM_PROMPT
        )

    def search(self, query, mode="ensemble"):
        """
        执行检索
        mode: "bm25" | "vector" | "ensemble" | "multi_query"
        """
        retriever_map = {
            "bm25": self.bm25,
            "vector": self.vector,
            "ensemble": self.ensemble,
            "multi_query": self.multi_query,
        }
        if mode not in retriever_map:
            raise ValueError(f"不支持的检索模式: {mode}，可选: {list(retriever_map.keys())}")
        return retriever_map[mode].invoke(query)

使用示例：

# 初始化系统
rag_system = MultiRetrieverSystem(valid_docs, embedding_model, llm)

# 对比不同检索策略的效果
query = "在多少速度范围内，HDC才会激活？"

bm25_results = rag_system.search(query, "bm25")
vector_results = rag_system.search(query, "vector")
ensemble_results = rag_system.search(query, "ensemble")
multi_results = rag_system.search(query, "multi_query")

print(f"BM25: {len(bm25_results)} 条 | 向量: {len(vector_results)} 条")
print(f"混合: {len(ensemble_results)} 条 | 多查询: {len(multi_results)} 条")

第六章：查询侧优化——让"问题"本身变好(重点)

前面几章我们一直在优化"检索"和"切片"，但很多时候，问题本身才是瓶颈。用户的原始问题可能模糊、口语化、信息量不足，直接拿去检索效果自然不好。

本章介绍四种查询侧优化技术：查询重写、查询分解、查询澄清、HyDE 查询扩展。它们的共同目标是——在检索之前，先把问题"打磨"好。

6.1 查询重写（Query Rewriting）

问题场景

用户的原始提问往往带有口语化、歧义性或信息不足的问题，直接用于检索效果不佳：

原始问题	问题	优化后
"这个东西怎么用"	太模糊，"东西"指什么？	"该产品的安装使用方法"
"保修坏了咋办"	口语化 + 歧义	"产品在保修期内出现故障的维修流程"
"和 XX 比呢"	缺乏上下文	"本产品与 XX 产品在性能参数上的对比"

原理

查询重写的核心思想：用 LLM 将用户的口语化问题，改写成更精确、更适合检索的形式。

与 MultiQueryRetriever 的区别：

MultiQuery：一个问题 → 多个改写版本，重点是多角度覆盖
Query Rewriting：一个问题 → 一个优化版本，重点是精准提升

两者可以组合使用：先重写优化，再多角度扩展。

实现

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi

llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# 查询重写 Prompt
rewrite_prompt = PromptTemplate(
    input_variables=["query"],
    template="""你是一个查询优化助手。请将用户的口语化问题改写为更适合信息检索的精确查询。

改写要求：
1. 补充隐含的上下文信息
2. 将口语化表达转为专业表述
3. 消除歧义，明确查询意图
4. 保持原意不变，不要添加原问题未提及的内容
5. 直接输出改写后的查询，不要解释

用户问题: {query}

改写后的查询:"""
)

# 构建重写链
rewrite_chain = rewrite_prompt | llm | StrOutputParser()

# 测试
original_query = "这个东西怎么用啊"
rewritten = rewrite_chain.invoke({"query": original_query})
print(f"原始问题: {original_query}")
print(f"重写结果: {rewritten}")
# 输出示例: "该产品的安装配置和使用操作方法"

在 RAG 链路中集成

将查询重写作为检索前的预处理步骤：

class RAGWithQueryRewriting:
    """集成查询重写的 RAG 系统"""

    def __init__(self, retriever, llm, rewrite_chain):
        self.retriever = retriever
        self.llm = llm
        self.rewrite_chain = rewrite_chain

    def invoke(self, question):
        # 第一步：重写查询
        rewritten_query = self.rewrite_chain.invoke({"query": question})
        print(f"原始问题: {question}")
        print(f"重写后:   {rewritten_query}")

        # 第二步：用重写后的查询进行检索
        docs = self.retriever.invoke(rewritten_query)

        # 第三步：用原始问题 + 检索结果生成答案
        context = "\n\n".join([doc.page_content for doc in docs])
        answer_prompt = f"""基于以下上下文回答用户问题。如果上下文中没有相关信息，请说明。

上下文：{context}

用户问题：{question}
答案："""
        answer = self.llm.invoke(answer_prompt).content

        return {
            "original_query": question,
            "rewritten_query": rewritten_query,
            "retrieved_docs": docs,
            "answer": answer
        }

# 使用示例
rag = RAGWithQueryRewriting(
    retriever=ensemble_retriever,
    llm=llm,
    rewrite_chain=rewrite_chain
)
result = rag.invoke("在多少速度范围内，HDC才会激活？")
print(result["answer"])

6.2 查询分解（Query Decomposition）

问题场景

有些问题包含多个子问题，或者需要多步推理才能回答：

复杂问题	分解后
"产品 A 和产品 B 的保修政策有什么区别？"	① 产品 A 的保修政策是什么？② 产品 B 的保修政策是什么？③ 对比两者差异
"这个产品的安装流程和注意事项分别是什么？"	① 安装流程是什么？② 有哪些注意事项？
"为什么推荐用 X 型号，它比 Y 好在哪？"	① X 型号有哪些优势？② Y 型号有哪些不足？③ 推荐 X 的理由是什么？

原理

查询分解的核心思想：用 LLM 将复杂问题拆解为多个简单的子问题，分别检索后合并结果。

这比直接检索复杂问题效果更好，因为：

子问题更聚焦，检索精度更高
每个子问题可以从不同文档中获取信息
最终合并的答案更全面

实现

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi
import json

llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# 查询分解 Prompt
decompose_prompt = PromptTemplate(
    input_variables=["question"],
    template="""你是一个问题分解助手。请将用户的复杂问题分解为 2-4 个独立的子问题，
每个子问题应该能独立检索和回答。

要求：
1. 子问题之间互不依赖，可以并行检索
2. 子问题覆盖原始问题的所有方面
3. 每个子问题简洁明确
4. 以 JSON 数组格式输出

用户问题: {question}

输出格式示例: ["子问题1", "子问题2", "子问题3"]

子问题列表:"""
)

decompose_chain = decompose_prompt | llm | StrOutputParser()


def decompose_query(question: str) -> list[str]:
    """将复杂问题分解为子问题"""
    result = decompose_chain.invoke({"question": question})
    # 解析 JSON 数组
    try:
        sub_queries = json.loads(result.strip())
        return sub_queries
    except json.JSONDecodeError:
        # 兜底：按行分割
        return [line.strip() for line in result.strip().split("\n") if line.strip()]


# 测试
question = "产品 A 和产品 B 的保修政策和价格有什么区别？"
sub_queries = decompose_query(question)
print(f"原始问题: {question}")
print(f"分解结果:")
for i, sq in enumerate(sub_queries, 1):
    print(f"  {i}. {sq}")
# 输出示例:
# 1. 产品 A 的保修政策是什么？
# 2. 产品 B 的保修政策是什么？
# 3. 产品 A 和产品 B 的价格分别是多少？

分解后并行检索与合并

from concurrent.futures import ThreadPoolExecutor
#使用 ThreadPoolExecutor 创建线程池，并行执行检索
#每个子问题独立检索，互不干扰
#按内容哈希值去重（避免重复文档）

def parallel_retrieve_and_merge(sub_queries, retriever, k_per_query=5):
    """
    对每个子问题并行检索，然后合并去重
    """
    all_docs = []
    seen_contents = set()

    def retrieve_one(query):
        return retriever.invoke(query)

    # 创建线程池，线程数等于子问题数量
    with ThreadPoolExecutor(max_workers=len(sub_queries)) as executor:
        #提交所有检索任务
        futures = [executor.submit(retrieve_one, q) for q in sub_queries]
        ## 收集结果
        for future in futures:
            docs = future.result()
            for doc in docs:
                # 按内容去重
                content_hash = hash(doc.page_content)
                if content_hash not in seen_contents:
                    seen_contents.add(content_hash)
                    all_docs.append(doc)

    return all_docs


# 完整的分解-检索-生成流程
class RAGWithDecomposition:
    """集成查询分解的 RAG 系统"""

    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def invoke(self, question):
        # 第一步：分解问题
        sub_queries = decompose_query(question)
        print(f"分解为 {len(sub_queries)} 个子问题")

        # 第二步：并行检索并合并
        all_docs = parallel_retrieve_and_merge(sub_queries, self.retriever)
        print(f"合并后共 {len(all_docs)} 个文档片段")

        # 第三步：用所有上下文生成综合答案
        context = "\n\n".join([doc.page_content for doc in all_docs])
        answer_prompt = f"""基于以下上下文，全面回答用户的问题。
请综合所有相关信息，给出完整、有条理的答案。

上下文：{context}

用户问题：{question}
答案："""
        answer = self.llm.invoke(answer_prompt).content

        return {
            "question": question,
            "sub_queries": sub_queries,
            "doc_count": len(all_docs),
            "answer": answer
        }

# 使用示例
rag_decomp = RAGWithDecomposition(retriever=ensemble_retriever, llm=llm)
result = rag_decomp.invoke("在多少速度范围内，HDC才会激活？")
print(result["answer"])

查询分解 vs 多查询检索：两者看起来相似，但目的不同。多查询检索是同一个问题的不同表述（平铺），查询分解是把一个问题拆成多个子问题（纵深）。在实际系统中可以组合使用。

6.3 查询澄清（Query Clarification）

问题场景

有时用户的提问过于模糊或缺乏关键信息，强行检索不如先问清楚：

模糊问题	问题所在	应该澄清
"怎么退货？"	哪个渠道买的？什么品类？	"请问您是在哪个渠道购买的？线上还是线下？"
"这个多少钱？"	"这个"指什么？	"请问您想了解哪款产品的价格？"
"出了问题怎么办？"	什么问题？硬件还是软件？	"请问您遇到的是什么类型的问题？"

原理

查询澄清的核心思想：用 LLM 判断问题是否信息充分，不充分则生成澄清反问。

这是一种"不急于回答，先搞清楚再行动"的策略，能显著提升用户体验——与其给一个答非所问的答案，不如先问清楚。

实现

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi
import json

llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# 查询澄清 Prompt
clarify_prompt = PromptTemplate(
    input_variables=["query"],
    template="""你是一个智能客服助手。请判断用户的提问是否包含足够的信息来进行准确检索和回答。

判断标准：
1. 问题是否有明确的主语（指代的对象是否清晰）
2. 问题是否包含必要的上下文（时间、地点、产品型号等）
3. 问题是否存在歧义（可能有多种理解方式）

请以 JSON 格式返回：
- 如果问题清晰：{{"need_clarify": false, "reason": "问题清晰的原因"}}
- 如果需要澄清：{{"need_clarify": true, "clarification": "向用户提出的澄清问题", "assumption": "如果必须回答时的合理假设"}}

用户问题: {query}

结果:"""
)

clarify_chain = clarify_prompt | llm | StrOutputParser()


def check_and_clarify(query: str) -> dict:
    """检查是否需要澄清，返回判断结果"""
    result = clarify_chain.invoke({"query": query})
    try:
        return json.loads(result.strip())
    except json.JSONDecodeError:
        # 兜底：假设不需要澄清
        return {"need_clarify": False, "reason": "解析失败，默认直接回答"}


# 测试
test_queries = [
    "怎么退货？",
    "iPhone 16 Pro 的保修期是多久？",
    "这个多少钱？",
]

for q in test_queries:
    result = check_and_clarify(q)
    print(f"\n问题: {q}")
    print(f"需要澄清: {result['need_clarify']}")
    if result['need_clarify']:
        print(f"澄清反问: {result['clarification']}")
        print(f"默认假设: {result.get('assumption', '无')}")

在对话流中集成

class RAGWithClarification:
    """集成查询澄清的 RAG 系统（支持多轮对话）"""

    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
        self.clarify_chain = clarify_chain

    def invoke(self, question, auto_clarify=False):
        """
        auto_clarify=True: 需要澄清时直接用假设继续（适合 API 调用）
        auto_clarify=False: 需要澄清时返回澄清问题（适合交互式对话）
        """
        # 第一步：检查是否需要澄清
        clarify_result = check_and_clarify(question)

        if clarify_result["need_clarify"]:
            if auto_clarify:
                # 自动模式：用假设重新表述问题
                question = clarify_result.get("assumption", question)
                print(f"[自动澄清] 假设用户意图为: {question}")
            else:
                # 交互模式：返回澄清问题，等待用户补充
                return {
                    "status": "need_clarify",
                    "clarification": clarify_result["clarification"],
                    "original_question": question
                }

        # 第二步：正常检索 + 生成
        docs = self.retriever.invoke(question)
        context = "\n\n".join([doc.page_content for doc in docs])

        answer_prompt = f"""基于以下上下文回答用户问题。

上下文：{context}

用户问题：{question}
答案："""
        answer = self.llm.invoke(answer_prompt).content

        return {
            "status": "answered",
            "question": question,
            "answer": answer
        }

# 使用示例
rag_clarify = RAGWithClarification(retriever=ensemble_retriever, llm=llm)

# 交互模式
result = rag_clarify.invoke("在多少速度范围内，HDC才会激活？", auto_clarify=False)
if result["status"] == "need_clarify":
    print(f"🤖: {result['clarification']}")  # "请问您是在哪个渠道购买的？"

# 自动模式（API 场景）
result = rag_clarify.invoke("在多少速度范围内，HDC才会激活？", auto_clarify=True)
print(result["answer"])

6.4 HyDE 查询扩展（Hypothetical Document Embeddings）

问题场景

用户的查询通常是短句或关键词，而知识库中的文档是长段落或完整句子。两者在向量空间中的分布存在天然差距——短查询的 Embedding 往往无法准确指向长文档。

原理

HyDE（Hypothetical Document Embeddings）的思路非常巧妙：

传统流程:  用户问题 → Embedding → 向量检索
HyDE 流程: 用户问题 → LLM 生成假设性回答 → Embedding → 向量检索

核心洞察：一个"假设性回答"的 Embedding，比原始问题的 Embedding 更接近真实文档的向量。

为什么？因为假设性回答在形式上（长度、用词、句式）更接近知识库中的真实文档，向量空间中的距离自然更近。

实现

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings

llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"
)

# HyDE Prompt：生成假设性文档
hyde_prompt = PromptTemplate(
    input_variables=["question"],
    template="""请根据以下问题，写一段可能包含答案的文档片段。
要求：
1. 像真实文档一样专业、详细
2. 包含具体的数据、步骤或事实
3. 长度在 100-200 字之间
4. 即使你不确定答案，也要根据问题合理推测，写出一段"看起来像真的"文档

问题: {question}

假设性文档:"""
)

hyde_chain = hyde_prompt | llm | StrOutputParser()


def hyde_search(question: str, vectorstore, k=5):
    """
    HyDE 检索流程：
    1. 生成假设性文档
    2. 对假设性文档做 Embedding
    3. 用假设性文档的向量去检索
    """
    # 第一步：生成假设性文档
    hypothetical_doc = hyde_chain.invoke({"question": question})
    print(f"假设性文档: {hypothetical_doc[:100]}...")

    # 第二步 + 第三步：用假设性文档的向量检索
    # LangChain 的 similarity_search 会自动对文本做 Embedding
    results = vectorstore.similarity_search(
        query=hypothetical_doc,  # 用假设性文档而非原始问题
        k=k
    )

    return {
        "original_question": question,
        "hypothetical_doc": hypothetical_doc,
        "retrieved_docs": results
    }


# 测试
result = hyde_search(
    question="产品保修期是多久？",
    vectorstore=vectorstore,
    k=3
)

print(f"\n原始问题: {result['original_question']}")
print(f"\n检索到 {len(result['retrieved_docs'])} 个文档:")
for i, doc in enumerate(result['retrieved_docs'], 1):
    print(f"  {i}. {doc.page_content[:80]}...")

HyDE 的局限性与应对

HyDE 并非万能，需要注意以下问题：

局限	说明	应对方案
LLM 幻觉风险	假设性文档可能包含错误信息，导致检索方向偏离	结合原始问题和假设性文档的向量做加权融合
额外延迟	多了一次 LLM 调用	用小模型生成假设性文档，大模型生成最终答案
领域依赖	通用领域的假设性文档质量不如专业领域	在 Prompt 中加入领域上下文

进阶：HyDE + 原始查询融合

同时用原始问题和假设性文档检索，融合结果：

from langchain.retrievers import EnsembleRetriever

def hybrid_hyde_search(question, vectorstore, bm25_retriever, k=10):
    """HyDE 向量检索 + BM25 关键词检索的混合方案"""
    # 生成假设性文档
    hypothetical_doc = hyde_chain.invoke({"question": question})

    # 创建两个向量检索器：一个用原始问题，一个用假设性文档
    # 通过 EnsembleRetriever 加权融合
    # 注意：这里需要在 Milvus 层面做两次搜索再合并

    # 方案一：用假设性文档检索（偏语义）
    hyde_docs = vectorstore.similarity_search(hypothetical_doc, k=k)

    # 方案二：用原始问题检索（偏精确）
    original_docs = vectorstore.similarity_search(question, k=k)

    # 合并去重
    seen = set()
    merged_docs = []
    for doc in hyde_docs + original_docs:
        if hash(doc.page_content) not in seen:
            seen.add(hash(doc.page_content))
            merged_docs.append(doc)

    return merged_docs[:k]

6.5 查询侧优化策略对比

策略	核心思想	适用场景	额外开销	效果提升
查询重写	一个问题 → 一个更精准的问题	口语化、模糊查询	1 次 LLM 调用	⭐⭐⭐
查询分解	一个问题 → 多个子问题	多方面、对比类复杂问题	N 次 LLM 调用 + N 次检索	⭐⭐⭐⭐
查询澄清	判断是否需要追问	信息不足的模糊问题	1 次 LLM 调用	⭐⭐⭐
HyDE	生成假设性文档再检索	短查询 vs 长文档的鸿沟	1 次 LLM 调用	⭐⭐⭐⭐
多查询检索	一个问题 → 多个改写版本	提升召回覆盖率	N 次检索	⭐⭐⭐

组合建议：在实际生产中，这些策略不是互斥的。推荐组合：查询澄清（入口过滤）→ 查询重写（标准化）→ HyDE 或多查询（扩展检索）→ 精排（兜底排序），形成完整的查询优化链路。

第七章：Reranker 精排模型接入(重点)

7.1 两阶段检索架构

RAG 系统的检索通常分两步走：

粗排（召回阶段）：用向量检索 / BM25 快速从海量文档中召回 Top-20 候选集。速度快，但精度有限。
精排（重排阶段）：用 Reranker 模型对这 20 条候选逐一计算精细相关性分数，重新排序后选出 Top-3。

粗排像是"海选"，精排像是"面试"——海选看个大概，面试才看真本事。

Reranker 和 Embedding 模型的本质区别：

Embedding 模型：把文本压缩成一个向量，用余弦相似度比较。信息有损，速度快。
Reranker 模型：直接把"问题 + 文档"一起输入模型，输出一个相关性分数。信息无损，速度慢，但精度高。

7.2 排序的关键维度

Reranker 不只是看语义相关性，还会综合考虑：

维度	说明	示例
语义相关性	文档是否真的回答了问题	问"饮食禁忌"→ 回答"胰岛素注射"的文档排后面
时效性	越新的文档通常越可靠	2024 年指南 > 2015 年资料
多样性	避免返回内容高度重复的文档	3 篇都讲"糖分控制"不如 2 篇"糖分控制" + 1 篇"运动管理"

7.3 DashScope Rerank 实战

阿里通义提供了开箱即用的 Rerank 服务，接入非常简单：

import dashscope
from http import HTTPStatus

docs = list()
for result in multi_results:
    docs.append(result.page_content)


resp = dashscope.TextReRank.call(
    model="qwen3-rerank",# 通义的重排模型
    api_key="your_api_key_here",
    query="在多少速度范围内，HDC才会激活？",
    documents=docs,
    top_n=2, #精排后只保留 Top-2
    return_documents=True,
    )
reRankdoc=list()
if resp.status_code == HTTPStatus.OK and len(resp['output']['results'])>0:
    for reRankoutput in resp['output']['results']:
        reRankdoc.append(reRankoutput['document']['text'])
        print(f"相关性分数: {reRankoutput['relevance_score']:.4f}")
        print(f"内容: {reRankoutput['document']['text']}")
        print("---")
else:
    print(resp)

7.4 精排后送入大模型生成答案

精排完成后，把最相关的文档送入大模型生成最终答案：

from langchain_core.prompts import ChatPromptTemplate

# 构建 Prompt 模板
prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识库助手。请根据以下检索到的上下文回答用户问题。

**规则：**
- 只基于提供的上下文回答，不要编造
- 如果上下文中没有相关信息，直接说「根据现有资料，我找不到这个问题的答案」
- 回答要简洁直接，引用原文时用引号

**检索到的上下文：**
{context}

**用户问题：**
{question}
""")

from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser

# 初始化大模型（这里用通义千问，也可以换成 DeepSeek）
llm = ChatTongyi(
    model="qwen-plus",                 # 模型名称
    dashscope_api_key="your_api_key_here"   # 替换为你的真实 Key
)

# 拼装上下文
context_text = "\n\n---\n\n".join([doc for doc in reRankdoc])

# 构建 Chain 并调用
chain = prompt | llm | StrOutputParser()

response = chain.invoke({
    "context": context_text,
    "question": query
})

print(response)

第八章：RAG 效果评估体系

10.1 为什么需要评估

RAG 系统调优最大的痛点是：你不知道改了某个参数后，效果到底变好了还是变差了。没有量化评估，调优就是盲人摸象。

评估方法：基于 LLM 的评估（RAGAS）：用大模型来判断检索和生成的质量

10.2 RAGAS 四大核心指标

RAGAS 是目前最主流的 RAG 评估框架，提供四个关键指标：

指标一：Faithfulness（忠实度）⭐⭐⭐⭐⭐

作用：防止模型幻觉，确保答案有据可依。

原理：将生成的答案拆成多个独立陈述，检查每个陈述是否能从检索的上下文中找到支撑。

公式：Faithfulness = 被支撑的陈述数 / 总陈述数

示例：
  问题："Python 的创始人是谁？"
  上下文："Python 由 Guido van Rossum 在 1991 年创建"
  答案："Python 的创始人是 Guido van Rossum，创建于 1991 年"
  → 两个陈述都有支撑 → Faithfulness = 1.0 ✓

指标二：Answer Relevancy（答案相关性）⭐⭐⭐⭐⭐

作用：确保答案直接回答问题，不跑题。

原理：反向操作——让 LLM 根据答案反推可能的问题，然后计算反推问题和原始问题的语义相似度。相似度越高，说明答案越切题。

示例：
  原始问题："如何安装 Python？"
  生成答案："访问 python.org 下载安装包..."
  反推问题："怎样安装 Python？" → 高相似度 → Answer Relevancy 高 ✓

指标三：Context Precision（上下文精确度）⭐⭐⭐⭐

作用：评估检索结果的排序质量——相关的文档是否排在前面。

原理：类似信息检索中的 Precision@K，检查每个位置的上下文是否与问题相关。

示例：
  检索到 5 个文档片段：
  位置1: 相关 ✓
  位置2: 相关 ✓
  位置3: 不相关 ✗
  位置4: 相关 ✓
  位置5: 不相关 ✗
  → 前面相关度高 → Context Precision 高

指标四：Context Recall（上下文召回率）⭐⭐⭐⭐

作用：评估检索系统是否找全了所需信息。

原理：需要标准答案（ground truth），检查标准答案中的信息能否从检索的上下文中推导出来。

公式：Context Recall = 可归因的陈述数 / 标准答案总陈述数

示例：
  标准答案："北京是中国首都，人口 2100 万，面积 16410 平方公里"
  检索上下文包含：✓ 北京是首都  ✓ 人口 2100万  ✗ 面积信息缺失
  → Context Recall = 2/3 ≈ 0.67

10.3 RAGAS 评估实战

#安装依赖: pip install ragas -i https://pypi.tuna.tsinghua.edu.cn/simple
from ragas import evaluate, RunConfig
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models import ChatTongyi

# 初始化模型
embedding_model = DashScopeEmbeddings(
    model="text-embedding-v3",
    dashscope_api_key="your_api_key_here"
)
llm = ChatTongyi(model="qwen-plus", api_key="your_api_key_here")

# ---- 准备测试数据 ----
test_questions = [
    {
        "question": "车顶行李架最大载荷是多少？",
        "ground_truth": "车顶行李架最大载荷为70kg"
    },
    {
        "question": "自适应巡航系统的工作速度范围是多少？",
        "ground_truth": "自适应巡航系统（ACC）可以在0-150km/h范围内工作"
    },
    {
        "question": "车辆推荐的轮胎胎压是多少（空载前轮）？",
        "ground_truth": "空载时前轮推荐胎压为230 kPa"
    },
    {
        "question": "保养周期是多少公里或多长时间？",
        "ground_truth": "保养周期为15,000公里或12个月（以先到者为准）"
    }
]

# ---- 构建评估数据集 ----
evaluation_data = {
    "question": [],
    "answer": [],
    "contexts": [],
    "ground_truth": []
}

print("开始生成评估数据...")
for i, item in enumerate(test_questions, 1):
    question = item["question"]
    ground_truth = item["ground_truth"]
    
    print(f"处理第 {i}/{len(test_questions)} 个问题: {question}")

    # 1. 检索相关文档
    retrieved_docs = rag_system.search(question, "multi_query")
    contexts = [doc.page_content for doc in retrieved_docs]

    # 2. 用大模型生成答案
    context_text = "\n\n".join(contexts)
    prompt = f"""基于以下上下文回答问题。如果没有相关信息，请说"无法从提供的信息中回答"。

    上下文：{context_text}

    问题：{question}
    答案："""
    answer = llm.invoke(prompt).content

    # 3. 保存评估数据
    evaluation_data["question"].append(question)
    evaluation_data["answer"].append(answer)
    evaluation_data["contexts"].append(contexts)
    evaluation_data["ground_truth"].append(ground_truth)

# ---- 执行评估 ----
dataset = Dataset.from_dict(evaluation_data)

# 配置评估运行参数
run_config = RunConfig(
    timeout=600,       # 单次操作超时时间（秒）
    max_workers=1,     # 并发数，本地模型建议设为1
    max_retries=10,    # 最大重试次数
    max_wait=120,      # 重试间隔上限（秒）
)

# 定义要使用的评估指标
metrics = [
    faithfulness,        # 忠实度：答案是否基于上下文
    answer_relevancy,    # 答案相关性：答案是否与问题相关
    context_precision,   # 上下文精确度：检索到的上下文是否相关
    context_recall       # 上下文召回率：是否检索到所有必要信息
]

print("\n开始评估...")
result = evaluate(
    dataset,
    metrics=metrics,
    embeddings=embedding_model,
    llm=llm,
    run_config=run_config
)

# ---- 输出评估结果 ----
print("评估结果：",result)

10.4 池化标注法：降低标注成本

评估需要人工标注"哪些文档是相关的"——如果文档库有 1000 篇，逐篇标注成本太高。

池化标注法的思路：先用多个检索器分别召回 Top-K，取并集形成"候选池"，只标注候选池中的文档。

# 各检索器的召回结果
bm25_results = ["doc1", "doc3", "doc5", "doc7", "doc9"]
vector_results = ["doc1", "doc2", "doc4", "doc6", "doc8"]
hybrid_results = ["doc1", "doc2", "doc3", "doc10", "doc11"]

# 创建候选池（并集）
candidate_pool = set(bm25_results + vector_results + hybrid_results)
print(f"候选池大小: {len(candidate_pool)}")  # 比如 11 个文档

# 只需要标注这 11 个文档，而不是全部 1000 个
# 标注完成后计算召回率
def calculate_recall(retrieved, relevant):
    """计算召回率"""
    retrieved_set = set(retrieved)
    relevant_set = set(relevant)
    true_positives = len(retrieved_set & relevant_set)
    return true_positives / len(relevant_set) if relevant_set else 0

# 示例：假设标注后确认 doc1, doc2, doc4 是相关的
relevant_docs = ["doc1", "doc2", "doc4"]
retrieved = ["doc1", "doc3", "doc5"]  # 某个系统的检索结果

recall = calculate_recall(retrieved, relevant_docs)
print(f"Recall: {recall:.2f}")  # 1/3 ≈ 0.33

4.RAG全链路优化：多路召回、查询测优化、Reranker.md 64 KB Permalink Riwayat Mentahan

RAG 全链路优化

第一章：RAG常见问题

第二章：Milvus 向量数据库实战

2.1 为什么选 Milvus

2.2 安装与连接

2.3 创建集合（Collection）

2.4 插入数据

2.5 向量检索

2.6 与 LangChain 集成

2.7 检索工具与 Agent 集成

把检索器变成工具

第三章：Embedding 模型选型

3.1 为什么 Embedding 选型很重要

3.2 主流 Embedding 模型对比

MTEB 基准

主流中文模型

主流多语言/API 模型

3.3 关键选择维度

维度一：向量维度（Dimension）

维度二：最大文本长度

维度三：Matryoshka 自适应维度

维度四：API 服务 vs 本地部署

3.4 如何在自己的数据上评估

3.5 模型推荐速查

第四章：文档智能切片策略

4.1 固定切片的局限

4.2 语义切块（SemanticChunker）

4.3 百分位数法原理详解

通俗解释：从"差生划线"说起

什么是百分位数

算法步骤

LangChain 的优化：局部最小值策略

参数调优

第五章：多路召回与混合检索(重点)

5.1 为什么需要多路召回

5.2 BM25 原理：图书馆管理员的打分规则

5.3 混合检索（EnsembleRetriever）

5.4 多查询检索（MultiQueryRetriever）

5.5 完整的多路召回系统

第六章：查询侧优化——让"问题"本身变好(重点)

6.1 查询重写（Query Rewriting）

问题场景

原理

实现

在 RAG 链路中集成

6.2 查询分解（Query Decomposition）

问题场景

原理

实现

分解后并行检索与合并

6.3 查询澄清（Query Clarification）

问题场景

原理

实现

在对话流中集成

6.4 HyDE 查询扩展（Hypothetical Document Embeddings）

问题场景

原理

实现

HyDE 的局限性与应对

进阶：HyDE + 原始查询融合

6.5 查询侧优化策略对比

第七章：Reranker 精排模型接入(重点)

7.1 两阶段检索架构

7.2 排序的关键维度

7.3 DashScope Rerank 实战

7.4 精排后送入大模型生成答案

第八章：RAG 效果评估体系

10.1 为什么需要评估

10.2 RAGAS 四大核心指标

指标一：Faithfulness（忠实度）⭐⭐⭐⭐⭐

指标二：Answer Relevancy（答案相关性）⭐⭐⭐⭐⭐

指标三：Context Precision（上下文精确度）⭐⭐⭐⭐

指标四：Context Recall（上下文召回率）⭐⭐⭐⭐

10.3 RAGAS 评估实战

10.4 池化标注法：降低标注成本

附录：完整 RAG 链路总览

4.RAG全链路优化：多路召回、查询测优化、Reranker.md 64 KB

Permalink Riwayat Mentahan