本实践练习将构建一个检索增强生成(RAG)管道,它在基础向量搜索上有所提升,融入了高级文档处理、混合检索和重排技术,适用于生产环境。目标是构建一个系统,通过运用多种检索信号并在生成前优化检索到的上下文,从而提供更贴切和准确的回答。假设你有一批文档(例如,PDF或Markdown格式的技术文章、项目文档或研究论文),你想将它们用作问答系统的知识库。先决条件: 确保你已安装所需的库:pip install langchain langchain-openai langchain-community sentence-transformers faiss-cpu tiktoken pypdf rank_bm25 # 如果有CUDA,可以使用faiss-gpu # 或者用其他向量存储客户端(如chromadb或pinecone-client)替代faiss pip install chromadb # 以Chroma为例 pip install unstructured # 用于处理更多文档类型你还需要访问大语言模型(如OpenAI的模型),并可能需要将API密钥设置为环境变量。1. 高级文档加载与分块与简单的固定大小分块不同,我们将使用RecursiveCharacterTextSplitter,它首先尝试根据语义边界(段落、句子)进行分割,然后才按字符计数。这通常能更好地保持上下文。我们还需要处理加载可能多种多样的文档类型。import os from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings # 或者使用其他嵌入提供商 from langchain_community.vectorstores import Chroma # 或者Pinecone、FAISS等 # 配置你的文档路径和类型 DOCS_PATH = "./your_documents" # 使用DirectoryLoader增加灵活性,如果需要,可按文件类型配置加载器 # 示例:此处为简化,只加载PDF loader = DirectoryLoader(DOCS_PATH, glob="**/*.pdf", loader_cls=PyPDFLoader, show_progress=True) documents = loader.load() # 高级分块策略 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, # 重叠有助于保持分块之间的上下文 length_function=len, add_start_index=True, # 对于后续可能的父文档检索很有用 ) chunks = text_splitter.split_documents(documents) print(f"已加载 {len(documents)} 份文档。") print(f"已分割成 {len(chunks)} 个分块。") # 初始化嵌入模型(如果使用OpenAI,请确保已设置OPENAI_API_KEY) embeddings = OpenAIEmbeddings(model="text-embedding-3-small")如果你有不同文件类型,可以考虑使用UnstructuredMarkdownLoader或langchain_community.document_loaders中的其他特定加载器。RecursiveCharacterTextSplitter的参数(chunk_size、chunk_overlap)通常需要根据你的文档特点和下游大语言模型上下文窗口大小进行调整。2. 混合搜索的索引构建一个性能好的RAG系统通常能从结合密集(向量)和稀疏(基于关键词)检索中获益。我们将设置这两种方式。a) 密集检索(向量存储)我们将文档分块索引到向量存储中。在此,我们使用Chroma作为本地示例,但根据需要,可以替换为你在生产环境中偏好的向量存储(Pinecone、Weaviate等)。# 初始化向量存储并索引分块 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db_hybrid" # 选择一个目录来保存索引 ) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10}) # 最初检索更多结果用于后续重排 print("向量存储已初始化,分块已索引。")b) 稀疏检索(BM25)BM25(Best Matching 25)是一种常用的基于关键词的算法。LangChain为其提供了检索器。from langchain.retrievers import BM25Retriever # BM25需要分块的原始文本 # 确保你的“chunks”包含page_content属性 bm25_retriever = BM25Retriever.from_documents(chunks) bm25_retriever.k = 10 # 根据关键词检索前10个结果 print("BM25检索器已初始化。")3. 使用EnsembleRetriever实现混合搜索现在,使用EnsembleRetriever结合密集和稀疏检索器。这使得可以为每种方法的贡献设置权重。最佳权重通常取决于具体的数据集和查询类型,需要通过实验确定。from langchain.retrievers import EnsembleRetriever # 初始化集成检索器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.4, 0.6] # 示例权重:略微优先考虑向量搜索 ) print("集成检索器已创建。") # 测试检索(可选) # sample_query = "什么是提示工程的最佳实践?" # retrieved_docs = ensemble_retriever.invoke(sample_query) # print(f"为示例查询检索到 {len(retrieved_docs)} 份文档。")4. 实现重排混合搜索会检索出多样化的文档集,但其中一些可能相关性不高。使用计算量更大的交叉编码器模型进行重排,可以显著提升传递给大语言模型的最终上下文质量。我们使用ContextualCompressionRetriever,它会封装我们的集成检索器并应用重排模型。from langchain.retrievers.document_compressors import CrossEncoderReranker from langchain_community.cross_encoders import HuggingFaceCrossEncoder from langchain.retrievers import ContextualCompressionRetriever # 初始化一个交叉编码器模型 # 像'cross-encoder/ms-marco-MiniLM-L-6-v2'这样的模型既高效又实用。 model = HuggingFaceCrossEncoder(model_name="cross-encoder/ms-marco-MiniLM-L-6-v2") # 初始化重排压缩器 compressor = CrossEncoderReranker(model=model, top_n=5) # 重排后保留前5个最相关的文档 # 创建压缩检索器 compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=ensemble_retriever # 使用混合检索器作为基础 ) print("重排检索器已配置。") # 测试重排检索(可选) # reranked_docs = compression_retriever.invoke(sample_query) # print(f"为示例查询重排到 {len(reranked_docs)} 份文档。") # 比较reranked_docs和retrieved_docs的内容top_n参数控制重排后保留的文档数量。这有助于大语言模型专注于最有价值的信息,并适应上下文窗口的限制。5. 使用LCEL构建完整的RAG链最后,使用LangChain表达式语言(LCEL)将优化过的检索器集成到完整的RAG链中。该链使用我们的compression_retriever获取上下文,将其格式化为提示,发送给大语言模型,并解析输出。from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 初始化大语言模型 llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) # 或gpt-4、claude等 # 定义提示模板 template = """你是一个问答助手。 请仅根据以下检索到的上下文来回答问题。 如果从上下文中无法得知答案,请直接回答不知道。 请保持回答简洁,并直接基于提供的信息。 上下文: {context} 问题:{question} 回答:""" prompt = ChatPromptTemplate.from_template(template) # 格式化检索到文档的辅助函数 def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) # 构建RAG链 rag_chain = ( {"context": compression_retriever | format_docs, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) print("优化过的RAG链已构建。")这种LCEL结构清晰地定义了数据流:用户的查询(RunnablePassthrough())被传递,同时也被compression_retriever用于获取和格式化上下文。之后,这些内容在提示模板中结合,由大语言模型处理,并解析最终的字符串输出。6. 调用和评估考量你现在可以用用户问题来调用该链:query = "解释RAG中混合搜索的含义。" final_answer = rag_chain.invoke(query) print("\n--- 最终回答 ---") print(final_answer) query_2 = "法国的首都是哪里?" # 示例:与上下文无关的问题 final_answer_2 = rag_chain.invoke(query_2) print("\n--- 最终回答 2 ---") print(final_answer_2)流程可视化:这是我们构建的管道图:digraph RAG_Pipeline { rankdir=LR; node [shape=box, style=rounded, fontname="Arial", fontsize=10]; edge [fontname="Arial", fontsize=9]; UserQuery [label="用户查询"]; QueryTransform [label="可选:\n查询转换\n(例如:HyDE)"]; BM25 [label="BM25检索器\n(稀疏)", style="filled", fillcolor="#a5d8ff"]; VectorStore [label="向量存储\n检索器 (密集)", style="filled", fillcolor="#bac8ff"]; Ensemble [label="集成检索器\n(组合与加权)", style="filled", fillcolor="#d0bfff"]; Reranker [label="交叉编码器\n重排器", style="filled", fillcolor="#eebefa"]; Format [label="格式化上下文"]; LLM [label="大语言模型\n(生成回答)", style="filled", fillcolor="#b2f2bb"]; FinalAnswer [label="最终回答"]; UserQuery -> QueryTransform [style=dashed, label="如果使用"]; UserQuery -> BM25; UserQuery -> VectorStore; QueryTransform -> BM25 [style=dashed]; QueryTransform -> VectorStore [style=dashed]; BM25 -> Ensemble; VectorStore -> Ensemble; Ensemble -> Reranker; Reranker -> Format; Format -> LLM [label="上下文"]; UserQuery -> LLM [label="原始查询"]; LLM -> FinalAnswer; }优化过的RAG管道流程,在大语言模型生成最终回答之前,融入了混合搜索和重排阶段。虚线表示可选的查询转换。评估: 这个优化过的管道应能比仅使用基础向量搜索的RAG产生更好的结果。然而,严格的评估是必需的。如第五章所述,使用LangSmith等工具来追踪执行情况,并定义指标(例如,RAGAS指标,如忠实度、上下文精确度、上下文召回率、回答相关性),这些指标需在一个具有代表性的问答对数据集上进行评估。将此管道的性能与更简单的基线进行比较,以量化混合搜索和重排所带来的提升。调整集成权重(EnsembleRetriever中的weights)以及重排后保留的文档数量(CrossEncoderReranker中的top_n)是根据评估结果进行的常见优化步骤。本练习为面向生产环境的RAG系统提供了一个范本。你可以通过集成查询转换(如HyDE)、实现父文档检索策略以获得更好的上下文,以及根据你的文档语料库精细化数据加载/分块来进一步改进它。请记住,监控(第五章)和部署实践(第七章)对于在生产中维持性能和可靠性非常重要。