|
|
@@ -0,0 +1,1692 @@
|
|
|
+# RAG 全链路优化
|
|
|
+> 从"能用"到"好用"。
|
|
|
+>
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 第一章:RAG常见问题
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|
|
|
+做过 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)} 个文档切片")
|
|
|
+```
|
|
|
+
|
|
|
+<font style="color:rgb(15, 17, 21);">在浏览器访问 </font><font style="color:#DF2A3F;background-color:rgb(235, 238, 242);">http://{你的Milvus IP}:9091/webui</font><font style="color:rgb(15, 17, 21);"> 打开Milvus 内置的图形化界面查看数据库和</font>**<font style="color:rgb(15, 17, 21);">集合 (Collection)</font>**<font style="color:rgb(15, 17, 21);"> 的列表及详细信息。</font>
|
|
|
+
|
|
|
+检索就更简单了:
|
|
|
+
|
|
|
+```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% 的边界切出来。这样无论文档的整体相似度是高是低,都能动态、自适应地找到话题转变最剧烈的地方。
|
|
|
+
|
|
|
+#### 什么是百分位数
|
|
|
+**<font style="color:rgb(15, 17, 21);">在一组按从小到大排列的数据中,第p百分位数表示:有p%的数据小于或等于这个数值,同时有(100-p)%的数据大于或等于这个数值。</font>**百分位数描述的是"数据分布中的相对位置",用考试分数来举例最直观:
|
|
|
+
|
|
|
+```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." | 对精确关键词不敏感,(例如:**<font style="color:rgb(15, 17, 21);">搜</font>**<font style="color:rgb(15, 17, 21);">:</font>`特斯拉 Model Y 2024款 后轮驱动版 报价`<br/>**<font style="color:rgb(15, 17, 21);">向量检索召回</font>**<font style="color:rgb(15, 17, 21);">:Model 3报价、Model Y 2023款、蔚来ES6报价</font>) |
|
|
|
+| **关键词检索(BM25)** | 精确匹配能力强,速度快 | 无法理解同义词和语义相似性(例如:"感冒"和"风寒") |
|
|
|
+
|
|
|
+
|
|
|
+多路召回的核心思想:**两条腿走路,取长补短**。
|
|
|
+
|
|
|
+### 5.2 BM25 原理:图书馆管理员的打分规则
|
|
|
+BM25 是经典的关键词检索算法,理解它有助于调优。用一个图书馆的例子来解释:
|
|
|
+
|
|
|
+你在图书馆找一本"怎么做蛋糕"的书,管理员(搜索引擎)用三条规则打分:
|
|
|
+
|
|
|
+**规则一:关键词出现越多,越可能是好书**
|
|
|
+
|
|
|
+书 A 提了 1 次"蛋糕",书 B 提了 10 次 → 书 B 更相关。
|
|
|
+
|
|
|
+但不是越多越好!如果一本书写了 100 次"蛋糕蛋糕蛋糕"全是废话,BM25 会"封顶"——到一定程度就不加分了。这叫**词频饱和**。
|
|
|
+
|
|
|
+**规则二:短书里提到一次,比长书里提到一次更有价值**
|
|
|
+
|
|
|
+书 C 只有 5 页,认真讲了"戚风蛋糕做法";书 D 有 500 页,在第 387 页顺带提了一句"蛋糕也是甜点"。BM25 会惩罚又长又水的书,给短小精悍的书更高分。这叫**文档长度归一化**。
|
|
|
+
|
|
|
+**规则三:越少见的词,越重要**
|
|
|
+
|
|
|
+查询"低糖无麸质蛋糕"——"蛋糕"很常见,很多书都有,重要性一般;"无麸质"很少见,只有几本书提到,那这些书很可能就是你要的。这叫**逆文档频率(IDF)**:越稀有,越珍贵。
|
|
|
+
|
|
|
+> **一句话总结**:BM25 喜欢内容相关、简洁不啰嗦、用词精准的文档,讨厌又长又水的"注水文"。
|
|
|
+>
|
|
|
+
|
|
|
+### 5.3 混合检索(EnsembleRetriever)
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|
|
|
+`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` 的解决思路很巧妙:**让大模型把一个问题改写成多个不同角度的问题,分别检索,合并去重**。
|
|
|
+
|
|
|
+相当于从多个"视角"去探索知识库,大大增加命中率。
|
|
|
+
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|
|
|
+```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 完整的多路召回系统
|
|
|
+把上面的组件封装成一个可切换的检索系统:
|
|
|
+
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|
|
|
+```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。
|
|
|
+
|
|
|
+粗排像是"海选",精排像是"面试"——海选看个大概,面试才看真本事。
|
|
|
+
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|
|
|
+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 系统调优最大的痛点是:**你不知道改了某个参数后,效果到底变好了还是变差了**。没有量化评估,调优就是盲人摸象。
|
|
|
+
|
|
|
+<font style="color:#DF2A3F;">评估方法:</font>**<font style="color:#DF2A3F;">基于 LLM 的评估</font>**<font style="color:#DF2A3F;">(RAGAS):用大模型来判断检索和生成的质量</font>
|
|
|
+
|
|
|
+### 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 链路总览
|
|
|
+<!-- 这是一张图片,ocr 内容为: -->
|
|
|
+
|
|
|
+
|