LangChain RAG PDF 问答 Demo

一、任务目标

本示例实现一个基于 RAG（检索增强生成）的 PDF 问答应用：用户先导入一份 PDF，再针对文档内容提问，系统会先检索与问题相关的片段，再结合这些片段由大模型生成答案，而不是仅靠模型自身知识回答。

核心能力：文档入库 → 按问题检索相关段落 → 将检索结果与问题一起交给大模型 → 返回基于文档的答案。

---

二、技术栈与依赖

模块	作用
Chroma	向量数据库，存储文档块对应的向量，支持相似度检索
ChatZhipuAI	大模型接口（智谱），用于根据「上下文 + 问题」生成回答
FastEmbedEmbeddings	文本嵌入模型，将文本转为向量供检索
PyPDFLoader	加载 PDF，解析为文本
RecursiveCharacterTextSplitter	将长文本按块切分，便于检索与上下文限制
PromptTemplate / StrOutputParser	提示模板与输出解析，组成 LCEL 调用链

---

三、代码结构概览

• 类：ChatPDF，封装「入库 + 检索 + 生成」全流程。
• 主要方法：
  • __init__：初始化大模型、文本分割器、RAG 提示模板。
  • ingest(pdf_file_path)：加载 PDF → 切块 → 写入向量库 → 构建检索器与调用链。
  • ask(query)：对当前已入库的 PDF 进行提问，走 RAG 链得到答案。
  • clear()：清空向量库、检索器与链，便于重新加载文档。

---

四、逻辑说明

4.1 初始化 __init__

• 大模型：使用智谱 glm-4-flash。API Key 建议通过环境变量或配置文件传入，不要写死在代码中。
• 文本分割：块长 1024 字符，块间重叠 100 字符，在保留上下文与控制单块长度之间做平衡。
• 提示模板：采用「系统角色 + 用户问题 + 检索上下文」的结构，要求模型严格依据上下文作答、不知道则明确说明、回答简洁（如最多三句话）。

def __init__(self):
    # 大模型：api_key 建议用 os.environ.get("ZHIPU_API_KEY") 等从环境变量读取
    self.model = ChatZhipuAI(model="glm-4-flash", api_key=None)
    # 文本分割：块长 1024，重叠 100
    self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)
    # 提示模板：系统角色 + 用户问题 + 检索上下文
    self.prompt = PromptTemplate.from_template(
        """<|system|>
你是专业的问答助手，需严格根据提供的检索上下文回答问题，未知内容直接回复不知道，答案最多三句话且保持简洁。
<|user|>
问题：{question}
上下文：{context}
<|assistant|>
"""
    )

4.2 文档入库 ingest(pdf_file_path)

1. 用 PyPDFLoader 加载指定 PDF，得到页面级文档列表。
2. 用 RecursiveCharacterTextSplitter 将文档切分为多个块（chunks）。
3. 使用 filter_complex_metadata 过滤掉不利于向量化的复杂元数据。
4. 用 FastEmbedEmbeddings 为每个块生成向量，并写入 Chroma 向量库。
5. 基于向量库创建 retriever：
   • search_type="similarity_score_threshold"：按相似度阈值检索；
   • k=3：最多取 3 个相关块；
   • score_threshold=0.5：仅保留相似度 ≥ 0.5 的块。
6. 构建 RAG 链：输入为用户问题；context 由 retriever 根据问题检索得到；question 原样传入；再经 prompt → 大模型 → StrOutputParser() 得到最终字符串答案。

def ingest(self, pdf_file_path: str):
    # 1. 加载 PDF，得到页面级文档列表
    docs = PyPDFLoader(file_path=pdf_file_path).load()
    # 2. 切分为多个块（chunks）
    chunks = self.text_splitter.split_documents(docs)
    # 3. 过滤复杂元数据
    chunks = filter_complex_metadata(chunks)
    # 4. 生成向量并写入 Chroma
    vector_store = Chroma.from_documents(documents=chunks, embedding=FastEmbedEmbeddings())
    # 5. 创建 retriever：相似度阈值检索，k=3，score_threshold=0.5
    self.retriever = vector_store.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={"k": 3, "score_threshold": 0.5},
    )
    # 6. 构建 RAG 链：context 来自 retriever，question 原样传入
    self.chain = (
        {"context": self.retriever, "question": RunnablePassthrough()}
        | self.prompt
        | self.model
        | StrOutputParser()
    )

4.3 提问 ask(query)

• 若尚未调用 ingest（即 self.chain 为空），则提示「请先添加 PDF 文件」。
• 否则对 query 调用 self.chain.invoke(query)，链内部会先检索再生成，返回基于当前 PDF 的答案。

def ask(self, query: str):
    if not self.chain:
        return "请先添加PDF文件"
    return self.chain.invoke(query)

4.4 清空 clear()

• 将向量库、检索器、链引用置为 None，便于后续重新加载另一份 PDF 或重置状态。

def clear(self):
    self.vector_store = None
    self.retriever = None
    self.chain = None

---

五、RAG 数据流简述

用户问题 query
    ↓
retriever.invoke(query)  →  从 Chroma 中按相似度取 top-k 文档块（如 k=3）
    ↓
Prompt 填入：context = 检索到的文本，question = query
    ↓
大模型（ChatZhipuAI）根据 context + question 生成回答
    ↓
StrOutputParser() 得到字符串 → 返回给用户

即：检索（Retrieval）→ 用检索结果增强提示（Augmented）→ 大模型生成（Generation），构成完整 RAG 流程。

---

六、使用方式示例

if __name__ == "__main__":
    chatpdf = ChatPDF()
    # 先入库一份 PDF（路径按实际替换）
    chatpdf.ingest(r"你的PDF路径.pdf")
    # 针对该 PDF 内容提问
    answer = chatpdf.ask("请简要介绍这份文档的主要内容")
    print(answer)

注意：运行前需配置大模型 API Key（如智谱开放平台申请），并通过环境变量或 ChatZhipuAI(..., api_key=...) 传入，避免在代码或文档中明文出现密钥。

---

七、小结

项目	说明
任务	基于 RAG 的 PDF 问答：先检索再生成，答案依托指定文档
关键技术	向量库（Chroma）+ 嵌入（FastEmbed）+ 检索器 + 大模型（智谱）+ LCEL 链
使用顺序	先 ingest( pdf 路径 )，再 ask( 问题 )；可调用 clear() 后重新 ingest 其他 PDF

完整代码

from langchain_community.vectorstores import Chroma
from langchain_community.chat_models import ChatZhipuAI
from langchain_community.embeddings import FastEmbedEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_community.vectorstores.utils import filter_complex_metadata

# 可选：导入环境变量模块，推荐用环境变量存储密钥，更安全
# import os

class ChatPDF:
    vector_store = None
    retriever = None
    chain = None
    
    def __init__(self):
        self.model = ChatZhipuAI(model="glm-4-flash", api_key="YOUR_ZHIPU_API_KEY_HERE")
        self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=100)
        self.prompt = PromptTemplate.from_template(
			"""<|system|>
你是专业的问答助手，需严格根据提供的检索上下文回答问题，未知内容直接回复不知道，答案最多三句话且保持简洁。
<|user|>
问题：{question}
上下文：{context}
<|assistant|>
"""
		)
    
    def ingest(self, pdf_file_path : str):
        docs = PyPDFLoader(file_path=pdf_file_path).load()
        chunks = self.text_splitter.split_documents(docs)
        chunks = filter_complex_metadata(chunks)
        
        vector_store = Chroma.from_documents(documents = chunks, embedding = FastEmbedEmbeddings())
        self.retriever = vector_store.as_retriever(
			search_type = "similarity_score_threshold",
			search_kwargs = {
				"k" : 3,
				"score_threshold" : 0.5,
			},
		)
        
        self.chain = ({
			"context" : self.retriever,
            "question" : RunnablePassthrough()
		} | self.prompt | self.model | StrOutputParser())
    
    def ask(self, query : str):
        if not self.chain:
            return "请先添加PDF文件"
        
        return self.chain.invoke(query)
    
    def clear(self):
        self.vector_store = None
        self.retriever = None
        self.chain = None

if __name__ == "__main__":
    chatpdf = ChatPDF()
    chatpdf.ingest(r"F:\Documents\项目大纲-AI就业班.pdf")
    answer = chatpdf.ask("请介绍一下这份AI就业班的项目大纲")
    print(answer)