一. 多检索查询
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)
View Code
!
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。