第1节：认识RAG

RAG定义

RAG是一种将外部知识检索与大语言模型生成能力结合的技术，通过动态注入权威数据解决模型幻觉问题和时效性问题

核心价值：不予大语言模型实时访问更新知识的能力，突破训练数据的时间限制

基本技术原理

架构：用户提问 -> 检索模块查询向量化+数据库匹配 ->增强模块,构建提示词 ->生成模块:大语言生成答案

RAG三步走：

1. 数据预处理，构建向量索引库

1. 1. 知识库构建：收集整理网页文档、数据库等多元数据信息，构建外部知识库
   2. 文档切块：将文档切分为适合大小的片段，以便后续检索。分段策略需要在语义完整性与检索效率之间取得平衡
   3. 向量化处理：使用嵌入模型将文本转换为向量，并存储在向量数据库中

1. 检索：当用户提出问题时，系统从外部知识库中检索与相关的文本片段

1. 1. 向量检索：将问题和文档转换为向量，通过相似度计算，匹配语义相关内容
   2. 关键词检索：基于关键词匹配快速定位内容，如BM25算法
   3. 混合检索：结合向量和关键词检索，提高召回率和准确性

1. 生成：将检索到的文本与用户提问合并为提示词，输入大语言模型，生成自然语言答案。模型基于外部知识库，而非仅依赖训练数据生成内容回答，显著降低虚构信息的风险

RAG架构演进

Native RAG（基础RAG）

核心流程：索引构建->检索召回->直接对话

用途：适用于简单的问答

缺陷：

• 检索噪音干扰：语义相似，但主题无关的内容
• 信息碎片化：固定分块割裂长上下文
• 多跳推理失效：无法关联跨文档信息

典型应用场景：企业产品手册查询，基础聊天机器人

Advanced RAG（高级RAG）

优化策略：

• 检索前：查询改写（Query Rewriting），子查询分解（HyDE），源数据过滤
• 检索中：混合查询向量加关键字，图数据库增强
• 检索后：重排序，上下文压缩

用途：提升高精度场景的答案质量，如法律合同、医疗诊断支持

优势：解决基础的噪声问题，支持复杂语义对齐

Modular RAG（模块化RAG）

架构特性：解耦为可插拔模块，检索器，重排器，路由代理等

核心能力：

• 迭代检查：逐步优化结果
• 动态路由：将查询分配到最佳模块，如API，数据库

用途：企业及知识管理平台需整合多元异构数据（如金融风控系统）

Agentic RAG（智能体RAG）

技术分支：

• 单质能体路由：动态分配任务至专属模块
• 多智能体协同：并行调用多个工具，如数据库，搜索引擎

用途：超复杂任务处理，如全球新闻汇总，多轮跨工具推理

企业为什么需要RAG

传统语言模型的局限性推动了RAG发展：

• 知识滞后：模型训练后，无法获取新知识
• 幻觉风险：对未知问题可能造成看似合理，实则错误的答案
• 专业领域局限：缺乏特定专业领域知识

RAG的优势：

• 实时性：通过更新知识库获取最新信息
• 准确性：答案基于检索内容生成，减少幻觉
• 可解释性：答案追溯可致知识原文，方便验证
• 轻量化：无需重新训练大语言模型，成本更低

RAG快速上手

工具选型

• LangChain：提供预制RAG链，支持快速集成LLM与向量数据库
• LlamaIndex：专为知识索引优化简单化文档分块与嵌入流程
• Milvus：开源高性能向量数据库
• FIASS：轻量级向量搜索库
• Pinecone：云服务向量数据库

极简RAG

1. 数据准备

1. 1. 格式支持：pdf网页文本等
   2. 分块策略：按语意或固定长度切分，避免信息碎片化

1. 索引构建

1. 1. 嵌入模型：选用开源的模型或者领域专用模型
   2. 向量化：将文本分块转换为向量存入数据库

1. 减索优化

1. 1. 混合检索：结合关键词（BM25）与语义搜索（向量相似度），提升召回率
   2. 重排序：用小模型筛选topk相关片段

1. 生成集合

1. 1. 提示工程：设计模板引导大语言模型，融合检索相关内容
   2. 大语言模型选型：GPT，Ollama等

# 使用LangChain构建最小化RAG
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings

# 1. 加载知识库（PDF/TXT/CSV），改成自己的知识库
docs = load_documents("企业知识库.zip")  

# 2. 文本分块与向量化
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")
db = FAISS.from_documents(docs, embeddings)

# 3. 检索增强问答
retriever = db.as_retriever()
qa_chain = RetrievalQA.from_chain_type(llm=ChatGPT, retriever=retriever)
print(qa_chain.run("Q：我司产品X的技术参数？"))

第2节：学习准备工作

搭建Conda环境

地址：https://www.anaconda.com/download/succes

使用conda -v 查看版本号

创建环境

conda create -n rag-study python=3.10 -y

激活环境

conda activate myenv

退出环境

conda deactivate

查看环境

conda env list

阿里百炼API申请

官方地址：https://bailian.console.aliyun.com/

第3节：极简RAG搭建

相关包安装

pip install pypdf2    # 用途：PDF 文件基础操作（纯文本提取、页面级操作）  
pip install dashscope  # 用途：调用阿里云大模型 API（如通义千问） 
  # 用途：构建大模型（LLM）应用的开发框架  
pip install langchain-openai  # - `langchain-openai`：专用于 OpenAI 模型（如 GPT 系列）的适配器
pip install langchain-community  # - `langchain-community`：社区贡献的扩展工具（向量库/文件加载器等）
pip install faiss-cpu  # 用途：高效向量相似性搜索（CPU 优化版）

必要库类导入

#提供跨平台的操作系统交互接口，允许开发者执行文件管理、进程控制、环境配置等底层操作
import os
#专业日志管理核心工具，可以监控生产环境、调试、运行状态等
import logging
#Python中的序列化库，可以将Python对象序列化二进制字节流，保存到文件在中，同时可以反序列化还原对象
import pickle
#用于读取PDF文件内容，提取文本、元数据（作者/标题）、页面对象等基础信息
from PyPDF2 import PdfReader
#加载预定义的问答任务链，自动化实现“检索→生成”流程
from langchain.chains.question_answering import load_qa_chain
#调用OpenAI的GPT系列模型,实现文本生成、问答、摘要等任务
from langchain_openai import OpenAI, ChatOpenAI
#阿里云DashScope提供的嵌入模型，适合中文场景优化，功能类似OpenAI Embeddings
from langchain_community.embeddings import DashScopeEmbeddings
#监控OpenAI API调用开销（消耗的tokens数量、费用），调试性能瓶颈
from langchain_community.callbacks.manager import get_openai_callback
#将长文本按语义递归分割为小片段（如按段落、句子），适配语言模型的上下文长度限制
from langchain.text_splitter import RecursiveCharacterTextSplitter
#Facebook开源的向量数据库库，支持快速相似性搜索（最近邻检索）
from langchain_community.vectorstores import FAISS
#Python类型标注库，用于声明函数参数/返回值的类型（如List[str]表示字符串列表），提升代码可读性与静态检查支持
from typing import List, Tuple

核心代码

数据读取

#函数一：读取pdf文件，提取每行内容和每行对应的页码，同时拼接成一个大文本
def extract_text_with_page_numbers(pdf) -> Tuple[str,List[int]]:
    """
    从pdf中提取文本并记录每行文本对应的页码

    参数：
        pdf: PDF文件对象

    返回:
        text: 提取的文本内容
        page_numbers: 每行文本对应的页码列表
    """

    text = ""
    page_numbers = []

    #遍历PDF的每一页，enumerate从1开始计数页码
    for page_number,page in enumerate(pdf.pages, start=1):
        #提取当前页的文本
        extracted_text = page.extract_text()
        #如果该页面有文本
        if extracted_text:
            #将当前页文本追加到总文本中（字符串拼接文本）
            text += extracted_text
            #记录每行文本对应的页码（按换行符分隔）
            page_numbers.extend([page_numbers] * len(extracted_text.split("\n")))
        else:
            #处理无文本页面的情况
            logging.warning(f"NO TEXT FOUND ON page {page_number}.")
    return text ,page_numbers

向量化处理

#向量化处理函数，对识别到的PDF文件进行分割，将分割好的文本块构建成向量存储
def process_text_with_splitter(text: str, page_numbers: List[int], save_path: str = None):
    """
       处理文本并创建向量存储

       参数:
           text: 提取的文本内容
           page_numbers: 每行文本对应的页码列表
           save_path: 可选，保存向量数据库的路径

       返回:
           knowledgeBase: 基于FAISS的向量存储对象
       """
    #创建递归式文本分割器
    text_splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", "\t", "."," "], #按语义单位分割
        chunk_size=512,  #区块最大长度
        chunk_overlap= 128,   #区块间重叠长度
        length_function= len,   #长度计算函数
    )

    #文本分块处理
    chunks = text_splitter.split_text(text)
    #print(f"文本被分割成: {len(chunks)}个块。")

    #使用阿里云DashScope生成文本嵌入向量
    embeddings = DashScopeEmbeddings(model="text-embedding-v2")

    #构建FAISS向量数据库
    knowledgeBase = FAISS.from_texts(chunks,embeddings)
    #print("已从文本块创建知识库...")
    #建立区块-页码映射关系,enumerate(chunks)同时获取分块的索引 i和内容 chunk。
    #page_numbers[i]表示第 i个分块在源文档中的起始页码
    page_info = {chunk: page_numbers[i] for i, chunk in enumerate(chunks)}
    knowledgeBase.page_info = page_info   #附加元数据

    #持久化存储
    if save_path:
        #确保目录存在
        os.makedirs(save_path, exist_ok=True)
        # 保存向量索引
        knowledgeBase.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 knowledgeBase

加载向量化数据库

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-embeddings-v2")


    #加载FAISS向量库（需要启用反序列化安全选项）
    knowledgeBase = FAISS.load_local(
        load_path,
        embeddings,
        allow_dangerous_deserialization=True
    )

    #加载页码元数据
    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:
            knowledgeBase.page_info = pickle.load(f)

    return knowledgeBase

运行代码

#初始化PyPDF2读取器，这个pdf文件可以放到py文件同一个地址下面，这样就不要找文件地址了（真正开发的时候不要这样，还是要另起文件夹，专门放数据文件）
pdf_reader = PdfReader('大学生创新创业三年行动计划.pdf')

#执行文本提取load_knowledge_base
text, page_numbers = extract_text_with_page_numbers(pdf_reader)

#处理文本并创建知识库，同时保存到磁盘
save_dir = "vertor_db"
knowledgeBase = process_text_with_splitter(text, page_numbers, save_path=save_dir)

#处理文本并创建知识库
knowledgeBase =  process_text_with_splitter(text, page_numbers)
# print(knowledgeBase)

查询内容生成

#设置查询问题
query = "大学生可以领取什么补贴？可以领多少钱？"
if query:
    #执行相似度搜索，找到与查询相关的文档
    docs = knowledgeBase.similarity_search(query)

    #初始化对话大模型
    chatLLM = ChatOpenAI(
    #大模型中还有一些参数，如：temperature、max_tokens等这里没有配置，参数的具体用法可以看dashscope官网或者看这个网站：https://www.promptingguide.ai/zh/introduction/settings
        #若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
        api_key="你的key",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        model="deepseek-v3"
    )

    #加载问答链
    chain = load_qa_chain(chatLLM, chain_type="stuff")process_text_with_splitter

    #准备输入数据
    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"])