# RAG 全链路优化
> 从"能用"到"好用"。
>
---
## 第一章：RAG常见问题
做过 RAG 的同学大概都踩过这三个坑：
| 病灶 | 症状 | 根因 |
| :--- | :--- | :--- |
| **切片粗暴** | 检索回来的内容支离破碎，上下文断裂 | 按固定字符数硬切，把一句话切成两半 |
| **检索不精准** | 召回的内容跟问题关系不大 | 只用向量相似度，忽略了关键词匹配 |
| **没有大局观** | 找到了局部细节，但遗漏了全局信息 | 单次检索只能看到"一小块"，缺乏多视角探索 |
这些问题不是孤立的，它们会层层叠加：切片质量差 → 检索命中率低 → 大模型拿不到有效上下文 → 生成的答案牛头不对马嘴。
---
## 第二章：Milvus 向量数据库实战
### 2.1 为什么选 Milvus
Milvus 是目前国内生态最成熟的开源向量数据库之一，支持亿级向量检索，社区活跃，文档齐全。对比 FAISS（纯内存、无持久化）和 Chroma（轻量但功能有限），Milvus 在生产环境中更有优势。
> 官方文档：[https://milvus.io/docs/zh](https://milvus.io/docs/zh)
>
### 2.2 安装与连接
[docker-compose.yml](https://www.yuque.com/attachments/yuque/0/2026/yml/21571931/1782813228889-e4fb1613-1e0d-4798-a463-0bc67019df08.yml)
```python
使用docker安装
docker compose up -d
```
```bash
# 安装 Python SDK
uv pip install pymilvus -i https://mirrors.aliyun.com/pypi/simple
```
```python
from pymilvus import MilvusClient, DataType
# 连接 Milvus 服务（默认端口 19530）
client = MilvusClient(uri="http://localhost:19530")
```
### 2.3 创建集合（Collection）
Milvus 的"集合"类似于关系型数据库的"表"。我们需要定义字段结构，然后创建集合和索引。
```python
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。
```python
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）搜索。
```python
# 加载集合到内存（检索前必须执行）
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 的封装来简化流程。
```python
# 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)** 的列表及详细信息。
检索就更简单了：
```python
# 创建检索器，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 自主决定何时调用。
```python
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 模型的核心任务：**把文本映射到高维向量空间，使得语义相近的文本在向量空间中距离也近**。
```plain
文本: "Milvus 是一个高性能向量数据库"
↓ Embedding 模型
向量: [0.23, -0.15, 0.87, 0.04, ..., -0.42] (如 1024 维)
```
选择 Embedding 模型时，需要同时考虑**效果、速度、成本、部署方式**四个维度。
### 3.2 主流 Embedding 模型对比
#### MTEB 基准
[MTEB](https://huggingface.co/spaces/mteb/leaderboard)（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 表示学习。核心思想：**高维向量天然包含低维信息**。
```plain
维度 1024 的向量
↓ 取前 512 维（丢弃后 512 维）
维度 512 的向量（精度损失 < 2%）
↓ 取前 256 维
维度 256 的向量（精度损失 < 5%）
```
这意味着：**一套 Embedding，多套索引**。
+ 粗排用 256 维，速度快
+ 精排用 1024 维，精度高
+ 存储 1024 维，检索时按需截断
```python
# 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 排名不能代替你的真实场景。一个轻量的评估流程：
```python
# 评估框架示意（简化版）
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. 在语义发生"突变"的地方切开——相似度低说明话题转了
```python
#安装依赖：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)%的数据大于或等于这个数值。**百分位数描述的是"数据分布中的相对位置"，用考试分数来举例最直观：
```plain
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：计算相邻句子的余弦相似度**
```plain
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 去比对原始的相邻相似度：
```plain
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. **精确定位**：在每个候选断点附近（前后若干句子），找到相似度的**局部最小值**（真正的"谷底"）作为最终切分位置
```plain
相似度曲线示意：
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 中混合检索的核心组件，它做的事情很简单：**并行调用多个检索器，然后加权融合结果**，解决单一检索器的局限性，提升检索的精准度、召回率和鲁棒性。
```python
#安装依赖: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` 的解决思路很巧妙：**让大模型把一个问题改写成多个不同角度的问题，分别检索，合并去重**。
相当于从多个"视角"去探索知识库，大大增加命中率。
```python
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 完整的多路召回系统
把上面的组件封装成一个可切换的检索系统：
```python
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)
```
使用示例：
```python
# 初始化系统
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**：一个问题 → 一个优化版本，重点是**精准提升**
两者可以组合使用：先重写优化，再多角度扩展。
#### 实现
```python
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 链路中集成
将查询重写作为检索前的预处理步骤：
```python
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 将复杂问题拆解为多个简单的子问题，分别检索后合并结果**。
这比直接检索复杂问题效果更好，因为：
+ 子问题更聚焦，检索精度更高
+ 每个子问题可以从不同文档中获取信息
+ 最终合并的答案更全面
#### 实现
```python
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 的价格分别是多少？
```
#### 分解后并行检索与合并
```python
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 判断问题是否信息充分，不充分则生成澄清反问**。
这是一种"不急于回答，先搞清楚再行动"的策略，能显著提升用户体验——与其给一个答非所问的答案，不如先问清楚。
#### 实现
```python
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', '无')}")
```
#### 在对话流中集成
```python
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）的思路非常巧妙：
```plain
传统流程: 用户问题 → Embedding → 向量检索
HyDE 流程: 用户问题 → LLM 生成假设性回答 → Embedding → 向量检索
```
核心洞察：**一个"假设性回答"的 Embedding，比原始问题的 Embedding 更接近真实文档的向量**。
为什么？因为假设性回答在形式上（长度、用词、句式）更接近知识库中的真实文档，向量空间中的距离自然更近。
#### 实现
```python
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 + 原始查询融合
同时用原始问题和假设性文档检索，融合结果：
```python
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 服务，接入非常简单：
```python
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 精排后送入大模型生成答案
精排完成后，把最相关的文档送入大模型生成最终答案：
```python
from langchain_core.prompts import ChatPromptTemplate
# 构建 Prompt 模板
prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识库助手。请根据以下检索到的上下文回答用户问题。
**规则：**
- 只基于提供的上下文回答，不要编造
- 如果上下文中没有相关信息，直接说「根据现有资料，我找不到这个问题的答案」
- 回答要简洁直接，引用原文时用引号
**检索到的上下文：**
{context}
**用户问题：**
{question}
""")
```
```python
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（忠实度）⭐⭐⭐⭐⭐
**作用**：防止模型幻觉，确保答案有据可依。
**原理**：将生成的答案拆成多个独立陈述，检查每个陈述是否能从检索的上下文中找到支撑。
```plain
公式：Faithfulness = 被支撑的陈述数 / 总陈述数
示例：
问题："Python 的创始人是谁？"
上下文："Python 由 Guido van Rossum 在 1991 年创建"
答案："Python 的创始人是 Guido van Rossum，创建于 1991 年"
→ 两个陈述都有支撑 → Faithfulness = 1.0 ✓
```
#### 指标二：Answer Relevancy（答案相关性）⭐⭐⭐⭐⭐
**作用**：确保答案直接回答问题，不跑题。
**原理**：反向操作——让 LLM 根据答案反推可能的问题，然后计算反推问题和原始问题的语义相似度。相似度越高，说明答案越切题。
```plain
示例：
原始问题："如何安装 Python？"
生成答案："访问 python.org 下载安装包..."
反推问题："怎样安装 Python？" → 高相似度 → Answer Relevancy 高 ✓
```
#### 指标三：Context Precision（上下文精确度）⭐⭐⭐⭐
**作用**：评估检索结果的排序质量——相关的文档是否排在前面。
**原理**：类似信息检索中的 Precision@K，检查每个位置的上下文是否与问题相关。
```plain
示例：
检索到 5 个文档片段：
位置1: 相关 ✓
位置2: 相关 ✓
位置3: 不相关 ✗
位置4: 相关 ✓
位置5: 不相关 ✗
→ 前面相关度高 → Context Precision 高
```
#### 指标四：Context Recall（上下文召回率）⭐⭐⭐⭐
**作用**：评估检索系统是否找全了所需信息。
**原理**：需要标准答案（ground truth），检查标准答案中的信息能否从检索的上下文中推导出来。
```plain
公式：Context Recall = 可归因的陈述数 / 标准答案总陈述数
示例：
标准答案："北京是中国首都，人口 2100 万，面积 16410 平方公里"
检索上下文包含：✓ 北京是首都 ✓ 人口 2100万 ✗ 面积信息缺失
→ Context Recall = 2/3 ≈ 0.67
```
### 10.3 RAGAS 评估实战
```python
#安装依赖: 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，取并集形成"候选池"，只标注候选池中的文档。
```python
# 各检索器的召回结果
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 链路总览
