RAG工程-基于LangChain 实现 Advanced RAG（预检索优化）

Advanced RAG 概述

        Advanced RAG 被誉为 RAG 的第二范式，它是在 Naive RAG 基础上发展起来的检索增强生成架构，旨在解决 Naive RAG 存在的一些问题，如召回率低、组装 prompt 时的冗余和重复以及灵活性不足等。它重点聚焦在检索增强，通过增加 Pre - Retrieval 预检索和 Post - Retrieval 后检索阶段，以及优化索引结构和原始查询来提高被索引内容的质量。

        在预检索处理优化方面，Advanced RAG 采用多种策略，如摘要索引、父子索引、假设性问题索引、元数据索引等。检索阶段会优化检索过程，主要方法包括混合检索等策略。检索后处理则使用复杂技术，如重排序、上下文压缩策略等，以确保生成的内容更准确、相关和简洁，更好地满足用户需求。

预检索优化

摘要索引

        在处理大量文档时，如何快速地找到所需信息是一个常见挑战。摘要索引是一种针对大量文档的结构化索引机制，核心目标是提升信息检索效率，解决传统检索中存在的信息定位慢、精准度低、查找繁琐等问题。其本质是通过对文档内容进行提炼、结构化处理和索引构建，形成一套高效的信息检索 “导航系统”，帮助用户快速、精准地定位和获取所需信息。

        在RAG 流程将文档分块后，我们可以使用大模型对每个分块内容进行总结，然后生成摘要索引。而后在检索时，我们可以根据问题找到摘要索引，再由摘要索引找到原文档；或者我们也可以直接回复用户摘要内容。

        接下来，我们用一段基于Langchain 框架的代码实现摘要索引的功能：

import os
from langchain.chat_models import ChatOpenAI
from langchain.document_loaders import TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate# 设置 OpenAI API 密钥
os.environ["OPENAI_API_KEY"] = "your_openai_api_key"def generate_summary_index(file_path):try:# 加载文档loader = TextLoader(file_path)documents = loader.load()# 文档分块text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)chunks = text_splitter.split_documents(documents)# 使用大模型生成摘要llm = ChatOpenAI(temperature=0)# 创建摘要生成链chain = ({"chunk": lambda x: x.page_content}| ChatPromptTemplate.from_template("请总结以下内容:\n\n{chunk}")| llm| StrOutputParser())# 批量生成文档摘要（最大并发数5）summaries = chain.batch(chunks, {"max_concurrency": 5})# 打印每个块的部分摘要内容for i, summary in enumerate(summaries):print(f"块 {i + 1} :   {repr(summary[:50])}...")# 创建摘要索引embeddings = OpenAIEmbeddings()summary_index = FAISS.from_texts(summaries, embeddings)return summary_index, chunks, summariesexcept Exception as e:print(f"生成摘要索引时出现错误: {e}")return None, None, Nonedef retrieve_info(summary_index, texts, summaries, question):if summary_index is None:return None, Nonetry:# 根据问题检索摘要索引docs = summary_index.similarity_search(question)if docs:summary = docs[0].page_content# 根据摘要找到原文档summary_index = summaries.index(summary)original_doc = texts[summary_index]return summary, original_doc.page_contentexcept ValueError:print("未在摘要列表中找到匹配的摘要。")except Exception as e:print(f"检索信息时出现错误: {e}")return None, Noneif __name__ == "__main__":file_path = "your_text_file.txt"summary_index, texts, summaries = generate_summary_index(file_path)question = "你想要查询的问题"summary, original_content = retrieve_info(summary_index, texts, summaries, question)if summary and original_content:print("摘要内容：", summary)print("原文档内容：", original_content)else:print("未找到相关信息。")

        这段代码使用了一个LangChain 框架的chain 链，封装了使用大模型对文档进行总结的流程。后续使用chain.batch()方法对文档进行总结摘要，并将最终结果保存到FAISS 向量数据库中，方便后续的检索查询使用。

父子索引

     在文档检索与处理的技术链中，文档块大小的设定始终是一个关键且充满矛盾的环节。一方面，为了确保 Embedding 过程能够精准锚定文档语义内核，我们需要将文档拆解为较小的单元。这是因为，当文档块被控制在合适的较小规模时，Embedding 模型可以聚焦于单一的语义主题或逻辑片段，生成的向量表征能够更纯粹、准确地映射文档的核心含义。

        但另一方面，当用户基于具体问题进行检索，期望获得详实答案时，较小的文档块却难以满足需求。因为 LLM 在生成答案时，需要依托丰富的上下文信息构建完整的逻辑链条。此时，较大的文档块凭借其承载的更多背景知识、完整的推理路径等内容，能为 LLM 提供更充分的 “弹药”，助力其输出全面且准确的回答。

        在应对这组看似不可调和的矛盾时，父子索引成为了破局的有效方案。父子索引通过构建层级化的文档关联体系，将文档信息进行结构化拆解与关联。它允许我们将大文档设定为父文档，将其拆解出的较小语义单元作为子文档，父子文档既保持各自独立存储，又通过特定的标识建立紧密联系。在检索环节，若侧重于 Embedding 的精准度，可优先检索子文档，利用其精确的语义向量快速定位相关信息；当用户需要完整的上下文来获取全面答案时，则可以通过父子索引的关联关系，迅速将相关的子文档聚合，形成内容丰富的大文档块提供给 LLM。如此一来，既保证了 Embedding 对文档语义的精准捕捉，又满足了 LLM 对完整上下文的需求，巧妙化解了文档检索中关于文档块大小的矛盾困境。

        接下来，我们用一段基于Langchain 框架的代码实现父子索引的功能：

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain_core.stores import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever# 定义文本文件路径，需要根据实际情况修改
TXT_DOCUMENT_PATH = "your_text_file.txt"# 初始化嵌入模型
embeddings_model = HuggingFaceEmbeddings()# 数据加载
# 初始化文档加载器
loader = TextLoader(TXT_DOCUMENT_PATH, encoding='utf-8')
# 加载文档
docs = loader.load()# 分割器准备
# 创建主文档分割器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1024)
# 创建子文档分割器
child_splitter = RecursiveCharacterTextSplitter(chunk_size=256)# 存储准备
# 存储小块
vectorstore = Chroma(collection_name="split_parents", embedding_function=embeddings_model
)
# 创建内存存储对象，存储大块
store = InMemoryStore()# 创建检索器
# 创建父文档检索器
retriever = ParentDocumentRetriever(vectorstore=vectorstore,docstore=store,child_splitter=child_splitter,parent_splitter=parent_splitter,search_kwargs={"k": 1}
)# 添加文档到检索器
retriever.add_documents(docs)# 搜索子块
print("------------搜索子块------------------------")
try:sub_docs = vectorstore.similarity_search("介绍下DeepSeek和市场占用情况")if sub_docs:print(sub_docs[0].page_content)else:print("未找到相关子块内容。")
except Exception as e:print(f"搜索子块时出现错误: {e}")# 搜索子块，返回关联大块
print("------------搜索子块，返回关联大块------------------------")
try:retrieved_docs = retriever.invoke("介绍下DeepSeek和市场占用情况")if retrieved_docs:print(retrieved_docs[0].page_content)else:print("未找到相关大块内容。")
except Exception as e:print(f"搜索关联大块时出现错误: {e}")

        在这段代码中，我们使用ParentDocumentRetriever创建了一个父文档检索器，并指定了向量存储对象、文档存储对象、子文档分割器、父文档分割器和搜索参数。这样在后续的retriever.invoke方法中，我们通过这个retriever就可以实现父子索引的效果，从细粒度的文档块中检索，然后返回并获取粗粒度的文档块。

假设性问题索引

        在传统RAG系统中，用户输入的查询往往是简短的问题，而待检索的文档块长度则相对较长，这种长度上的显著差异使得语义相似性计算面临挑战。当使用传统方法计算简短查询与长文档块之间的语义相似度时，由于长文档块包含大量冗余信息，极易导致语义匹配失准，从而使检索结果难以精准契合用户需求。

        为有效解决这一问题，我们可以引入大语言模型（LLM）进行辅助优化。具体而言，让 LLM 针对每个文档块，基于其内容生成 N 个假设性问题。这些假设性问题能够从不同角度提炼文档块的核心语义信息，将其转化为更贴近用户实际查询形式的文本表述。随后，对这些假设性问题进行向量化处理，将其嵌入向量空间构建索引。在系统运行过程中，当接收到用户的查询时，不再直接将其与原始长文档块进行匹配，而是与假设性问题的向量索引进行相似度查询。通过这种方式，能够找到与用户查询语义最为接近的假设性问题，并将该问题对应的原始文档块提取出来，作为上下文传递给 LLM。由于假设性问题已经对文档块的关键语义进行了精准提炼，LLM 基于此接收到的上下文，能够更准确地理解用户意图，从而生成高质量、高相关性的答案。

        接下来，上代码。

from typing import List
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from langchain.chat_models import ChatOpenAI  # 假设使用 OpenAI 模型，可按需替换
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import uuid
from langchain.vectorstores import Chroma
from langchain.retrievers import MultiVectorRetriever
from langchain_core.documents import Document
from langchain.storage import InMemoryByteStore# 定义文本文件路径，需要根据实际情况修改
TXT_DOCUMENT_PATH = "your_text_file.txt"# 初始化嵌入模型
from langchain.embeddings import HuggingFaceEmbeddings
embeddings_model = HuggingFaceEmbeddings()# 初始化大语言模型，这里使用 ChatOpenAI 作为示例，可根据需求替换
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)# 定义假设性问题模型
class HypotheticalQuestions(BaseModel):"""生成假设性问题"""questions: List[str] = Field(..., description="List of questions")# 创建提示模板
prompt = ChatPromptTemplate.from_template("""生成一个包含3个假设问题的列表，以下文档可用于回答这些问题:{doc}"""
)# 创建假设性问题链
chain = ({"doc": lambda x: x.page_content}| prompt# 将LLM输出构建为字符串列表| llm.with_structured_output(HypotheticalQuestions)# 提取问题列表| (lambda x: x.questions)
)# 数据加载
# 初始化文档加载器
loader = TextLoader(TXT_DOCUMENT_PATH, encoding='utf-8')
# 加载文档
docs = loader.load()# 分割文档（如果需要）
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(docs)# 针对文档1生成假设性问题
# chain.invoke(docs[0])# 批量处理所有文档生成假设性问题（最大并行数5）
hypothetical_questions = chain.batch(docs, {"max_concurrency": 5})
# print(hypothetical_questions)# 初始化Chroma向量数据库（存储生成的问题向量）
vectorstore = Chroma(collection_name="hypo-questions", embedding_function=embeddings_model
)
# 初始化内存存储（存储原始文档）
store = InMemoryByteStore()id_key = "doc_id"  # 文档标识键名# 配置多向量检索器
retriever = MultiVectorRetriever(vectorstore=vectorstore,byte_store=store,id_key=id_key,search_kwargs={"k": 1}
)# 为每个原始文档生成唯一ID
doc_ids = [str(uuid.uuid4()) for _ in docs]# 将生成的问题转换为带元数据的文档对象
question_docs = []
for i, question_list in enumerate(hypothetical_questions):question_docs.extend([Document(page_content=s, metadata={id_key: doc_ids[i]}) for s in question_list])# 将问题文档存入向量数据库
retriever.vectorstore.add_documents(question_docs)
# 将原始文档存入字节存储（通过ID关联）
retriever.docstore.mset(list(zip(doc_ids, docs)))# 进行检索
sub_docs = retriever.vectorstore.similarity_search("deepseek受到哪些攻击？")print("==========================检索到的相似问题==========================")
if sub_docs:print(sub_docs[0].page_content)
else:print("未检索到相似问题")print("==========================自动匹配问题文档块==========================")
retrieved_docs = retriever.invoke("deepseek受到哪些攻击？")
if retrieved_docs:print(retrieved_docs[0].page_content)
else:print("未检索到匹配的文档块")

        这段代码的重点在于HypotheticalQuestions 结合 langchain 链的使用，将文档拆分为多个假设性问题。而后我们使用了MultiVectorRetriever多向量检索器存储了两个库，一个是向量库，用来存假设性问题，一个是内存库，用来存问题对应的文档块。

元数据索引

        与传统依赖文本内容语义分析的检索方式不同，元数据索引利用预先定义的标签、属性、类别等结构化元数据信息，构建起一套轻量化、结构化的检索体系。

        例如，在企业知识库场景中，文档可能包含产品手册、技术规范、市场报告等多种类型，通过为每个文档添加 “文档类型”“创建时间”“所属部门”“关键词标签” 等元数据，系统能够快速定位特定条件下的文档集合。当用户想要检索 “过去一年由研发部门创建的技术规范” 时，元数据索引可直接基于 “创建时间”“所属部门”“文档类型” 等元数据进行快速筛选，在无需深入分析文本内容语义的情况下，即可高效返回目标数据集。这种检索方式不仅大幅提升了数据检索的效率，还能有效降低计算资源消耗，尤其适用于数据集规模庞大、数据类型多样且检索需求具有明确结构化特征的场景，为 RAG 系统快速锁定相关数据、提升检索响应速度提供了有力支持 。

        接下来，让我们写一段关于元数据索引检索的代码示例：

from langchain.schema import Document
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.retrievers.self_query.base import SelfQueryRetriever# 初始化嵌入模型
embeddings_model = HuggingFaceEmbeddings()# 初始化大语言模型，这里使用 ChatOpenAI 作为示例，可根据需求替换
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)# 定义文档列表
docs = [Document(page_content="小米智能手环6",metadata={"品牌": "小米",  "价格": 249, "评分": 4.6}),Document(page_content="华为FreeBuds Pro无线耳机",metadata={"品牌": "华为",  "价格": 999, "评分": 4.8}),Document(page_content="小米移动电源3",metadata={"品牌": "小米",  "价格": 99, "评分": 4.4}),Document(page_content="华为Mate 40 Pro智能手机",metadata={"品牌": "华为",  "价格": 6999, "评分": 5.0}),Document(page_content="小米AirDots Pro蓝牙耳机",metadata={"品牌": "小米",  "价格": 299, "评分": 4.5}),Document(page_content="华为智能手表GT 2",metadata={"品牌": "华为",  "价格": 1288, "评分": 4.7}),Document(page_content="小米小爱音箱Play",metadata={"品牌": "小米",  "价格": 169, "评分": 4.3})
]# 元数据字段定义（指导LLM如何解析查询条件）
metadata_field_info = [{"name": "品牌", "type": "string", "description": "产品的品牌名称"},{"name": "价格", "type": "integer", "description": "产品的价格"},{"name": "评分", "type": "float", "description": "产品的用户评分"},
]# 文档内容描述（指导LLM理解文档内容）
document_content_description = "电子产品的信息"# 创建向量存储
vectorstore = Chroma.from_documents(docs, embeddings_model, collection_name="self-query")# 创建自查询检索器（核心组件）
retriever = SelfQueryRetriever.from_llm(llm,vectorstore,document_content_description,metadata_field_info,
)# 示例查询
query = "查找价格低于 500 且评分高于 4.5 的小米产品"
results = retriever.get_relevant_documents(query)# 输出查询结果
print(f"查询: {query}")
for result in results:print(f"产品: {result.page_content}, 品牌: {result.metadata['品牌']}, 价格: {result.metadata['价格']}, 评分: {result.metadata['评分']}")

        这段代码的核心是通过 SelfQueryRetriever.from_llm 方法创建的自查询检索器 retriever。该方法接受大语言模型 llm、向量存储 vectorstore、文档内容描述 document_content_description 和元数据字段信息 metadata_field_info 作为参数，构建出一个能够根据用户输入的自然语言查询，自动解析查询条件并在向量存储中进行检索的检索器。

        为了辅助大模型进行元数据的处理转换，我们还需要一段比较标准的，可以将问题转换为元数据的提示词，像这样：

你的目标是按照以下提供的结构化请求模式，对用户的查询进行结构化处理。

请以 Markdown 代码片段的形式回复，其中包含符合以下格式的 JSON 对象：

 

{"query": string, // 用于与文档内容进行比较的文本字符串"filter": string // 用于筛选文档的逻辑条件语句
}

 

注意事项：

• 查询字符串：应仅包含期望与文档内容匹配的文本。筛选条件中的任何条件都不应在查询字符串中出现。 

• 逻辑条件语句：由一个或多个比较和逻辑运算语句组成。

• 比较语句格式：comp（attr，val）

  • comp（取值为 eq、ne、gt、gte、lt、lte、contain、like、in、nin）：分别表示等于、不等于、大于、大于等于、小于、小于等于、包含、模糊匹配、在、不在这些比较操作。

  • attr（字符串类型）：要应用比较操作的属性名称。

  • val（字符串类型）：比较时使用的值。

• 逻辑运算语句格式：op（statement1，statement2，…）

  • op（取值为 and、or、not）：分别表示逻辑与、逻辑或、逻辑非运算符。

  • statement1, statement2, 等（可以是比较语句或逻辑运算语句）：是一个或多个要应用逻辑运算的语句。

• 运算符使用限制：确保仅使用上述列出的比较器和逻辑运算符，不要使用其他运算符。

• 属性引用：确保筛选器中引用的属性是数据源中实际存在的属性。

• 函数使用规范：确保筛选条件仅使用函数名称及其对应的属性名称（当对属性应用函数时）。

• 日期格式：当筛选条件涉及处理日期类型值时，仅使用 “YYYY-MM-DD” 格式。

• 数据类型匹配：确保筛选条件根据属性的描述进行操作，仅进行与存储数据类型相符的可行比较。

• 筛选条件的使用：确保仅在有需要时使用筛选条件。如果不存在需要应用的筛选条件，请将 filter 的值设置为 "NO_FILTER"。

示例 1

数据来源：

{"content": "歌曲歌词","attributes": {"artist": {"type": "string","description": "歌曲艺术家的名称"},"length": {"type": "integer","description": "歌曲时长（秒）"},"genre": {"type": "string","description": "歌曲类型，可以是 \"pop\", \"rock\" 或 \"rap\""}}
}

 

用户查询：泰勒・斯威夫特或凯蒂・佩里创作的关于青少年爱情的时长少于 3 分钟的舞曲流行歌曲有哪些？

 

结构化请求：

 

{"query": "青少年爱情","filter": "and(or(eq(\"artist\", \"Taylor Swift\"), eq(\"artist\", \"Katy Perry\")), lt(\"length\", 180), eq(\"genre\", \"pop\"))"
}

示例 2

数据来源：

 

{"content": "歌曲歌词","attributes": {"artist": {"type": "string","description": "歌曲艺术家的名称"},"length": {"type": "integer","description": "歌曲时长（秒）"},"genre": {"type": "string","description": "歌曲类型，可以是 \"pop\", \"rock\" 或 \"rap\""}}
}

 

用户查询：
哪些歌曲没有在 Spotify 上发布

 

结构化请求：

 

{"query": "","filter": "NO_FILTER"
}

示例 3

数据来源：

{"content": "电子产品的信息","attributes": {"品名": {"type": "string","description": "产品的品牌名称"},"价格": {"type": "integer","description": "产品的价格"},"评分": {"type": "float","description": "产品的用户评分"}}
}

用户查询：
小米价格高于200元的耳机

结构化请求：

 对应测试的代码片段：

# 由大模型进行提取问题的提示词from langchain.chains.query_constructor.base import get_query_constructor_prompt# 构建查询解析器（调试用）
prompt = get_query_constructor_prompt(document_content_description,metadata_field_info,
)
print(prompt.format(query="小米价格高于200元的耳机"))

总结

        最后，我们对本次介绍的Advanced RAG 的预检索优化方法进行小结，本次提到了四种预检索优化的方法，包括摘要索引、父子索引、假设性问题索引、元数据索引。

索引类型	适用场景	案例
摘要索引	适用于需要快速检索和生成简洁上下文的场景。	在新闻资讯平台中，系统需要快速从海量新闻中提取关键信息，通过摘要索引可以迅速生成简洁的上下文，帮助用户快速了解新闻的核心内容。
父子索引	适用于需要确保语义完整性和层次化检索的场景。	在法律检索系统中，用户查询法律条款时，父子索引通过分层检索精准查找相关内容，并召回对应大文档块确保上下文的完整性，避免因分块过细导致语义丢失。
假设性问题索引	适用于需要处理复杂查询和多样化表达的场景。	在药品咨询系统中，用户查询症状时可能会问：“感冒了吃什么药？”。假设性问题索引通过为每种药品生成一系列假设性问题，帮助用户更准确地检索到相关信息。
元数据索引	适用于需要快速筛选和分类的场景。	在电商推荐系统中，系统通过元数据索引快速筛选出符合用户偏好的商品信息，提高推荐效率和准确性。