RAG初步实战：从 PDF 到问答：我的第一个轻量级 RAG 系统（附详细项目代码内容与说明）

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

项目背景与目标

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

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

✅ 学习目标

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

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

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

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

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

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

技术架构概览

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

模块拆解与组件说明

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

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

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

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

核心工具与模型选型

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

🧮 2. 向量数据库：FAISS

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

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

核心功能实现详解

📄 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

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 os

class 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 os

def 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", "。", "！", "？", ".", " ", ""]
    )

    # 对每一页内容进行切割，保留 metadata
    documents = 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 np

import pickle
from typing import List, Tuple
from embedding import EmbeddingModel
from langchain_core.documents import Document

class VectorStore:
    def __init__(self, dim: int = 512, db_path: str = "faiss_index"):
        self.dim = dim
        self.db_path = db_path
        self.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 np
    from 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，支持多轮对话式语义问答。