RAG 检索增强生成

为什么你要学它

给一个大模型（比如 Qwen2.5-7B、Llama-3-8B）直接提问，如果问题涉及它训练数据之外的知识——比如"我们公司 2025 年 Q4 的员工手册第 3.2 条说了什么"或者"最近这篇内部论文的实验结论是什么"——它会一本正经地编造答案。这就是 LLM 的核心局限性：它不知道自己不知道什么。

RAG（Retrieval-Augmented Generation，检索增强生成） 解决这个问题。它的做法非常直观：

1. 把你的文档（PDF / 网页 / Markdown / Confluence）切分成若干片段
2. 用 Embedding 模型把每个片段编码成向量
3. 用户提问时，把问题也编码成向量，在文档向量库中搜出最相关的几个片段
4. 把"问题 + 搜到的相关片段"拼到 prompt 里，交给 LLM 生成答案

关键是答案附带出处。如果模型说"2025 年开始实行弹性工作制"，你可以点击查看它引用的那个文档片段，验证真实性。

RAG 对比全量微调：

一句话，RAG 是当前企业把 LLM 落地到私有知识库问答成本最低、迭代最快的方案。学习它，不只是学一个 pipeline，而是掌握一套"把静态文档变成可对话智能体"的系统工程。

一句话概览（快速版）

• 标准 pipeline：文档加载 → 切分 chunk → 向量化 → 建索引 → 检索（向量+关键词）→ rerank → 注入 prompt → 生成。
• 关键决策：chunk size / overlap 大小（语义完整性 vs 检索粒度）、embedding 模型（中文用 BGE-M3 或 text-embedding-3-large）、向量库（FAISS 轻量、pgvector 与 SQL 生态结合、Qdrant/Milvus 分布式）。
• 进阶技巧：HyDE（先让模型生成假设答案再检索）、Query Rewriting（改写用户口语化提问）、Metadata Filtering（先按日期/来源过滤再搜）、Self-RAG（模型自己判断需不需要检索）。

核心拆解

🔑 Chunking：怎么切文档，直接决定检索上限

Chunking 是 RAG 里最被低估的一环。切得太大→噪声多、单次 prompt 塞不下；切得太小→跨 chunk 的语义信息被截断、检索到的片段上下文不完整。

常见策略：

1. 固定大小 + overlap（最常用）

• 每个 chunk 200~500 tokens，overlap 20~100 tokens
• 直觉：overlap 保证一个完整句子不会被从中间切断
• 中文建议：chunk size 500~1000 字（中文 token/字 约 1:2），overlap 100~200 字

2. 按语义结构切

• Markdown：按 ##、### 等标题切，保留章节结构
• PDF：先识别段落/表格，再按段落切
• 代码文件：按函数/类边界切
• 优点：切分天然对齐语义，检索时相关性更强

3. 递归切分

• 先用分隔符（\n\n / \n / . / 。 / / 空）切大段
• 如果某段超出最大 chunk size，再递归往下切
• LangChain 的 RecursiveCharacterTextSplitter 就是这么工作的

经验法则：如果你不知道怎么选，从 chunk size=500 tokens、overlap=50 起步。一个有效的诊断方法是：拿 20 个已知答案的问题做回归测试，如果检索到的 top-3 chunk 里总是缺少答案——要么增大 chunk size，要么增大 overlap。

🔑 Embedding：把语义压缩到向量里

Embedding 模型把一段文本编码成一个固定维度的向量（通常 768/1024/1536 维）。语义相近的文本，向量在空间中距离更近。

相似度度量：

中文推荐模型：

一个关键细节：如果你用 BGE 系列，它们要求在检索时对问题加前缀 "Represent this sentence for searching relevant passages:"，不添加会掉点。使用前务必看模型卡片的要求。

🔑 向量数据库：存储与 ANN 搜索

embedding 完成后，你需要把 (vector, metadata) 存到一个可以做近似最近邻（ANN）搜索的系统里。为什么是"近似"？因为精确搜索（暴力计算 query 与所有向量的相似度）在百万级以上规模太慢。ANN 牺牲一点点可接受的精度（通常 <1%）换取 10~100 倍速度。

选型对比：

起步建议：先用 FAISS 跑通流程，再根据规模迁移。FAISS 的优点是零运维——你只需要保存一个 index 文件；缺点是没有内置元数据过滤（需要自己在 Python 层实现）。

🔑 Rerank：二次排序，把真正相关的拿到前面

向量检索返回 top-K（比如 K=20），但这些结果未必按相关性严格排序。Rerank 模型把问题和每个候选 chunk 做一次更精准的打分，重新排序后取 top-N（比如 N=3~5）注入 prompt。

原理区别：

• Embedding：把问题和 chunk 各自编码成向量，一次计算离线索引
• Rerank（Cross-Encoder）：把"[CLS] 问题 [SEP] chunk" 一起编码，输出一个相关性分数。模型更大更深，质量更高，但必须在线计算，无法预先索引

中文推荐：BAAI/bge-reranker-large

为什么 rerank 这么关键？经验上，加上 rerank 能让一个 baseline 系统的 top-k 召回率相对提升 15%~30%。用户感知就是"答案突然变得靠谱了"。

🔑 Query Rewriting：用户问的不是他真正想问的

真实用户的提问往往口语化、模糊、缺上下文：

用户：上次那个新员工入职流程是什么来着？

直接拿这句话去搜，向量相似度可能很低。Query Rewriting 让 LLM 先把问题改写为标准查询语句：

code4 lines
请把用户的问题改写为更适合文档检索的标准查询，输出 3 个改写版本：
1. 新员工入职流程包括哪些步骤？
2. 公司入职手续、培训、系统开通的流程是什么？
3. 2025 年新员工入职指南中规定的流程？

然后用这 3 个改写查询分别做检索，把结果合并去重再 rerank。

HyDE（Hypothetical Document Embedding） 是另一个方向：让模型先"凭空"写一段它认为会是答案的文本（哪怕是错的），用这段文本去检索。由于假设答案和真实答案的语义空间更接近，检索效果会比直接用问题检索更好。

🔑 Prompt 模板：怎么把检索到的内容交给模型

一个合格的 RAG prompt 至少要明确三点：

1. 角色：你是一个基于给定文档回答问题的助手
2. 规则：只能使用提供的上下文；不确定时说"文档中未提及"；引用来源编号
3. 上下文和问题：把检索到的 chunk（带编号）和问题填进去

code22 lines
你是一个严谨的文档助手，必须仅基于下面提供的上下文片段回答用户问题。

规则：
- 如果上下文中没有相关信息，直接回答"根据提供的文档，无法回答该问题"。
- 不要编造内容，不要引用不在上下文中的信息。
- 每个回答后用 [来源 #n] 的形式标注信息来自哪个片段。

---

上下文：

#1
(此处插入 chunk1 的原文)

#2
(此处插入 chunk2 的原文)

...

---

用户问题：{question}

🔑 常见评估指标

做 RAG 一定要有回归测试集（至少 20 道，建议 50~100 道），每道题带一个黄金答案和黄金 chunk 编号。每次改 chunk 大小、换 embedding 模型或加 rerank，跑一遍测试集。

常用指标：

• Context Precision：top-k 检索到的 chunk 中，有多少比例包含答案
• Faithfulness：生成答案中有多少比例的信息确实来自上下文（不是模型编的）
• Answer Relevance：生成答案对问题的实际帮助程度

Faithfulness 通常需要另一个 LLM 当"裁判"。

完整跑通方案

下面我们从零搭建一个本地 PDF 问答助手。使用的技术栈：LangChain（流程编排）+ FAISS（向量检索）+ BGE-M3（embedding，开源）+ 任意 LLM（推理生成）。

第一步：安装依赖

bash3 lines
pip install langchain langchain-community langchain-text-splitters \
             langchain-huggingface faiss-cpu pypdf sentence-transformers \
             torch transformers

第二步：把 PDF 切分成 chunk 并建索引

python38 lines
import os
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings

PDF_DIR = "./pdfs"
INDEX_PATH = "./faiss_index"

# 1. 加载所有 PDF
docs = []
for fname in os.listdir(PDF_DIR):
    if fname.lower().endswith(".pdf"):
        loader = PyPDFLoader(os.path.join(PDF_DIR, fname))
        docs.extend(loader.load())  # 每页作为一个 Document
print(f"加载了 {len(docs)} 页 PDF")

# 2. 切分 chunk（中文建议 500~1000 字一段，带 overlap）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,        # 每段 800 字符（约 400 tokens）
    chunk_overlap=120,     # 相邻段重叠 120 字符
    separators=["\n\n", "\n", "。", "！", "？", ";", ".", " ", ""],
)
splits = splitter.split_documents(docs)
print(f"切分成 {len(splits)} 个 chunk")

# 3. 用 BGE-M3 做 embedding（中文效果好、开源免费）
#    首次运行会自动下载模型权重（约 2GB）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda" if __import__("torch").cuda.is_available() else "cpu"},
    encode_kwargs={"normalize_embeddings": True},
)

# 4. 建 FAISS 索引并保存
vectorstore = FAISS.from_documents(splits, embeddings)
vectorstore.save_local(INDEX_PATH)
print(f"索引已保存到 {INDEX_PATH}")

BGE-M3 的特殊说明：检索问题时，BGE 系列要求你给问题文本加前缀 "Represent this sentence for searching relevant passages: "。稍后我们在检索时会处理。

第三步：加载索引 + 实现检索 + rerank

python56 lines
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from sentence_transformers import CrossEncoder
import re

INDEX_PATH = "./faiss_index"

# 加载 embedding 模型
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda" if __import__("torch").cuda.is_available() else "cpu"},
    encode_kwargs={"normalize_embeddings": True},
)

# 加载 FAISS
vectorstore = FAISS.load_local(
    INDEX_PATH, embeddings, allow_dangerous_deserialization=True
)

# 加载 rerank 模型（约 2GB，CPU 也能跑，GPU 更快）
reranker = CrossEncoder(
    "BAAI/bge-reranker-large",
    device="cuda" if __import__("torch").cuda.is_available() else "cpu",
)


def clean_text(s: str) -> str:
    """简单清理多余空白"""
    return re.sub(r"\s+", " ", s).strip()


def retrieve(question: str, k_initial: int = 20, k_final: int = 5):
    """两步检索：先向量粗召回，再 rerank 细排序"""
    # BGE 要求的检索前缀
    query = f"Represent this sentence for searching relevant passages: {question}"

    # 1) 向量粗召回
    docs = vectorstore.similarity_search(query, k=k_initial)
    if not docs:
        return []

    # 2) Cross-Encoder rerank
    pairs = [(question, clean_text(d.page_content)) for d in docs]
    scores = reranker.predict(pairs, convert_to_tensor=False)

    # 按分数降序排序，取 top-k_final
    scored = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [(d, float(s)) for d, s in scored[:k_final]]


# 小测试
if __name__ == "__main__":
    results = retrieve("这份文档的主要内容是什么？")
    for i, (doc, score) in enumerate(results):
        print(f"\n--- 候选 #{i+1} (score={score:.3f}, 来源: {doc.metadata.get('source','?')})")
        print(doc.page_content[:300] + ("..." if len(doc.page_content) > 300 else ""))

第四步：生成答案

这里给两个 LLM 调用选项：本地开源模型（Ollama） 和 HuggingFace pipeline（直接本地跑小模型）。

选项 A：Ollama 方式（推荐新手）

先安装 Ollama（https://ollama.com），然后 ollama pull qwen2.5:7b。

python55 lines
import requests

OLLAMA_URL = "http://localhost:11434/api/generate"

PROMPT_TEMPLATE = """你是一个严谨的文档助手，必须仅基于下面提供的上下文片段回答用户问题。

规则：
- 如果上下文中没有相关信息，直接回答"根据提供的文档，无法回答该问题"。
- 不要编造内容，不要引用不在上下文中的信息。
- 每个回答后用 [来源 #n] 的形式标注信息来自哪个片段。

---

上下文：

{context}

---

用户问题：{question}
"""


def answer_with_rag(question: str) -> str:
    # 检索
    retrieved = retrieve(question, k_initial=20, k_final=5)
    if not retrieved:
        return "知识库中未找到相关内容。"

    # 拼上下文
    context_parts = []
    for i, (doc, score) in enumerate(retrieved):
        source = doc.metadata.get("source", "未知来源")
        page = doc.metadata.get("page", "?")
        context_parts.append(
            f"#{i+1} [来源文件: {source}, 第 {page} 页, 相关度 {score:.2f}]\n{clean_text(doc.page_content)}"
        )
    context = "\n\n".join(context_parts)

    prompt = PROMPT_TEMPLATE.format(context=context, question=question)

    # 调本地 Ollama
    resp = requests.post(OLLAMA_URL, json={
        "model": "qwen2.5:7b",
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.2, "num_ctx": 8192},
    })
    return resp.json()["response"]


if __name__ == "__main__":
    q = "这份文档的主要观点是什么？"
    print("Q:", q)
    print("A:\n", answer_with_rag(q))

选项 B：HuggingFace pipeline 本地跑小模型

python40 lines
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch

MODEL_NAME = "Qwen/Qwen2.5-1.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME, torch_dtype=torch.bfloat16, device_map="auto"
)
generator = pipeline("text-generation", model=model, tokenizer=tokenizer)


def answer_with_rag_hf(question: str) -> str:
    retrieved = retrieve(question, k_initial=20, k_final=5)
    if not retrieved:
        return "知识库中未找到相关内容。"

    context_parts = []
    for i, (doc, score) in enumerate(retrieved):
        context_parts.append(
            f"#{i+1} [来源: {doc.metadata.get('source','?')}, page {doc.metadata.get('page','?')}]\n{clean_text(doc.page_content)}"
        )
    context = "\n\n".join(context_parts)

    messages = [
        {"role": "system", "content": (
            "你是一个严谨的文档助手，必须仅基于提供的上下文回答问题。"
            "如果上下文中没有相关信息，直接回答'根据提供的文档，无法回答该问题'。"
            "不要编造内容。"
        )},
        {"role": "user", "content": f"上下文:\n{context}\n\n问题: {question}"},
    ]

    out = generator(
        messages,
        max_new_tokens=512,
        temperature=0.2,
        top_p=0.9,
        pad_token_id=tokenizer.eos_token_id,
    )
    return out[0]["generated_text"][-1]["content"]

第五步：加上 Query Rewriting 提升难问题的召回

用户的原始问题可能口语化、缺上下文。Query Rewriting 让 LLM 把它拆成多个更适合检索的标准查询，再用"并集检索 + rerank"：

python60 lines
def rewrite_query(question: str) -> list[str]:
    """让 LLM 把一个问题改写为 3 个角度不同的检索查询"""
    prompt = f"""请把用户的问题改写为 3 个不同角度、更适合做文档检索的标准查询语句。
每个查询单独一行，不要加编号，不要输出其他内容。

用户问题：{question}
"""
    resp = requests.post(OLLAMA_URL, json={
        "model": "qwen2.5:7b",
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.0},
    })
    return [line.strip() for line in resp.json()["response"].splitlines() if line.strip()]


def answer_with_rag_v2(question: str) -> str:
    queries = [question] + rewrite_query(question)

    # 并集检索：用每个查询都搜一遍，合并去重
    seen = set()
    all_docs = []
    for q in queries:
        docs = vectorstore.similarity_search(
            f"Represent this sentence for searching relevant passages: {q}",
            k=10,
        )
        for d in docs:
            key = clean_text(d.page_content)[:100]
            if key not in seen:
                seen.add(key)
                all_docs.append(d)

    # rerank 合并后的结果
    if not all_docs:
        return "知识库中未找到相关内容。"
    pairs = [(question, clean_text(d.page_content)) for d in all_docs]
    scores = reranker.predict(pairs)
    scored = sorted(zip(all_docs, scores), key=lambda x: x[1], reverse=True)[:5]

    # 注入 prompt
    context_parts = [
        f"#{i+1} [来源: {d.metadata.get('source','?')}, page {d.metadata.get('page','?')}]\n{clean_text(d.page_content)}"
        for i, (d, s) in enumerate(scored)
    ]
    context = "\n\n".join(context_parts)

    prompt = PROMPT_TEMPLATE.format(context=context, question=question)
    resp = requests.post(OLLAMA_URL, json={
        "model": "qwen2.5:7b",
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.2, "num_ctx": 8192},
    })
    return resp.json()["response"]


if __name__ == "__main__":
    print("改写后的查询:", rewrite_query("上次那个新员工入职流程是什么？"))
    print(answer_with_rag_v2("新员工入职流程是什么？"))

常见误区

误区 1：chunk size 拍脑袋，不做回归测试 → 解释：chunk size / overlap 对最终质量影响极大，但没有普适答案。建议准备一个 20~50 道题的测试集，对比 chunk size=300 / 500 / 1000 的 context precision，再决定。

误区 2：只做向量检索，不加关键词搜索/rerank → 解释：向量相似度对术语匹配很差（比如用户搜一个精确的产品型号名，向量相似度可能输给一个语义接近但字面上不一样的段落）。生产系统一定要做 混合检索（BM25 + 向量），再套一层 rerank。LangChain 的 EnsembleRetriever 直接支持。

误区 3：用了 BGE embedding 却没加检索前缀 → 解释：BGE 系列在训练时让问题带前缀，让模型区分"这是一个 query 还是一个 document"。推理时不加前缀，检索召回会明显下降。

误区 4：prompt 里不禁止模型编内容，也不要求标注来源 → 解释：不加约束时，LLM 会把它自己的训练知识和检索到的信息混在一起输出，你无法分辨哪些是真的。"根据提供的文档，无法回答该问题" 这条退路一定要写进 system prompt。

误区 5：把整张 PDF 当作一个 chunk → 解释：文档长于 2000 tokens 时，embedding 会丢失大部分细节。正确做法是切分成 500~1000 tokens 的小段，带 overlap，必要时再做"文档摘要 chunk"把文档宏观信息存一份。

误区 6：用用户提问直接生成答案，不做 prompt 注入上下文 → 解释：那就不是 RAG 了——你只是在调用 LLM 的训练知识，私有文档信息根本没到模型那里。

误区 7：不做评估，"感觉还不错"就上线 → 解释：RAG 的质量高度依赖具体文档和具体问题分布，直觉非常不可靠。至少要有一个静态回归测试集，每次改动后跑一遍，记录 context precision 和 faithfulness 的变化。否则你永远不知道"这次改 chunk size 是变好还是变坏了"。