RAG 原理与实现：从零搭建企业知识库

本文适合谁：刚接触 RAG 的程序员，或者知道 RAG 是什么但没有亲手跑过完整流程的人。读完本文，你能理解 RAG 的底层逻辑，并跑通一个完整的企业知识问答系统。

为什么 LLM 不能直接回答你的公司问题

在动手写代码之前，先理解问题的本质——为什么需要 RAG。

想象你是一家电商公司，刚上线了 AI 客服。用户问："我买的电子产品还能退货吗？"

直接用 LLM 回答会发生什么？

它会说："根据中国消费者权益保护法，一般商品有七天无理由退货权利……" ——然后给出一段通用的法律知识，完全不知道你们公司具体的退货规则：是 7 天还是 15 天、电子产品有没有特殊说明、特价商品能不能退。

这不是 LLM 笨，而是它真的不知道。你的退货政策从来没有出现在任何公开的训练数据里。

这就是 RAG 要解决的问题：让 LLM 在回答之前，先从你的知识库里找到相关信息，然后基于这些真实信息来回答。

RAG 是什么：用一个图书馆比喻

你去图书馆查资料写报告，有两种方式：

闭卷方式：不查任何资料，完全凭记忆写。优点是快，缺点是可能记错，也不知道最新进展。

开卷方式：先去书架找相关书籍，翻到相关章节，边看边写，引用原文。优点是准确、有据可查，缺点是需要先找资料。

RAG 就是让 LLM 做开卷考试。

具体流程：

你的知识库是"书架"——提前把公司文档整理好，存进向量数据库
用户提问时，先在"书架"里找相关资料（检索）
把找到的资料摆到 LLM 面前
LLM 看着资料回答

RAG 全称 Retrieval-Augmented Generation，检索增强生成。

根据 AWS 的定义：RAG 对 LLM 输出进行优化，使其在生成之前能够引用训练数据来源之外的权威知识库。

三个关键技术概念

理解 RAG 之前，需要先搞清楚三个概念。不用担心，我会用类比来解释。

1. Embedding（向量化）：把文字变成坐标

计算机不懂文字的语义，但它能计算数字的距离。

Embedding 模型做的事情是：把一段文字变成一组数字（向量），而且语义相近的文字，变成的向量也相近。

举个例子：

"苹果手机多少钱" → [0.2, 0.8, -0.3, ...] （1536维数字）
"iPhone 售价" → [0.18, 0.79, -0.31, ...] （数字非常接近）
"今天天气真好" → [-0.5, 0.1, 0.7, ...] （数字差别很大）

这样，通过计算向量之间的距离，就能判断两段文字的语义有多相关。

可以想象成：Embedding 把文字"投射"到一个巨大的语义空间里，语义相似的词和句子落在空间里相近的位置。

2. Chunking（分块）：把文档切成合适大小的片段

为什么要切？两个原因：

原因一：LLM 有上下文窗口限制，一次能处理的文字有上限。一份 500 页的 PDF 塞不进去。

原因二：检索时不需要整本文档，只需要跟问题最相关的那几段。如果文档太大，相关信息被大量无关内容稀释，反而找不准。

切蛋糕类比：一整个大蛋糕切成小块，吃的时候只取你想吃的那块。太大一块吃不完，太小一块又吃不够。512-1024 个字符（token）通常是合适的"块大小"。

3. 向量相似度检索：在语义空间里找"最近的邻居"

有了向量，怎么找相关文档？

把用户问题也转成向量，然后在向量数据库里找"最近的邻居"——向量距离最小的那些文档块，就是最相关的内容。

最常用的是余弦相似度（cosine similarity）：计算两个向量的夹角，夹角越小（越接近 0 度），两个向量越相似。取相似度最高的 top-k 个块返回。

RAG 的完整流程

RAG 分两个阶段：

离线索引阶段（做一次，或知识更新时做）

code

原始文档 → 加载 → 清洗 → 分块 → 向量化 → 存入向量数据库

在线检索阶段（每次查询都跑一遍）

code

用户问题 → 向量化 → 相似度检索 → 取 top-k 文档块 → 拼入 Prompt → LLM 生成答案

RAG 完整工作流程——离线构建阶段（文档切割+向量化入库）与在线检索阶段（检索+生成）

动手实现：从零搭建一个 RAG 系统

理论够了，开始动手。我们要搭建一个能回答公司退货政策问题的 RAG 系统。

需要安装的依赖：

bash

pip install langchain langchain-community langchain-openai chromadb tiktoken

如果不想用 OpenAI（需要付费），下面提供 HuggingFace 免费 Embedding 模型的替代方案。

第一步：准备知识库文档

python

# 实际项目里，这里换成 PyPDFLoader、UnstructuredWordDocumentLoader 等
# 读取真实的 PDF 或 Word 文档

# 我们先用一段示例文本模拟公司退货政策文档
company_policy = """
退货政策（最后更新：2024年3月）

一、电子产品
1.1 自购买之日起 7 天内可申请无理由退货
    - 前提：商品保持原包装完好，附件齐全
    - 不支持退货情形：包装损坏、有明显使用痕迹、缺少配件
1.2 7天以上、30天以内：仅支持产品质量问题换货
1.3 超过30天：需联系厂商走质保流程

二、服装鞋帽
2.1 自购买之日起 15 天内可申请无理由退货
    - 前提：吊牌未剪、未洗涤、未穿着
2.2 促销特价商品：不支持无理由退货，仅支持质量问题换货
2.3 内衣、泳衣类：出于卫生原因，不支持退货

三、食品饮料
3.1 不支持无理由退货
3.2 质量问题：请联系客服，提供照片和购买凭证，3个工作日内处理

四、退货流程
第一步：登录 App，进入"我的 → 我的订单"
第二步：找到目标订单，点击"申请售后"
第三步：选择"退货退款"，填写退货原因
第四步：等待客服审核（工作日 24 小时内）
第五步：审核通过后，快递员上门取件
第六步：商品入库验收后，3-5 个工作日退款至原支付账户

五、特殊说明
- 节假日期间退款时间可能延长至 7-10 个工作日
- 大家电类商品退货需联系配送人员上门处理
- 如遇退款纠纷，可通过 400-XXX-XXXX 联系客服中心
"""

第二步：加载文档，进行分块

python

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

# 把文本包装成 LangChain Document 对象
# 实际项目里，TextLoader/PyPDFLoader 会直接生成这个对象
documents = [Document(
    page_content=company_policy,
    metadata={"source": "company_policy.txt", "version": "2024-03"}
)]

# RecursiveCharacterTextSplitter 是 LangChain 最常用的分块器
# 它按优先级递归切割：先按段落（\n\n），再按换行（\n），再按句子，最后按字符
splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,      # 每块最大 400 个字符
    chunk_overlap=50,    # 相邻块重叠 50 个字符，防止信息在边界处丢失
    separators=["\n\n", "\n", "。", "，", " ", ""]  # 中文文档加上句号和逗号
)

chunks = splitter.split_documents(documents)

print(f"原文档 {len(documents)} 篇，分块后 {len(chunks)} 个 chunk")
print("\n=== 各 chunk 预览 ===")
for i, chunk in enumerate(chunks):
    print(f"\nChunk {i+1}（{len(chunk.page_content)} 字符）：")
    print(chunk.page_content[:80] + "...")

预期输出：

code

原文档 1 篇，分块后 6 个 chunk

=== 各 chunk 预览 ===

Chunk 1（387 字符）：
退货政策（最后更新：2024年3月）

一、电子产品
1.1 自购买之日起 7 天内可申请无理由退货
    - 前提：商品保持原包装完好，附件齐全...

Chunk 2（342 字符）：
二、服装鞋帽
2.1 自购买之日起 15 天内可申请无理由退货...
...

第三步：向量化并存入 Chroma

python

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 方案一：使用 OpenAI Embedding（需要 OPENAI_API_KEY）
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # 性价比最高的选择
)

# 方案二：使用 HuggingFace 本地 Embedding（免费，需要下载约 500MB 模型）
# from langchain_community.embeddings import HuggingFaceEmbeddings
# embeddings = HuggingFaceEmbeddings(
#     model_name="BAAI/bge-small-zh-v1.5",  # 专为中文优化
#     model_kwargs={"device": "cpu"}
# )

# 创建向量数据库，把 chunks 向量化后存入
# Chroma 是一个轻量级开源向量数据库，适合开发和原型验证
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 本地持久化，重启程序不丢失数据
)

print(f"向量数据库构建完成，共 {vectorstore._collection.count()} 个向量")

预期输出：

code

向量数据库构建完成，共 6 个向量

第四步：构建检索问答链

python

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

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",  # 也可以换成 DeepSeek
    temperature=0,        # temperature=0 让回答更确定性，适合知识问答
)

# 自定义 Prompt 模板，明确告诉 LLM 基于提供的文档回答
custom_prompt = PromptTemplate(
    input_variables=["context", "question"],
    template="""你是一个专业的客服助手，根据以下提供的公司政策文档来回答用户问题。

重要规则：
1. 只根据下面的文档内容回答，不要添加文档中没有的信息
2. 如果文档中没有明确答案，请如实说明"该问题在当前文档中未找到相关规定"
3. 回答要简洁清晰，必要时列出步骤

【相关文档内容】
{context}

【用户问题】
{question}

【回答】"""
)

# 构建 RetrievalQA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",   # "stuff" = 把所有检索结果塞进一个 Prompt（最简单）
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}  # 每次检索 top-3 个 chunk
    ),
    chain_type_kwargs={"prompt": custom_prompt},
    return_source_documents=True  # 返回来源文档，方便调试和展示
)

第五步：测试问答效果

python

def ask(question: str):
    """测试一个问题"""
    print(f"\n{'='*50}")
    print(f"问：{question}")
    result = qa_chain.invoke({"query": question})
    print(f"答：{result['result']}")
    print(f"\n来源文档（共 {len(result['source_documents'])} 个 chunk）：")
    for i, doc in enumerate(result['source_documents']):
        print(f"  [{i+1}] {doc.page_content[:60]}...")
    return result

# 测试问题
ask("我买了衣服，15天了还能退货吗？")
ask("电子产品的无理由退货期是多少天？")
ask("退货需要多久才能退款？")
ask("内衣可以退货吗？")
ask("食品买了可以退货吗？")

预期输出：

code

==================================================
问：我买了衣服，15天了还能退货吗？
答：根据公司退货政策，服装鞋帽自购买之日起 15 天内可申请无理由退货，前提是吊牌未剪、未洗涤、未穿着。

如果您是在第15天申请退货，还在政策允许范围内，可以操作。但如果已超过15天，则只能在存在质量问题时申请换货，无法申请无理由退货。

来源文档（共 3 个 chunk）：
  [1] 二、服装鞋帽
2.1 自购买之日起 15 天内可申请无理由退货
  [2] 四、退货流程...
  [3] 一、电子产品...

==================================================
问：退货需要多久才能退款？
答：根据退货流程说明，商品入库验收后，退款将在 3-5 个工作日内退回原支付账户。

需要注意的是，节假日期间退款时间可能延长至 7-10 个工作日。

来源文档（共 3 个 chunk）：
  [1] 四、退货流程...
  [2] 五、特殊说明...
  [3] ...

RAG 系统工作正常：它检索到了相关的政策文档块，LLM 基于文档内容给出了准确的回答，不是通用的泛泛而谈。

常见问题和解决方案

问题一：检索结果和问题不相关

可能的原因：

Embedding 模型选择不当（中文场景用了英文模型）
chunk_size 太大，语义信息被稀释

解决方案：

python

# 中文场景用 BGE 系列 Embedding 模型效果更好
from langchain_community.embeddings import HuggingFaceEmbeddings
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")

# 减小 chunk_size，提高语义聚焦
splitter = RecursiveCharacterTextSplitter(chunk_size=256, chunk_overlap=30)

问题二：回答被截断，上下文不完整

可能的原因：chunk 太小，一个完整的规则被切成了两段，每段单独看语焉不详。

解决方案：增大 chunk_size 或使用父子分块（第 02 篇详细讲）：

python

# 增大块大小
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)

问题三：模型回答了文档里没有的内容（幻觉）

可能的原因：Prompt 没有强约束模型只能基于文档回答。

解决方案：在 Prompt 里明确指令：

python

template="""仅根据以下文档内容回答问题。如果文档中没有相关内容，请明确回答"根据现有文档，该问题没有明确规定"。不要使用你自己的知识来补充。

文档内容：
{context}

问题：{question}

回答："""

问题四：处理 PDF 等复杂格式文档

bash

pip install pypdf unstructured

python

from langchain_community.document_loaders import PyPDFLoader

# 加载 PDF
loader = PyPDFLoader("/path/to/your/document.pdf")
documents = loader.load()

# 或者加载整个目录的文档
from langchain_community.document_loaders import DirectoryLoader
loader = DirectoryLoader("/path/to/docs", glob="**/*.pdf")
documents = loader.load()

RAG 检索质量调优的关键参数

搭好基础版 RAG 之后，影响效果的几个关键参数：

k（检索数量）：每次返回多少个 chunk。k=3 是常见起点，太少可能漏掉关键信息，太多则占用上下文窗口，LLM 注意力分散。

python

# k 值调优
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})   # 默认，大多数场景
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})   # 文档量大时可以增加
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})   # 问题非常精确时可以减少

score_threshold（相似度阈值）：只返回相似度超过阈值的结果，过滤不相关内容。

python

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 5}  # 只返回相似度 > 0.7 的结果
)

chunk_size 和 chunk_overlap：这是影响 RAG 效果最大的参数，第 02 篇专门讲分块策略。

一个生产可用版本的结构

上面的示例是最简化版本，生产环境需要补充几个关键能力：

python

class ProductionRAG:
    """生产就绪的 RAG 系统骨架"""

    def __init__(self, vectorstore, llm):
        self.vectorstore = vectorstore
        self.llm = llm
        self.qa_chain = self._build_chain()

    def _build_chain(self):
        return RetrievalQA.from_chain_type(
            llm=self.llm,
            retriever=self.vectorstore.as_retriever(
                search_type="similarity",
                search_kwargs={"k": 3}
            ),
            return_source_documents=True
        )

    def add_documents(self, new_docs: list[Document]):
        """增量添加文档（不需要重建整个索引）"""
        splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        chunks = splitter.split_documents(new_docs)
        self.vectorstore.add_documents(chunks)
        print(f"已添加 {len(chunks)} 个新文档块")

    def ask(self, question: str, user_id: str = None) -> dict:
        """带日志和错误处理的问答"""
        try:
            result = self.qa_chain.invoke({"query": question})
            # 生产环境：记录问答日志，用于后续质量评估
            # log_qa(user_id, question, result, timestamp=datetime.now())
            return {
                "answer": result["result"],
                "sources": [doc.metadata for doc in result["source_documents"]],
                "success": True
            }
        except Exception as e:
            # 生产环境：告警通知、降级回答
            return {
                "answer": "抱歉，系统暂时无法回答您的问题，请联系人工客服。",
                "error": str(e),
                "success": False
            }

小结

RAG 的核心逻辑非常简单：在回答之前先查资料，基于资料来回答。

实现上分两个阶段：

离线：加载文档 → 分块 → 向量化 → 存入向量数据库
在线：问题向量化 → 检索相似块 → 拼入 Prompt → LLM 生成答案

这个最简版本能解决大多数企业知识问答场景。但要做到生产级别，还有很多可以优化的地方：分块策略（第 02 篇）、Embedding 选型（第 03 篇）、向量数据库选型（第 04 篇）、混合检索与重排序（第 05 篇）……

先跑通这个最简版本，再逐步优化。