在前面的文章里,我们已经把文档切分、向量检索和本地 RAG 闭环串了起来。但“检索到了内容”不等于“回答就可靠”:候选可能太少而漏掉答案,也可能包含许多相似却无关的片段。本篇只聚焦一个核心知识点:如何把召回、重排和引用组织成更可靠的 RAG 检索阶段。示例不依赖网络、密钥或第三方包,先用可观察的词项评分模拟流程。
召回与重排分别解决什么问题
召回(retrieval)追求的是覆盖率:从较大的知识库中快速找出一批“可能相关”的候选。向量数据库通常按照向量相似度返回前 k 条,k 取值较大时不容易漏掉证据,但也会带来更多噪声和上下文开销。
重排(reranking)发生在召回之后,目标是提高候选前几名的精确度。它可以使用更细致的相似度模型,也可以结合标题、标签、时间、权限等业务信号。两阶段设计的关键是:召回负责“别漏”,重排负责“排好”。如果正确片段没有进入候选集,重排再强也无法找回它。
引用则是另一条链路。每个片段从入库开始就应携带稳定的 source、标题或文档编号;生成上下文时保留这些标识,回答时要求模型只引用实际提供的来源。引用不是装饰,而是让读者和开发者能够回到原文核对,也方便定位“检索错了”还是“生成说错了”。
一个最小可运行示例
下面的程序用词项重叠模拟第一阶段召回,再用标题和正文的不同权重模拟第二阶段重排。它不代表生产级算法,却能清楚展示数据流。保存为 rerank_demo.py 后运行:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58
| from dataclasses import dataclass from re import findall
@dataclass(frozen=True) class Chunk: source: str title: str text: str
KNOWLEDGE = [ Chunk("deploy.md#env", "部署配置", "部署前应把数据库连接配置放入环境变量,不要写进代码。"), Chunk("deploy.md#logs", "部署日志", "排查部署异常时,先查看应用日志、请求时间和配置变更。"), Chunk("cache.md#ttl", "缓存过期", "缓存应设置合理的过期时间,并观察命中率和内存占用。"), Chunk("security.md#secret", "密钥保护", "访问令牌只能从环境变量读取,不能提交到代码仓库。"), ]
def terms(text: str) -> set[str]: """用中文双字符和英文单词构造教学用词项。""" words = set(findall(r"[a-zA-Z0-9_]+", text.lower())) compact = "".join(text.lower().split()) words.update(compact[i:i + 2] for i in range(len(compact) - 1)) return words
def overlap(query: str, text: str) -> float: query_terms = terms(query) text_terms = terms(text) return len(query_terms & text_terms) / max(len(query_terms), 1)
def retrieve(query: str, limit: int = 4) -> list[tuple[float, Chunk]]: """宽召回:保留较多候选,避免过早丢掉证据。""" ranked = [(overlap(query, chunk.title + chunk.text), chunk) for chunk in KNOWLEDGE] return sorted(ranked, key=lambda item: item[0], reverse=True)[:limit]
def rerank(query: str, candidates: list[tuple[float, Chunk]], limit: int = 2): """精排:标题匹配更重要,同时保留初始召回分。""" result = [] for recall_score, chunk in candidates: title_score = overlap(query, chunk.title) body_score = overlap(query, chunk.text) final_score = 0.5 * title_score + 0.4 * body_score + 0.1 * recall_score result.append((final_score, chunk)) return sorted(result, key=lambda item: item[0], reverse=True)[:limit]
question = input("请输入问题:").strip() candidates = retrieve(question) selected = rerank(question, candidates)
print("\n重排后的证据:") for score, chunk in selected: print(f"- {score:.3f} | [{chunk.source}] {chunk.title}: {chunk.text}")
|
运行命令如下:
输入“数据库连接配置放在哪里?”时,程序会打印带分数的候选和来源。这里没有预先写死运行结果;分数由本地 Python 根据输入实际计算。retrieve 的 limit=4 是宽召回,rerank 的 limit=2 是最终送入上下文的数量。调大或调小它们,可以观察覆盖率和噪声之间的取舍。
如何把引用带进生成上下文
实际接入模型时,不要只把 chunk.text 拼起来,而应把来源一起格式化。一个简单的上下文构造函数如下:
1 2 3 4 5
| def build_context(selected: list[tuple[float, Chunk]]) -> str: blocks = [] for index, (_, chunk) in enumerate(selected, start=1): blocks.append(f"[证据 {index} | 来源: {chunk.source}]\n{chunk.text}") return "\n\n".join(blocks)
|
随后将问题和 build_context(selected) 作为一次请求的输入,并在提示词中明确三条约束:只能依据证据回答;证据不足时直接说明无法确定;每个关键结论后标注对应的证据编号。模型输出的引用编号仍然需要程序校验:编号必须存在于本次上下文,不能允许模型凭空创造文档链接。若需要展示原文页面,应用程序应通过 source 在自己的索引中生成链接,而不是信任模型提供的 URL。
常见问题
召回数量越大越好吗? 不是。数量太小会漏掉答案,数量太大则增加噪声、延迟和输入成本。可以先用离线问题集观察“正确片段是否进入候选”,再选择一个足够覆盖的召回数量;之后通过重排压缩到较小的上下文。
向量分数高就一定相关吗? 不一定。语义相似度只是信号,短文本、重复内容和术语相近的错误片段都可能得分很高。业务过滤、关键词约束、时间范围和权限判断应在检索流程中明确处理。
为什么答案有引用仍可能错误? 模型可能引用了正确来源,却错误理解或拼接了原文。因此要区分“引用存在”和“结论被证据支持”两种检查。对重要场景,可让模型逐条列出证据,再由规则或人工复核结论。
该在重排前做权限过滤吗? 是。用户无权访问的片段不应进入候选,更不能依赖提示词要求模型忽略它们。权限过滤应在服务端根据可信身份完成,并在日志中记录过滤结果。
小结
召回、重排和引用是三个相互衔接但职责不同的环节:召回扩大覆盖,重排改善候选顺序,引用让答案能够回到原文核对。本文用纯 Python 演示了宽召回、加权重排和来源保留的最小实现。生产环境可以替换为 Embedding 检索和专门的重排模型,但应保留同样的边界:先验证证据是否被召回,再验证排序和生成,最后检查引用是否真实存在且能够支持结论。这样排查 RAG 问题时,才不会把所有错误都归咎于提示词。