RAG 知识库从零搭建：文档解析、向量检索到生成回答

为什么需要 RAG

大模型很强，但有两个硬伤：知识截止日期和幻觉。RAG（Retrieval-Augmented Generation）通过外挂知识库解决这两个问题——先从你的文档里检索相关内容，再让大模型基于检索结果生成回答。

后端开发者做 RAG 的优势：你懂数据管道、懂 API 设计、懂工程化，这些恰好是 RAG 从 Demo 走向生产的关键。

RAG 全链路架构

用户提问 → Query 改写 → 向量检索 → 重排序 → 构造 Prompt → LLM 生成 → 回答
                ↑                                    ↑
          Embedding 模型                      知识库文档

核心环节：

1. 文档处理：解析 → 清洗 → 分块（Chunking）
2. 向量化：Embedding 模型将文本转为向量
3. 存储检索：向量数据库存储 + 相似度检索
4. 生成回答：将检索结果拼入 Prompt，交给 LLM 生成

第一步：文档解析与分块

文档解析

支持 PDF、Word、Markdown 等格式，Python 生态有现成工具：

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    UnstructuredMarkdownLoader
)

def load_documents(file_path: str):
    """根据文件类型加载文档"""
    if file_path.endswith('.pdf'):
        loader = PyPDFLoader(file_path)
    elif file_path.endswith('.docx'):
        loader = Docx2txtLoader(file_path)
    elif file_path.endswith('.md'):
        loader = UnstructuredMarkdownLoader(file_path)
    else:
        raise ValueError(f"不支持的文件格式: {file_path}")
    return loader.load()

Chunk 策略

分块是 RAG 效果的决定性因素之一。太大检索不精准，太小丢失上下文。

def split_documents(documents, chunk_size=500, chunk_overlap=50):
    """递归分块，保留语义连续性"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""]
    )
    return splitter.split_documents(documents)

实战经验

中文文档建议 chunk_size=300~500，overlap=50。太大会导致检索时混入无关内容，太小会打断语义。Markdown 文档优先按标题层级分块。

第二步：Embedding 选型

模型	维度	中文效果	部署方式	成本
OpenAI text-embedding-3-small	1536	良好	API	$0.02/1M tokens
BAAI/bge-large-zh-v1.5	1024	优秀	本地部署	免费（需 GPU）
BAAI/bge-m3	1024	优秀	本地/API	免费/低
阿里通义 Embedding	1536	优秀	API	有免费额度

from langchain.embeddings import HuggingFaceEmbeddings

# 本地部署 bge-large-zh，零成本、数据不出域
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True}
)

踩坑记录

normalize_embeddings=True 很重要！不归一化的话，余弦相似度和点积结果不一致，召回效果会明显下降。

第三步：向量数据库

Milvus vs Qdrant vs pgvector

维度	Milvus	Qdrant	pgvector
性能	最强，亿级向量	强，百万级丝滑	一般，十万级够用
部署复杂度	中等（Docker）	低（单二进制）	最低（PG 插件）
生态	丰富	良好	PostgreSQL 生态
适用场景	大规模生产	中小项目首选	已有 PG 的项目

我的选择：Qdrant。部署简单、性能够用、API 友好，中小项目最佳平衡点。

from langchain.vectorstores import Qdrant
from qdrant_client import QdrantClient

client = QdrantClient(host="localhost", port=6333)

vectorstore = Qdrant.from_documents(
    documents=chunks,
    embedding=embeddings,
    url="http://localhost:6333",
    collection_name="knowledge_base",
    force_recreate=True
)

第四步：检索与重排序

向量检索

def retrieve(query: str, k: int = 5):
    """向量检索 Top-K 相关文档"""
    results = vectorstore.similarity_search_with_score(query, k=k)
    return results

重排序（Reranker）

向量检索召回的粗排结果往往不够精准，加一层 Reranker 精排效果提升明显：

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

def rerank(query: str, documents: list, top_n: int = 3):
    """对检索结果重排序"""
    pairs = [[query, doc.page_content] for doc in documents]
    scores = reranker.compute_score(pairs, normalize=True)
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return ranked[:top_n]

效果对比

实测加 Reranker 后，回答准确率从 ~70% 提升到 ~88%。代价是增加一次模型推理，延迟多 200-500ms。对质量要求高的场景值得。

第五步：构造 Prompt 生成回答

def generate_answer(query: str, context_docs: list) -> str:
    """基于检索结果生成回答"""
    context = "\n\n".join([
        f"[文档{i+1}] {doc.page_content}"
        for i, doc in enumerate(context_docs)
    ])

    prompt = f"""你是一个专业的知识库助手。请基于以下检索到的参考资料回答用户问题。

要求：
1. 只基于参考资料回答，不要编造信息
2. 如果参考资料中没有相关内容，请明确说明"根据现有知识库无法回答"
3. 回答要简洁准确，条理清晰

参考资料：
{context}

用户问题：{query}
"""

    response = llm.invoke(prompt)
    return response.content

完整 RAG Pipeline

把上面的环节串起来：

def rag_pipeline(query: str) -> dict:
    """RAG 完整流程"""
    # 1. 向量检索
    raw_results = retrieve(query, k=10)

    # 2. 重排序
    reranked = rerank(query, [doc for doc, _ in raw_results], top_n=3)
    context_docs = [doc for doc, _ in reranked]

    # 3. 生成回答
    answer = generate_answer(query, context_docs)

    return {
        "query": query,
        "answer": answer,
        "sources": [
            {"content": doc.page_content[:100], "score": float(score)}
            for doc, score in reranked
        ]
    }

生产环境踩坑总结

1. 文档质量 > 模型能力

垃圾进、垃圾出。花时间清洗文档比换更大的模型有效得多。去掉页眉页脚、目录、参考文献等噪声内容。

2. Chunk 策略要因文档而异

• 技术文档：按标题层级分块
• FAQ：一条一个 Chunk
• 长文：递归分块 + overlap
• 表格：整表作为一个 Chunk，不要拆

3. 多路召回比单路强

同时做向量检索 + 关键词检索（BM25），结果融合后 Rerank，召回率能提升 15-20%。

4. 缓存 Embedding 结果

文档入库后 Embedding 不变，缓存下来避免重复计算。尤其用 API 时能省钱。

5. 监控回答质量

生产环境一定要加回答质量评估：来源引用率、用户反馈、人工抽检。没有监控的 RAG 就是黑盒。

下一步

• Query 改写：用户提问往往模糊，用 LLM 先改写 Query 再检索
• 多轮对话：维护对话历史，做上下文 Query 补全
• 知识图谱：对于复杂关联关系，KG + RAG 效果远超纯向量检索
• 流式输出：SSE 流式返回，用户体验大幅提升

---

本文记录的是实战经验，不是理论教程。每个环节都经过真实项目验证，有问题欢迎交流。