从零搭建可评估 RAG

很多 RAG 教程停在「向量库 + top-k」。这个教程从 PDF 解析一路做到带引用回答和评估报告，重点讲清楚每一步为什么会失败。

最终效果

bash

python -m rag ingest ./docs
python -m rag ask "退款申请超过 7 天还能处理吗？"
python -m rag eval

输出：

text

可以申请，但需要人工审核。政策规定签收后 7 天内可自动退款；
超过 7 天的申请需专员根据商品状态判断。[1][2]

来源：
[1] 退款政策 v3 · p.2
[2] 客服 FAQ · p.1

评估报告：

text

questions: 50
Recall@5: 0.92
MRR: 0.78
citation precision: 0.88
answer contains: 0.84
avg latency: 1.8s

技术栈

bash

pip install pymupdf sentence-transformers rank-bm25 jieba numpy openai tqdm

PDF 解析：PyMuPDF
Embedding：BAAI/bge-small-zh-v1.5（本地 CPU 可跑）
关键词召回：BM25 + jieba
重排：BAAI/bge-reranker-base（可选，首次会下载模型）
生成：任意 OpenAI 兼容接口

text

rag/
├─ __init__.py
├─ __main__.py
├─ ingest.py       # 解析 + 切分
├─ index.py        # 向量 + BM25 索引
├─ retrieve.py     # 混合召回 + 重排
├─ answer.py       # 带引用生成
└─ eval.py         # 评估
data/
├─ chunks.jsonl
├─ embeddings.npy
├─ bm25.pkl
└─ golden.jsonl
docs/

Step 1：解析 PDF，保留页码

ingest.py：

python

from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path

import fitz


def parse_pdf(path: Path):
    doc = fitz.open(path)
    for page_no, page in enumerate(doc, start=1):
        text = page.get_text("text").strip()
        if not text:
            continue
        text = re.sub(r"[ \t]+", " ", text)
        text = re.sub(r"\n{3,}", "\n\n", text)
        yield {
            "doc_id": hashlib.sha1(path.name.encode()).hexdigest()[:8],
            "title": path.stem,
            "page": page_no,
            "text": text,
        }


def split_sentences(text: str) -> list[str]:
    parts = re.split(r"(?<=[。！？!?；;])\s*", text)
    return [p.strip() for p in parts if p.strip()]


def chunk_page(page: dict, max_chars: int = 600, overlap: int = 80) -> list[dict]:
    sentences = split_sentences(page["text"])
    chunks: list[dict] = []
    buf = ""
    for sent in sentences:
        if len(buf) + len(sent) > max_chars and buf:
            chunks.append({**page, "text": buf, "chunk_no": len(chunks)})
            buf = buf[-overlap:] + sent
        else:
            buf += sent
    if buf.strip():
        chunks.append({**page, "text": buf, "chunk_no": len(chunks)})
    return chunks


def main(docs_dir: str, out: str = "data/chunks.jsonl") -> None:
    rows: list[dict] = []
    for path in sorted(Path(docs_dir).glob("*.pdf")):
        for page in parse_pdf(path):
            for chunk in chunk_page(page):
                chunk["id"] = f"{chunk['doc_id']}#p{chunk['page']}#{chunk['chunk_no']}"
                rows.append(chunk)
    Path(out).parent.mkdir(parents=True, exist_ok=True)
    with open(out, "w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    print(f"写入 {len(rows)} 个片段 -> {out}")


if __name__ == "__main__":
    main("./docs")

要点：

页码必须保留，否则无法引用；
先按句子切，再按长度兜底，避免把一句话劈开；
片段带 doc_id、title、page、chunk_no，后续评估和引用都靠它；
扫描件没有文本层时，这里会得到空字符串，需要先 OCR。

Step 2：建向量与 BM25 索引

index.py：

python

from __future__ import annotations

import json
import pickle
from pathlib import Path

import jieba
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

MODEL_NAME = "BAAI/bge-small-zh-v1.5"


def load_chunks(path: str = "data/chunks.jsonl") -> list[dict]:
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]


def build(chunks_path: str = "data/chunks.jsonl") -> None:
    chunks = load_chunks(chunks_path)
    texts = [c["text"] for c in chunks]

    model = SentenceTransformer(MODEL_NAME)
    vectors = model.encode(
        texts,
        batch_size=32,
        normalize_embeddings=True,
        show_progress_bar=True,
    )
    np.save("data/embeddings.npy", vectors.astype("float32"))

    tokenized = [list(jieba.cut(t)) for t in texts]
    bm25 = BM25Okapi(tokenized)
    with open("data/bm25.pkl", "wb") as f:
        pickle.dump({"bm25": bm25, "tokens": tokenized}, f)

    print(f"索引完成：{len(chunks)} 片段，维度 {vectors.shape[1]}")


if __name__ == "__main__":
    build()

建议把 embeddings.npy 和 chunks.jsonl 的行顺序视为强约束：任何重排都要同时重建两者，否则 ID 会错位。

Step 3：混合召回 + RRF

retrieve.py：

python

from __future__ import annotations

import json
import pickle

import jieba
import numpy as np
from sentence_transformers import CrossEncoder, SentenceTransformer

DENSE_MODEL = SentenceTransformer("BAAI/bge-small-zh-v1.5")
RERANKER = CrossEncoder("BAAI/bge-reranker-base")

with open("data/chunks.jsonl", encoding="utf-8") as f:
    CHUNKS = [json.loads(line) for line in f]
VECTORS = np.load("data/embeddings.npy")
with open("data/bm25.pkl", "rb") as f:
    BM25_DATA = pickle.load(f)
BM25 = BM25_DATA["bm25"]


def rrf(rank_lists: list[list[int]], k: int = 60) -> list[int]:
    scores: dict[int, float] = {}
    for ranks in rank_lists:
        for rank, idx in enumerate(ranks, start=1):
            scores[idx] = scores.get(idx, 0.0) + 1.0 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)


def search(query: str, top_k: int = 8, use_rerank: bool = True) -> list[dict]:
    qv = DENSE_MODEL.encode([query], normalize_embeddings=True)[0]
    dense_scores = VECTORS @ qv
    dense_idx = np.argsort(-dense_scores)[:30].tolist()

    sparse_scores = BM25.get_scores(list(jieba.cut(query)))
    sparse_idx = np.argsort(-sparse_scores)[:30].tolist()

    fused = rrf([dense_idx, sparse_idx])
    candidates = [CHUNKS[i] for i in fused[:20]]

    if use_rerank and candidates:
        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 c: c["rerank_score"], reverse=True)

    return candidates[:top_k]

为什么用 RRF 而不是把分数相加：向量相似度和 BM25 分数不在同一个尺度，直接相加会被某一路支配。RRF 只看排名，简单稳定。

Step 4：带引用生成

answer.py：

python

from __future__ import annotations

import json
import os

from openai import OpenAI

from .retrieve import search

client = OpenAI(
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    api_key=os.getenv("OPENAI_API_KEY"),
)
MODEL = os.getenv("MODEL", "gpt-4.1-mini")

PROMPT = """你是一个严谨的文档问答助手。
只使用提供的资料回答，不要使用资料之外的知识。
每个事实性句子后标注来源编号，如 [1]。
如果资料不足，把 insufficient 设为 true，并说明缺什么。

资料：
{sources}

问题：{question}

输出 JSON：
{{"answer": "...", "citations": [1,2], "insufficient": false}}"""


def format_sources(hits: list[dict]) -> str:
    lines = []
    for i, c in enumerate(hits, start=1):
        lines.append(f"[{i}] {c['title']} · p.{c['page']} · {c['id']}\n{c['text']}")
    return "\n\n".join(lines)


def answer(question: str, top_k: int = 6) -> dict:
    hits = search(question, top_k=top_k)
    if not hits:
        return {"answer": "没有检索到相关资料。", "citations": [], "insufficient": True, "sources": []}

    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "只输出合法 JSON。"},
            {
                "role": "user",
                "content": PROMPT.format(sources=format_sources(hits), question=question),
            },
        ],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=800,
    )
    data = json.loads(resp.choices[0].message.content)

    valid = {i for i in range(1, len(hits) + 1)}
    data["citations"] = [i for i in data.get("citations", []) if i in valid]
    data["sources"] = [
        {"no": i, "title": hits[i - 1]["title"], "page": hits[i - 1]["page"], "id": hits[i - 1]["id"]}
        for i in data["citations"]
    ]
    return data

注意：引用校验不能省。模型可能引用一个不存在的编号，或者引用了但原文不支持结论。至少校验编号存在，进阶做法是逐句让评审模型判断「这段原文是否支持这句话」。

Step 5：构造评估集

data/golden.jsonl，每条一行：

json

{"q":"退款申请超过 7 天还能处理吗？","relevant_ids":["3f2a1b7c#p2#0"],"must_contain":["人工审核"],"max_latency":5}
{"q":"跨境订单的运费怎么算？","relevant_ids":["9c1d2e3f#p1#2"],"must_contain":["按实际重量"],"max_latency":5}

至少 50 条，覆盖：

常见问题；
需要跨页/跨文档的问题；
文档中没有答案的问题；
带编号、专有名词的问题；
容易混淆的相似政策。

Step 6：评估脚本

eval.py：

python

from __future__ import annotations

import json
import statistics
from pathlib import Path

from .answer import answer
from .retrieve import search


def recall_at_k(hits: list[dict], relevant: set[str], k: int = 5) -> float:
    got = {h["id"] for h in hits[:k]}
    return 1.0 if got & relevant else 0.0


def mrr(hits: list[dict], relevant: set[str]) -> float:
    for i, h in enumerate(hits, start=1):
        if h["id"] in relevant:
            return 1.0 / i
    return 0.0


def main() -> None:
    cases = [json.loads(line) for line in Path("data/golden.jsonl").read_text(encoding="utf-8").splitlines() if line]
    rows = []
    for case in cases:
        hits = search(case["q"], top_k=8)
        result = answer(case["q"], top_k=6)

        cited = {s["id"] for s in result.get("sources", [])}
        relevant = set(case["relevant_ids"])
        rows.append(
            {
                "q": case["q"],
                "recall5": recall_at_k(hits, relevant),
                "mrr": mrr(hits, relevant),
                "citation_precision": len(cited & relevant) / len(cited) if cited else 0.0,
                "contains": all(s in result["answer"] for s in case.get("must_contain", [])),
                "insufficient": result.get("insufficient", False),
            }
        )

    report = f"""questions: {len(rows)}
Recall@5: {statistics.mean(r['recall5'] for r in rows):.2f}
MRR: {statistics.mean(r['mrr'] for r in rows):.2f}
citation precision: {statistics.mean(r['citation_precision'] for r in rows):.2f}
answer contains: {statistics.mean(r['contains'] for r in rows):.2f}
"""
    print(report)
    Path("data/eval-report.md").write_text(report, encoding="utf-8")


if __name__ == "__main__":
    main()

先看 Recall@5。如果它低于 0.8，调生成提示词没有意义——模型根本没拿到正确资料。

Step 7：把命令串起来

__main__.py：

python

import sys

from . import answer as answer_mod
from . import eval as eval_mod
from . import index, ingest


def main() -> None:
    cmd = sys.argv[1] if len(sys.argv) > 1 else "help"
    if cmd == "ingest":
        ingest.main(sys.argv[2] if len(sys.argv) > 2 else "./docs")
    elif cmd == "index":
        index.build()
    elif cmd == "ask":
        result = answer_mod.answer(" ".join(sys.argv[2:]))
        print(result["answer"])
        for s in result.get("sources", []):
            print(f"[{s['no']}] {s['title']} · p.{s['page']}")
    elif cmd == "eval":
        eval_mod.main()
    else:
        print("用法: python -m rag [ingest|index|ask|eval] ...")


if __name__ == "__main__":
    main()

完整流程：

bash

python -m rag ingest ./docs
python -m rag index
python -m rag ask "退款申请超过 7 天还能处理吗？"
python -m rag eval

调优顺序

不要一上来换向量库。按这个顺序：

解析：随机抽 10 页，看文本是否完整、表格是否错乱；
切分：抽 20 个片段，看是否切断语义；
Recall@5：正确片段有没有进前 5；
重排：加了重排，MRR 是否提升；
引用：引用是否支持结论；
生成：最后才调提示词。

常见失败

页码丢了

引用只能写「某文档」，用户无法核对。解析时就把页码写进每个片段。

只做向量检索

编号、专有名词、代码几乎检不到。必须加 BM25，再做 RRF。

chunk 太大

一段 2000 字，模型只用了其中一句。按句子/标题切，控制在 400–800 字。

用回答质量代替检索评估

回答错可能只是生成问题。先测 Recall@k，再测引用与忠实性。

验收清单

能解析 20 份以上 PDF，页码正确
chunk 保留标题、页码、文档 ID
向量 + BM25 混合检索
有重排，top-k 控制在 4–8
回答带引用，引用编号经过校验
有 50 条评估集，能输出 Recall@5 / MRR / 引用准确率
能看到每个问题的失败原因

扩展任务

用 bge-m3 替换小模型，比较 Recall@5；
加父子块：小块检索，大块回填；
加查询改写，处理「那它的价格呢」；
加 OCR，支持扫描件；
把评估接进 CI，低于基线就失败。

小结

RAG 的瓶颈通常在解析、切分和召回，而不是生成。
混合检索 + 重排是性价比最高的一组改进。
引用要能定位到页码，且必须校验编号。
先测 Recall@k，再谈回答质量。

下一篇把工具能力标准化：写一个 MCP Server 和 Client。

#最终效果

#技术栈

#Step 1：解析 PDF，保留页码

#Step 2：建向量与 BM25 索引

#Step 3：混合召回 + RRF

#Step 4：带引用生成

#Step 5：构造评估集

#Step 6：评估脚本

#Step 7：把命令串起来

#调优顺序

#常见失败

#验收清单

#扩展任务

#小结

相关阅读

最终效果

技术栈

Step 1：解析 PDF，保留页码

Step 2：建向量与 BM25 索引

Step 3：混合召回 + RRF

Step 4：带引用生成

Step 5：构造评估集

Step 6：评估脚本

Step 7：把命令串起来

调优顺序

常见失败

验收清单

扩展任务

小结