RAG 检索不再答非所问:混合检索 + 智能重排序的四站流水线
【免费下载链接】RAG_TechniquesThis repository showcases various advanced techniques for Retrieval-Augmented Generation (RAG) systems. Each technique has a detailed notebook tutorial.项目地址: https://gitcode.com/GitHub_Trending/ra/RAG_Techniques
给内部支持机器人提了个单:“出差打车能报多少?”机器人回了一段产品功能介绍。语料库里明明躺着完整的报销政策,向量相似度却把产品手册的“打车”段落顶到了第一位——词对上了,意思全跑偏。这类翻车不是生成模型的错,是检索阶段就没把对的段落递过去:召回错了,后面再强的 LLM 也只能在错误的前提上编。RAG 高级检索要补的就是这一环:用混合检索把该找的找全,用智能重排序把顺序掰正,再配上过滤与分层索引控制成本。
下面按一条真实查询走过的流水线拆四站:召回 → 精排 → 清洗 → 导航,每站给最小可跑的代码,参数怎么拧、坑在哪里一并说清。
第一讲 · 混合检索:BM25 和向量各管一段,alpha 该给多少
单独用向量检索,"报销标准"这种精确词容易召回一堆"报销流程简介";单独用关键词检索,"出差住宿能报几块"这种换词表达又抓不到。两个通道一个认词一个认意,谁也不能替谁干活,加权混排是最划算的修法。仓库里的all_rag_techniques/fusion_retrieval.ipynb就是这条路线的实现。
最小实现,约 24 行:
import numpy as np from rank_bm25 import BM25Okapi def minmax(arr): lo, hi = arr.min(), arr.max() return (arr - lo) / (hi - lo + 1e-9) def blended_hits(vs, bm25, q, k=5, alpha=0.5): """alpha 越大越偏向量;两路分数先归一再加权""" pool = vs.similarity_search(q, k=60) # 全库取出一次,方便按 BM25 下标对齐 corpus = vs.similarity_search("", k=vs.index.ntotal) kw = minmax(bm25.get_scores(q.split())) vec = 1 - minmax(np.array([s for _, s in vs.similarity_search_with_score(q, k=len(corpus))])) merged = alpha * vec + (1 - alpha) * kw idx = np.argsort(merged)[::-1][:k] return [corpus[i] for i in idx]注意一个细节:FAISS 的分数是"距离",越小越近,所以向量那路做了反向归一化。两路量纲不同,归一化必须在"当前候选池"上做,跨池子拿历史 min/max 会串味。分块端建议相邻块留 10%~20% 重叠,防止跨边界的句子被拦腰截断。
🎛️ alpha 怎么拧:
| 你的查询长什么样 | 起手 alpha | 拧法 |
|---|---|---|
| 精确词多:编号、型号、法条 | 0.3 | BM25 压阵,别让它被语义漂走 |
| 概念多、同义改写多 | 0.6~0.7 | 语义通道挑大梁 |
| 判不准 | 0.5 | 先跑 20 条真实 query 看命中再微调 |
alpha 别指望调到"最优解"就收工——线上查询分布会漂,留个按查询类型分档的开关。
候选进来了,顺序还是乱的——评委上场。
第二讲 · 智能重排序:LLM 当评委,候选池留多大
混合检索捞回来的 20 个分块,前两名未必是真答案。让 LLM 逐对读"查询 × 分块",比双塔向量多一层交互,看得更细。两阶段架构的思路很直接:向量通道负责便宜地"找得广",LLM 负责贵一点地"排得准"。
评分提示词是重排序里最容易被低估的零件。要点三条:temperature 设 0 保证同一对内容打分稳定;用结构化输出接分数,别做正则解析;让模型同时给一句判断理由,调试时能直接看出它为什么打高分。仓库参考all_rag_techniques/reranking.ipynb。
from pydantic import BaseModel, Field from langchain_openai import ChatOpenAI class Verdict(BaseModel): """评委的打分单""" score: float = Field(ge=1, le=10, description="与查询的贴合度") reason: str = Field(description="一句话说明给分依据") JUDGE = ( "你是文档检索的评审。只依据内容判断,不要常识脑补。\n" "档位参考:9-10 直接回答问题;5-8 提供部分支撑;1-4 跑题。\n" "查询:{q}\n分块:{text}\n" "给出评分和一句理由。" ) judge = ChatOpenAI(model="gpt-4o-mini", temperature=0) def llm_rerank(q, pool, top=5): """逐对送审,只留分数过线的""" chain = judge.with_structured_output(Verdict) scored = [(d, chain.invoke(JUDGE.format(q=q, text=d.page_content)).score) for d in pool] scored.sort(key=lambda t: t[1], reverse=True) return [d for d, s in scored[:top] if s >= 5]20 个候选逐个打分是串行的 20 次调用,延迟会很难看。分批并发是标准解法:
from concurrent.futures import ThreadPoolExecutor def parallel_judge(q, pool, top=5, batch=4): def _one(doc): v = judge.with_structured_output(Verdict).invoke(JUDGE.format(q=q, text=doc.page_content)) return doc, v.score with ThreadPoolExecutor(max_workers=4) as ex: pairs = list(ex.map(_one, pool)) pairs.sort(key=lambda t: t[1], reverse=True) return [d for d, s in pairs[:top] if s >= 5]📦 候选池(Top-K)留多大:
| 池子 | 延迟 | 风险 |
|---|---|---|
| 10 | 低 | 真答案没被召回进来时,精排也救不回来 |
| 20 | 中 | 常规知识库的稳妥默认 |
| 40 | 高(约 200 次 LLM 调用) | 只有长尾查询或超大语料才值得 |
原则是池子宁大勿小但别贪心:精排只能救排序问题,救不了召回问题。
评委排完,还有一批"沾边"的——过闸。
第三讲 · 多维过滤:阈值取 0.6 还是 0.8
重排之后常见的残留:同一页的三个近义分块、来源不对的文档、相似度 0.61 的边缘结果。三道闸解决这三类,顺序不能乱:先元数据、再阈值、后打散——元数据最便宜,放在前面能省掉后面两道闸要算的分数。
🚦 元数据闸:进任何打分之前先把不属于这个查询的文档剔掉——按来源、日期、权限。这一步是纯字段比对,零模型调用,省下的都是实打实的 embedding 和 LLM 开销。
📏 相似度闸:0.6 还是 0.8 别拍脑袋,跟查询类型和库大小走。
import math def similarity_floor(qtype, n_docs): base = 0.72 if qtype == "fact" else 0.62 bump = 0.04 * math.log10(n_docs + 1) return round(min(base + bump, 0.9), 2)🧺 打散闸:MMR 把"和已入选分块有多像"算进得分,同一段话的三个版本只留一个。
import numpy as np def scatter(docs, embs, k=5, lam=0.75): """相关性 lam、多样性 1-lam,避免近义分块挤占窗口""" embs = np.array(embs, dtype=float) embs = embs / (np.linalg.norm(embs, axis=1, keepdims=True) + 1e-9) picked = [int(np.argmax([d.score for d in docs]))] rest = list(range(len(docs))) while len(picked) < k and rest: best, best_v = None, -1e9 for i in rest: red = max(embs[i] @ embs[j] for j in picked) v = lam * docs[i].score + (1 - lam) * (1 - red) if v > best_v: best, best_v = i, v picked.append(best) rest.remove(best) return [docs[i] for i in picked]第四讲 · 分层索引:摘要层、概览层、明细层怎么逐级下钻
语料库上千个文件时,扁平检索有两个毛病:贵(全量分块都要打分)和吵(摘要级的意图被细节分块淹没)。分层索引的思路是把文档切成三层:摘要层管"这份文档讲什么",概览层管"哪个章节相关",明细层管"原文怎么说的"。检索时从上往下钻,每层只放行过线的文档,把打分范围越压越小。仓库里all_rag_techniques/hierarchical_indices.ipynb给的是两层版:先查摘要、再按页码回捞明细,原理完全一致。
def drill_down(summary_vs, detail_vs, q, k=5): """摘要层先筛,再回捞对应文档的明细分块""" top_sum = summary_vs.similarity_search(q, k=10) keep = {s.metadata["doc_id"] for s in top_sum if s.metadata["score"] <= 0.25} # 分数是距离,越小越近 hits = [] for did in list(keep)[:3]: # 只下钻前 3 份文档 hits += detail_vs.similarity_search(q, k=k, filter={"doc_id": did}) return hits下钻策略跟查询类型走:事实型问题直接查明细层,省掉一层往返;需要跨章节综合的问题才从摘要层一路钻下去。动态下钻不是"每次都走满三层",而是按问题的粒度选入口层。
落地速查:参数一次看全
📋 参数速查
| 旋钮 | 起手值 | 管什么 | 拧的方向 |
|---|---|---|---|
| alpha | 0.5 | 混合检索里向量通道的权重 | 精确词多降、概念多升 |
| 候选池 | 20 | LLM 评委要看多少个分块 | 精度吃紧加、成本吃紧减 |
| 相似度下限 | 0.62~0.72 | 相似度闸的放行线 | 按查询类型动态算 |
| lambda_mmr | 0.75 | 打散时相关性与多样性的配比 | 结果重复多就调大权重给多样性 |
| 分块大小 | 800~1000 字符 | 明细层的颗粒度 | 配合 10%~20% 重叠 |
三个最常踩的坑:
🕳️ 分数不归一直接相加:BM25 分是 0 到几十,余弦是 0 到 1,不加归一化时 alpha 形同虚设,BM25 一路把结果焊死。
🕳️ 评委看全库:跳过过滤直接把全部分块送 LLM 打分,成本翻一个数量级。正确顺序永远是:元数据 → 阈值 → 打散 → 精排。
🕳️ 阈值一刀切:事实型问题用 0.8 的线,探索型问题用 0.6 的线,混着用两边都不讨好。
四站都是可插拔的插件,不是焊死的整体。先把混合检索的 alpha 拧到 20 条真实查询不再翻车,再一站一站往上叠——顺序错了,后面的精排和分层都是在给前面的错误买单。
【免费下载链接】RAG_TechniquesThis repository showcases various advanced techniques for Retrieval-Augmented Generation (RAG) systems. Each technique has a detailed notebook tutorial.项目地址: https://gitcode.com/GitHub_Trending/ra/RAG_Techniques
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考