第 4 章 · RAG 检索增强生成

本章目标：解决大模型"不知道你的私有数据、会乱编"的问题，做出企业里最值钱的应用——私有知识库问答。

4.1 为什么需要 RAG

回顾第 2 章：大模型的知识来自训练数据，而且是"截止到某天"的。于是它有三个硬伤：

不知道你的私有数据：你公司的智粮系统文档、内部制度、项目代码，它都没见过。
知识会过时：训练之后发生的事，它不知道。
会幻觉：你问它"根据我们公司的手册，玉米含水率超标怎么处理"，它会一本正经地编。

有一个最直接的"笨办法"：把整个手册塞进 prompt。但手册动辄几万字，会撞上下文窗口，而且塞得越满模型越抓不住重点。

RAG（Retrieval-Augmented Generation，检索增强生成）的思路更聪明：

不要"全塞进去"，而是"先检索出最相关的那几段，再喂给模型"。

流程分两段：

离线（入库时）：把文档切成小段 → 每段转成向量 → 存进向量库。
在线（提问时）：把问题也转成向量 → 在向量库里找最相似的几段 → 把这些段落拼进 prompt → 让模型基于它们回答。

用一张图理解：

[文档] → 切分 → [段落1][段落2][段落3]... → Embedding → [向量库]
                                                        ↑
用户提问 → Embedding → 向量 → 相似度检索 → 找到最相关的几段 → 拼进 prompt → LLM → 回答

4.2 Embedding：把文字变成"能算距离"的向量

模型没法直接比较两段文字"意思像不像"。解决办法是 Embedding（向量化）：

用一个专门的模型，把一段文字变成一个固定长度的数字向量（比如 1024 维的数组），语义相近的文字，向量在空间里的距离就近。

比如"玉米入库"和"小麦入库"的向量会比较接近，而和"工资发放"的向量离得很远。

这个"距离"通常用**余弦相似度（Cosine Similarity）**衡量，值在 -1 到 1 之间，越接近 1 越相似：

import numpy as np

def cosine_similarity(a, b):
    a = np.array(a); b = np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 两个语义相近的句子，相似度会高；不相关的会低
print(cosine_similarity(vec("玉米入库了"), vec("小麦入库了")))   # 高，比如 0.9
print(cosine_similarity(vec("玉米入库了"), vec("发工资了")))     # 低，比如 0.2

你不需要自己写这些——有现成的 Embedding 模型帮你算向量。中文场景推荐用开源的 BAAI/bge-small-zh-v1.5（本地跑、免费、效果好）：

pip install sentence-transformers chromadb

from sentence_transformers import SentenceTransformer

embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")

vec = embedder.encode("玉米入库了")
print(vec.shape)   # (512,)  —— 一个 512 维的向量

选 Embedding 模型的关键：它必须"理解"你的语言和领域。中文业务文档就用中文模型（BGE、M3E 等），别用英文模型硬套。Embedding 模型和生成模型（DeepSeek）是两回事——前者负责"找相关"，后者负责"写答案"。

4.3 文本切分：RAG 效果的第一道坎

文档不能整篇塞进向量库，要切成小块（chunk）。切分策略直接决定检索质量，这是 RAG 里最容易被忽略、又最影响效果的地方。

两个核心参数：

chunk_size（块大小）：一块多少字/多少 token。太大会混入无关内容、稀释语义；太小会丢失上下文、答不全。中文场景常见 300~800 字一块。
overlap（重叠）：相邻两块重叠多少。留一点重叠（比如 10%~20%），避免一句话正好被切断、语义断裂。

一个直观例子。原文：

"称重系统在并发写入时会出现数据不一致。解决方案是为地磅串口读取增加重试机制，并在入库时加分布式锁。"

如果按 20 字切、无重叠，可能把"解决方案是为地磅串口…"切断，检索时这段就失去了"方案"的完整性。
如果整段算一块，语义完整，检索命中率就高。

实用的切分做法（用 LangChain 的切分器，或自己写）：

def simple_split(text, chunk_size=300, overlap=50):
    """按字符切分，块大小 300 字，重叠 50 字。"""
    chunks = []
    start = 0
    while start < len(text):
        chunks.append(text[start:start + chunk_size])
        start += chunk_size - overlap
    return chunks

# 更专业的做法是"按段落/标题结构切"，用 LangChain 的 RecursiveCharacterTextSplitter：
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
chunks = splitter.split_text(long_document)

经验：优先按文档的自然结构（标题、段落、章节）切，再按长度兜底。纯按固定字数切会切断语义。你的文档如果是结构化的（有标题层级），一定利用起来。

4.4 向量数据库：Chroma 快速上手

切好的段落，向量化之后要存起来、能按相似度查——这就是向量数据库。

选型结论（前面章节提过，这里展开）：

工具	定位	何时用
Chroma	轻量嵌入式，`pip` 即用	学习、原型、个人项目
FAISS	向量算法库（非数据库）	算法研究、离线检索
Milvus	企业级分布式	生产、高并发、私有化

本章用 Chroma 学习，它 API 极简：

import chromadb

client = chromadb.PersistentClient(path="./my_kb")   # 持久化到本地目录
collection = client.get_or_create_collection("company_docs")

# 写入：文档 + 对应向量 + 元数据（元数据可用来做过滤，很实用）
collection.add(
    documents=["玉米入库标准含水率是 12%。", "小麦入库标准含水率是 13%。"],
    embeddings=[embedder.encode("玉米入库标准含水率是 12%。").tolist(),
                embedder.encode("小麦入库标准含水率是 13%。").tolist()],
    metadatas=[{"type": "标准"}, {"type": "标准"}],
    ids=["doc1", "doc2"],
)

# 查询：按向量相似度返回最相关的几条
res = collection.query(
    query_embeddings=[embedder.encode("玉米的含水率要求是多少？").tolist()],
    n_results=1,
)
print(res["documents"][0])   # ['玉米入库标准含水率是 12%。']

4.5 完整 RAG 项目：企业知识库问答

把上面串起来，做一个真正能用的知识库问答：

import os
from openai import OpenAI
from sentence_transformers import SentenceTransformer
import chromadb

# ---------- 初始化 ----------
llm = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"), base_url="https://api.deepseek.com")
embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
db = chromadb.PersistentClient(path="./my_kb")
col = db.get_or_create_collection("company_docs")

# ---------- 离线：入库文档 ----------
def add_documents(docs: list[str]):
    """把一批文档切片、向量化、入库。"""
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)

    ids, documents, embeddings, metas = [], [], [], []
    for i, doc in enumerate(docs):
        for j, chunk in enumerate(splitter.split_text(doc)):
            cid = f"d{i}-c{j}"
            ids.append(cid)
            documents.append(chunk)
            embeddings.append(embedder.encode(chunk).tolist())
            metas.append({"source": f"文档{i}"})

    col.add(ids=ids, documents=documents, embeddings=embeddings, metadatas=metas)
    print(f"入库完成，共 {len(ids)} 个片段。")

# ---------- 在线：检索 + 生成 ----------
def ask(question: str, top_k=3):
    # 1. 检索最相关的片段
    res = col.query(
        query_embeddings=[embedder.encode(question).tolist()],
        n_results=top_k,
    )
    context_chunks = res["documents"][0]

    # 2. 拼成上下文
    context = "\n\n".join(f"【资料{i+1}】{c}" for i, c in enumerate(context_chunks))

    # 3. 让模型基于资料回答，且答不出就直说
    resp = llm.chat.completions.create(
        model="deepseek-v4-flash",
        temperature=0.3,
        messages=[
            {
                "role": "system",
                "content": (
                    "你是企业知识库助手。请只根据下面提供的【资料】回答用户问题。\n"
                    "规则：\n"
                    "1. 如果资料里有答案，就准确回答，并注明依据的是哪条资料；\n"
                    "2. 如果资料里没有答案，直接回答'资料中未找到相关信息'，不要编造。"
                ),
            },
            {
                "role": "user",
                "content": f"资料：\n{context}\n\n用户问题：{question}",
            },
        ],
    )
    print("---- 命中的资料 ----")
    for i, c in enumerate(context_chunks):
        print(f"[{i+1}] {c}")
    print("---- 模型回答 ----")
    print(resp.choices[0].message.content)

# ---------- 跑起来 ----------
add_documents([
    "收粮入库流程：车辆过磅 → 抽样化验含水率 → 按标准折算 → 入库登记。玉米标准含水率 12%，小麦 13%。",
    "含水率折算公式：折合重量 = 实测重量 × (1 - 实测含水率) / (1 - 标准含水率)。",
    "称重系统并发写入可能出现数据不一致，需为地磅串口读取增加重试机制，并加分布式锁。",
])

ask("玉米入库含水率超标了怎么折算？")
ask("公司年会什么时候开？")   # 资料里没有，模型应诚实说不知道

观察两件事（这就是 RAG 的价值）：

第一个问题，模型会"引用"你给的资料回答，而不是凭训练记忆瞎说。
第二个问题，资料里没有，模型会诚实回答"未找到"，而不是编一个日期——因为我们在 system 里明确禁止了编造。

4.6 检索质量优化：RAG 从"能用"到"好用"

上面是最朴素的 RAG（Naive RAG）。真实项目里，检索质量是瓶颈，常见三个优化：

1. 混合检索（向量 + 关键词）

纯向量检索有时会漏掉"精确关键词匹配"（比如专有名词、型号、编号）。混合检索 = 向量相似度 + 关键词（BM25）同时搜，结果合并，兼顾"语义相近"和"字面相同"。

2. 重排序（Rerank）

第一轮检索先粗召回 20 条，再用一个专门的重排序模型精排，取最相关的 3 条。这比"一次就精搜"效果好很多，因为粗召回能保证不遗漏，精排能保证精准。

3. 查询改写

用户问得太口语、太模糊（"那玩意儿怎么整？"），先让 LLM 把问题改写/扩展成更明确的检索词，再拿去检索。

这三招不是让你现在都实现，而是让你知道：RAG 的功力不在"会调 API"，而在"检索这条链怎么调优"。而这部分——数据组织、切分、召回、重排——恰恰是你传统工程能力最能发挥的地方。

4.7 怎么评估 RAG 效果

不能靠"感觉答得不错"来评判。要有指标：

召回率（Recall）：该找出来的相关段落，找出来了多少。召回漏了，后面模型再强也答不对。
答案准确率：模型最终回答对不对。
进阶可用 RAGAS 这类自动评估框架（https://docs.ragas.io/），自动打"忠实度、相关性"等分。

最低成本的评估办法：准备 20~50 个"标准问题 + 标准答案"，每次改了切分策略/检索策略，就批量跑一遍看准确率变化。用指标说话，是你区别于"会抄 Demo"的人的关键。

4.8 本章小结与练习

你该记住的：

RAG = 先检索、再生成，解决私有数据 + 时效 + 幻觉三大问题。
Embedding 把文字变成向量，语义相近则向量相近，用余弦相似度衡量。
切分策略（块大小 + 重叠 + 按结构切）直接决定检索质量。
学习/原型用 Chroma，生产用 Milvus。
必须写"找不到就直说"的规则来压幻觉。

练习：

把 3.5 的代码跑通，换成你自己手头的一份真实文档（制度、规范、README 都行）。
分别用 chunk_size=100 和 chunk_size=800 切同一份文档，观察回答质量差异，理解"切分影响效果"。
故意问一个资料里没有的问题，验证模型是否诚实回答"未找到"。

下一章：让模型从"只动嘴"变成"能动手"——Function Calling 与 Agent 智能体。