实战：构建 RAG 知识库问答系统

文档加载 -> 分块 -> Embedding -> 向量存储 -> 检索 -> LLM 生成 -> 引用溯源 全流程实战

项目概述

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与大语言模型相结合的架构，能够有效解决 LLM 知识滞后、幻觉等问题。本项目实现一个完整的 RAG 知识库问答系统，支持 PDF、Markdown、HTML 等多格式文档的导入与问答。

功能目标

• 支持多种格式文档的加载与解析
• 对文档进行合理分块，保留上下文连贯性
• 将文本块转换为向量并存入向量数据库
• 基于语义相似度检索相关文档片段
• 结合 LLM 生成带有引用来源的准确回答
• 支持引用溯源，标注答案来源的文档和页码

技术栈

组件	选型
文档加载	LangChain DocumentLoader
文本分块	RecursiveCharacterTextSplitter
Embedding	OpenAI Embeddings / HuggingFace Embeddings
向量数据库	Chroma
检索策略	相似度检索 + MMR
LLM	OpenAI GPT / 本地模型
框架	LangChain

系统架构

用户提问
    |
    v
[Embedding 向量化] --> [语义检索] --> [Chroma 向量库]
    |                       |
    |                       v
    |               [检索结果 + 来源元数据]
    |                       |
    v                       v
[Prompt 组装] --> [LLM 生成回答] --> [返回带引用的答案]

---

环境准备

首先安装所需的依赖包：

pip install langchain langchain-community langchain-openai chromadb openai pypdf tiktoken

各包说明：

• langchain：核心框架，提供文档加载、分块、链式调用等能力
• langchain-community：社区维护的第三方集成
• langchain-openai：OpenAI Embeddings 和 LLM 的 LangChain 封装
• chromadb：Chroma 向量数据库
• pypdf：PDF 解析引擎
• tiktoken：OpenAI 分词器，用于准确计算 token 数量

---

Step 1：文档加载

LangChain 提供统一的 DocumentLoader 接口，屏蔽了不同格式文档的解析差异。每个 Loader 将文档解析为 Document 对象，包含 page_content（文本内容）和 metadata（来源、页码等元数据）。

支持的格式

from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader, UnstructuredHTMLLoader

# PDF 加载
pdf_loader = PyPDFLoader("knowledge_base/产品手册.pdf")
pdf_docs = pdf_loader.load()  # 每页一个 Document

# Markdown 加载
md_loader = UnstructuredMarkdownLoader("knowledge_base/技术文档.md")
md_docs = md_loader.load()

# HTML 加载
html_loader = UnstructuredHTMLLoader("knowledge_base/帮助中心.html")
html_docs = html_loader.load()

PDF 加载示例

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("data/产品手册.pdf")
documents = loader.load()

print(f"共加载 {len(documents)} 页")
for doc in documents[:2]:
    print(f"页码: {doc.metadata.get('page', 'N/A')}")
    print(f"内容预览: {doc.page_content[:100]}...")

PyPDFLoader 按页分割 PDF，每页生成一个独立的 Document，并在 metadata 中记录 source 和 page 信息，为后续引用溯源奠定基础。

---

Step 2：文档分块

LLM 的上下文窗口有限，且过长的文档会影响检索精度，因此需要将文档切割成适当大小的块。RecursiveCharacterTextSplitter 是 LangChain 推荐的通用分块器，它按优先级依次尝试不同的分隔符（\n\n -> \n -> 。 -> 空格），尽可能保持语义完整性。

分块配置

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,          # 每块最大字符数
    chunk_overlap=128,       # 相邻块重叠字符数
    length_function=len,     # 长度计算方式
    separators=["\n\n", "\n", "。", ".", " ", ""],
)

chunks = text_splitter.split_documents(documents)
print(f"分块后共 {len(chunks)} 个文本块")

chunk_overlap 参数非常关键，它让相邻块之间存在部分重叠，避免因分割边界导致重要信息断裂。例如，一个段落在第一块末尾被截断，其剩余部分会在下一块的开头出现。

---

Step 3：Embedding 向量化

Embedding 模型将文本转换为高维向量，使语义相近的文本在向量空间中彼此靠近。这里提供两种方案：

使用 OpenAI Embeddings

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",  # 性价比最高的模型
    openai_api_key="sk-your-api-key",
)

使用 HuggingFace Embeddings（本地部署）

from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",  # 中文场景推荐
    encode_kwargs={"normalize_embeddings": True},  # 归一化便于余弦相似度计算
)

OpenAI Embeddings 准确度高但需 API 调用；HuggingFace Embeddings 可本地运行，适合数据敏感场景。

---

Step 4：向量存储

Chroma 是一款轻量级向量数据库，支持本地持久化，无需额外部署服务。它将文本向量及其对应的原始文本和元数据一同存储。

从文档创建向量存储

from langchain_chroma import Chroma

vector_store = Chroma.from_documents(
    documents=chunks,          # 分块后的 Document 列表
    embedding=embeddings,      # Embedding 模型
    persist_directory="./chroma_db",  # 持久化目录
)

# 持久化到磁盘
vector_store.persist()

# 后续可直接加载已有向量库
# vector_store = Chroma(
#     embedding_function=embeddings,
#     persist_directory="./chroma_db",
# )

persist_directory 将向量数据保存到本地磁盘，避免每次启动都重新 Embedding，节省时间和 API 费用。

---

Step 5：检索

从向量库中检索与用户问题最相关的文档片段。

相似度检索

query = "产品的保修政策是什么？"
retrieved_docs = vector_store.similarity_search(query, k=4)

for doc in retrieved_docs:
    print(f"来源: {doc.metadata.get('source', 'unknown')}")
    print(f"页码: {doc.metadata.get('page', 'N/A')}")
    print(f"内容: {doc.page_content[:150]}...\n")

MMR 多样化检索

如果希望检索结果在相关性和多样性之间取得平衡，可以使用 MMR（Maximum Marginal Relevance）：

mmr_docs = vector_store.max_marginal_relevance_search(
    query, k=4, fetch_k=20, lambda_mult=0.5
)

fetch_k 指定初始候选取回的文档数，lambda_mult 控制多样性权重（0 完全多样，1 完全相关）。

重排序（进阶）

为进一步提升检索质量，可以用交叉编码器对初筛结果重新排序：

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

reranker = CrossEncoderReranker(
    model=HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3"),
    top_n=3,
)
compressor = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vector_store.as_retriever(search_kwargs={"k": 10}),
)
reranked_docs = compressor.invoke(query)

---

Step 6：LLM 生成

将检索到的文档片段与用户问题组装成 Prompt，送入 LLM 生成回答。

自定义 Prompt 模板

from langchain.prompts import ChatPromptTemplate

prompt_template = ChatPromptTemplate.from_messages([
    ("system", """你是一个基于知识库的问答助手。请根据以下上下文内容回答问题。
如果上下文中找不到答案，请如实告知，不要编造。

上下文：
{context}

回答时请引用来源文档名称和页码，格式为【来源：{source}，第{page}页】。"""),
    ("human", "{question}"),
])

RetrievalQA 链

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 将所有检索结果放入一次 Prompt
    retriever=vector_store.as_retriever(search_kwargs={"k": 4}),
    chain_type_kwargs={"prompt": prompt_template},
    return_source_documents=True,  # 返回来源文档
)

chain_type="stuff" 适用于检索结果总 token 数不超过模型上下文窗口的场景。若文档过长，可改用 map_reduce 或 refine 策略。

---

Step 7：引用溯源

在实际应用中，用户不仅需要答案，还需要知道答案来自哪里。引用溯源让系统输出具有可验证性，增强用户信任。

带引用的 QA 链

from langchain.prompts import PromptTemplate

# 自定义 Prompt，强制要求引用来源
qa_prompt = PromptTemplate(
    template="""使用以下上下文内容回答用户问题。

上下文：
{context}

问题：{question}

回答要求：
1. 基于上下文给出准确回答
2. 在每一处关键信息后标注来源，格式：【来源：{source}，第{page}页】
3. 如果信息不充分，请说明缺少哪些信息

回答：""",
    input_variables=["context", "question", "source", "page"],
)

def query_with_citation(question: str) -> dict:
    """执行带引用溯源的问答"""
    # 1. 检索相关文档
    docs = vector_store.similarity_search(question, k=4)

    # 2. 组装上下文和来源信息
    context_parts = []
    source_info = []
    for i, doc in enumerate(docs):
        context_parts.append(f"[文档{i + 1}] {doc.page_content}")
        source = doc.metadata.get("source", "未知文档")
        page = doc.metadata.get("page", "N/A")
        source_info.append({"source": source, "page": page})

    context_str = "\n\n".join(context_parts)
    sources_str = "\n".join([f"- {s['source']} (第{s['page']}页)" for s in source_info])

    # 3. 构建 Prompt
    prompt_value = qa_prompt.format(
        context=context_str,
        question=question,
        source=sources_str,
        page="",
    )

    # 4. 调用 LLM
    response = llm.invoke(prompt_value)

    return {
        "answer": response.content,
        "sources": source_info,
        "raw_docs": docs,
    }

# 使用示例
result = query_with_citation("产品的保修期限是多久？")
print(f"回答: {result['answer']}")
print(f"来源: {result['sources']}")

输出示例

回答：根据产品手册，该产品的保修期限为两年（自购买之日起计算）。
【来源：产品手册.pdf，第12页】在保修期内，非人为损坏可享受免费维修服务。
【来源：产品手册.pdf，第15页】

来源: [{'source': '产品手册.pdf', 'page': 12}, {'source': '产品手册.pdf', 'page': 15}]

---

完整代码

将以上所有步骤整合为一个完整的 rag_query.py 脚本：

"""
RAG 知识库问答系统 - 完整实现
支持 PDF / Markdown / HTML 文档
"""

import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma

# ---------- 配置 ----------
PDF_PATH = "data/产品手册.pdf"
PERSIST_DIR = "./chroma_db"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "sk-your-api-key")

# ---------- Step 1: 文档加载 ----------
print("[1/6] 加载文档...")
loader = PyPDFLoader(PDF_PATH)
documents = loader.load()
print(f"  -> 共 {len(documents)} 页")

# ---------- Step 2: 文档分块 ----------
print("[2/6] 文档分块...")
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=128,
    separators=["\n\n", "\n", "。", ".", " ", ""],
)
chunks = text_splitter.split_documents(documents)
print(f"  -> 共 {len(chunks)} 个文本块")

# ---------- Step 3: Embedding 向量化 ----------
print("[3/6] 创建 Embedding 模型...")
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=OPENAI_API_KEY,
)

# ---------- Step 4: 向量存储 ----------
print("[4/6] 创建向量存储...")
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory=PERSIST_DIR,
)
vector_store.persist()
print(f"  -> 向量库已持久化至 {PERSIST_DIR}")

# ---------- Step 5 & 6: 检索 + LLM 生成 ----------
print("[5/6] 初始化检索链...")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, openai_api_key=OPENAI_API_KEY)

from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

qa_prompt = PromptTemplate(
    template="""使用以下上下文内容回答用户问题。
如果找不到答案，请如实告知，不要编造。

上下文：
{context}

问题：{question}

请在回答中引用来源文档名称和页码。回答：""",
    input_variables=["context", "question"],
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_store.as_retriever(search_kwargs={"k": 4}),
    chain_type_kwargs={"prompt": qa_prompt},
    return_source_documents=True,
)

# ---------- Step 7: 交互问答 ----------
print("[6/6] 系统就绪，输入问题（输入 exit 退出）\n")

while True:
    query = input("问题: ")
    if query.lower() in ("exit", "quit"):
        break

    result = qa_chain.invoke({"query": query})

    print(f"\n回答: {result['result']}\n")
    print("参考来源:")
    seen = set()
    for doc in result["source_documents"]:
        source = doc.metadata.get("source", "unknown")
        page = doc.metadata.get("page", "N/A")
        key = f"{source}:{page}"
        if key not in seen:
            seen.add(key)
            print(f"  - {source} (第{page}页)")
    print()

运行方式

export OPENAI_API_KEY="sk-your-api-key"    # Linux / macOS
set OPENAI_API_KEY="sk-your-api-key"       # Windows
python rag_query.py

---

总结

本文从零到一构建了一个完整的 RAG 知识库问答系统，涵盖文档加载、分块、向量化、存储、检索、生成和引用溯源七个核心步骤。该系统可直接应用于企业内部知识库问答、产品文档智能客服、学术文献检索等场景。

关键优化方向包括：引入重排序提升检索精度、使用多路召回融合不同检索策略、对输出结果进行事实性校验等。