学习目标

1. RAG技术概述

2. RAG核心原理与流程

3. NaiveRAG

4. LangChain快速搭建本地知识库-使用NaiveRAG

在上一篇文章中《大模型应用开发（四）：主流三种开发模式（范式）》，RAG是大模型应用开发的范式，我们今天一起来认识一下RAG，并使用RAG快熟搭建一个本地知识库

一、什么是RAG

RAG（Retrieval-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！

二、NaiveRAG

NaiveRAG的一个最基本的RAG阶段，具体步骤如下：

基本的输入/输出：用户直接输入问题给LLM，若不使用检索增强生成（RAG），则直接由LLM对问题进行应答

使用RAG的流程：

• Indexing => 如何更好地把知识存起来？将文本文档切块，然后向量化并存储到向量数据库，用于检索

• Retrieval => 如何在大量的知识中，找到一小部分有用的，给到模型参考？通过向量相似度找到有用的块

• Generation => 如何结合用户的提问和检索到的知识，让模型生成有用的答案？将问题和有用的块给LLM生成最终答案

划重点：上面三个步骤虽然看似简单，但在 RAG 应用从构建到落地实施的整个过程中，涉及较多复杂的工作内容！

三、LangChain快速搭建本地知识库检索-使用NaiveRAG

3.1. 环境准备

1. 本地安装好 Conda 环境

2. 推荐使用阿里大模型平台百炼：https://bailian.console.aliyun.com/

3. 百炼平台使用

   • 注册登录

   • 申请api key

3.2. 搭建流程

构建知识库

1. 加载文档数据

2. 文档分块

3. 文本块向量化处理并存储向量数据库

检索+生成

1. 封装检索接口

2. 构建调用流程：Query -> 检索 -> Prompt -> LLM -> 回复

3.3 代码案例

使用Conda创建虚拟环境并激活

conda create -n naive_rag python=3.12
conda activate naive_rag

激活后，终端开头会显示 (naive_rag)，表示已进入该虚拟环境，此时安装的所有包都只属于这个环境

安装相关依赖

python版本使用的是3.12

PyPDF2==3.0.1
dashscope==1.23.3
langchain==0.3.25
langchain-community==0.3.24
langchain-openai==0.3.18
faiss-cpu==1.11.0

构建知识库代码

build_knowledge_store.py

# -*- coding=utf-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, start=1):
        # 提取文本
        extracted_text = page.extract_text()
        if extracted_text:
            text += extracted_text
            page_numbers.extend([page_number] * len(extracted_text.split("\n")))
        else:
            logging.warning(f"No 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_size=512,  # 每个块的字符数据，根据各大模型的效果，512个字符时模型很好理解语意的合适值
        chunk_overlap=128,  # 块与块之间的重叠部分大小，一般设置为块大小的10%～20%，即上一个chunk末尾的128个字符与下一个chunk开始的128个字符是一样的，为了保证上下文语意的连贯性
        length_function=len,
    )
​
    # 分割文本
    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(
        model="text-embedding-v4"
    )
    # 从文本块创建知识库
    knowledge_base = FAISS.from_texts(chunks, embeddings)
    print("已从文本块创建知识库...")
    knowledge_base.page_info = page_info
​
    # 如果提供了保存路径，则保存向量数据库和页码信息
    if save_path:
        # 确保目录存在
        os.makedirs(save_path, exist_ok=True)
​
        # 保存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_path=VECTOR_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 ​

# -*- coding=utf-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, embeddings=None) -> FAISS:
    """
    从磁盘加载向量数据库和页码信息
​
    参数:
        load_path: 向量数据库的保存路径
        embeddings: 可选，嵌入模型。如果为None，将创建一个新的DashScopeEmbeddings实例
​
    返回:
        knowledgeBase: 加载的FAISS向量数据库对象
    """
    # 如果没有提供嵌入模型，则创建一个新的
    if embeddings is None:
        embeddings = DashScopeEmbeddings(
            model="text-embedding-v4"
        )
​
    # 加载FAISS向量数据库，添加allow_dangerous_deserialization=True参数以允许反序列化
    knowledge_base = FAISS.load_local(load_path, embeddings, allow_dangerous_deserialization=True)
    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_key="sk-xxx",
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
            model="deepseek-v3"
        )
​
        # 加载问答链
        chain = load_qa_chain(chatLLM, chain_type="stuff")
​
        # 准备输入数据:将问题和相似性检索的答案chunks 交给LLM----->新的提示词。--- 这里就是RAG的A-增强
        input_data = {"input_documents": docs, "question": query}
​
        # 使用回调函数跟踪API调用成本
        with get_openai_callback() as cost:
            # 执行问答链
            response = chain.invoke(input=input_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调用成本

• 展示问答结果和来源页码，方便用户验证信息