ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

RAG 进阶:重排序 + 证据过滤闭环,从根源解决大模型幻觉

RAG 进阶:重排序 + 证据过滤闭环,从根源解决大模型幻觉 前言为什么你的混合 RAG 依然产生幻觉上一章我们实现了向量检索 BM25 关键词检索 RRF 结果融合的混合检索。 很多同学做到这里以为 RAG 已经完成但实际业务跑起来依旧幻觉严重。混合检索输出的只是候选证据不是可用证据。 召回只求 “不要漏掉有用文档”并不保证排在前面的文档一定适合回答用户问题。如果把一堆低相关、无关的候选直接喂给大模型模型在缺少有效依据时就会脑补编造内容这就是 RAG 幻觉一大来源。本章要解决两件事重排序 (Rerank)使用 Cross‑Encoder 重新计算问题和每条候选证据的真实相关性把真正有用的排到最前面证据过滤 拒答设置分数阈值过滤低质量证据没有合格证据就直接拒答绝不强行回答最终形成链路多路召回 → RRF融合 → Cross‑Encoder重排序 → 分数归一化 → 阈值过滤 → 证据不足拒答涉及文件清单backend/app/config.py重排序相关配置backend/app/services/pipeline_utils.py归一化、证据过滤工具backend/app/services/retrieval.py重排序、检索主逻辑、query 改写backend/scripts/check_rerank_filter.py多模式对比测试脚本一、召回 vs 重排序两者目标完全不一样1. 召回阶段向量 / BM25目标尽量不要漏掉可能有用的内容允许把一些相关性一般的文档捞回来追求高召回率。2. 重排序阶段Cross‑Encoder目标从候选里面挑选真正适合回答问题的内容追求精准。举个业务例子用户问题数字化补贴申报材料混合检索召回多条结果第二条 支持范围第三条 补贴标准第四条 申报材料 ✅真正需要第五条 申报时间向量 / BM25 可能把 “补贴标准” 排到第一 重排序会识别出第四条申报材料和问题匹配度最高把它置顶。召回负责 “大海捞候选”重排序负责 “候选里面挑精品”。二、Cross‑Encoder 与 Embedding 的区别表格模型使用方式特点Embedding向量模型问题、文档分别编码向量再算相似度速度快适合全库检索精度一般Cross‑Encoder 重排序模型输入(query, document)问答对直接输出相关性分数精度更高速度慢只能对少量候选打分不能直接全库检索工程范式Embedding/BM25 做召回Cross‑Encoder 做小集合重排序。三、四种检索模式整体架构vector仅向量检索无 BM25无重排序无过滤hybrid向量 BM25 RRF 融合不重排序、不过滤hybrid_rerank混合召回 RRFCross‑Encoder 重排序不做阈值过滤full完整链路混合召回 RRF 重排序 分数归一化 阈值过滤 证据不足拒答生产推荐四、工程实现完整落地4.1 配置文件 config.pyfrom pydantic_settings import BaseSettings class Settings(BaseSettings): # 重排序模型可使用BAAI/bge‑reranker-base / qwen3‑rerank reranker_model: str BAAI/bge-reranker-base # 是否开启重排序 enable_reranker: bool True # 重排序之后保留多少条证据 rerank_top_k: int 4 # 证据最低分数阈值低于该分数会被过滤 min_evidence_score: float 0.25 class Config: env_file .env.env环境变量示例RERANKER_MODELqwen3‑rerank ENABLE_RERANKERtrue RERANK_TOP_K4 MIN_EVIDENCE_SCORE0.25注意第一次启动会自动下载模型或者把模型下载到本地使用本地路径加载避免线上环境无法访问 huggingface。4.2 工具函数 pipeline_utils.py 归一化 证据过滤Cross‑Encoder 输出的是 logit 原始分数范围不是 0‑1我们需要归一化统一打分口径。import math from typing import Any def normalize_logit(value: float) - float: sigmoid归一化把任意实数映射到 [0,1] max(-30, min(30, value))限制区间防止math.exp数值溢出 return 1 / (1 math.exp(-max(-30, min(30, value)))) def filter_evidence(evidence: list[dict[str, Any]], min_score: float) - list[dict]: 根据阈值过滤证据只保留分数min_score的证据 设计思想宁可拒答也不要低质量证据进入大模型 return [ item for item in evidence if float(item.get(evidence_score, 0)) min_score ]4.3 retrieval.py重排序模型加载、rerank 函数、查询改写模型加载 lru_cache 全局单例重排序模型体积大不能每次请求重复加载使用缓存只加载一次。local_files_onlyTrue强制只用本地模型文件不在线拉取。from functools import lru_cache from sentence_transformers import CrossEncoder from app.config import get_settings lru_cache def get_reranker(): 获取重排序模型单例全局只加载一次 settings get_settings() return CrossEncoder(settings.reranker_model, local_files_onlyTrue) # 如果需要自定义本地绝对路径 # def get_reranker(): # model_path rD:\models--BAAI--bge-reranker-base\snapshots\2cfc18c9415c912f9d8155881c133215df768a70 # return CrossEncoder(model_path, local_files_onlyTrue)rerank 重排序主函数带异常降级模型 OOM、文件损坏、加载失败不能让整个接口崩溃使用 RRF 分数做降级兜底。import asyncio from app.services.pipeline_utils import normalize_logit async def rerank(query: str, candidates: list[dict], top_k: int) - list[dict]: Cross‑Encoder重排序 :param query: 用户改写后的查询 :param candidates: RRF融合后的候选证据列表 :param top_k: 重排序后保留条数 :return: 重排序完成的证据列表 if not candidates: return [] settings get_settings() if not settings.enable_reranker: # 开关关闭直接截取前N条返回 return candidates[:top_k] try: reranker_model get_reranker() # 组装问答对 [(query,doc_content), ...] pairs [(query, item[content]) for item in candidates] # 使用to_thread把同步predict放到线程运行不阻塞asyncio事件循环 scores await asyncio.to_thread(reranker_model.predict, pairs) for item, score in zip(candidates, scores): raw_score float(score) item[rerank_raw_score] raw_score item[rerank_score] normalize_logit(raw_score) # 按归一化分数倒序排序取top_k return sorted(candidates, keylambda x: x[rerank_score], reverseTrue)[:top_k] except Exception as e: # 异常降级模型出错时复用RRF分数放大充当伪重排分数保证链路可用 print(f[rerank warn] 重排序异常触发降级: {e}) for item in candidates: item[rerank_score] min(1.0, item.get(rrf_score, 0) * 30) return candidates[:top_k]Query 改写函数口语化问题转为检索查询用户输入口语LLM 改写为适合知识库检索的简短 query失败直接返回原问题。from app.services.model_factory import create_chat_model, invoke_text async def rewrite_query(question: str) - str: 把口语化问题改写成更适合知识库检索的简洁查询异常退回原问题 prompt 将下面的政务咨询改写为一个适合知识库检索的简洁中文查询。 不得添加问题中没有的地区、政策或条件只输出改写结果。 问题{question} try: llm create_chat_model(get_settings(), temperature0) content, _ await invoke_text(llm, prompt.format(questionquestion)) return content.strip() except Exception: return question4.4 retrieve 主流程整合重排序、分数统一、证据过滤、拒答关键流程片段完整代码参考原文async def retrieve( session: Session, question: str, filters: RetrievalFilters, top_k: int 8, evidence_top_k: int 4, min_score: float 0.25, mode: str full ) - RetrievalResult: # 1.查询改写 new_question await rewrite_query(question) # 2.向量检索 vector_results await VectorStoreService().search( new_question, top_k * 2, document_idsfilters.document_ids or None ) vector_results await apply_metadata_filters(vector_results, filters) for item in vector_results: item[source] vector keyword_list [] if mode ! vector: # 3.BM25关键词检索 keyword_list await keyword_search(session, new_question, top_k*2, filters) # 4.RRF融合两路召回结果 fused [vector_results[:top_k], keyword_list] rrf_list reciprocal_rank_fusion(fused) candidates rrf_list[:top_k] # 5.父块上下文扩展子块补齐父文档内容 evidence [] for item in candidates: score max(0.0, min(1.0, float(item.get(score, 0)))) if score min_score: item[evidence_score] score evidence.append(item) evidence await expand_parent_context(session, evidence) # 保证送入重排序的候选数量足够不能太少 evidence evidence[:max(evidence_top_k, top_k)] # 6.分模式是否执行重排序 if mode in [hybrid_rerank, full]: evidence await rerank(new_question, evidence, evidence_top_k) else: evidence evidence[:evidence_top_k] # 7.统一 evidence_score 字段多模式兼容 # 优先级rerank_score score rrf_score*30 for item in evidence: normalized item.get(rerank_score) if normalized is None: normalized item.get(score, item.get(rrf_score, 0) * 30) item[evidence_score] max(0.0, min(1.0, float(normalized))) # 8.full模式执行证据阈值过滤 if mode full: evidence filter_evidence(evidence, min_score) # 9.证据为空则标记拒答 refused len(evidence) 0 refusal_reason 没有查询到合适的数据 if refused else None trace { original_query: new_question, vector_results: vector_results, keyword_results: keyword_list, fused_results: rrf_list, final_evidence: evidence } return RetrievalResult( querynew_question, candidatescandidates, evidenceevidence, refusedrefused, refusal_reasonrefusal_reason, tracetrace )关键点说明max(evidence_top_k, top_k)给重排序提供足够候选候选太少重排序没有意义。evidence_score统一字段不管是向量分数、RRF 分数、重排序分数全部收敛到同一个字段上层服务不需要关心底层模式。只有full模式才开启过滤hybrid_rerank 只调顺序不丢数据方便调试对比。refusedTrue交给上层接口当为 True 时直接返回拒答文案不要送入 LLM 生成。五、测试脚本对比 hybrid /hybrid_rerank/full /vectorbackend/scripts/check_rerank_filter.pyimport asyncio import sys from pathlib import Path BACKEND_ROOT Path(__file__).resolve().parents[1] if str(BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(BACKEND_ROOT)) from app.database import SessionLocal from app.schemas import RetrievalFilters from app.services.retrieval import retrieve def print_evidence(mode: str, result) - None: print(f 模式{mode} ) print(f是否拒答{result.refused}) print(f拒答原因{result.refusal_reason}) print(f候选数量{len(result.candidates)}) print(f证据数量{len(result.evidence)}) for item in result.evidence: metadata item.get(metadata, {}) info { chunk_id: item.get(chunk_id), source: item.get(source), rrf_score: round(float(item.get(rrf_score, 0)), 4), rerank_score: round(float(item.get(rerank_score, 0)), 4), evidence_score: round(float(item.get(evidence_score, 0)), 4), filename: metadata.get(filename), section: metadata.get(section), content_preview: item.get(content, )[:80] } print(info) print() async def main() - None: question 数字化补贴申报材料 filters RetrievalFilters() with SessionLocal() as session: hybrid await retrieve(session, question, filters, modehybrid, top_k6, min_score0.2) hybrid_rerank await retrieve(session, question, filters, modehybrid_rerank, top_k6, min_score0.2) full await retrieve(session, question, filters, modefull, top_k6, min_score0.6) vector await retrieve(session, question, filters, modevector, top_k6, min_score0.6) print(f用户问题{question}\n) print_evidence(hybrid, hybrid) print_evidence(hybrid_rerank, hybrid_rerank) print_evidence(full, full) print_evidence(vector, vector) if __name__ __main__: asyncio.run(main())六、运行测试 结果解读执行命令.\.venv\Scripts\python.exe scripts\check_rerank_filter.py示例输出参考用户问题数字化补贴申报材料 模式hybrid 是否拒答 False 候选数量 6 证据数量 3 {chunk_id:xxx,rrf_score:0.0164,rerank_score:0.0,evidence_score:0.4918,...} 模式hybrid_rerank 是否拒答 False 证据数量3 {chunk_id:xxx,rerank_score:0.8742,evidence_score:0.8742,section:第四条 申报材料,...} 模式full 是否拒答 False 证据数量2 {rerank_score:0.8742,evidence_score:0.8742,section:第四条 申报材料}观测要点hybrid_rerank的输出出现rerank_score并且真正相关文档分数显著更高顺序发生变化full模式会按照min_score过滤掉低分证据如果过滤完证据为空refusedTrue拒答原因填充。首次运行加载重排序模型会慢属于正常现象。七、工程踩坑与细节说明Cross‑Encoder 不能全库打分速度很慢只能作用于召回后的小批量候选集合。sigmoid 截断‑30~30原始 logit 分数绝对值过大直接传给math.exp会数值溢出报错。降级逻辑必不可少生产环境模型 OOM、文件损坏、GPU 内存不足不能让检索接口直接 500。多 mode 模式的意义开发调试用 hybrid/hybrid_rerank 看中间结果线上生产打开 full 模式。拒答不是能力弱RAG 系统质量标准不知道就说不知道不要编造。evidence_score 统一字段上层服务、评估脚本不需要区分底层是向量 / BM25 / 重排序只看这一个字段。八、总结本章完成 RAG 检索层质量闭环多路召回 → RRF融合 → Cross‑Encoder重排序 → sigmoid分数归一化 → 阈值过滤 → 证据不足拒答RAG 系统从 “能够找到文档” 升级为 “只使用高质量文档回答”从源头降低幻觉风险。
返回列表