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

Multi-Query 多路召回

多路召回流程图

        多路召回策略利用大语言模型（LLM）对原始查询进行拓展，生成多个与原始查询相关的问题，再将原始查询和生成的所有相关问题一同发送给检索系统进行检索。它适用于用户查询比较宽泛、模糊或者需要从多个角度获取信息的场景。当用户提出一个较为笼统的问题时，通过多路召回可以从不同维度去检索相关信息，以全面满足用户需求。

多路召回的实现流程

1. 利用 LLM 生成相关问题​：当用户输入原始查询时，LLM 会对查询进行深度语义解析，识别其中的关键词、主题和潜在需求。

2. 将所有问题发送给检索系统​：完成相关问题的生成后，原始查询与 LLM 生成的 N 个相关问题会一同被发送至检索系统。

3. 获取更多检索文档​：通过 Multi-Query 多路召回，检索系统能够从向量库中获取到更多与用户需求相关的文档。传统单一查询检索受限于用户输入的表述方式和详细程度，可能会遗漏一些关键信息。而 Multi-Query 多路召回利用多个问题进行多次检索，每个问题都能检索到一批相关文档，这些文档集合相互补充，涵盖了更广泛的内容和角度。​

基于LangChain 框架的多路召回代码实现

import logging
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.retrievers import MultiQueryRetriever
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI# 加载文档
def load_documents(file_path):try:loader = TextLoader(file_path, encoding='utf-8')return loader.load()except FileNotFoundError:print(f"错误：未找到文件 {file_path}")return []except Exception as e:print(f"加载文件时出现错误: {e}")return []# 分割文档
def split_documents(docs, chunk_size=600, chunk_overlap=100):text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)return text_splitter.split_documents(docs)# 创建向量数据库
def create_vectorstore(chunks, embedding_model):return Chroma.from_documents(documents=chunks, embedding=embedding_model, collection_name="multi-query")# 初始化检索器
def initialize_retriever(vectorstore, llm):retriever = vectorstore.as_retriever()QUERY_PROMPT = PromptTemplate(input_variables=["question"],template="""You are an AI language model assistant. Your task is to generate 5 different versions of the given user question to retrieve relevant documents from a vector  database. By generating multiple perspectives on the user question, your goal is to help the user overcome some of the limitations of distance-based similarity search. Provide these alternative questions separated by newlines. Original question: {question}""")return MultiQueryRetriever.from_llm(prompt=QUERY_PROMPT,retriever=retriever,llm=llm,include_original=True)# 打印文档
def pretty_print_docs(docs):for doc in docs:print(doc.page_content)if __name__ == "__main__":# 定义文件路径TXT_DOCUMENT_PATH = "your_text_file.txt"# 初始化日志logging.basicConfig()logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)# 初始化嵌入模型embeddings_model = HuggingFaceEmbeddings()# 初始化大语言模型，这里使用 ChatOpenAI 作为示例，你可以根据需要替换llm = ChatOpenAI()# 加载文档docs = load_documents(TXT_DOCUMENT_PATH)if docs:# 分割文档chunks = split_documents(docs)# 创建向量数据库vectorstore = create_vectorstore(chunks, embeddings_model)# 初始化检索器retrieval_from_llm = initialize_retriever(vectorstore, llm)# 执行检索unique_docs = retrieval_from_llm.invoke("详细介绍DeepSeek")# 打印文档pretty_print_docs(unique_docs)

        在 LangChain 框架中，MultiQueryRetriever 是实现多路召回功能的强大工具。它的核心原理是通过与精心设计的多路召回提示词相结合，突破传统单一检索的局限。

        具体来说，当用户输入查询时，MultiQueryRetriever 会将该查询传递给语言模型，在多路召回提示词的引导下，语言模型会从多个维度和视角对原始查询进行拓展，生成一系列相关的衍生问题。这些衍生问题与原始查询一同组成查询集合，随后被发送至检索系统。检索系统基于这个查询集合，在向量数据库中执行多次检索操作，每次检索都对应一个查询，如同从不同方向探索信息宝库，从而从向量库中召回更多与用户需求潜在相关的文档。这种方式能够有效弥补用户原始查询可能存在的表述模糊、语义局限等问题，极大地丰富了检索结果的数量和多样性，为后续的信息处理和答案生成提供更全面、充足的素材，显著提升了整个检索增强生成系统的性能和用户体验，尤其适用于复杂、模糊或需要深度信息挖掘的查询场景。

Decomposition 问题分解

        问题分解（Decomposition）策略通过将复杂、模糊的用户查询拆解为多个更易处理的子问题，能够显著提升检索的准确性与全面性，进而为生成高质量答案奠定基础。问题分解策略的核心目标是：

• 提升召回率：通过子问题覆盖更多相关文档片段

• 降低处理难度：每个子问题聚焦单一语义单元

• 增强可解释性：明确展示问题解决路径

典型适用场景：

• 多条件复合问题（"同时满足A和B的方案"）

• 多步骤推理问题（"实现X需要哪些步骤"）

• 对比分析问题（"A与B的优劣比较"）

常见分解策略

逻辑结构分解

        将问题拆分为逻辑关联的子模块：

示例：

原问题：
"如何设计一个支持高并发支付的电商系统？"

子问题分解：

• 电商支付系统的核心组件有哪些？

• 高并发场景下的数据库选型建议

• 支付接口的限流熔断方案

• 分布式事务一致性保障方法

时间序列分解

        按时间维度拆分阶段性问题：

示例：

原问题：
"从立项到上线的APP开发全流程"

子问题分解：

• 移动应用立项阶段的需求分析方法

• 敏捷开发中的迭代管理实践

• APP上线前的测试验收标准

• 应用商店发布审核注意事项

多视角分解

        从不同角度生成互补性问题：

示例：

原问题：
"深度学习在医疗影像中的应用"

子问题分解：

• （技术视角）医疗影像分析的常用深度学习模型架构

• （临床视角）三甲医院实际部署案例中的准确率数据

• （合规视角）医学AI模型的法律监管要求

分解流程

串行分解流程

并行分解流程

技术实现方案

基于规则模板的分解

        根据不同领域和问题类型，预先制定规则和模板。这种方式适用于结构化程度高、问题模式相对固定的场景，优点是简单直接、执行效率高，但灵活性较差，难以应对复杂多变或新型的问题。

# 定义分解规则模板
DECOMPOSE_RULES = {"设计类问题": ["核心组件", "架构模式", "技术选型"],"比较类问题": ["优势分析", "劣势分析", "适用场景"]
}def rule_based_decompose(query, category):return [f"{query}的{aspect}" for aspect in DECOMPOSE_RULES.get(category, [])]

基于LLM的智能分解

        借助自然语言处理（NLP）技术，尤其是深度学习模型（如 Transformer 架构），对用户问题进行深度语义分析。这种方法能够处理复杂语义，适应性强，但对模型的训练和计算资源要求较高。

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplatedecompose_prompt = PromptTemplate.from_template("""将复杂问题拆分为3-5个原子子问题：要求：1. 每个子问题可独立检索2. 覆盖原始问题的所有关键方面3. 使用明确的问题句式4. 你的目标是帮助用户克服基于距离的相似性搜索的一些局限性输入问题：{query}生成子问题："""
)decompose_chain = LLMChain(llm=llm, prompt=decompose_prompt)# 执行分解
sub_questions = decompose_chain.run("如何构建企业级数据中台？").split("\n")
# 输出：["数据中台的核心架构组成？", "数据治理的最佳实践？", "..."]

混合分解策略

        结合规则与LLM的优势：

def hybrid_decomposition(query):# 第一步：分类问题类型classifier_chain = LLMChain(...) category = classifier_chain.run(query)# 第二步：根据类型选择策略if category in DECOMPOSE_RULES:return rule_based_decompose(query, category)else:return decompose_chain.run(query)

层次化多阶段分解

        对于极其复杂的问题，采用分层次、多阶段的分解策略。首先进行宏观层面的初步分解，得到几个主要的子问题方向，然后针对每个子问题再进行进一步细化分解。通过这种逐步细化的方式，确保问题的每个方面都能得到充分考虑，检索也更加精准。

        在这里，我让DeepSeek 帮我生成了一个基于LangChain ，构建小说的代码，使用的就是层次化多阶段分解的策略。具体代码如下：

from langchain.chains import SequentialChain, TransformChain
from langchain.prompts import PromptTemplate
from langchain.schema import StrOutputParser
from typing import Dict, List# 第一阶段：生成小说大纲
def create_outline_chain(llm):outline_template = """你是一个专业小说家，根据用户需求创作小说大纲。
用户需求：{user_input}
请按以下格式生成包含5个章节的小说大纲：
1. 章节标题：章节核心事件
2. 章节标题：章节核心事件
...
输出示例：
1. 命运相遇：主角在拍卖会意外获得神秘古剑
2. 真相初现：发现古剑与失传王朝的关联"""return LLMChain(llm=llm,prompt=PromptTemplate.from_template(outline_template),output_key="outline")# 第二阶段：章节分解器
def chapter_decomposer(outline: str) -> List[Dict]:"""将大纲解析为章节结构"""chapters = []for line in outline.split("\n"):if "." in line:num, rest = line.split(".", 1)title, event = rest.split("：", 1) if "：" in rest else (rest.strip(), "")chapters.append({"chapter_num": num.strip(),"title": title.strip(),"core_event": event.strip()})return chapters# 第三阶段：分章节内容生成
def create_chapter_chain(llm):chapter_template = """根据以下大纲创作小说章节内容：
整体大纲：
{outline}当前章节：{chapter_num} {title}
核心事件：{core_event}
要求：
1. 包含3个关键场景
2. 保持文学性描写
3. 字数800-1000字章节内容："""return LLMChain(llm=llm,prompt=PromptTemplate.from_template(chapter_template),output_key="chapter_content")# 构建完整工作流
class NovelGenerator:def __init__(self, llm):self.outline_chain = create_outline_chain(llm)self.chapter_chain = create_chapter_chain(llm)# 定义转换链处理章节分解self.decompose_chain = TransformChain(input_variables=["outline"],output_variables=["chapters"],transform=chapter_decomposer)# 构建主流程self.master_chain = SequentialChain(chains=[self.outline_chain,self.decompose_chain,self._build_chapter_expansion()],input_variables=["user_input"],output_variables=["final_novel"])def _build_chapter_expansion(self):"""构建章节扩展子链"""def expand_chapters(inputs: Dict) -> Dict:full_novel = []for chapter in inputs["chapters"]:# 合并上下文context = {"outline": inputs["outline"],**chapter}# 生成单章内容result = self.chapter_chain(context)full_novel.append(f"## {chapter['chapter_num']} {chapter['title']}\n{result['chapter_content']}")return {"final_novel": "\n\n".join(full_novel)}return TransformChain(input_variables=["outline", "chapters"],output_variables=["final_novel"],transform=expand_chapters)def generate(self, user_input: str) -> str:"""执行完整生成流程"""return self.master_chain({"user_input": user_input})["final_novel"]# 使用示例
if __name__ == "__main__":from langchain.chat_models import ChatOpenAIllm = ChatOpenAI(model="gpt-4", temperature=0.7)generator = NovelGenerator(llm)novel = generator.generate("创作一部以民国古玩商人为背景的悬疑爱情小说")# 保存生成结果with open("novel_draft.md", "w", encoding="utf-8") as f:f.write(novel)

基于知识图谱的分解

        我们可以利用知识图谱中实体与关系的结构化信息，辅助问题分解。借助知识图谱的语义关联，使分解结果更具逻辑性和针对性。代码示例如下：

from langchain.chains import LLMChain, TransformChain
from langchain.prompts import PromptTemplate
from typing import Dict, List# 示例知识图谱（实际应用时可替换为Neo4j等图数据库接口）
MEDICAL_KNOWLEDGE_GRAPH = {"nodes": {"糖尿病": {"type": "疾病","properties": {"别名": ["消渴症"],"分类": ["代谢性疾病"]},"relationships": [{"type": "治疗方法", "target": "胰岛素治疗", "properties": {"有效性": "高"}},{"type": "并发症", "target": "糖尿病肾病", "properties": {"概率": "30%"}},{"type": "检查项目", "target": "糖化血红蛋白检测"}]},"胰岛素治疗": {"type": "治疗方法","properties": {"给药方式": ["注射"],"副作用": ["低血糖"]}}}
}class KnowledgeGraphDecomposer:def __init__(self, llm):# 实体识别链self.entity_chain = LLMChain(llm=llm,prompt=PromptTemplate.from_template("""从医疗问题中提取医学实体：问题：{question}输出格式：["实体1", "实体2", ...]"""),output_key="entities")# 子问题生成链self.decompose_chain = LLMChain(llm=llm,prompt=PromptTemplate.from_template("""基于知识图谱关系生成子问题：原始问题：{question}相关实体：{entities}知识关联：{kg_info}请生成3-5个逻辑递进的子问题，要求：1. 覆盖诊断、治疗、预防等方面2. 使用明确的医学术语3. 体现知识图谱中的关联关系输出格式：- 子问题1- 子问题2"""),output_key="sub_questions")def query_kg(self, entity: str) -> Dict:"""知识图谱查询"""return MEDICAL_KNOWLEDGE_GRAPH["nodes"].get(entity, {})def build_pipeline(self):"""构建处理管道"""return TransformChain(input_variables=["question"],output_variables=["sub_questions"],transform=self._process)def _process(self, inputs: Dict) -> Dict:# 步骤1：实体识别entities = eval(self.entity_chain.run(inputs["question"]))# 步骤2：知识图谱查询kg_info = {}for entity in entities:node_info = self.query_kg(entity)if node_info:kg_info[entity] = {"属性": node_info.get("properties", {}),"关联": [rel["type"]+"→"+rel["target"] for rel in node_info.get("relationships", [])]}# 步骤3：子问题生成return {"sub_questions": self.decompose_chain.run({"question": inputs["question"],"entities": str(entities),"kg_info": str(kg_info)})}# 使用示例
if __name__ == "__main__":from langchain.chat_models import ChatOpenAIllm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.5)decomposer = KnowledgeGraphDecomposer(llm)pipeline = decomposer.build_pipeline()# 执行问题分解result = pipeline.invoke({"question": "糖尿病患者应该如何进行治疗？"})print("生成的子问题：")print(result["sub_questions"])

混合检索

        混合检索流程       

        混合检索策略是一种将不同检索方式优势相结合的检索方法，在实际应用中展现出强大的灵活性与高效性。其核心在于融合向量检索（稠密检索）和关键字检索（稀疏检索），常见的是以 BM25 关键词检索等稀疏检索策略与向量检索搭配使用。

        向量检索（稠密检索）基于深度学习模型，将文本转化为高维向量，通过计算向量之间的相似度，检索出语义相近的文本。它擅长捕捉文本的语义信息，能理解用户查询背后的含义，即使查询语句与文档表述不完全一致，只要语义相似，也能实现精准检索，适合处理语义复杂、模糊的查询场景。而关键字检索（稀疏检索），如 BM25 关键词检索，依据用户输入的关键词在文本中进行匹配，通过统计关键词在文档中的出现频率、位置等信息，评估文档与查询的相关性。这种方式简单直接，对结构化数据、精确匹配的检索需求有很好的效果，能够快速定位包含特定关键词的文档。

        以下是一段基于BM25 检索结合向量检索的代码示例：

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever# 假设这些函数和变量已经定义
def pretty_print_docs(docs):for doc in docs:print(doc.page_content)def perform_hybrid_retrieval(chunks, vector_retriever, query):try:# BM25关键词检索BM25_retriever = BM25Retriever.from_documents(chunks, k=3)BM25Retriever_doc = BM25_retriever.invoke(query)print("BM25关键词检索结果:")pretty_print_docs(BM25Retriever_doc)# 向量检索vector_retriever_doc = vector_retriever.invoke(query)print("\n向量检索结果:")pretty_print_docs(vector_retriever_doc)# 向量检索和关键词检索的权重各0.5，两者赋予相同的权重retriever = EnsembleRetriever(retrievers=[BM25_retriever, vector_retriever], weights=[0.5, 0.5])ensemble_result = retriever.invoke(query)print("\n混合检索结果:")pretty_print_docs(ensemble_result)except Exception as e:print(f"检索过程中出现错误: {e}")

        在这里，我们运用EnsembleRetriever方法实现了高效的混合检索策略。具体而言，通过将基于 BM25 算法的关键词检索器BM25_retriever与向量检索器vector_retriever相结合，并为二者赋予相等的权重 0.5，使得检索过程既能发挥 BM25 关键词检索在精确匹配方面的优势，快速定位包含目标关键词的文档；又能借助向量检索在语义理解层面的特长，捕捉语义相近的相关内容。这种强强联合的方式，最终整合出兼具准确性和全面性的混合检索结果，有效提升了检索系统对复杂查询的处理能力和响应质量。