从零实现一个最小可用 RAG 系统

完整链路

RAG 不是”接个向量库”这么一件事，它是一条有八个环节的流水线，任何一环掉链子都会体现在最终答案上：

文档解析 → 切块 → embedding → 入库 → 检索 → 重排 → 拼装 → 生成

解析环节最容易被低估。PDF 转文本会丢掉表格结构、把双栏排版读成左右交错、把页眉页脚混进正文。扫描件还得过 OCR。解析质量差，后面所有优化都是白做——检索到的原文本身就是乱的。

切块决定检索的最小粒度。embedding 决定语义空间的几何。入库环节要带上元数据（来源、时间、权限标签），因为过滤要在检索时做。检索环节产出候选。重排把候选里真正相关的提到前面。拼装决定模型看到什么。生成决定答案怎么用这些材料。

切块参数怎么定

三个参数：块大小、重叠、切分依据。

块大小是召回率和精度的拉扯。块太小，语义不完整，一段话被切断后单独编码出来的向量可能指向完全不同的方向，检索时召回的是碎片，模型拼不出完整答案。块太大，一个块里混入多个主题，向量是多个语义的平均，跟任何一个具体问题的相似度都不高，召回时被更”纯粹”的其他块挤掉。

经验起点是 300 到 800 个 token，具体值取决于文档类型：FAQ 类一条一块最合适；技术手册按小节切；法律合同按条款切。没有通用最优值，得用评测集扫一遍。

重叠的作用是防止语义被切断。相邻块之间保留 10% 到 20% 的重复内容，让跨边界的语句至少在一个块里完整出现。重叠的代价是索引膨胀和检索结果冗余——同一个句子可能出现在两三个块里，全部塞进上下文会浪费窗口。检索后做一次去重能缓解。

按语义切比按长度切好：优先在段落、标题、句号处断开，实在太长才强制切。实现上先按标题分层，层内再按段落合并，直到接近目标长度。这份额外的代码量换来的召回提升通常很明显。

def chunk(text, encode, decode, size=512, overlap=64):
    """按 token 切块，带重叠。encode/decode 为占位接口。"""
    ids = encode(text)
    step = size - overlap
    chunks, i = [], 0
    while i < len(ids):
        seg = ids[i:i + size]
        if seg:
            chunks.append(decode(seg))
        if i + size >= len(ids):
            break
        i += step
    return chunks

检索召回不足的排查顺序

“问了没答上来”是 RAG 最常见的故障，但原因可能在五个不同的位置。按顺序排查，别跳步：

一、答案所在的块到底有没有被索引。直接拿问题里的关键词去库里做一次字符串匹配。如果匹配不到，问题在解析或切块——可能是那份文档根本没入库，或者表格内容在解析时丢了。这一步能排除掉相当比例的问题，而且成本极低。

二、块在库里，但检索排不进来。把 top_k 临时放大到 100，看目标块排在第几位。如果排在几十位开外，说明向量相似度确实低，要查 embedding 是否合适（模型是否支持该语言、查询和文档是否用了同一个模型编码）。

三、查询和文档的语义鸿沟。用户问的是口语化短句，文档写的是书面长句。这时候查询改写（用模型把问题扩写成若干候选查询，分别检索后合并）或者 HyDE 式的假设性文档生成能拉近距离。

四、过滤条件误伤。检查元数据过滤是不是筛掉了正确文档。带过滤的检索很容易出现召回塌陷，见向量索引那篇。

五、切块粒度问题。答案需要跨三个块才能拼出来，而只召回了其中一个。这时要么增大块，要么引入”父块检索”——用小粒度块做检索，命中后返回它所属的大粒度父块给模型。

排查的前提是有可测量的召回率，所以评测集要先建。

重排为什么改善明显

向量检索是双塔结构：查询和文档各自独立编码成一个向量，相似度就是一次点积。查询和文档之间没有任何交互，所有语义压缩都发生在编码阶段。这个设计决定了它能做海量数据的快速检索，也决定了它的精度上限。

重排（rerank）是交叉编码器：把查询和文档拼在一起过一遍模型，让两者的 token 互相做注意力，直接输出一个相关性分数。

# 双塔：score = f(q) · g(d)          查询文档无交互，可预计算文档向量
# 交叉：score = h(q ⊕ d)             查询文档充分交互，必须在线计算

代价是交叉编码无法预计算文档表示，每对 (查询, 文档) 都要跑一次前向，所以只能用在候选集上。典型两阶段：向量检索召回 top-50 到 top-200，重排取 top-5 送进上下文。

效果改善明显的原因在于，双塔的向量相似度容易被表层词汇相似度主导，而交叉编码能识别”这个问题和这段虽然用词不同但说的是一件事”。实测里重排对长尾问题和需要推理的匹配提升最大。

提示词模板怎么约束”仅依据资料回答”

光写一句”仅依据上述资料回答”不够，模型会在资料不足时自动补全。有效的约束要给出明确的失败出口：

PROMPT = """依据下方编号资料回答问题。

规则：
1. 不得使用资料之外的知识。资料不足以回答时，输出 {{"answer": null, "reason": "缺少XX信息"}}。
2. 每个事实性陈述后标注来源编号，格式 [3]，可多标。
3. 资料中出现的具体数值、日期、名称必须原样引用，不得换算或改写。
4. 不同资料冲突时，指出冲突并说明各自来源，不要自行选择。
5. 只输出 JSON，不要解释。格式：{{"answer": "...", "sources": [1, 3]}}

资料：
{context}

问题：{question}
"""

规则 2 的引用标注不只是可溯源，它还有个副作用：模型为了标引用必须回溯到具体片段，注意力会更集中于资料本身， hallucination 概率下降。

规则 1 里的 null 出口是关键。如果不给”答不上来”的合法路径，模型在资料不足时只能硬编一个答案。给了之后，宁可答 null 也不编的比例会明显上升。

拼装上下文时，把重排得分最高的片段放在开头和结尾，中间放次要的。模型对上下文首尾位置的内容更敏感，中段容易被忽略。

代码骨架

from dataclasses import dataclass, field
from typing import Protocol, List, Optional

class Embedder(Protocol):
    def embed(self, texts: List[str]) -> List[List[float]]: ...

class VectorStore(Protocol):
    def upsert(self, ids: List[str], vectors, payloads: List[dict]) -> None: ...
    def search(self, vector, top_k: int, filters: Optional[dict] = None) -> List[tuple]: ...

class Reranker(Protocol):
    def rerank(self, query: str, docs: List[str], top_k: int) -> List[int]: ...

class LLM(Protocol):
    def generate(self, prompt: str) -> str: ...

@dataclass
class Config:
    chunk_size: int = 512
    chunk_overlap: int = 64
    retrieve_k: int = 50      # 向量检索候选数
    rerank_k: int = 5         # 重排后进入上下文的数量

class MiniRAG:
    def __init__(self, embedder: Embedder, store: VectorStore,
                 reranker: Reranker, llm: LLM, cfg: Config = field(default_factory=Config)):
        self.e, self.s, self.r, self.m, self.cfg = embedder, store, reranker, llm, cfg

    def index(self, doc_id: str, text: str, meta: dict,
              encode, decode) -> int:
        chunks = chunk(text, encode, decode, self.cfg.chunk_size, self.cfg.chunk_overlap)
        if not chunks:
            return 0
        vecs = self.e.embed(chunks)
        ids = [f"{doc_id}#{i}" for i in range(len(chunks))]
        payloads = [{"doc_id": doc_id, "text": c, **meta} for c in chunks]
        self.s.upsert(ids, vecs, payloads)
        return len(chunks)

    def retrieve(self, question: str, filters: Optional[dict] = None):
        qv = self.e.embed([question])[0]
        cands = self.s.search(qv, self.cfg.retrieve_k, filters)
        docs = [p["text"] for _, p in cands]
        keep = self.r.rerank(question, docs, self.cfg.rerank_k)
        return [cands[i][1] for i in keep]

    def answer(self, question: str, filters: Optional[dict] = None) -> str:
        hits = self.retrieve(question, filters)
        ctx = "nn".join(f"[{i+1}] {h['text']}" for i, h in enumerate(hits))
        return self.m.generate(PROMPT.format(context=ctx, question=question))

接口全部用 Protocol 声明，接哪家实现都不影响上层逻辑。这样做的另一个好处是评测时能轻松替换组件——把 Reranker 换成恒等映射，就能量化重排到底贡献了多少。

最小评估方案

二十到五十条问答对就够了，关键在怎么造和怎么测。

取样要覆盖四类：能用单个片段直接回答的（测基础召回）；需要跨片段拼接的（测切块与重排）；资料里根本没有答案的（测是否会编）；资料里信息相互冲突的（测约束是否生效）。第四类最容易被漏掉，也最能暴露提示词问题。

每条问答对标注它对应的正确文档 id（或正确片段）。有了这个标注，检索侧就能算出可自动化的指标：

def recall_at_k(rag, qa_pairs, k=50):
    hit = 0
    for q, gold_doc_id in qa_pairs:
        cands = rag.retrieve(q)
        hit += any(c["doc_id"] == gold_doc_id for c in cands)
    return hit / len(qa_pairs)

生成侧的指标分开算：格式合规率（能否 parse 成 JSON）、引用准确率（标注的编号是否真的支持该陈述）、拒答正确率（无答案组里有多少输出了 null）。前两个能自动算，后两个抽一部分人工看。

有了这套基线，改切块大小、换 embedding、加不加重排，每个改动的影响都能量化。没有基线的话，所有优化都是凭感觉。

上线前补上的几件事

超时和降级：检索超时就跳过重排直接用向量结果，生成超时就返回检索片段列表。半截答案比错误答案有用。

去重：重排后多个片段可能高度重叠，做一次相似度去重再拼装，能省下不少上下文。

缓存：相同问题的检索结果可以缓存，但缓存键要包含知识库版本号，否则文档更新后还在返回旧片段。

日志：每次问答记录被召回的片段 id 和重排分数。线上出问题时，这批日志是唯一的排查依据。