lqq-agent-study/02_RAG_study/4.RAG全链路优化：多路召回、查询测优化、Reranker.md

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 的思路完全不同：先按句子拆分，再按语义合并。

核心原理：

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

#安装依赖：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. 粗筛：用百分位数阈值找出所有"可疑"的候选断点（相似度低于阈值的位置）
2. 精确定位：在每个候选断点附近（前后若干句子），找到相似度的局部最小值（真正的"谷底"）作为最终切分位置

相似度曲线示意：

  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 系统的检索通常分两步走：

1. 粗排（召回阶段）：用向量检索 / BM25 快速从海量文档中召回 Top-20 候选集。速度快，但精度有限。
2. 精排（重排阶段）：用 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

---

附录：完整 RAG 链路总览