✅ 理解 RAG 的完整流水线流程 ✅ 掌握 LangChain 中常用的文档加载器 ✅ 能独立实现「文档加载 → 文本分块 → 向量化 → 检索问答」的完整流程 ✅ 写出可运行的基础 RAG 代码

---

一、RAG 流水线核心概念

1. 什么是 RAG？

检索增强生成（Retrieval-Augmented Generation）

• 作用：让大模型基于私有文档 / 本地知识库回答问题，解决 “模型不知道的内容”
• 核心流程：
  1. 文档加载：从文件 / 网页 / 数据库读取数据
  2. 文本分块：把长文档切分成小片段（Chunk）
  3. 向量化：把文本片段转成向量存入向量库
  4. 用户提问：问题也转为向量
  5. 检索召回：从向量库找到和问题最相似的文档片段
  6. 生成答案：把问题 + 召回片段一起交给大模型，生成精准回答

---

二、文档加载器精讲（核心考点）

LangChain 提供了多种加载器，适配不同数据源：

表格

加载器	适用场景	代码示例
TextLoader	纯文本文件（.txt）	from langchain.document_loaders import TextLoader
PyPDFLoader	PDF 文档	from langchain.document_loaders import PyPDFLoader
CSVLoader	CSV 表格文件	from langchain.document_loaders import CSVLoader
WebBaseLoader	网页内容爬取	from langchain.document_loaders import WebBaseLoader
DirectoryLoader	批量加载文件夹内文件	from langchain.document_loaders import DirectoryLoader

代码示例：加载本地 PDF 文件

python

运行

from langchain.document_loaders import PyPDFLoader

# 1. 初始化加载器
loader = PyPDFLoader("./your_document.pdf")

# 2. 加载文档
documents = loader.load()

# 3. 查看加载结果
print(f"加载了 {len(documents)} 页文档")
print("第一页内容：", documents[0].page_content[:100])

代码示例：加载网页内容

python

运行

from langchain.document_loaders import WebBaseLoader

# 加载指定网页
loader = WebBaseLoader("https://example.com/article")
documents = loader.load()

print("网页内容：", documents[0].page_content[:200])

---

三、文本分块（Chunking）

文档加载后，需要切成小片段才能存入向量库，常用工具：

• RecursiveCharacterTextSplitter：按字符递归切割，适配大多数文本

python

运行

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 初始化切割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 每个片段的最大字符数
    chunk_overlap=50  # 片段之间的重叠字符数（防止上下文断裂）
)

# 切割文档
splits = text_splitter.split_documents(documents)

print(f"切割成了 {len(splits)} 个片段")
print("第一个片段：", splits[0].page_content)

---

四、向量化与向量库

把文本片段转成向量，存入向量库，用于后续检索：

python

运行

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# 初始化嵌入模型（文本转向量）
embeddings = OpenAIEmbeddings()

# 存入 FAISS 向量库
vector_store = FAISS.from_documents(splits, embeddings)

# 转为检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 3})  # 每次检索返回3个最相关片段

---

五、完整 RAG 问答链实现

python

运行

from langchain.chains import RetrievalQA
from langchain_openai import OpenAI

# 初始化模型
llm = OpenAI(temperature=0)

# 构建 RAG 问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 把所有召回片段直接拼接到prompt中
    retriever=retriever
)

# 提问
question = "这份文档的核心结论是什么？"
result = qa_chain.run(question)
print("回答：", result)

三种 chain_type （stuff /map_reduce/refine）

1. chain_type="stuff"（最基础、默认）

• 逻辑：把所有检索到的片段全部拼在一起，做成一条长提示词，一次性发给大模型
• 优点：代码简单、只调用 1 次大模型、速度快
• 缺点：文档多的时候，提示词超长，容易超出模型上下文限制
• 适用：片段少、短文档（日常测试、小型知识库）

2. chain_type="map_reduce"（分片汇总）

• 逻辑分两步：
  1. Map：每条文档片段单独丢给大模型，各自生成摘要 / 答案
  2. Reduce：把所有片段的结果再汇总一次，生成最终答案
• 优点：支持超多文档，不怕超长上下文
• 缺点：调用多次大模型，速度慢、消耗 token 多
• 适用：海量长文档检索

3. chain_type="refine"（逐段迭代优化）

• 逻辑：先拿第一条片段生成答案，再用第二条片段修正、优化上一轮答案，逐条迭代
• 优点：答案连贯性最好、细节最丰富
• 缺点：调用次数最多、速度最慢
• 适用：对答案精度要求极高的场景

---

六、课后作业（必做）

1. 用 PyPDFLoader 加载一个本地 PDF 文件
2. 用 RecursiveCharacterTextSplitter 把文档切成片段
3. 存入 FAISS 向量库，构建一个基础 RAG 问答链
4. 简答题：简述 RAG 流水线的完整流程（从文档加载到生成答案）

5. 作业 1-3：加载 PDF + 分块 + 构建 RAG 链

   这三题是连贯的，写在同一个文件里，

   # --- 1. 导入所有需要的库 ---
   from langchain.document_loaders import PyPDFLoader
   from langchain.text_splitter import RecursiveCharacterTextSplitter
   from langchain_openai import OpenAIEmbeddings, OpenAI
   from langchain_community.vectorstores import FAISS
   from langchain.chains import RetrievalQA
   
   # --- 2. 配置API Key（换成你自己的）---
   import os
   os.environ["OPENAI_API_KEY"] = "你的API Key"
   os.environ["OPENAI_API_BASE"] = "你的代理地址（如果需要）"
   
   # --- 3. 加载PDF文档 ---
   # 把下面的路径换成你自己的PDF文件路径
   loader = PyPDFLoader("./your_document.pdf")
   documents = loader.load()
   
   print(f"✅ 加载完成，共 {len(documents)} 页")
   
   # --- 4. 文本分块 ---
   text_splitter = RecursiveCharacterTextSplitter(
       chunk_size=500,      # 每个片段最大字符数
       chunk_overlap=50     # 片段之间的重叠字符数
   )
   splits = text_splitter.split_documents(documents)
   
   print(f"✅ 分块完成，共 {len(splits)} 个片段")
   
   # --- 5. 向量化并存入向量库 ---
   embeddings = OpenAIEmbeddings()
   vector_store = FAISS.from_documents(splits, embeddings)
   retriever = vector_store.as_retriever(search_kwargs={"k": 3})
   
   print("✅ 向量库构建完成")
   
   # --- 6. 构建RAG问答链 ---
   llm = OpenAI(temperature=0)
   qa_chain = RetrievalQA.from_chain_type(
       llm=llm,
       chain_type="stuff",  # 把召回的片段直接拼接到prompt中
       retriever=retriever
   )
   
   print("✅ RAG问答链准备就绪")
   
   # --- 7. 测试提问 ---
   question = "这份文档的核心内容是什么？"
   result = qa_chain.run(question)
   print(f"\n问题：{question}")
   print(f"回答：{result}")
   

   ---

   作业 4：简答题标准答案

   题目：简述 RAG 流水线的完整流程（从文档加载到生成答案）

   RAG 流水线的完整流程分为两个阶段：

6. 离线构建阶段（只做一次）

   • 文档加载：使用 PyPDFLoader 等加载器读取 PDF、网页、TXT 等数据源。
   • 文本分块：使用 RecursiveCharacterTextSplitter 将长文档切割成短片段（Chunk）。
   • 向量化：通过 OpenAIEmbeddings 等嵌入模型，将文本片段转为向量。
   • 存入向量库：将向量存入 FAISS、Chroma 等向量数据库，用于后续检索。

7. 在线问答阶段（用户提问时执行）

   • 用户提问：用户输入问题。
   • 问题向量化：将用户的问题也转为向量。
   • 语义检索：从向量库中找到与问题向量最相似的文档片段。
   • 拼接上下文：将问题和召回的相关片段拼接成完整 Prompt。
   • 生成答案：将 Prompt 交给大模型，生成基于文档内容的回答。