Python后端AI专题22：产品编号搜不到？把关键词与向量用 RRF 融合

Python后端AI专题22产品编号搜不到把关键词与向量用 RRF 融合用户搜索“KF-2048 端口”向量模型可能认为“产品安装说明”很相关却把精确包含KF-2048的短表格排在后面。Embedding 擅长同义表达产品编号、错误码和合同号则更适合词法检索。混合检索不是把两个分数直接相加而是先承认它们的量纲不同。Top K 后过滤攻击答案租户 B 有 10 条相似度 1.0A 只有一条 0.8。全局 Top 5 会全部选 BPython 过滤后 A 得到空列表SQL 先限制tenantA再 Top 5A 能得到自己的 0.8 记录。内存测试构造 10 条 B 记录查询仍只返回alice-1awaitstore.upsert([VectorRecord(alice-1,tenant-a,kb-1,退款期限七天,[0.8,0.6][0.0]*6),*[VectorRecord(fbob-secret-{i},tenant-b,kb-1,B 的机密退款规则,query_vector)foriinrange(10)],])resultawaitstore.search(tenant_idtenant-a,knowledge_base_idkb-1,vectorquery_vector,limit5,)assert[item.idforiteminresult][alice-1]检索单测真实结果3 passed in 0.13s。仍需 PostgreSQL 集成测试因为内存实现无法证明 SQL 的 WHERE、JOIN、pgvector 操作符和 HNSW 行为正确。为什么不能把分数直接相加向量相似度可能在 -1 到 1词法分数可能是命中词权重之和范围随查询变化Rerank 又可能输出 0—1。这样写finalvector_scorekeyword_score意味着哪个分值范围更大哪个就支配结果并非真正融合。归一化也需要稳定分布面对不同查询往往脆弱。RRF 只相信名次Reciprocal Rank Fusion 对每个结果在每条列表中的排名累计score(d) Σ 1 / (k rank_i(d))k60缓和第一名与后续名次差距。某块在向量第 2、关键词第 1它会得到1/62 1/61只在向量第 1 的块只有1/61双路都认可的结果通常上升。完整 RRF 模块from__future__importannotationsfromcollections.abcimportSequencefromapp.services.retrieval.typesimportRetrievedChunkdefreciprocal_rank_fusion(vector_results:Sequence[RetrievedChunk],keyword_results:Sequence[RetrievedChunk],*,k:int60,)-list[RetrievedChunk]:Fuse incomparable scores using only within-list ranks.by_id:dict[str,RetrievedChunk]{}scores:dict[str,float]{}forsource,namein((vector_results,vector),(keyword_results,keyword),):forrank,iteminenumerate(source,start1):ifitem.idnotinby_id:by_id[item.id]RetrievedChunk(item.id,item.text,0.0,item.document,item.page,item.section,dict(item.diagnostics),)scores[item.id]scores.get(item.id,0.0)1.0/(krank)by_id[item.id].diagnostics[f{name}_rank]rank by_id[item.id].diagnostics.update(item.diagnostics)forid_,iteminby_id.items():item.scorescores[id_]item.diagnostics[rrf_score]scores[id_]returnsorted(by_id.values(),keylambdaitem:(-item.score,item.id))这里复制一个新的RetrievedChunk避免直接修改召回列表中的对象造成测试和日志相互污染diagnostics 保存各路名次线上才能解释某个结果为何上升。词法基线怎样照顾产品编号本项目的透明基线识别英文数字及连字符编号re.findall(r[A-Za-z0-9](?:-[A-Za-z0-9])*|[\u4e00-\u9fff]{2,},text)完整 token 命中加 3 分中文字符交集提供较小补充。它不是生产级中文搜索引擎但行为可解释足以展示混合链。大规模数据应把关键词搜索下推到 PostgreSQL FTS/OpenSearch而不是records_for()把全部 chunk 读入 Python。双路候选必须应用同一权限向量查询已过滤租户、知识库和 ready 状态关键词读取的records_for()也使用完全相同条件。若其中一路权限较松RRF 会把越权结果堂而皇之地融合回来。安全约束不能只检查“主召回”。产品编号端到端用例语料含退款、KF-2048 的安装端口是 8080、差旅。查询“KF-2048 使用什么端口”断言第一名为产品记录且 diagnostics 有keyword_rank1。测试不强迫 Fake Embedding 故意失败它证明无论向量排序如何精确编号路径参与融合并可解释。本篇最终完整模块fusion.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsfromcollections.abcimportSequencefromapp.services.retrieval.typesimportRetrievedChunkdefreciprocal_rank_fusion(vector_results:Sequence[RetrievedChunk],keyword_results:Sequence[RetrievedChunk],*,k:int60,)-list[RetrievedChunk]:Fuse incomparable scores using only their within-list ranks.by_id:dict[str,RetrievedChunk]{}scores:dict[str,float]{}forsource,namein((vector_results,vector),(keyword_results,keyword)):forrank,iteminenumerate(source,start1):ifitem.idnotinby_id:by_id[item.id]RetrievedChunk(item.id,item.text,0.0,item.document,item.page,item.section,dict(item.diagnostics),)scores[item.id]scores.get(item.id,0.0)1.0/(krank)by_id[item.id].diagnostics[f{name}_rank]rank by_id[item.id].diagnostics.update(item.diagnostics)forid_,iteminby_id.items():item.scorescores[id_]item.diagnostics[rrf_score]scores[id_]returnsorted(by_id.values(),keylambdaitem:(-item.score,item.id))本篇练习手算并验证 RRF向量列表[A,B,C]关键词列表[B,D,A]取k60手算 A/B/C/D 得分与最终顺序说明 B 为什么第一。然后写 pytest 构造这两条RetrievedChunk列表断言B 第一A 同时包含vector_rank1和keyword_rank3C 没有keyword_rank输入对象的原始 score 不被修改。下一篇会给出完整答案并加入 Rerank为什么 RRF 第一名仍不一定最能回答当前问题。