大模型 RAG 检索管道工程 切分、召回与重排

RAG 的效果差异九成不在模型，而在检索管道。同一个模型配不同的切分与召回策略，答案质量可以差好几倍。本文按管道顺序讲清五个环节的工程决策：切分、向量化、混合召回、重排、上下文组装，以及每一环怎么量化验收。

一、先建立评估集，再动管道

没有评估集的 RAG 调优是玄学。最小评估集只需要 50 到 100 条真实问答对，三个指标就够：

recall@k    正确文档是否出现在前 k 个召回结果里（召回环节）
MRR         正确文档排位倒数的平均值（排序环节）
answer faithfulness  答案是否只依据检索到的上下文（生成环节）

前两个可以纯离线自动评估，不消耗 LLM 调用额度，要挂在 CI 上。第三类可用另一个模型做裁判（LLM as judge），成本高些，按周跑。有了这条基线，每次改参数才知道是变好还是变坏。

二、切分：结构优先于长度

最普遍的初级问题就是按固定字符数暴力切分，把一段完整说明切成两半。

优先级顺序是：按文档结构切（标题、段落、代码块、表格边界），结构不可用时按段落聚合，最后才按长度兜底。

def split_by_heading(doc: str, max_chars=1200, overlap=150) -> list[dict]:
    chunks, cur, path = [], [], []
    for line in doc.splitlines():
        if line.startswith("#"):
            level = len(line) - len(line.lstrip("#"))
            path = path[: level - 1] + [line.lstrip("# ").strip()]
        cur.append(line)
        if len("\n".join(cur)) > max_chars:
            chunks.append({"text": "\n".join(cur).strip(),
                           "heading": " > ".join(path),
                           "source": doc[:64]})
            cur = cur[-overlap:] if overlap else []
    if cur.strip():
        chunks.append({"text": "\n".join(cur).strip(), "heading": " > ".join(path)})
    return chunks

三个经验值：块大小 400 到 1200 字符（中文可到 800 到 1500 字），重叠 10 到 15%，每个块必须携带元数据（来源、标题路径、更新时间）。

标题路径要拼进待嵌入文本，而不是只存元数据：

embed_text = f"{c['heading']}\n{c['text']}"

这让"这一节在讲什么"进入向量，对短块的召回质量提升明显，成本为零。

代码、表格这类结构不要按长度切。表格按行组切并把表头每段重复带上；代码按函数边界切。切坏的表格与代码是低质量答案的主要来源。

三、混合召回：向量不够，必须加词法

纯向量召回有一个稳定缺陷：精确匹配差。型号名、错误码、函数名、专有名词这类查询，embedding 相似度经常输给语义相近但语义不符的段落。

正确做法是向量召回 + BM25（或全文索引）并做融合，用 RRF 最省事且不需要调分数标度：

def rrf(vector_ids, bm25_ids, k=60, weight=(0.6, 0.4)):
    scores = {}
    for ids, w in zip((vector_ids, bm25_ids), weight):
        for rank, did in enumerate(ids):
            scores[did] = scores.get(did, 0) + w / (k + rank + 1)
    return [did for did, _ in sorted(scores.items(),
            key=lambda x: -x[1])]

多数数据库现在已经内置了混合检索，可以直接用：

-- PostgreSQL: 向量 + 全文，应用层 RRF 或数据库内加权
SELECT id, content,
       1 - (embedding <=> $1::vector) AS vec_score,
       ts_rank(search_tsv, plainto_tsquery('simple', $2)) AS lex_score
FROM chunks
WHERE embedding <=> $1::vector < 0.45
   OR search_tsv @@ plainto_tsquery('simple', $2)
ORDER BY (0.6 * (1 - (embedding <=> $1::vector))
        + 0.4 * ts_rank(search_tsv, plainto_tsquery('simple', $2))) DESC
LIMIT 30;

召回候选取 30 到 100 条，宁可宽一点，把精度交给重排。这一层的优化目标是 recall@50，不是精度。

四、重排：性价比最高的一步

召回要快，粗排有误差；Cross-Encoder 把 query 与候选拼在一起打分，精度高但慢，因此只对召回后的 30 到 50 条打分。典型收益是在 recall 不变的前提下把正确文档推进前 3 位，而代价只有几十毫秒。

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", device="cuda")

def rerank(query, cands, top=5):
    pairs = [(query, c["text"]) for c in cands]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, cands), key=lambda x: -x[0])
    return [c for _, c in ranked[:top]]

自托管模型成本高时可用云端 rerank API；再降级可用"LLM 打分"做替代，但要注意时延。另一个几乎免费的技巧是分数阈值截断：重排后把明显低于阈值的候选丢弃，宁缺毋滥——上下文塞进无关内容导致答案被污染，比检索不到更糟。

五、上下文组装与失效治理

组装阶段三件事：

按相关性排序后仍给稳定顺序（例如按文档时间或标题排序），避免同一问题因为召回顺序抖动而给出不同答案。

加来源标注：每段带来源与时间，便于生成答案时引用，也便于人工核对。

prompt_ctx = "\n\n".join(
    f"[{i+1}] 来源: {c['source']} · {c['heading']}\n{c['text']}"
    for i, c in enumerate(top_chunks)
)

处理知识更新：块级元数据带 updated_at 与版本，检索按最新版本过滤。旧版本内容与新版混在一起，是"答案看着对但已过时"的直接原因。

生成侧再配一层兜底：提示词明确要求"上下文不含答案时必须回答不知道"，并检测答案里的引用是否真实存在于上下文，引用不存在的块要能在监控里被统计出来——这是幻觉率最便宜的代理指标。

六、监控与迭代顺序

线上要盯四组指标：召回为空率、重排后最高分分布、答案带引用率、人工点踩率。前两个纯统计即可，能极早发现问题（比如换 embedding 模型导致分数分布整体漂移，相似度阈值突然失效）。

阈值要随分布调整，不能一次写死。经验做法是定期用评估集重新标定，把阈值存配置中心而非硬编码。

优化投入的优先级（按性价比排序）非常稳定：先修切分质量，再加 BM25 混合召回，然后上重排，最后才考虑微调 embedding 或换模型。绝大多数团队在第二步就拿到最大收益，而把预算花在第四步上。

小结

RAG 的工程要点是五个环节各自的正确做法：结构优先的切分并把标题路径嵌入向量、向量加词法的混合召回并用 RRF 融合、召回宽、重排严、组装带来源与版本并对无关内容截断、以及上线 recall@k 与 MRR 的离线评估集。顺序上先治切分再加混合召回再上重排，最后才是换模型。把评估集挂进 CI，管道改造就变成可度量的工程问题，而不是靠感觉调参。