. RAG 检索增强生成，让大模型拥有外部知识

文档加载多源数据的统一接入文档加载是 RAG 离线索引的第一步目标是从各类异构数据源中提取纯文本内容为后续处理提供标准化的输入。常见的数据源包括本地文件TXT、CSV、JSON、Markdown、PDF 等、网页内容、数据库、第三方服务接口等。在 LangChain 框架中提供了丰富的 DocumentLoader 类不同格式的文件经过对应加载器处理后会统一输出为 Document 对象该对象包含page_content正文内容与metadata元数据两个核心属性方便后续流程统一处理。实战多格式文档加载实现我们以服装产品相关数据为例实现 TXT、CSV、JSON、Markdown、PDF 五种格式文件的加载。首先安装所需依赖pip install pypdf jq完整加载代码如下复制代码from langchain_community.document_loaders import (TextLoader, CSVLoader, JSONLoader, PyPDFLoader)方式1: 从 TXT 文件加载服装目录txt_loader TextLoader(“clothing_data/clothing_catalog.txt”, encoding“utf-8”)txt_docs txt_loader.load()for i, doc in enumerate(txt_docs[:2]):print(doc.page_content[:100])方式2: 从 CSV 文件加载产品信息csv_loader CSVLoader(“clothing_data/products.csv”, encoding“utf-8”)csv_docs csv_loader.load()for i, doc in enumerate(csv_docs[:2]):print(doc.page_content[:100])方式3: 从 JSON 文件加载产品详情json_loader JSONLoader(file_path“clothing_data/product_details.json”,jq_schema“.[].description”,text_contentFalse)json_docs json_loader.load()for i, doc in enumerate(json_docs[:2]):print(doc.page_content[:100])方式4: 从 Markdown 文件加载尺码指南md_loader TextLoader(“clothing_data/size_guide.md”, encoding“utf-8”)md_docs md_loader.load()for i, doc in enumerate(md_docs[:2]):print(doc.page_content[:100])方式5: 从 PDF 文件加载服装产品手册pdf_loader PyPDFLoader(“clothing_data/clothing_catalog.pdf”)pdf_docs pdf_loader.load()for i, doc in enumerate(pdf_docs[:2]):print(doc.page_content[:100])复制代码代码中JSON加载相对特殊需要通过jq_schema参数指定提取字段的路径示例中.[].description表示提取JSON数组中每个元素的description字段text_contentFalse则表示内容需要按jq规则解析而非直接读取纯文本。image动画视频在《32. 文档智能分割拆解长文本轻量化分块》文档智能分割长文本的轻量化分块完成原始文档的加载后长文本并不能直接用于后续的向量嵌入与检索环节。一方面大语言模型的上下文窗口存在上限无法一次性处理超长文本另一方面过大的文本块会混入无关内容稀释语义降低检索精度同时向量嵌入模型本身对输入长度也有限制通常在512到8192个Token之间。因此我们需要对长文档进行智能分割将其拆分为合适大小的文本块。文档分割Text Splitting是离线索引的关键环节直接影响后续检索的精准度。LangChain提供了多种分割策略适配不同的场景需求RecursiveCharacterTextSplitter按递归字符分隔符切分优先使用段落、句子等自然边界是最推荐的通用分割方案TokenTextSplitter严格按照Token数量进行切分CharacterTextSplitter按用户指定的字符分隔符进行切分。分割过程中有两个核心参数需要关注chunk_size表示单个文本块的最大长度chunk_overlap表示相邻文本块之间的重叠长度用于避免语义在切块边界处断裂保证上下文的连贯性。实战递归字符分割实现我们使用RecursiveCharacterTextSplitter对加载完成的所有文档进行统一分块设置块大小为200字符重叠长度为40字符并按段落、句号、逗号、空格的优先级选择分隔符。复制代码from langchain_community.document_loaders import (TextLoader, CSVLoader, JSONLoader, PyPDFLoader)from langchain_text_splitters import RecursiveCharacterTextSplitter加载 TXT 文件txt_loader TextLoader(“clothing_data/clothing_catalog.txt”, encoding“utf-8”)txt_docs txt_loader.load()加载 CSV 文件csv_loader CSVLoader(“clothing_data/products.csv”, encoding“utf-8”)csv_docs csv_loader.load()加载 JSON 文件json_loader JSONLoader(file_path“clothing_data/product_details.json”,jq_schema“.[].description”,text_contentFalse)json_docs json_loader.load()加载 Markdown 文件md_loader TextLoader(“clothing_data/size_guide.md”, encoding“utf-8”)md_docs md_loader.load()加载 PDF 文件pdf_loader PyPDFLoader(“clothing_data/clothing_catalog.pdf”)pdf_docs pdf_loader.load()合并所有文档all_docs txt_docs csv_docs json_docs md_docs pdf_docs创建分割器text_splitter RecursiveCharacterTextSplitter(chunk_size200, # 每个块最大200个字符chunk_overlap40, # 块之间重叠40个字符separators[“\n\n”, “。”, “”, , “”] # 按优先级尝试的分隔符)对加载的所有文档进行分割chunks text_splitter.split_documents(all_docs)print(f原始文档数: {len(all_docs)}“)print(f分割后的文本块数: {len(chunks)}”)for i, chunk in enumerate(chunks[:3]):print(f块 {i1} ({len(chunk.page_content)} 字符): {chunk.page_content[:60]}…“)print(f元数据: {chunk.metadata}”)复制代码调用split_documents()方法时会自动保留每个文本块对应的元数据如来源文件名方便后续检索时追溯信息来源。image动画视频在《33. 向量嵌入文本语义的数字化表达》向量嵌入文本语义的数字化表达得到尺寸合适的文本块后我们需要将自然语言文本转化为计算机可以计算的高维向量这一过程就是向量嵌入Embedding它是实现语义检索的基础。Embedding是将文本转换为固定维度高维向量的过程其核心特性是语义越相似的文本在向量空间中的距离越近。基于这一特性我们可以通过计算向量间的距离来衡量两段文本的语义相似度。向量具备三个核心特性维度向量的维度由嵌入模型决定常见为768、1536、3072维语义相似性通常通过余弦相似度来衡量两段文本的语义接近程度可计算性向量之间支持距离计算、加减运算等数学操作。目前主流的嵌入模型包括OpenAI的text-embedding-3-small、text-embedding-3-largeCohere的embed-english-v3.0、embed-multilingual-v3.0以及开源的BGE、GTE、E5系列可本地部署无需API调用费用。实战开源模型批量嵌入与相似度验证我们使用开源中文嵌入模型bge-small-zh-v1.5对分割后的文本块进行批量嵌入并实现余弦相似度计算与Top N相似文本匹配。首先安装依赖pip install sentence-transformers完整实现代码如下复制代码from langchain_community.document_loaders import (TextLoader, CSVLoader, JSONLoader, PyPDFLoader)from langchain_text_splitters import RecursiveCharacterTextSplitterfrom sentence_transformers import SentenceTransformerimport numpy as nptxt_loader TextLoader(“clothing_data/clothing_catalog.txt”, encoding“utf-8”)txt_docs txt_loader.load()csv_loader CSVLoader(“clothing_data/products.csv”, encoding“utf-8”)csv_docs csv_loader.load()json_loader JSONLoader(file_path“clothing_data/product_details.json”, jq_schema“.[].description”,text_contentFalse)json_docs json_loader.load()md_loader TextLoader(“clothing_data/size_guide.md”, encoding“utf-8”)md_docs md_loader.load()pdf_loader PyPDFLoader(“clothing_data/clothing_catalog.pdf”)pdf_docs pdf_loader.load()合并所有文档all_docs txt_docs csv_docs json_docs md_docs pdf_docs创建分割器text_splitter RecursiveCharacterTextSplitter(chunk_size200, # 每个块最大200个字符chunk_overlap40, # 块之间重叠40个字符separators[“\n\n”, “。”, “”, , “”] # 按优先级尝试的分隔符)对所有文档进行分割chunks text_splitter.split_documents(all_docs)model SentenceTransformer(‘BAAI/bge-small-zh-v1.5’)提取所有文本块的内容过滤掉空块chunk_texts [chunk.page_content for chunk in chunks if chunk.page_content.strip()]批量嵌入所有文本块vectors model.encode(chunk_texts)print(f文本块数量: {len(chunks)}“)print(f向量数量: {len(vectors)}”)print(f每个向量的维度: {len(vectors[0])})定义余弦相似度计算函数def cosine_similarity(v1, v2):return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))计算查询文本与所有文本块的相似度返回 Top N 结果def get_top_similar(query_text, vectors, chunk_texts, top_n3):query_vector model.encode([query_text])[0]similarities []for i, v in enumerate(vectors):score cosine_similarity(query_vector, v)similarities.append((i, score))similarities.sort(keylambda x: x[1], reverseTrue)return similarities[:top_n]验证嵌入效果for query in [“冬天穿的保暖外套”, “夏天穿的凉快T恤”]:print(f\n’{query}’ 与各文本块的相似度 (Top 3):“)for idx, score in get_top_similar(query, vectors, chunk_texts):print(f” 相似度 {score:.4f}: {chunk_texts[idx][:60]})复制代码通过测试查询可以直观看到语义相近的文本会获得更高的相似度分数验证了嵌入模型的语义表征能力。image动画视频在《34. 向量存储RAG 架构的核心组件》向量存储RAG架构的核心组件当文本数量较少时我们可以像上面一样直接在内存中计算相似度但当知识库规模扩大到十万、百万级别时就需要专门的向量数据库来实现高效的存储与检索。向量数据库是专门用于存储和高效检索向量数据的系统是RAG架构中的核心组件。它通过特殊的索引算法能够在海量向量中快速找到与查询向量最接近的结果大幅提升检索效率。目前主流的向量数据库包括Chroma开源轻量级产品开箱即用非常适合本地开发和原型验证FAISSFacebook开源的向量检索库性能极高但仅提供检索功能不具备完整的数据库管理能力Pinecone全托管云服务无需运维适合生产环境直接使用Weaviate功能丰富的开源向量数据库Milvus分布式向量数据库适合大规模生产环境部署。选型向量数据库时通常需要考虑数据规模、查询性能要求、部署方式本地/云端托管、是否支持元数据过滤与混合检索等因素。实战Chroma向量数据库集成我们选择轻量级的Chroma作为向量存储方案将本地嵌入模型与Chroma进行集成实现向量的持久化存储。首先安装依赖pip install langchain-chroma由于Chroma默认的嵌入接口与本地SentenceTransformer模型不直接兼容我们需要自定义一个封装类实现embed_documents与embed_query两个方法完整代码如下复制代码from langchain_community.document_loaders import (TextLoader, CSVLoader, JSONLoader, PyPDFLoader)from langchain_text_splitters import RecursiveCharacterTextSplitterfrom sentence_transformers import SentenceTransformerimport numpy as npfrom langchain_chroma import Chromatxt_loader TextLoader(“clothing_data/clothing_catalog.txt”, encoding“utf-8”)txt_docs txt_loader.load()csv_loader CSVLoader(“clothing_data/products.csv”, encoding“utf-8”)csv_docs csv_loader.load()json_loader JSONLoader(file_path“clothing_data/product_details.json”, jq_schema“.[].description”,text_contentFalse)json_docs json_loader.load()md_loader TextLoader(“clothing_data/size_guide.md”, encoding“utf-8”)md_docs md_loader.load()pdf_loader PyPDFLoader(“clothing_data/clothing_catalog.pdf”)pdf_docs pdf_loader.load()合并所有文档all_docs txt_docs csv_docs json_docs md_docs pdf_docs创建分割器text_splitter RecursiveCharacterTextSplitter(chunk_size200, # 每个块最大200个字符chunk_overlap40, # 块之间重叠40个字符separators[“\n\n”, “。”, “”, , “”] # 按优先级尝试的分隔符)对所有文档进行分割chunks text_splitter.split_documents(all_docs)model SentenceTransformer(‘bge-small-zh-v1.5’)chunk_texts [chunk.page_content for chunk in chunks if chunk.page_content.strip()]vectors model.encode(chunk_texts)print(f文本块数量{len(chunks)}“)print(f向量数量{len(vectors)}”)print(f每个向量的维度{len(vectors[0])})def cosine_similarity(vec1, vec2):return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))def get_top_similar(query_text, vectors, chunk_texts, top_n3):query_vector model.encode([query_text])[0]similarities []for i, v in enumerate(vectors):score cosine_similarity(query_vector, v)similarities.append((i, score))similarities.sort(keylambda x: x[1], reverseTrue)return similarities[:top_n]persist_directory “chroma_clothing_db”class LocalEmbedding:def embed_documents(self, texts):return model.encode(texts).tolist()def embed_query(self, text): return model.encode([text])[0].tolist()embedding_function LocalEmbedding()vectorstore Chroma.from_documents(documentschunks,embeddingembedding_function,persist_directorypersist_directory)print(f向量库存储条数{vectorstore._collection.count()})复制代码调用Chroma.from_documents()方法后系统会自动完成所有文本块的向量化与存储并将数据持久化到指定的本地目录中后续可直接加载复用。image动画视频在《35. 向量检索快速匹配相关文档》向量检索快速匹配相关文档完成向量库的构建后就进入了RAG的在线查询阶段核心是根据用户的问题从向量库中快速匹配出最相关的文档这一步的检索策略直接决定了最终回答的质量。检索策略决定了如何从向量数据库中找到与用户问题最相关的文档是RAG流程中的关键环节。我们可以将检索方式分为基础检索与高级检索两大类。基础检索方式(1) 相似度检索Similarity Search最基础的检索方式直接返回与查询向量余弦距离最近的K个文档(2) MMR检索Maximal Marginal Relevance在相关性与多样性之间取得平衡避免返回大量内容重复的文档适合需要覆盖更多信息维度的场景。高级检索技术(1) 多查询检索Multi-Query由大模型从不同角度改写原始问题生成多个查询向量后合并检索结果提升召回率(2) 父子文档检索Parent-Child检索时匹配细粒度的小块但返回包含该小块的更大的父文档保证上下文完整性(3) 上下文压缩Contextual Compression对检索到的文档进行片段提取或摘要只保留与问题最相关的部分减少无效上下文(4) 自查询检索Self-Query由大模型将用户的自然语言问题转换为结构化查询结合元数据过滤提升精准度(5) 混合检索Ensemble结合向量语义检索与传统关键词检索兼顾语义匹配与关键词匹配。实战多种检索方式实现基于前面构建的Chroma向量库我们演示相似度检索、MMR检索、元数据过滤检索以及Retriever接口的使用。复制代码query “有没有保暖的冬天外套”results vectorstore.similarity_search(query, k2)for i, doc in enumerate(results):print(f{i1}. {doc.page_content[:80]}“)print(f元数据: {doc.metadata}”)results_mmr vectorstore.max_marginal_relevance_search(query,k2,fetch_k10,lambda_mult0.7 )for i, doc in enumerate(results_mmr):print(f {i1}. {doc.page_content[:80]}…)results_filtered vectorstore.similarity_search(“保暖的外套”,k2,filter{“gender”: “女”})for doc in results_filtered:print(f{doc.page_content[:80]})retriever vectorstore.as_retriever(search_type“similarity”,search_kwargs{“k”: 3})print(retriever.invoke(“有没有保暖的冬天外套”))复制代码其中MMR检索的fetch_k参数表示先召回的候选文档数量lambda_mult用于控制多样性程度取值为0时追求最大多样性取值为1时追求最大相关性。通过as_retriever()方法可以将向量库封装为标准的Retriever接口方便后续与RAG链、Agent等组件集成。imageRAGAgent让检索更智能到这里一套基础的RAG检索流程就已经搭建完成。但基础RAG存在一个明显的问题无论用户输入什么内容哪怕是问候、闲聊、数学计算等与知识库无关的问题系统都会执行一次数据库检索造成不必要的资源浪费。为了解决这个问题我们可以将检索能力封装为工具交给Agent自主调度。Agent的核心能力是根据用户问题自主思考、判断并选择合适的工具来完成任务。将Retriever包装为Agent的一个工具后Agent会自行判断当问题涉及知识库内容时调用检索工具获取信息当问题可以直接回答或需要使用其他工具时则不会触发检索从而大幅提升资源利用率与交互效率。在LangChain中我们可以通过create_retriever_tool函数快速将Retriever包装为标准工具只需指定工具名称与描述Agent就会根据描述来判断何时调用该工具。实战智能导购Agent实现我们将检索工具与计算、文件操作等工具整合搭建一个具备服装导购能力的智能Agent它可以自主判断是否需要检索产品信息也能完成数学计算、文件读写等操作。完整代码如下复制代码import osfrom deepagents.backends import FilesystemBackendfrom dotenv import load_dotenvfrom langchain.agents import create_agentfrom langchain.chat_models import init_chat_modelfrom langchain_core.tools import tool, BaseToolfrom langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryToolfrom langchain_community.document_loaders import (TextLoader, CSVLoader, JSONLoader, PyPDFLoader)from langchain_text_splitters import RecursiveCharacterTextSplitterfrom sentence_transformers import SentenceTransformerimport numpy as npfrom langchain_chroma import Chromafrom langchain_core.tools import create_retriever_tooltxt_loader TextLoader(“clothing_data/clothing_catalog.txt”, encoding“utf-8”)txt_docs txt_loader.load()csv_loader CSVLoader(“clothing_data/products.csv”, encoding“utf-8”)csv_docs csv_loader.load()json_loader JSONLoader(file_path“clothing_data/product_details.json”, jq_schema“.[].description”,text_contentFalse)json_docs json_loader.load()md_loader TextLoader(“clothing_data/size_guide.md”, encoding“utf-8”)md_docs md_loader.load()pdf_loader PyPDFLoader(“clothing_data/clothing_catalog.pdf”)pdf_docs pdf_loader.load()合并所有文档all_docs txt_docs csv_docs json_docs md_docs pdf_docs创建分割器text_splitter RecursiveCharacterTextSplitter(chunk_size200, # 每个块最大200个字符chunk_overlap40, # 块之间重叠40个字符separators[“\n\n”, “。”, “”, , “”] # 按优先级尝试的分隔符)对所有文档进行分割chunks text_splitter.split_documents(all_docs)embedding_model SentenceTransformer(‘bge-small-zh-v1.5’)chunk_texts [chunk.page_content for chunk in chunks if chunk.page_content.strip()]vectors embedding_model.encode(chunk_texts)def cosine_similarity(vec1, vec2):return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))def get_top_similar(query_text, vectors, chunk_texts, top_n3):query_vector model.encode([query_text])[0]similarities []for i, v in enumerate(vectors):score cosine_similarity(query_vector, v)similarities.append((i, score))similarities.sort(keylambda x: x[1], reverseTrue)return similarities[:top_n]persist_directory “chroma_clothing_db”class LocalEmbedding:def embed_documents(self, texts):return embedding_model.encode(texts).tolist()def embed_query(self, text): return embedding_model.encode([text])[0].tolist()embedding_function LocalEmbedding()vectorstore Chroma.from_documents(documentschunks,embeddingembedding_function,persist_directorypersist_directory)query “有没有保暖的冬天外套”results vectorstore.similarity_search(query, k2)for i, doc in enumerate(results):print(f{i 1}.{doc.page_content[:80]}“)print(f元数据{doc.metadata}”)results_mmr vectorstore.max_marginal_relevance_search(query,k2,fetch_k10,lambda_mult0.7)for i, doc in enumerate(results_mmr):print(f{i 1}.{doc.page_content[:80]}“)results_filtered vectorstore.similarity_search(“保暖的外套”,k2,filter{“gender”: “女”})for doc in results_filtered:print(f”{doc.page_content[:80]})retrieve vectorstore.as_retriever(search_type“similarity”,search_kwargs{“k”: 3})load_dotenv()prefix “QWEN”model init_chat_model(model_provider“openai”,configurable_fields[“model”, “api_key”, “base_url”],config_prefixprefix).with_config({“configurable”: {f{prefix}_model: os.getenv(f{prefix}_MODEL),f{prefix}_api_key: os.getenv(f{prefix}_API_KEY),f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL)}})class CalculateTool(BaseTool):name: str “calculate”description: str “计算数学表达式的值”def _run(self, expression: str) - str: try: return f计算结果: {eval(expression)} except Exception as e: return f计算错误: {str(e)} async def _arun(self, expression: str) - str: return self._run(expression) # 初始化工具 calculate CalculateTool()write_file WriteFileTool()read_file ReadFileTool()list_dir ListDirectoryTool()search_tool create_retriever_tool(retrieverretrieve,name“search”,description“搜索服装产品信息当用户询问服装相关问题时使用此工具”)agent create_agent(modelmodel,tools[calculate, write_file, read_file, list_dir,search_tool],system_prompt“你是一个服装导购助手可以搜索产品信息、计算数值、读写文件、列出目录。”,debugTrue)queries [“你好”,“有没有保暖的冬天外套”,“计算 2024*12500然后把结果保存到 result.txt”,“读取 result.txt 的内容”,“列出当前目录文件”]for q in queries:print(f\n问{q}“)result agent.invoke({“messages”: [{“role”: “user”, “content”: q}]})print(f答{result[‘messages’][-1].content}”)复制代码通过多组测试问题可以验证面对普通问候时Agent会直接回复面对服装相关问题时会自动调用检索工具面对计算与文件操作需求时则会调用对应工具实现了能力的智能调度。总结