个人学习笔记
返回首页
案例RAG实战

构建知识库问答系统

基于 RAG 技术栈的完整文档问答系统

技术栈

PythonLangChainLlamaIndexChromaFastAPIOpenAI APISentence-TransformersPydantic

项目概述

知识库问答系统是当前 LLM 应用中最热门的落地场景之一。传统的 LLM 依赖训练数据中的知识，存在知识过时和幻觉问题。RAG（Retrieval-Augmented Generation，检索增强生成）通过在生成回答前先从外部知识库中检索相关文档，让 LLM 基于真实上下文回答问题，有效解决了这些痛点。

本案例将从零开始构建一个完整的文档问答系统。我们将处理一个包含技术文档的知识库，实现文档解析、智能切分、向量索引、混合检索、重排序、引用溯源等核心功能。最终交付一个可通过 FastAPI 访问的 REST API 服务。

技术架构

系统分为离线索引阶段和在线查询阶段两个主要流程：

离线索引阶段（数据准备管道）

原始文档 -> 文档解析 -> 智能切分 -> Embedding -> 向量数据库

在线查询阶段（问答检索管道）

用户提问 -> Query Embedding -> 混合检索 -> Reranker -> LLM 生成 -> 引用溯源

详细步骤

步骤 1：数据准备与文档解析

首先需要将各种格式的文档统一解析为纯文本。常见格式包括 PDF、Word、Markdown、HTML。对于 PDF，推荐使用 pdfplumber（文本型 PDF）或 Surya/Mathpix（扫描件）。解析后的文本需要清洗：去除页眉页脚、特殊字符、多余空白。

import pdfplumber
from pathlib import Path

def parse_pdf(file_path: str) -> str:
    """解析 PDF 为纯文本"""
    text_parts = []
    with pdfplumber.open(file_path) as pdf:
        for page in pdf.pages:
            text = page.extract_text()
            if text:
                # 清洗：去除多余空白
                text = "\n".join(line.strip() for line in text.split("\n") if line.strip())
                text_parts.append(text)
    return "\n\n".join(text_parts)

# 批量处理文档目录
docs_dir = Path("./knowledge_base")
documents = []
for file_path in docs_dir.glob("**/*.pdf"):
    text = parse_pdf(str(file_path))
    documents.append({
        "content": text,
        "source": str(file_path),
        "type": "pdf"
    })
print(f"解析完成：{len(documents)} 个文档")

步骤 2：文档智能切分

切分是 RAG 中最关键的环节之一。好的切分策略需要平衡两个目标：每个块包含完整的语义（不会在句子中间断开），同时不会太长（避免引入无关信息稀释检索质量）。推荐使用递归字符切分器，按段落 > 句子 > 子句的优先级逐级切分，chunk_overlap 保留相邻块的上下文。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,          # 每块最大字符数
    chunk_overlap=64,        # 相邻块重叠字符数
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "]
)

all_chunks = []
for doc in documents:
    chunks = splitter.create_documents(
        texts=[doc["content"]],
        metadatas=[{"source": doc["source"]}]
    )
    all_chunks.extend(chunks)

print(f"切分后总块数: {len(all_chunks)}")
print(f"平均块长度: {sum(len(c.page_content) for c in all_chunks) / len(all_chunks):.0f}")

步骤 3：Embedding 索引与向量存储

Embedding 模型将文本片段转化为高维向量，语义相似的文本在向量空间中距离更近。中文场景推荐使用 BAAI/bge 系列（bge-large-zh-v1.5），在 C-MTEB 排行榜上表现优秀。Chroma 是轻量级的向量数据库，适合原型开发和中小规模数据（百万级以下）。

from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma

# 初始化 Embedding 模型
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    encode_kwargs={"normalize_embeddings": True}
)

# 构建向量索引
vectorstore = Chroma.from_documents(
    documents=all_chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 测试检索
query = "如何配置系统的网络设置？"
results = vectorstore.similarity_search(query, k=5)
for i, doc in enumerate(results):
    print(f"\n--- 结果 {i+1} (来源: {doc.metadata['source']}) ---")
    print(doc.page_content[:200])

步骤 4：混合检索与重排序

向量检索擅长语义匹配（"损失函数"能匹配到"目标函数"），但可能在精确关键词匹配上不如传统方法。混合检索结合 BM25（关键词匹配）和向量检索（语义匹配），取两者的并集后用 Reranker 精排，能显著提升最终检索质量。推荐使用 bge-reranker-large 作为精排模型。

from langchain.retrievers import EnsembleRetriever, BM25Retriever

# BM25 检索器
bm25_retriever = BM25Retriever.from_documents(all_chunks, k=10)

# 向量检索器
vector_retriever = vectorstore.as_retriever(
    search_type="mmr",  # MMR 算法增加结果多样性
    search_kwargs={"k": 10, "fetch_k": 20}
)

# 混合检索（加权合并）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 向量检索权重稍高
)

# Reranker 精排
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
reranker = HuggingFaceCrossEncoder("BAAI/bge-reranker-large")

def rerank(query, docs, top_k=3):
    pairs = [(query, doc.page_content) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:top_k]]

步骤 5：生成回答与引用溯源

将检索到的相关文档片段构建为上下文，配合精心设计的 Prompt 模板送入 LLM。Prompt 模板需要明确要求：基于上下文回答、标注引用来源、对不知道的内容诚实回答。引用溯源让用户能够验证回答的准确性，是知识库问答系统区别于普通聊天机器人的核心特征。

from openai import OpenAI
from pydantic import BaseModel

class QAAnswer(BaseModel):
    answer: str
    sources: list[str]
    confidence: str

client = OpenAI()

def generate_answer(query: str, contexts: list) -> QAAnswer:
    context_text = "\n\n".join(
        f"[来源{i+1}] {ctx.page_content}\n来源文件: {ctx.metadata['source']}"
        for i, ctx in enumerate(contexts)
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """你是一个知识库问答助手。
规则：
1. 只基于提供的文档上下文回答
2. 回答时标注引用来源编号，如 [来源1]
3. 如果文档中没有相关信息，直接说"根据现有文档，无法回答该问题"
4. 给出回答的置信度：高/中/低"""},
            {"role": "user", "content": f"上下文：\n{context_text}\n\n问题：{query}"}
        ],
        temperature=0,
        response_format={"type": "json_object"}
    )

    return QAAnswer(**json.loads(response.choices[0].message.content))

步骤 6：FastAPI 服务封装

将问答管道封装为 REST API 服务，使用 FastAPI 构建高性能异步接口。包含请求验证（Pydantic）、错误处理、流式输出（SSE）等生产级特性。

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

app = FastAPI(title="知识库问答 API")

class QueryRequest(BaseModel):
    question: str
    top_k: int = 3

class QueryResponse(BaseModel):
    answer: str
    sources: list[str]
    confidence: str

@app.post("/api/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    try:
        # 1. 检索
        candidates = ensemble_retriever.invoke(request.question)
        contexts = rerank(request.question, candidates, request.top_k)

        # 2. 生成
        result = generate_answer(request.question, contexts)
        return QueryResponse(**result.model_dump())
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/health")
async def health():
    return {"status": "ok", "document_count": len(all_chunks)}

# 启动: uvicorn server:app --reload --port 8000

常见问题与解决方案

Q: 检索到的文档不相关怎么办？

优化方向：(1) 调整 chunk_size 和 chunk_overlap，尝试多种切分策略；(2) 使用 Query Rewrite 将用户问题改写为更适合检索的表述；(3) 使用 HyDE（Hypothetical Document Embedding）生成假设答案再做检索；(4) 增加 Reranker 进行精排，过滤低质量候选。

Q: 大规模文档如何处理？

Chroma 适合百万级以下文档。更大规模建议使用 Milvus（分布式向量数据库）或 Qdrant。同时考虑：(1) 使用 FAISS 的 IVF 索引加速；(2) 对文档进行分层索引；(3) 异步批量处理 Embedding 任务。

Q: 如何评估 RAG 系统的质量？

建议三维度评估：(1) 检索质量：Recall@K、MRR（Mean Reciprocal Rank）；(2) 生成质量：Faithfulness（忠实度，是否有幻觉）、Relevancy（相关性）；(3) 使用 RAGAS 框架进行自动化评估，覆盖完整管道。

学习要点总结

  1. RAG 系统的效果 80% 取决于数据质量和检索策略，而非 LLM 本身
  2. 文档切分是 RAG 中最容易被忽视但最关键的环节
  3. 混合检索（BM25 + 向量）+ Reranker 是当前最佳实践
  4. 引用溯源是知识库 QA 区别于普通聊天机器人的核心能力
  5. 评测体系比算法选择更重要——没有度量就没有优化方向
← Prompt 驱动的自动化工作流多模态内容审核系统 →