学习目标RAG技术概述RAG核心原理与流程NaiveRAGLangChain快速搭建本地知识库-使用NaiveRAG在上一篇文章中《大模型应用开发四主流三种开发模式范式》RAG是大模型应用开发的范式我们今天一起来认识一下RAG并使用RAG快熟搭建一个本地知识库一、什么是RAGRAGRetrieval-Augmented Generation检索增强生成是一种结合信息检索Retrieval和文本生成Generation的技术RAG技术通过实时检索相关文档或信息并将其作为上下文输入到生成模型中从而提高生成结果的时效性和准确性。上图是RAG的基本流程用户提出问题---去知识库检索问题的答案(Retrieval)---将问题和答案组装成新的提示词(Augmentation)---将新提示词交给LLM生成答案(Generation)1.1 RAG 的优势是什么解决知识时效性问题大模型的训练数据通常是静态的无法涵盖最新信息而RAG可以检索外部知识库实时更新信息确保回答的时效性减少模型幻觉通过引入外部知识回答基于检索到的真实文档RAG能够减少模型生成虚假或不准确内容的可能性提升专业领域回答质量RAG能够结合垂直领域的专业知识库生成更具专业深度的回答生成内容的溯源可解释性检索生成后的内容可追溯来源使回答增加可信度灵活性与可控性高知识库可独立更新不影响模型本身企业能精确把控输出内容实施成本较低无需大量标注数据或模型训练仅需构建知识库与检索系统。1.2 RAG 的核心原理与流程Step1数据预处理构建索引知识库知识整理及加载收集并整理文档、网页、数据库等多源数据构建外部知识库文档分块将文档切分为适当大小的片段chunks以便后续检索。分块策略需要在语义完整性与检索效率之间取得平衡向量化处理使用嵌入模型如BGE、M3E、Chinese-Alpaca-2等将文本块转换为向量并存储在向量数据库中Step2检索阶段查询处理将用户输入的问题转换为向量并在向量数据库中进行相似度检索找到最相关的文本片段使用向量数据库的相似度检索重排序对检索结果进行相关性排序选择最相关的片段作为生成阶段的输入(优化步骤)Step3生成阶段上下文组装将检索到的文本片段与用户问题结合形成增强的上下文输入生成回答大语言模型基于增强的上下文生成最终回答划重点RAG 本质上就是重构了一个新的 Prompt二、NaiveRAGNaiveRAG的一个最基本的RAG阶段具体步骤如下基本的输入/输出用户直接输入问题给LLM若不使用检索增强生成RAG则直接由LLM对问题进行应答使用RAG的流程Indexing 如何更好地把知识存起来将文本文档切块然后向量化并存储到向量数据库用于检索Retrieval 如何在大量的知识中找到一小部分有用的给到模型参考通过向量相似度找到有用的块Generation 如何结合用户的提问和检索到的知识让模型生成有用的答案将问题和有用的块给LLM生成最终答案划重点上面三个步骤虽然看似简单但在 RAG 应用从构建到落地实施的整个过程中涉及较多复杂的工作内容三、LangChain快速搭建本地知识库检索-使用NaiveRAG3.1. 环境准备本地安装好 Conda 环境推荐使用阿里大模型平台百炼https://bailian.console.aliyun.com/百炼平台使用注册登录申请api key3.2. 搭建流程构建知识库加载文档数据文档分块文本块向量化处理并存储向量数据库检索生成封装检索接口构建调用流程Query - 检索 - Prompt - LLM - 回复3.3 代码案例使用Conda创建虚拟环境并激活conda create -n naive_rag python3.12 conda activate naive_rag激活后终端开头会显示(naive_rag)表示已进入该虚拟环境此时安装的所有包都只属于这个环境安装相关依赖python版本使用的是3.12PyPDF23.0.1 dashscope1.23.3 langchain0.3.25 langchain-community0.3.24 langchain-openai0.3.18 faiss-cpu1.11.0构建知识库代码build_knowledge_store.py# -*- codingutf-8 -*- # description: 数据预处理构建知识库 # File : build_knowledge_store.py # author: zhouyer # time : 2026/3/10 22:05 import logging import os import pickle from typing import List, Tuple from PyPDF2 import PdfReader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS # 设置百炼平台的api_key到环境变量中 DASHSCOPE_API_KEY_YY os.environ[DASHSCOPE_API_KEY] os.getenv(DASHSCOPE_API_KEY_YY) # 向量数据库存储路基 VECTOR_STORE_PATH ./document/vector_db_01 def load_document(document_path: str) - Tuple[str, List[int]]: ** 数据预处理构建索引库-第一步加载文档数据** 从PDF中提取文本并记录每行文本对应的页码 参数: document_path: 文档路径 返回: text: 提取的文本内容 page_numbers: 每行文本对应的页码列表 text page_numbers [] # 读取pdf文档 pdf_reader PdfReader(document_path) for page_number, page in enumerate(pdf_reader.pages, start1): # 提取文本 extracted_text page.extract_text() if extracted_text: text extracted_text page_numbers.extend([page_number] * len(extracted_text.split(\n))) else: logging.warning(fNo text found on page {page_number}.) return text, page_numbers def process_text_with_splitter(text: str) - List[str]: ** 数据预处理构建索引库-第二步文档分块 ** 将文档切分为适当大小的片段chunks以便后续检索 参数: text: 提取的文本内容 page_numbers: 每行文本对应的页码列表 返回: chunks文本分块结果 # 创建文本分割器用于将长文本分割成小块 text_splitter RecursiveCharacterTextSplitter( separators[\n\n, \n, ., , ], chunk_size512, # 每个块的字符数据根据各大模型的效果512个字符时模型很好理解语意的合适值 chunk_overlap128, # 块与块之间的重叠部分大小一般设置为块大小的10%20%即上一个chunk末尾的128个字符与下一个chunk开始的128个字符是一样的为了保证上下文语意的连贯性 length_functionlen, ) # 分割文本 return text_splitter.split_text(text) def chunk_document_embedding(chunks: List[str], page_numbers: List[int], save_path: str None) - FAISS: ** 数据预处理构建索引库-第二步文本块向量化处理并保存到向量数据库 ** 在LangChain 中提供了 from_texts 和 from_documents 两个通用方法可以快捷地将数据从文本和文档中导入到向量数据库中。 将每个chunk向量化后保存的Faiss向量库同时保存页码信息方便后续溯源 :param chunks: 文档块 :param page_numbers: 每行文本对应的页码 :param save_path:可选保存向量数据库的路径 :return knowledgeBase: 基于FAISS的向量存储对象 # 存储每个文本块对应的页码信息 page_info {chunk: page_numbers[i] for i, chunk in enumerate(chunks)} # 调用阿里百炼平台文本嵌入模型配置环境变量 DASHSCOPE_API_KEY embeddings DashScopeEmbeddings( modeltext-embedding-v4 ) # 从文本块创建知识库 knowledge_base FAISS.from_texts(chunks, embeddings) print(已从文本块创建知识库...) knowledge_base.page_info page_info # 如果提供了保存路径则保存向量数据库和页码信息 if save_path: # 确保目录存在 os.makedirs(save_path, exist_okTrue) # 保存FAISS向量数据库 knowledge_base.save_local(save_path) print(f向量数据库已保存到: {save_path}) # 保存页码信息到同一目录 with open(os.path.join(save_path, page_info.pkl), wb) as f: pickle.dump(page_info, f) print(f页码信息已保存到: {os.path.join(save_path, page_info.pkl)}) return knowledge_base if __name__ __main__: pdf_document_path document/xxxx.pdf # 提取文本和页码信息 text, page_numbers load_document(pdf_document_path) # print(f页码: {page_numbers}) # print(f文本: {text}) print(f总字符数:{len(text)}) # 文档分块 chunks process_text_with_splitter(text) print(f文本被分割成 {len(chunks)} 个块。) # 向量化处理并存储 knowledge_store chunk_document_embedding(chunks, page_numbers, save_pathVECTOR_STORE_PATH) print(knowledge_store) # 返回一个FAISS对象---- langchain_community.vectorstores.faiss.FAISS object at 0x107710980 总字符数:3881 文本被分割成 10 个块。 已从文本块创建知识库... 向量数据库已保存到: ./document/vector_db_01 页码信息已保存到: ./document/vector_db_01/page_info.pkl langchain_community.vectorstores.faiss.FAISS object at 0x111835d90 目前当前构建知识库的代码是一个非常基础的结构仅能支持pdf文档如果有其他的也可以进行改在load_document加载文档数据process_text_with_splitter文档分块chunk_document_embedding文本块向量化处理并存储上面三个方法对应【构建知识库】搭建流程中的三个步骤检索生成代码retrieval_generation.py# -*- codingutf-8 -*- # description: 检索生成 # File : retrieval_generation.py # author: zhouyer # time : 2026/3/10 22:24 import os import pickle from langchain.chains.question_answering import load_qa_chain from langchain_community.callbacks.manager import get_openai_callback from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_openai import ChatOpenAI # 设置百炼平台的api_key到环境变量中 DASHSCOPE_API_KEY_YY os.environ[DASHSCOPE_API_KEY] os.getenv(DASHSCOPE_API_KEY_YY) # 向量数据库存储路基 VECTOR_STORE_PATH ./document/vector_db_01 def load_knowledge_base(load_path: str, embeddingsNone) - FAISS: 从磁盘加载向量数据库和页码信息 参数: load_path: 向量数据库的保存路径 embeddings: 可选嵌入模型。如果为None将创建一个新的DashScopeEmbeddings实例 返回: knowledgeBase: 加载的FAISS向量数据库对象 # 如果没有提供嵌入模型则创建一个新的 if embeddings is None: embeddings DashScopeEmbeddings( modeltext-embedding-v4 ) # 加载FAISS向量数据库添加allow_dangerous_deserializationTrue参数以允许反序列化 knowledge_base FAISS.load_local(load_path, embeddings, allow_dangerous_deserializationTrue) print(f向量数据库已从 {load_path} 加载。) # 加载页码信息 page_info_path os.path.join(load_path, page_info.pkl) if os.path.exists(page_info_path): with open(page_info_path, rb) as f: page_info pickle.load(f) knowledge_base.page_info page_info print(页码信息已加载。) else: print(警告: 未找到页码信息文件。) return knowledge_base def simple_query(query: str): 检索生成 检索传入问题---从向量数据库中检索相似性答案(问题向量化相似性检索)---得到相似性的文本块(Top k) 生成将问题检索的答案结合形成增强的上下文输入(Prompt)----LLM---输出最终回答 :param query: 问题 :return: if query: ###### 如何加载已保存的向量数据库 knowledge_store load_knowledge_base(VECTOR_STORE_PATH) # similarity_search()基础相似度搜索传递 query(搜索语句)、k(返回条数)、filter(过滤器)、fetch_k(富余条数) 等。 docs knowledge_store.similarity_search(query) # 初始化对话大模型,不建议使用推理模型因为推理模型还需要进一步逻辑推理耗时而且没用因为这里直接给咯答案只需要整理使用chat模型即生成模型 # deepseek r1就是推理不行不能使用 chatLLM ChatOpenAI( # 若没有配置环境变量请用百炼API Key将下行替换为api_keysk-xxx, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeldeepseek-v3 ) # 加载问答链 chain load_qa_chain(chatLLM, chain_typestuff) # 准备输入数据:将问题和相似性检索的答案chunks 交给LLM-----新的提示词。--- 这里就是RAG的A-增强 input_data {input_documents: docs, question: query} # 使用回调函数跟踪API调用成本 with get_openai_callback() as cost: # 执行问答链 response chain.invoke(inputinput_data) print(f查询已处理。成本: {cost}) print(response[output_text]) print(来源:) # 记录唯一的页码 unique_pages set() # 显示每个文档块的来源页码 for doc in docs: text_content getattr(doc, page_content, ) source_page knowledge_store.page_info.get( text_content.strip(), 未知 ) if source_page not in unique_pages: unique_pages.add(source_page) print(f文本块页码: {source_page}) if __name__ __main__: ############ 【检索】基于构建好的知识库进行 ############# # 设置查询问题 # query 客户经理被投诉了投诉一次扣多少分 # query 客户经理每年评聘申报时间是怎样的 query 资深客户经理的考核标准是什么 simple_query(query) query 客户经理被投诉了投诉一次扣多少分 查询已处理。成本: Tokens Used: 1253 Prompt Tokens: 1201 Prompt Tokens Cached: 0 Completion Tokens: 52 Reasoning Tokens: 0 Successful Requests: 1 Total Cost (USD): $0.0 根据工作质量考核标准中的服务质量考核部分客户经理被投诉一次扣2分。具体条款如下 2、客户服务效率低态度生硬或不及时为客户提供维护服务有客户投诉的每投诉一次扣2分。 query 客户经理每年评聘申报时间是怎样的 答案如下 查询已处理。成本: Tokens Used: 1287 Prompt Tokens: 1239 Prompt Tokens Cached: 0 Completion Tokens: 48 Reasoning Tokens: 0 Successful Requests: 1 Total Cost (USD): $0.0 根据第十一条的规定客户经理每年评聘的申报时间是**一月份**。分行人力资源部和个人业务部会在**二月份**组织统一的资格考试考试合格者将获得个金客户经理资格证书有效期为一年。 来源: 文本块页码: 1 query 资深客户经理的考核标准是什么 根据提供的考核标准资深客户经理的准入和考核要求如下 1. **准入标准** - 储蓄业务季日均余额500万元 - 个贷业务季新增发放个贷800万元 2. **考核分值对应级别** - 5级165分 - 4级170分 - 3级175分 - 2级180分 - 1级185分 3. **其他说明** - 个贷业务是中级以上客户经理含资深的考核进入标准。 - 超出最低标准的部分可按比例折算50万储蓄50万个贷50张有效卡5分。 注意资深客户经理还需满足基础素质要求如学历、工作经验等和工作质量考核如避免投诉、差错等扣分项。具体执行可能根据分行政策调整。 load_knowledge_base加载本地的知识库用于检索simple_query相似度检索LLM生成答案3.4 小结1. PDF文本提取与处理使用PyPDF2库的PdfReader从PDF文件中提取文本在提取过程中记录每行文本对应的页码便于后续溯源使用RecursiveCharacterTextSplitter将长文本分割成小块便于向量化处理2. 向量数据库构建使用OpenAIEmbeddings / DashScopeEmbeddings将文本块转换为向量表示使用FAISS向量数据库存储文本向量支持高效的相似度搜索为每个文本块保存对应的页码信息实现查询结果溯源3. 语义搜索与问答链基于用户查询使用similarity_search在向量数据库中检索相关文本块使用文本语言模型和load_qa_chain构建问答链将检索到的文档和用户问题作为输入生成回答4. 成本跟踪与结果展示使用get_openai_callback跟踪API调用成本展示问答结果和来源页码方便用户验证信息