实操 02 RAG 实战：从文档到可溯源问答系统

所属：AI 知识图谱 → 实操系列（practice/）。正文方法论见 02 篇 · RAG 九环节决策，本篇把九个环节逐个落到代码。

速览卡

项	内容
一句话定位	用 ~200 行 Python 跑通一个"带引用来源、答不上会拒答"的文档问答系统——10 篇项目梯度里投入产出比最高的练手项目
核心内容	文档解析与分块、Embedding 与向量库、混合检索 + Rerank、引用溯源、拒答机制、第一版评测
前置知识	实操 01 LLM API 实操入门（会调用、会结构化输出）
读完能做到	把任意一批 Markdown/PDF 文档变成可问答的知识库，答案附带出处
配套正文	RAG 与微调的选型判断见 02 篇；评测体系见 05 篇

技术选型约定：向量库用 Qdrant（Docker 一条命令起、有本地模式、生产可用）。同类替代：pgvector（已有 PostgreSQL 的团队）、Milvus（超大规模）、FAISS（单机实验）。选型维度对比见附录。Embedding 模型名标 [易变]。

---

一、系统全貌：RAG 到底在跑什么

                    ┌───────── 离线索引（慢，一次性/定期）─────────┐
文档 ──▶ 解析 ──▶ 分块 ──▶ Embedding ──▶ 存入向量库
                                        │
                    ┌───────── 在线问答（快，每次请求）────────────┐
用户提问 ──▶ Embedding ──▶ 向量检索 Top-K ──▶ (可选 Rerank)
          ──▶ 拼 Prompt（问题 + 带来源的片段）──▶ LLM 生成带引用的答案

两条链路完全独立。索引质量决定上限，检索质量决定下限，生成只负责组织语言——效果差时先查前两段，不要急着怪模型。

---

二、环境准备

pip install openai qdrant-client sentence-transformers pypdf

# 方式 A：Docker 起 Qdrant（推荐，与生产形态一致）
docker run -p 6333:6333 -v qdrant_data:/qdrant/storage qdrant/qdrant

# 方式 B：本地模式（零依赖，数据存本地目录，适合练手）
# 代码里 QdrantClient(path="./qdrant_data") 即可，无需 Docker

---

三、环节 1~3：解析 → 分块 → 向量化（离线索引）

3.1 分块：RAG 效果的第一决定因素

分块的核心矛盾：块太小 → 语义不完整；块太大 → 检索不精准 + 上下文塞不下。

对 Markdown 这类有结构的文档，按标题层级切远好于定长切：

# chunking.py
import re

def split_markdown(text: str, max_chars: int = 800, min_chars: int = 100) -> list[dict]:
    """按 Markdown 标题切块，超长再按段落二分。
    每块携带完整标题路径（面包屑），这是检索命中率的关键。"""
    lines, chunks, buf, header_path = text.splitlines(), [], [], []

    def flush():
        if buf and "".join(buf).strip():
            chunks.append({
                "text": "".join(buf).strip(),
                "breadcrumb": " > ".join(header_path),   # 如 "三、检索 > 3.1 混合检索"
            })

    for line in lines:
        m = re.match(r"^(#{1,4})\s+(.*)", line)
        if m:                                            # 遇到标题：结算上一块
            flush()
            level = len(m.group(1))
            header_path = header_path[:level - 1] + [m.group(2)]
            buf = [f"## {m.group(2)}\n"]                 # 块内保留标题行
        else:
            buf.append(line + "\n")
            if sum(len(x) for x in buf) >= max_chars:    # 超长兜底：结算
                flush(); buf = []
    flush()

    # 太碎的块向前合并
    merged = []
    for c in chunks:
        if merged and len(c["text"]) < min_chars:
            merged[-1]["text"] += "\n" + c["text"]
        else:
            merged.append(c)
    return merged

参数基线（先抄再调）：

参数	基线	调整方向
块大小	300~800 字符	问答细碎 → 调小；教程长段落 → 调大
块重叠	10~15%（定长切时必需，标题切可省）	检索断句严重 → 加重叠
保留面包屑	必做	—

九环节完整决策表见 02 篇。

3.2 Embedding 与入库

# indexing.py
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer

# [易变] embedding 模型：中文场景常用 BAAI/bge 系列，按当期榜单选
MODEL = "BAAI/bge-small-zh-v1.5"
model = SentenceTransformer(MODEL)
DIM = model.get_sentence_embedding_dimension()

client = QdrantClient(path="./qdrant_data")        # 本地模式；Docker 用 QdrantClient(url="http://localhost:6333")
COLL = "docs"

def ensure_collection():
    if not client.collection_exists(COLL):
        client.create_collection(
            COLL, vectors_config=VectorParams(size=DIM, distance=Distance.COSINE))

def index_documents(files: list[str]):
    ensure_collection()
    points, pid = [], 0
    for f in files:
        text = open(f, encoding="utf-8").read()
        for c in split_markdown(text):
            vec = model.encode(c["text"]).tolist()
            points.append(PointStruct(
                id=pid,
                vector=vec,
                payload={                           # payload 存元数据：溯源与过滤全靠它
                    "text": c["text"],
                    "breadcrumb": c["breadcrumb"],
                    "source": f,                    # 来源文件 → 答案里展示的出处
                },
            ))
            pid += 1
    # 建议分批 upsert（每批 ≤ 100）
    for i in range(0, len(points), 100):
        client.upsert(COLL, points[i:i + 100])
    print(f"索引完成：{pid} 块")

index_documents(["../00-基础概念与术语谱系.md", "../02-模型能力增强技术.md"])

为什么面包屑和 source 必须进 payload：后面"引用溯源"（第六节）和"元数据过滤"（第五节）都从这里取数据。建库时想清楚要什么，不要事后补。

---

四、环节 4：检索（在线链路开始）

# retrieval.py
def search(query: str, top_k: int = 5) -> list[dict]:
    vec = model.encode(query).tolist()
    hits = client.query_points(COLL, query=vec, limit=top_k).points
    return [{
        "score": h.score,
        "text": h.payload["text"],
        "source": h.payload["source"],
        "breadcrumb": h.payload["breadcrumb"],
    } for h in hits]

单路向量检索的已知短板（正文 02 篇有完整分析）：

• 查"02 篇"这类精确名词时，语义相似度不如关键词匹配 → 混合检索
• Top-5 里往往只有 2~3 块真正相关，排序噪声大 → Rerank

---

五、环节 5~6：混合检索 + Rerank（性价比最高的两步）

5.1 Rerank：一行代码级别的效果提升

原理：向量检索召回 Top-20（宽召回），用交叉编码器精排取 Top-3（严排序）。交叉编码器把"问题+候选"拼一起打分，精度远高于各自编码再算距离，只是慢——所以只能用在少量候选上。

# pip install sentence-transformers  （CrossEncoder 同库）
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-base")      # [易变]

def search_with_rerank(query: str, recall_k: int = 20, top_k: int = 3) -> list[dict]:
    candidates = search(query, top_k=recall_k)
    if not candidates:
        return []
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)
    for c, s in zip(candidates, scores):
        c["rerank_score"] = float(s)
    candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
    return candidates[:top_k]

5.2 混合检索：向量 + 关键词（BM25）

对编号、类名、专有名词等精确匹配场景，BM25 是向量检索的补集。小规模文档用纯 Python 实现：

pip install rank-bm25 jieba

import jieba
from rank_bm25 import BM25Okapi

class HybridRetriever:
    """向量(语义) + BM25(关键词) 双路召回 → RRF 融合 → Rerank 精排"""

    def __init__(self, all_chunks: list[dict]):
        self.chunks = all_chunks
        self.bm25 = BM25Okapi([list(jieba.cut(c["text"])) for c in all_chunks])

    def retrieve(self, query: str, top_k: int = 3) -> list[dict]:
        # 双路召回
        vec_hits = search(query, top_k=20)                       # 向量路
        bm_scores = self.bm25.get_scores(list(jieba.cut(query)))
        bm_top = sorted(range(len(bm_scores)), key=lambda i: -bm_scores[i])[:20]
        bm_hits = [dict(self.chunks[i], score=bm_scores[i]) for i in bm_top]

        # RRF 融合（倒数排名融合，k=60 是标准取值）
        k, pool = 60, {}
        for rank, h in enumerate(vec_hits + bm_hits):
            key = h["text"][:100]
            pool.setdefault(key, {**h, "rrf": 0})
            pool[key]["rrf"] += 1 / (k + rank + 1)
        fused = sorted(pool.values(), key=lambda x: -x["rrf"])[:10]

        # Rerank 精排
        pairs = [(query, c["text"]) for c in fused]
        scores = reranker.predict(pairs)
        fused.sort(key=lambda c: -float(scores[c and fused.index(c)]))
        return fused[:top_k]

成本收益对照（对照 02 篇四个高杠杆动作）：

动作	实现代价	典型收益
Rerank	~15 行	最明显，先做这个
混合检索	~40 行	精确名词类问题命中率大幅提升
查询改写	一次 LLM 调用	多轮对话"那它呢"类问题必需
元数据过滤	payload 条件	多租户/按文档隔离时必需

---

六、环节 7~8：引用溯源与拒答

这是"Demo"和"可交付系统"的分界线。核心：答案里的每个论断都能点回原文；证据不足时明确说不知道。

# rag_qa.py
from openai import OpenAI
import os

llm = OpenAI(api_key=os.environ["LLM_API_KEY"])

PROMPT = """你是严谨的知识库问答助手。仅依据下方【检索片段】回答问题。

规则：
1. 每个论断末尾标注来源，格式 [编号]，编号对应片段序号
2. 若片段不足以回答，直接回复"知识库中没有足够信息回答此问题"，不要编造
3. 答案简洁，不要复述片段原文

【检索片段】
{context}

【问题】
{question}"""

def format_context(chunks: list[dict]) -> str:
    parts = []
    for i, c in enumerate(chunks, 1):
        parts.append(f"[{i}] (来源: {c['source']} · {c['breadcrumb']})\n{c['text']}")
    return "\n\n".join(parts)

def ask(question: str, retriever) -> dict:
    chunks = retriever.retrieve(question, top_k=3)

    # 拒答闸门：检索分数过低直接不问模型
    if not chunks or chunks[0].get("rerank_score", 1) < 0.3:
        return {"answer": "知识库中没有找到相关内容。", "sources": []}

    resp = llm.chat.completions.create(
        model="gpt-4o-mini",                                     # [易变]
        messages=[{"role": "user", "content": PROMPT.format(
            context=format_context(chunks), question=question)}],
        temperature=0,
    )
    return {
        "answer": resp.choices[0].message.content,
        "sources": [{"source": c["source"], "breadcrumb": c["breadcrumb"],
                     "snippet": c["text"][:120]} for c in chunks],
    }

拒答的双闸门设计：

1. 检索层拒答（上面的分数阈值）：最省钱，垃圾进不来
2. 生成层拒答（Prompt 规则 2）：兜底，防止片段相关但答案不在其中时模型硬编

阈值定多少？不要拍脑袋，用评测集扫出来（下一节）。

---

七、环节 9：第一版评测

不用上来就搞大评测平台，20 个手写问答对就能挡住大部分改坏：

# eval_set.json —— 手工构造，覆盖三类问题
[
  {"q": "RAG 和微调怎么选？",          "expect_contains": ["知识", "更新", "成本"], "type": "事实"},
  {"q": "知识库里有没有写量子计算？",   "expect_contains": ["没有", "没有找到"],      "type": "拒答"},
  {"q": "温度设 0 有什么用？",         "expect_contains": ["确定性", "一致"],       "type": "事实"}
]

# quick_eval.py
import json

def run_eval(eval_path: str, retriever):
    cases = json.load(open(eval_path, encoding="utf-8"))
    hit = 0
    for c in cases:
        out = ask(c["q"], retriever)
        ok = any(k in out["answer"] for k in c["expect_contains"])
        hit += ok
        status = "PASS" if ok else "FAIL"
        print(f"[{status}] {c['q']}  →  {out['answer'][:50]}...")
    print(f"\n通过率: {hit}/{len(cases)}")
    # 把失败 case 存下来，修一条攒一条 —— 这就是 05 篇说的回归评测集雏形

改任何参数（块大小、模型、阈值）前后各跑一次，通过率不降才算改对。评测体系的完整方法论见 05 篇。

---

八、生产化检查清单

练手版 → 生产版，逐项核对：

• [ ] 文档解析：PDF 用 pypdf/pymupdf，扫描件需 OCR；表格考虑转 Markdown 再切
• [ ] 增量更新：文件 hash 对比，只重新索引变更文档（不要每次全量重建）
• [ ] 权限过滤：检索时按用户权限加 payload 条件（必须在检索层过滤，不能只在生成层说"不要泄露"——见 09 篇数据泄露 6 风险点）
• [ ] 多租户隔离：payload 加 tenant_id，检索强制过滤
• [ ] 缓存：相同问题（归一化后）直接返回缓存
• [ ] 可观测：记录每次的检索分数、引用命中、用户点赞/点踩（05 篇反馈闭环）
• [ ] 成本：检索+Rerank 几乎免费，成本大头是生成调用——上下文片段控制在 2~4 块

---

九、动手任务

• [ ] 任务 1：用本仓库 AI 目录的 3 篇文档建库，跑通 ask("RAG 九个环节是什么")，检查引用编号是否指向正确的来源文件
• [ ] 任务 2：造 5 个"知识库里没有"的问题（如"今天股市行情"），验证拒答闸门全部生效；把阈值调低再看差异
• [ ] 任务 3：对比 search()（纯向量）与 search_with_rerank()（加 Rerank）对同一批 10 个问题的 Top-3 质量，记录差异
• [ ] 任务 4（进阶）：写 20 条评测集跑 quick_eval，然后把块大小从 800 改成 400 重建索引，看通过率变化——这就是一次最小闭环的回归评测
• [ ] 自检：能不看代码说出——两条链路分别是什么？为什么 Rerank 只对少量候选做？拒答为什么要双闸门？权限过滤必须放在哪一层？

---

延伸方向

• 还不会调 LLM API → 实操 01 LLM API 实操入门
• 不想花钱调 API，本地跑开源模型 → 实操 03 模型本地部署与推理优化
• 答案要"动手做"而不只是"回答" → 实操 06 Agent 实战（规划中）
• RAG 的九环节决策表与微调边界 → 02 篇 · 模型能力增强技术
• 评测集怎么科学地扩、LLM-as-Judge 怎么写 → 05 篇 · 工程化与 LLMOps