Agent 开发实战（六）：RAG 进阶——文档检索增强做到生产级

第（二）篇的文档检索是"切 chunk → 向量化 → 余弦相似度"的最简版。能用，但容易漏召回、回答张冠李戴。本篇把它升级到生产可用的 RAG：混合检索、结果融合、重排、引文溯源。

1. 最简版的三个问题

只靠向量：语义相近能召回，但"精确关键词"（如型号 VCFv4.2、错误码 ERR_412）容易漏。向量对字面匹配反而弱。
没有重排：Top-K 是按相似度粗排的，最相关的不一定排第一，模型可能吃到"差一点"的片段。
没有引文：模型回答时引用了哪段文档？用户无法核实，出了问题难溯源。

下面逐个解决。

2. 分块策略：别无脑按字数切

第（二）篇用固定 size=800 切。更好的做法：

按结构切：Markdown 按标题（##、###）切，保留层级语义；
重叠保留：chunk 之间留 overlap，避免一句话被切断丢失上下文；
元数据：每个 chunk 记录来源文件名、章节、页码，供后续过滤与引文。

import re

def chunk_markdown(text, overlap=100):
    chunks, buf, cur_heading = [], [], "正文"
    for line in text.splitlines():
        if re.match(r"^#{1,3}\s", line):
            if buf: chunks.append((cur_heading, "\n".join(buf)))
            cur_heading = line.lstrip("# ").strip()
            buf = [line]
        else:
            buf.append(line)
    if buf: chunks.append((cur_heading, "\n".join(buf)))
    return chunks   # 再做 overlap 拼接，略

3. 混合检索：向量 + BM25

用 rank_bm25 做关键词检索，与向量检索各出一版排名，再用 RRF（Reciprocal Rank Fusion） 融合：

from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, chunks, embeddings):
        self.chunks = chunks
        self.embeddings = embeddings
        self.bm25 = BM25Okapi([c[1].split() for c in chunks])

    def search(self, query, top_k=5):
        # 向量召回
        q_emb = embed(query)
        vec_scores = [cosine(q_emb, e) for e in self.embeddings]
        vec_rank = self._rank(vec_scores)
        # 关键词召回
        bm_scores = self.bm25.get_scores(query.split())
        bm_rank = self._rank(bm_scores)
        # RRF 融合
        fused = {}
        for rank, idx in enumerate(vec_rank[:top_k*2]):
            fused[idx] = fused.get(idx, 0) + 1/(rank+1+60)
        for rank, idx in enumerate(bm_rank[:top_k*2]):
            fused[idx] = fused.get(idx, 0) + 1/(rank+1+60)
        return sorted(fused, key=fused.get, reverse=True)[:top_k]

RRF 不依赖分数绝对值，只依赖排名，所以能把"向量相似度"和"BM25 词频"这两种量纲不同的信号平稳融合。生产里这是性价比极高的一招。

4. 重排（Rerank）

融合后的 Top-K 仍可能"第一不是最相关"。加一层轻量重排：用 cross-encoder 对 (query, chunk) 打分重排。若不想引重模型，可用"向量相似度 × 0.7 + BM25 归一化 × 0.3"的加权重排作为廉价替代。

def rerank(query, chunks, top_k=3):
    # 示意：用 cross-encoder，无模型时退回融合分
    pairs = [(query, c[1]) for c in chunks]
    scores = cross_encoder.predict(pairs)   # 假设存在
    return [chunks[i] for i in sorted(range(len(chunks)), key=lambda i: scores[i], reverse=True)[:top_k]]

5. 引文溯源

检索到的 chunk 带上 doc_id 和位置，回答时要求模型标注引用，便于用户核实：

def build_context(chunks):
    out = []
    for i, (heading, text) in enumerate(chunks):
        out.append(f"[{i}]（来源：{heading}）\n{text}")
    return "\n\n".join(out)

并在 system prompt 里加一句："回答中涉及文档事实时，用 [n] 标注引用编号。" 这样最终回答形如"根据退款政策，渠道类订单 7 天无理由 [2]"，用户能直接跳回原文档核对。

6. 元数据过滤

用户问"只看 2026 年的实验文档"，可用 chunk 元数据（年份、类型）先过滤再检索，召回更准：

def search(self, query, filters=None):
    cand = [i for i,c in enumerate(self.chunks)
            if all(c.meta.get(k)==v for k,v in (filters or {}).items())]
    # 仅在 cand 内做向量+BM25
    ...

7. 评估检索质量

别凭感觉。固定 10~20 个真实问题 + 期望命中的文档，跑一遍看命中率：

def eval_retrieval(queries_with_expected):
    hits = 0
    for q, expected_doc in queries_with_expected:
        top = retriever.search(q, top_k=3)
        if any(expected_doc in c[0] for c in top):
            hits += 1
    return hits / len(queries_with_expected)

命中率低于预期，再调 chunk 策略或权重，而不是盲目上更重的模型。

8. 小结

把 RAG 做扎实，核心是四件事：

分块按结构、带元数据；
混合检索（向量 + BM25 + RRF）补上字面召回；
重排把最相关的顶到第一；
引文让回答可核实。

这些都不需要大模型参与检索本身，成本极低，却能把"答非所问"的概率显著压下去。

RAG 的瓶颈 90% 在"检索"，不在"生成"。先把检索做对，再谈模型。