当前位置：首页 > news >正文

第十一节：多检索查询、混合检索(多检索+RRF重排)、检索后优化(文档压缩)

news 2026/7/3 1:08:20

一. 多检索查询

1.功能

实现了基于多查询自动生成的预索引高级RAG智能问答。利用大模型对用户原始问题进行语义扩写，批量生成多条同义、相近问句，多次检索、合并去重，解决用户提问表达方式单一、和文档措辞不匹配导致漏检、少检的问题，提升文档召回率，可根据网页原文精准回答问题。

【一个问题搜不准 → 让大模型生成 3~5 个同义不同问法的问题 → 一起检索 → 合并去重 → 拿到最全上下文！】

2. 特点

无需人工构造问句、自动扩写问题、检索召回率高、自动合并去重、对比普通检索优化明显、代码简洁、内置日志可查看生成问句、适配口语化提问。

3. 核心流程【重点】

(一)、存储阶段（入库）

① 加载网页原始数据，对网页文本进行均等切分，生成文档碎片。

② 调用Embedding模型，对切分后的原文块进行向量化。

③ vectorstore（Chroma）：存储原文明文 + 原文向量，无额外存储结构。

(二)、问答阶段（检索）

① 用户输入原始问题，送入MultiQueryRetriever。

② LLM自动生成3条左右同义相似问句（扩写、改写、换种问法）。

③ 多条问句分别向量化，并行检索向量库。

④ 合并所有检索结果，自动去重，剔除重复文档。

⑤ 将去重后的最全上下文送入大模型，生成最终回答。

4. 重点补充

4.1 存储规则

本代码为普通原文入库：

page_content：明文+向量双存；无额外metadata存储、无人工绑定ID。

4.2 内部机制

MultiQueryRetriever：内部封装、全自动。无需手动编写生成问句逻辑，内部自带通用提示词生成多条相似问题。

4.3 LLM调用次数

- 优化前普通RAG：1次（仅生成回答）。

- 优化后多查询RAG：2次。

- ① 第一次：生成多条相似查询问句；② 第二次：大模型生成最终答案。

4.4 和假设问题RAG区别（必考）

- 多查询检索(MultiQuery)：问答阶段生成问题，临时生成、用完即丢、不存向量库。

- 假设问题(Hypothetical)：入库阶段生成问题，永久存入向量库、绑定原文ID。

5. 代码分享

# ===================== 1. 导入依赖包 =====================
from model_config import llm, get_local_embeddings
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableMap
from langchain_classic.retrievers import MultiQueryRetriever
import logging# ===================== 2. 全局配置 =====================
URL = "https://news.pku.edu.cn/mtbdnew/15ac0b3e79244efa88b03a570cbcbcaa.htm"
# 分块大小
CHUNK_SIZE = 600
CHUNK_OVERLAP = 100
# 开启日志（查看自动生成的问题）
logging.basicConfig(level=logging.INFO)# ===================== 3. 加载网页 + 文档分块 =====================
# 加载网页内容
loader = WebBaseLoader(URL)
raw_docs = loader.load()
# 文档切分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP)
split_docs = text_splitter.split_documents(raw_docs)# ===================== 4. 向量库 + 基础检索器 =====================
embedding_model = get_local_embeddings()
vectorstore = Chroma.from_documents(documents=split_docs, embedding=embedding_model)
base_retriever = vectorstore.as_retriever()# ===================== 5. 提示词模板 =====================
prompt = ChatPromptTemplate.from_template("""请根据下面给出的上下文来回答问题:{context}问题: {question}"""
)# ===================== 6. 优化前：普通检索 RAG =====================
print("-------------- 优化前（普通检索）-------------------")
normal_chain = (RunnableMap({"context": lambda x: base_retriever.invoke(x["question"]), "question": lambda x: x["question"]})| prompt| llm| StrOutputParser()
)
print(normal_chain.invoke({"question": "天才AI少女是谁"}))# ===================== 7. 优化后：多查询检索 =====================
print("\n-------------- 多查询检索（自动生成问题-------------------")
# 初始化多查询检索器：自动生成多条相似问题
# 作用：自动生成多个相似问题 → 并行检索 → 结果去重
# 自动生成：['天才AI少女是谁  ', '谁被称为天才AI少女  ', '天才AI少女指的是哪位人物']
multi_query_retriever = MultiQueryRetriever.from_llm(retriever=base_retriever, llm=llm)
# 测试检索结果
unique_docs = multi_query_retriever.invoke("天才AI少女是谁")
print("去重后文档数量：", len(unique_docs))# ===================== 8. 优化后问答链 =====================
print("\n-------------- 优化后（多查询检索）-------------------")
optimized_chain = (RunnableMap({"context": lambda x: multi_query_retriever.invoke(x["question"]), "question": lambda x: x["question"]})| prompt| llm| StrOutputParser()
)
print(optimized_chain.invoke({"question": "天才AI少女是谁"}))

View Code

二. 混合检索(多检索+RRF重排)

1. 功能

实现了多查询生成 + RRF 算法重排的高级 RAG 智能问答。自动生成多个相似问题，分别检索后通过 RRF 算法融合所有检索结果，解决单一问题表述偏差、检索不全、检索不准的问题，得到最精准、最全面的检索结果。

2. 核心原理

一个问题搜不全 → 本地固定提示词生成多个同义问题 → 多个问题分别执行向量检索 → 汇总全部检索结果 →用 RRF 算法按出现次数 + 排名统一打分重排 → 筛选出最优文档上下文 → 将最终上下文连同用户问题一起传入 LLM → 大模型生成最终回答

RRF 算法融合重排（核心）：

(1) 遍历所有结果：

(2) 文档出现次数越多 → 分越高

(3) 文档排名越靠前 → 分越高

(4) 最后按总分从高到低重新排序

3. 重点分析

LLM调用次数：共2次

① 生成多个查询：1 次

② 生成最终回答：1 次

4. 代码分享

# ===================== 导入依赖包 =====================
from langchain_chroma import Chroma
from model_config import llm, get_local_embeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.load import dumps, loads
from langchain_core.prompts import PromptTemplate# ===================== 1. 准备测试文本数据 =====================
texts = ["人工智能在医疗诊断中的应用。","人工智能如何提升供应链效率。","NBA季后赛最新赛况分析。","传统法式烘焙的五大技巧。","红楼梦人物关系图谱分析。","人工智能在金融风险管理中的应用。","人工智能如何影响未来就业市场。","人工智能在制造业的应用。","今天天气怎么样","人工智能伦理：公平性与透明度。",
]# ===================== 2. 初始化向量库 & 基础检索器 =====================
embeddings_model = get_local_embeddings()
vectorstore = Chroma.from_texts(texts=texts, embedding=embeddings_model)
base_retriever = vectorstore.as_retriever()# ===================== 3. 本地手写提示词（替代远程hub拉取） =====================
prompt = PromptTemplate(input_variables=["original_query"],template="""You are a helpful assistant that generates multiple search queries based on a single input query.Generate multiple related search queries to help retrieve relevant documents.Output each query on a new line, do not add extra explanations.        Original Query: {original_query}""",
)# 链结构：提示词 → LLM生成 → 文本解析 → 过滤空行切分问题
generate_queries = prompt | llm | StrOutputParser() | (lambda x: [q.strip() for q in x.split("\n") if q.strip()])# ===================== 4. RRF 互逆排序融合算法 =====================
def reciprocal_rank_fusion(results: list[list], k=60):"""RRF 重排核心：1. 文档出现次数越多 → 得分越高2. 文档排名越靠前 → 得分越高3. 合并所有结果，重新排序"""fused_scores = {}# 遍历每一组检索结果for docs in results:for rank, doc in enumerate(docs):doc_key = dumps(doc)  # 把文档转成唯一标识fused_scores[doc_key] = fused_scores.get(doc_key, 0) + 1 / (rank + k)# 按得分从高到低排序sorted_docs = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)return [(loads(doc_key), score) for doc_key, score in sorted_docs]# ===================== 5. 构建 RAG Fusion 完整检索链 =====================
# 流程：生成多问题 → 批量检索 → 合并重排
rag_fusion_chain = generate_queries | base_retriever.map() | reciprocal_rank_fusion# ===================== 6. 执行测试 =====================
if __name__ == "__main__":query = "人工智能的应用"# 1.测试生成的多个问题【调用LLM】multi_queries = generate_queries.invoke({"original_query": query})print("----- 生成的多个相似问题【调用LLM】 -----")for q in multi_queries:print(q)# 2.执行融合检索【纯本地检索】final_results = rag_fusion_chain.invoke({"original_query": query})# 输出最终结果print("\n----- RAG Fusion 最终重排结果【纯本地检索】 -----")for doc, score in final_results:print(f"【得分：{score:.4f}】 {doc.page_content}")# 3.基于重排结果回答【调用LLM】print("\n----- 基于重排结果回答【调用LLM】 -----")answer = llm.invoke(f"根据文档回答问题：{query}\n文档：{final_results}")print(answer.content)

View Code

三. 检索后优化(文档压缩)

1. 功能

基于检索后文档压缩优化 RAG 流程，在向量检索获取文档之后，通过多种压缩策略对文档进行二次处理，过滤无关文档、剔除重复内容、精简冗余语句，最终只把高价值文本送入大模型，降低 Token 消耗，提升问答精准度与响应速度。

总结：普通检索会返回一堆文档（有重复、有无关、内容长），压缩检索可以：

① 删掉无关的文档

② 删掉重复 / 冗余的文档

③ 只保留最关键、最精简的内容

④ 给 LLM 更少、更准的内容 → 省 token、回答更快更准

2. 核心流程

用户发起提问 → 向量库检索匹配文档 → 进入文档压缩流程筛选优化 → 输出纯净文本内容 → 传入LLM生成答案

3. 对比四种过滤器

(1) LLMChainExtractor 内容提取器

作用：在单篇文档里，摘取和用户问题强相关的核心语句，删减多余描述，缩短文本长度

是否调 LLM：需要

(2) LLMChainFilter 文档过滤器

作用：整体判断整篇文档是否和问题相关，直接剔除无关整篇文档

是否调 LLM：需要

(3) EmbeddingsFilter 相似度过滤器

作用：依靠嵌入模型计算问题与文档相似度，设置阈值自动筛选，纯本地运算，无大模型调用，速度快。

是否调 LLM：不需要，仅向量计算

(4) Pipeline 流水线（去冗余+去无关）

A: EmbeddingsRedundantFilter 冗余过滤器

作用：检索结果之间做比对，剔除内容重复、高度相似的冗余文档

是否调 LLM：不需要

B: DocumentCompressorPipeline 压缩流水线

作用：串联多个过滤组件，按顺序批量完成去重、筛选、精简

是否调 LLM：看内部组装组件，纯向量组合则不调用，含 LLM 组件则调用

4. 代码分享

from model_config import llm, get_local_embeddings
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_classic.retrievers import ContextualCompressionRetriever
from langchain_classic.retrievers.document_compressors import (LLMChainExtractor,  # LLM提取关键内容（精炼）EmbeddingsFilter,  # 向量相似度过滤（去无关）LLMChainFilter,  # LLM判断是否相关（去无关）DocumentCompressorPipeline,  # 压缩流水线（串联多个过滤器）
)
from langchain_community.document_transformers import EmbeddingsRedundantFilter  # 去冗余文档# ===================== 1. 全局配置 =====================
# 向量库本地存储路径
persist_directory = "./chroma_db"
# 加载向量嵌入模型
embeddings_model = get_local_embeddings()
# 全局问题
question = "deepseek的发展历程"# ===================== 2. 向量库构建（仅首次运行执行） =====================
# # 首次运行：加载文档 → 分块 → 入库
# documents = TextLoader("deepseek介绍.txt", encoding="utf-8").load()
# text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)  # 分块大小  # 重叠大小
# # 文档切分
# texts = text_splitter.split_documents(documents)
# # 创建并持久化向量库
# chroma = Chroma.from_documents(texts, embeddings_model, persist_directory=persist_directory)# ===================== 2. 加载向量库 （后续运行，需要注释掉上面2的代码） =====================
# 后续运行：直接加载本地向量库
chroma = Chroma(persist_directory=persist_directory, embedding_function=embeddings_model)# ===================== 3. 检索器 =====================
# 基础检索器（未压缩、未过滤）
retriever = chroma.as_retriever()# 统一打印方法：只输出纯文本内容，屏蔽向量、元数据等多余信息
def print_doc_content(doc_list):print("【文档纯文本内容】")for idx, doc in enumerate(doc_list, 1):print(f"{idx}. {doc.page_content.strip()}")print("-" * 80)# ===================== 4. 压缩前原始检索结果 =====================
print("=" * 60)
print("------------------- 【未压缩】原始检索结果 --------------------------")
# 1. 普通检索：直接返回所有匹配到的文档（包含冗余、无关内容）
docs = retriever.invoke(question)
print(f"未压缩文档数量：{len(docs)}")
print_doc_content(docs)# ===================== 5. 第一种压缩：LLM 提取关键内容（精炼） =====================
print("\n" + "=" * 60)
print("------------------- 【压缩1】LLMChainExtractor（LLM提取关键内容） --------------------------")
# 作用：LLM 从整篇文档中，只提取和问题相关的片段，删除无关内容（内容变短、变精）
compressor = LLMChainExtractor.from_llm(llm)
# 包装成压缩检索器
compression_retriever = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=retriever)
# 执行压缩检索
compressed_docs = compression_retriever.invoke(question)
print(f"压缩后文档数量：{len(compressed_docs)}")
print_doc_content(compressed_docs)# ===================== 6. 第二种压缩：LLM 过滤无关文档 =====================
print("\n" + "=" * 60)
print("------------------- 【压缩2】LLMChainFilter（LLM过滤无关文档） --------------------------")
# 作用：LLM 判断文档是否与问题相关，不相关直接丢弃，只保留相关文档（文档数量变少）
_filter = LLMChainFilter.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(base_compressor=_filter, base_retriever=retriever)
compressed_docs = compression_retriever.invoke(question)
print(f"过滤后文档数量：{len(compressed_docs)}")
print_doc_content(compressed_docs)# ===================== 7. 第三种压缩：向量相似度过滤 =====================
print("\n" + "=" * 60)
print("------------------- 【压缩3】EmbeddingsFilter（向量相似度过滤） --------------------------")
# 作用：计算问题与文档的向量相似度，低于阈值直接丢弃（纯向量计算，不调用LLM）
# similarity_threshold：相似度阈值（0~1），越高越严格
embeddings_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.71)
compression_retriever = ContextualCompressionRetriever(base_compressor=embeddings_filter, base_retriever=retriever)
compressed_docs = compression_retriever.invoke(question)
print(f"过滤后文档数量：{len(compressed_docs)}")
print_doc_content(compressed_docs)# ===================== 8. 第四种压缩：流水线压缩（最强组合） =====================
print("\n" + "=" * 60)
print("------------------- 【压缩4】Pipeline 流水线（去冗余+去无关） --------------------------")
# 1. 去冗余：删除内容相似、重复的文档
redundant_filter = EmbeddingsRedundantFilter(embeddings=embeddings_model)
# 2. 去无关：只保留高于相似度阈值的文档
relevant_filter = EmbeddingsFilter(embeddings=embeddings_model, similarity_threshold=0.66)
# 3. 流水线：先去冗余 → 再去无关（两步过滤）
pipeline_compressor = DocumentCompressorPipeline(transformers=[redundant_filter, relevant_filter])
# 最终压缩检索器
compression_retriever = ContextualCompressionRetriever(base_compressor=pipeline_compressor, base_retriever=retriever)
compressed_docs = compression_retriever.invoke(question)
print(f"流水线最终文档数量：{len(compressed_docs)}")
print_doc_content(compressed_docs)