RAG初步实战：从 PDF 到问答：我的第一个轻量级 RAG 系统

项目背景与目标

在大模型逐渐普及的今天，Retrieval-Augmented Generation（RAG，检索增强生成）作为连接“知识库”和“大语言模型”的核心范式，为我们提供了一个高效、实用的路径。为了快速学习RAG的原理，并掌握它的使用方法，我在这开发了一个pdf问答项目

这个项目的初衷，就是以“本地知识问答系统的动手实践”为目标，系统学习并串联以下几个关键知识模块：

✅ 学习目标

模块	学习重点
文档解析	如何从 PDF 文档中提取结构化文本，并保留元数据（如页码）
文本向量化	使用中文 embedding 模型，将自然语言转为向量表征
向量存储与检索	搭建本地 FAISS 向量数据库，掌握向量的存储、检索与匹配机制
前端交互	使用 Streamlit 构建简单前端，实现交互式问答体验

📌 项目特色
无需翻墙、全本地运行：选用了国内可用的向量化模型和 API 服务，便于部署。

完整链路闭环：从 PDF → Chunk → 向量化 → 检索 → 语言模型生成，一步不落。

结构清晰、易于拓展：代码结构模块化，方便后续更换模型、接入多个文档等。

🧩 项目适合人群
想学习 RAG 工作流程的开发者或学生

需要本地构建问答系统但不方便翻墙的用户

想搭建个人知识库搜索问答助手的 AI 学习者

技术架构概览

   ↓ 文本切分
📜 Chunk 文本 + 元信息↓ 向量化 Embedding（中文模型）
🔍 FAISS 向量数据库↓ 向量相似度匹配（Top-K）
🔁 提取匹配段落（内容 + 元数据）↓ 拼接上下文 Prompt
🧠 百炼智能体 API（对话生成）↓
🧾 用户界面展示（Streamlit）

模块拆解与组件说明

模块	使用组件	功能说明
文档解析	PyMuPDF（或 fitz）	从 PDF 中按页读取文本，并保留页码信息等元数据
文本切分	LangChain	将每页文本按段落或长度切分为小块，提高语义粒度
向量化	bge-small-zh 模型（HuggingFace）	将 Chunk 文本转为 512 维向量，用于语义匹配
向量存储	FAISS 本地向量库	构建并保存索引，实现快速相似度搜索
前端交互	Streamlit	提供简单直观的问答界面，支持用户输入与响应展示

模块之间的关系
向量数据库只保存向量 + 元数据，不保存完整语义；

每次用户输入时，实时提取向量、进行检索并构造上下文 Prompt；

构造后的 Prompt 被送入百炼智能体模型进行回复生成；

Streamlit 前端实时展示问答结果，形成闭环。

核心工具与模型选型

在本项目中，为了实现从 PDF 文档中提取段落，并基于语义进行匹配和问答的功能，选用了以下核心工具链与模型组件，确保系统具有较高的效率、准确性以及良好的可扩展性。
🧠 1. Embedding 模型：BAAI/bge-small-zh

项目	内容
模型名称	BAAI/bge-small-zh
模型来源	HuggingFace @ BAAI
是否开源	✅ 是
支持语言	中文（优化）
部署方式	本地部署（无需联网，免翻墙）
模型体积	小型（约 120MB）
向量维度	512
优势亮点	轻量、高速、适配中文语义匹配任务
调用方式	封装在 embedding.py 文件中，定义了 EmbeddingModel.embed_texts() 接口用于段落向量化处理

🧮 2. 向量数据库：FAISS

项目	内容
名称	FAISS（Facebook AI Similarity Search）
作用	存储并检索高维向量（用于语义匹配）
部署方式	本地离线，使用 .index 和 .meta 文件存储数据
使用方式	在 vector_store.py 中封装了 VectorStore 类，实现： • add() 向数据库添加向量与原始文本 • save()/load() 存储与加载 • search() 执行相似度检索
匹配方式	L2 距离（欧氏距离）索引器 IndexFlatL2

📚 3. 文本切分工具：LangChain TextSplitter

项目	内容
工具模块	RecursiveCharacterTextSplitter
来源	LangChain
作用	将原始 PDF 文档内容切分成多个适配 embedding 的小段（chunk）
分段策略	使用换行符、标点符号等多级分隔符，避免语义断裂
使用方式	在 rag_chain.py 的 load_and_split_pdf() 中使用

🖼 5. 可视化界面框架：Streamlit

项目	内容
框架名称	Streamlit
用途	构建简洁交互式 Web 应用界面
使用方式	主入口文件 app.py，支持用户输入问题、展示检索段落与大模型生成回复

核心功能实现详解

📄 1. PDF 文档加载与切分
文件：rag_chain.py
函数：load_and_split_pdf()

✅ 实现目标：
将整本 PDF 文档切分为可用于语义匹配的文本段（chunk），避免段落过长或断句不清导致 embedding 表达质量下降。

🔍 2. 文本向量化（Embedding）
文件：embedding.py
类名：EmbeddingModel

✅ 实现目标：
将每段文本转为稠密语义向量（float32），便于后续进行语义匹配检索。

✅ 模型选型：
使用 HuggingFace 上的 BAAI/bge-small-zh 本地模型，支持中文语义精度较高。

🧠 3. 向量数据库构建与检索
文件：vector_store.py
类名：VectorStore

✅ 实现目标：
将文本对应的向量存入 FAISS 数据库；
支持向量相似度检索，返回与用户 query 最相近的段落及其元信息。

✅ 数据结构：
.index 文件：存储 FAISS 索引（支持快速相似度查询）
.meta 文件：存储 chunk 原文与元数据（如页码）

✅ 核心方法：
add(texts, vectors, metadatas)
save() / load()
search(query, top_k)

🤖 4. 问答生成：接入百炼智能体 API
文件：baichuan_llm.py
类名：DashScopeChatBot

✅ 实现目标：
组合用户问题与匹配段落，通过大模型生成符合上下文的回答。

项目目录结构说明

rag_demo/
├── app.py                  # 主入口，基于 Streamlit 的问答交互界面
├── embedding.py            # 文本向量化模块，封装 BGE-small-zh 模型
├── vector_store.py         # 自定义向量数据库类，基于 FAISS 实现
├── rag_chain.py            # 文档加载与切分工具，支持 PDF 预处理
├── baichuan_llm.py         # 调用百炼智能体（Bailian）生成问答内容
├── docs/                   # 存放用户上传或处理的 PDF 文档
│   └── example.pdf         # 示例 PDF 文件
├── faiss_index.index       # FAISS 索引文件（自动生成，保存向量索引）
├── faiss_index.meta        # FAISS 元信息文件（保存每段文本及页码）
└── README.md               # 项目说明文档

📌 各模块说明

文件 / 目录	类型	作用描述
app.py	主程序	启动 Streamlit 应用，支持用户输入与问答
embedding.py	模型封装	加载本地 BAAI/bge-small-zh 模型，并执行文本向量化
vector_store.py	数据管理	构建、查询、保存 FAISS 向量数据库
rag_chain.py	工具模块	加载 PDF 并使用智能分段切割为 chunk
baichuan_llm.py	模型调用	调用百炼智能体 API，生成基于文档内容的回答
docs/	文档目录	存放所有待处理的 PDF 文件
faiss_index.index	索引数据	FAISS 保存的向量索引二进制文件
faiss_index.meta	元信息	存储每段文本的原文及其元数据（如页码）
README.md	文档	项目的功能与使用说明

主要文件内容
app.py

import streamlit as st
from vector_store import VectorStore
import os# 设置页面标题
st.set_page_config(page_title="RAG 问答助手", layout="wide")# 标题
st.title("📄 PDF 语义搜索助手")
st.markdown("使用 FAISS + bge-small-zh 向量模型，实现 PDF 文档语义检索")# 加载向量库
@st.cache_resource
def load_vector_store():store = VectorStore()store.load()return store# 主入口
def main():store = load_vector_store()# 用户输入user_query = st.text_input("🔍 请输入你的问题：", placeholder="例如：番茄叶片检测方法有哪些？")# 查询结果if user_query:results = store.search(user_query, top_k=5)st.subheader("🔎 匹配结果")for i, (text, meta, score) in enumerate(results):with st.expander(f"结果 {i+1} ｜页面：{meta.get('page_label', '未知')} ｜得分：{score:.4f}"):st.write(text)# 运行主程序
if __name__ == "__main__":main()

embedding.py

from sentence_transformers import SentenceTransformer
from typing import List
import osclass EmbeddingModel:def __init__(self, model_name: str = "BAAI/bge-small-zh"):print("✅ 正在加载 embedding 模型，请稍候...")self.model = SentenceTransformer(model_name)print("✅ 模型加载完成：", model_name)def embed_texts(self, texts: List[str]) -> List[List[float]]:# 进行批量编码（text → vector）embeddings = self.model.encode(texts, show_progress_bar=True)return embeddings# ✅ 示例调用代码
if __name__ == "__main__":# 示例文本sample_texts = ["人工智能正在改变世界。","番茄叶片病虫害检测方法研究","LangChain 是一个强大的 RAG 框架。"]# 实例化模型embedder = EmbeddingModel()# 获取向量vectors = embedder.embed_texts(sample_texts)for i, vec in enumerate(vectors):print(f"\n🔹 文本 {i + 1} 的向量维度: {len(vec)}，前 5 维预览: {vec[:5]}")

rag_chain.py

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import osdef load_and_split_pdf(pdf_path: str, chunk_size=500, chunk_overlap=50):# 加载 PDF 文档（每一页为一个 Document）loader = PyPDFLoader(pdf_path)pages = loader.load()# 使用递归切割器分段（可按字符长度+换行符智能分段）splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap,separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""])# 对每一页内容进行切割，保留 metadatadocuments = splitter.split_documents(pages)return documents# ✅ 测试运行入口
if __name__ == "__main__":test_pdf_path = r"D:\Desktop\AI\rag_demo\docs\RT-TLTR番茄叶片病虫害检测方法研究_胡成峰.pdf"  # 请确保路径正确if not os.path.exists(test_pdf_path):print(f"❌ 文件未找到：{test_pdf_path}")else:chunks = load_and_split_pdf(test_pdf_path)print(f"✅ 共切分出 {len(chunks)} 个段落\n")# 打印前 3 个 chunk 的内容与元信息for i, chunk in enumerate(chunks[:3]):print(f"🔹 Chunk {i + 1}")print("内容片段：", chunk.page_content[:200].replace("\n", " ") + "...")print("元信息：", chunk.metadata)print("-" * 60)

vector_store.py

import faiss
import os
import numpy as npimport pickle
from typing import List, Tuple
from embedding import EmbeddingModel
from langchain_core.documents import Documentclass VectorStore:def __init__(self, dim: int = 512, db_path: str = "faiss_index"):self.dim = dimself.db_path = db_pathself.index = faiss.IndexFlatL2(dim)  # L2 距离索引器self.texts = []      # 存储 chunk 原文self.metadatas = []  # 存储 chunk 的元信息def add(self, texts: List[str], vectors: List[List[float]], metadatas: List[dict]):self.index.add(np.array(vectors).astype("float32"))self.texts.extend(texts)self.metadatas.extend(metadatas)def save(self):faiss.write_index(self.index, f"{self.db_path}.index")with open(f"{self.db_path}.meta", "wb") as f:pickle.dump({"texts": self.texts, "metadatas": self.metadatas}, f)print(f"✅ 向量数据库已保存到：{self.db_path}.index / .meta")def load(self):self.index = faiss.read_index(f"{self.db_path}.index")with open(f"{self.db_path}.meta", "rb") as f:meta = pickle.load(f)self.texts = meta["texts"]self.metadatas = meta["metadatas"]print("✅ 向量数据库已加载")def search(self, query: str, top_k: int = 3) -> List[Tuple[str, dict, float]]:embedder = EmbeddingModel()query_vec = embedder.embed_texts([query])[0]D, I = self.index.search(np.array([query_vec]).astype("float32"), top_k)results = []for idx, dist in zip(I[0], D[0]):results.append((self.texts[idx], self.metadatas[idx], dist))return results# ✅ 测试入口
if __name__ == "__main__":import numpy as npfrom rag_chain import load_and_split_pdf# 1. 加载 PDF 并切分docs: List[Document] = load_and_split_pdf(r"D:\Desktop\AI\rag_demo\docs\RT-TLTR番茄叶片病虫害检测方法研究_胡成峰.pdf")texts = [doc.page_content for doc in docs]metadatas = [doc.metadata for doc in docs]# 2. 向量化embedder = EmbeddingModel()vectors = embedder.embed_texts(texts)# 3. 存入 FAISS 向量库store = VectorStore()store.add(texts, vectors, metadatas)store.save()# 4. 进行语义搜索store.load()results = store.search("番茄叶片检测方法")for i, (txt, meta, score) in enumerate(results):print(f"\n🔍 匹配结果 {i + 1}：")print("得分：", score)print("页面：", meta.get("page_label", "N/A"))print("内容片段：", txt[:200], "...")

查询算法解析：基于 FAISS 的 L2 距离向量检索

query_vec = embedder.embed_texts([query])[0]

功能：将用户输入的自然语言 query 通过 embedding 模型转化为一个向量（query_vec）。

底层模型：你使用的是 BAAI/bge-small-zh，输出的是一个 512 维的向量。

D, I = self.index.search(np.array([query_vec]).astype("float32"), top_k)

功能：调用 FAISS.IndexFlatL2 的 search() 方法，返回：

D：每个候选结果与 query 向量之间的 L2 距离（欧氏距离平方）；

I：每个距离对应的原始文本在库中的索引位置。

底层算法：暴力遍历全部向量，通过 欧几里得距离（L2） 找出最近的 top_k 个向量

项目成果演示

📌 项目成果演示
本项目最终实现了一个可交互的 PDF 文档语义问答系统，集成了 Streamlit 页面、FAISS 向量数据库、中文 embedding 模型（bge-small-zh）和文档切分等组件，具备了完整的 RAG（Retrieval-Augmented Generation）基础架构。以下为演示亮点：

总结与思考

📌 项目总结
本项目以“从0搭建一个轻量级 RAG（Retrieval-Augmented Generation）语义搜索原型系统”为目标，围绕 LangChain 的文档处理工具链，结合 HuggingFace 本地向量化模型 BAAI/bge-small-zh 和高效的向量数据库 FAISS，完成了一个完整闭环的流程：

✅ 从 PDF 文档中提取文本并进行智能切分；
✅ 使用本地 embedding 模型对文本块进行向量化；
✅ 构建 FAISS 本地向量数据库，实现高效查询；
✅ 使用 Streamlit 实现简单而直观的交互页面；
✅ 支持全中文处理，部署门槛低、响应速度快、成本接近为零。

该项目在结构上清晰、功能上实用，非常适合初学者上手 RAG 系统构建，同时也具备进一步拓展 LLM 调用、问答生成、多文档多模态处理等能力的基础。

💡 学习思考
向量检索系统构建并不复杂，但细节决定效果：
文本的切分策略对匹配质量有很大影响；
向量化模型选择直接决定语义召回质量；
搜索算法（如使用 FAISS 的 L2 距离）虽简单，但结果解释性差，需合理展示。
本地部署是理解 RAG 的最好方式：
避免过度依赖 API，提升系统理解力；
便于调试 embedding 模型、切分器、检索逻辑等各个组件；
有助于构建对 embedding 语义空间的直觉认识。
国产 embedding 模型正在崛起：

像 bge-small-zh、acge_text_embedding、FlagEmbedding 等模型已经在中文语义匹配上达到非常好的效果；

对于轻量级任务，small 版本完全够用，延迟低、精度可接受。

向量数据库并不是黑盒：

像 FAISS 支持查看索引结构、手动添加/查询/删除向量；

元数据（如页面编号）可以与结果绑定，极大增强可解释性。

🚀 后续方向
集成 LLM，构建基于召回结果的回答生成（即真正的 RAG 问答）；
支持多文档、多格式（txt/doc/html）的语义索引；
引入关键词过滤、正则抽取等补充匹配方式；
调整 UI，支持多轮对话式语义问答。