掌柜智库项目(RAG)实战9. 检索数据节点实现与测试9.5 结果融合重排node_rrfRRF介绍融合重排 (Reciprocal Rank Fusion) 做去重融合打分输出一个统一的 TopN 列表rrf_chunks供后续回答/重排使用。目标为多路召回的同一数据源向量数据库两路的切片进行排序筛选。核心策略就是把统一切片每一路排名的倒数和该路权重相乘作为累加起来然后把多路的切片汇总排序截取前 n 名。也就是说出现次数越多分越高排名越高分越多该路权重越大分越高。核心代码score_map.get(chunk_id, 0.0) 1.0 / (k pos) * weight其中score_map保存每个切片的累计分数。pos就是该路的当前排名。weight是该路的权重。k是一个衰减系数越小则排名越重要越大则出现次数越重要这里默认值为 60则偏大认为多次出现更加重要。处理流程1获取上游检索节点返回的文档2为不同来源设置权重节点代码实现步骤1导入基础依赖importsysfromtypingimportList,Dict,Anyfromapp.utils.task_utilsimportadd_running_task,add_done_taskfromapp.core.loggerimportlogger步骤2主流程编写defnode_rrf(state): RRF (Reciprocal Rank Fusion) 倒数排名融合节点 功能 将来自不同检索源如 Embedding 检索、HyDE 检索、知识图谱检索等的结果进行融合排序。 RRF 是一种无需训练的算法仅根据文档在不同列表中的排名来计算最终得分。 步骤 1. 提取各路检索结果从 state 中获取 embedding_chunks 和 hyde_embedding_chunks。 2. 结果标准化将不同格式的检索结果统一转换为包含 chunk_id 的实体列表。 3. 设置权重为不同来源分配权重当前配置Embedding1.0, HyDE1.0。 4. 执行 RRF计算融合分数并重新排序。 5. 结果截断保留 Top K 个结果。 6. 更新状态将融合后的结果存入 state[rrf_chunks]。 logger.info(---RRF (倒数排名融合) 开始处理---)add_running_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))# 第一步获取上游检索节点返回的文档# 上游检索节点Milvus hybrid_search返回的通常是 hit 列表# {entity: {...fields...}, distance: ...}# RRF 需要使用 chunk_id 做去重与计分因此这里必须保留 entity而不是仅抽取 content 字符串。embedding_chunks_as_entity_list(state.get(embedding_chunks))hyde_embedding_chunks_as_entity_list(state.get(hyde_embedding_chunks))logger.info(fRRF 输入统计: Embedding源{len(embedding_chunks)}条, HyDE源{len(hyde_embedding_chunks)}条)# Debug 日志打印部分 ID 以便核对ifembedding_chunks:logger.debug(fEmbedding源 chunk_ids (前5个):{[c.get(chunk_id)forcinembedding_chunks[:5]]})ifhyde_embedding_chunks:logger.debug(fHyDE源 chunk_ids (前5个):{[c.get(chunk_id)forcinhyde_embedding_chunks[:5]]})# 第二步为不同来源设置权重# 当前策略两路召回权重相等均为 1.0source_weights[(embedding_chunks,1.0),(hyde_embedding_chunks,1.0)]# 第三步应用带权重的RRF计算最终得分# k60 是 RRF 算法的经典常数max_results10 限制最终召回数量rrf_resreciprocal_rank_fusion(source_weights,k60,max_results10)# 第四步解包结果提取文档和分数rrf_chunks[docfordoc,scoreinrrf_res]# 记录任务结束add_done_task(state[session_id],sys._getframe().f_code.co_name,state.get(is_stream))return{rrf_chunks:rrf_chunks}步骤3编写解析向量数据库结果函数def_as_entity_list(state_list)-List[Dict[str,Any]]: 将上游节点输出统一规整为 entity dict 列表。 兼容 - dict: {entity: {..属性名和对应的字.}, distance: ...} 或直接就是 {...} - pymilvus Hit: 不是 dict但通常支持 hit.get(entity) 或 hit.entity - 其他当作 chunk_id out:List[Dict[str,Any]][]fordocin(state_listor[]):ifnotdoc:continuefinal_ent{}# 情况A: doc 是 Pymilvus 的 Hit 对象 (具有 entity 属性)# Hit 对象结构通常是: idxxx, distancexxx, entity{field1: val1, ...}# 这里的 id 是 Milvus 内部的主键 ID (int64 或 str)ifhasattr(doc,entity)andhasattr(doc,id):# 1. 提取 entity 中的业务字段 (如 content, item_name, chunk_id 等)# 注意: doc.entity 可能是一个 Entity 对象也可能直接是 dictentity_contentdoc.entityifhasattr(entity_content,to_dict):final_ententity_content.to_dict()elifisinstance(entity_content,dict):final_ententity_content.copy()else:# 尝试直接作为 dict 访问 (某些版本 sdk)try:final_entdict(entity_content)except:pass# 2. 补充最外层的 id 和 distance# 优先保留 entity 内部已有的 chunk_id/id如果没有则把外层的 id 补进去ifidnotinfinal_entandchunk_idnotinfinal_ent:final_ent[id]doc.id# 补充 distance (score)ifhasattr(doc,distance):final_ent[score]doc.distance# 情况B: doc 已经是字典 (模拟数据或已处理数据)elifisinstance(doc,dict):# 尝试获取 entity 字段 (嵌套结构 {entity: {...}, id: ...})ifentityindoc:entdoc[entity]ifisinstance(ent,dict):final_entent.copy()# 尝试从外层补充 id/scoreifidindocandidnotinfinal_ent:final_ent[id]doc[id]ifdistanceindoc:final_ent[score]doc[distance]else:# 扁平结构直接使用final_entdoc# 情况C: 其他对象 (尝试 get 方法)elifhasattr(doc,get):entdoc.get(entity)ordocifisinstance(ent,dict):final_entent# 最终校验必须是非空字典iffinal_entandisinstance(final_ent,dict):out.append(final_ent)returnout步骤4通用带权重的RRF算法实现该代码实现的是带权重的倒数排名融合Reciprocal Rank Fusion, RRF算法核心作用是将多个不同来源的文档排序结果结合各来源的权重进行融合最终输出一个综合所有来源排序信息、按融合得分降序排列的统一文档列表解决多来源排序结果的融合与重排序问题。输入参数source_weights核心输入列表类型每个元素是(来源文档列表, 权重)的元组。其中来源文档列表单个来源对文档的排序结果有序靠前的文档在该来源中相关性更高权重该来源的重要性系数权重越高该来源的排序结果对最终融合的影响越大。kRRF 算法专属常数默认 60用于平衡文档排名对得分的影响避免排名过前的文档得分无限制偏高是 RRF 的标准超参数。max_results可选参数限制最终返回的文档数量None表示返回全部融合结果。输出结果列表类型每个元素是(文档对象, RRF融合得分)的元组按融合得分降序排列得分越高表示文档在综合所有来源后的相关性越强。defreciprocal_rank_fusion(source_weights:list,k:int60,max_results:intNone,)-List[tuple]: 通用带权重的RRF算法实现 :param source_weights: 列表每个元素是(来源文档列表, 权重)的元组 例如: [([doc1, doc2], 1.0), ([doc2, doc3], 0.8)] :param k: RRF 常数默认 60。用于平滑排名影响避免高排名文档占据过大优势。 :param max_results: 只返回前 N 个None 表示全部 :return: [(元素, RRF 得分), ...] 按得分降序排列 # score_map: 记录 chunk_id 到 RRF 累加得分的映射score_map{}# chunk_map: 记录 chunk_id 到文档实体对象的映射用于最终返回chunk_map{}# 1. 遍历所有来源计算每个文档的 RRF 分数# source_weights 结构: [(doc_list, weight), ...]fordocs,weightinsource_weights:# enumerate(docs, start1): 获取排名 (rank)从 1 开始forrank,iteminenumerate(docs,start1):# 获取文档唯一标识 ID# Milvus 设计上把主键字段在 API 层面统一叫 id不管你在 schema 里定义的字段名是 pk、id 还是其他# 这是为了保持 API 兼容性无论用户怎么命名主键SDK 都用 id 来指代 “这条数据的唯一标识”# 你在向量数据库 UI 里看到的 pk 是表结构定义名而代码里拿到的 id 是API 返回的统一主键别名chunk_iditem.get(chunk_id)oritem.get(id)ifnotchunk_id:# 如果找不到 ID记录警告并跳过避免程序崩溃logger.warning(fRRF Warning: item missing chunk_id/id:{list(item.keys())ifisinstance(item,dict)elseitem})continue# RRF 核心公式: score weight * (1 / (k rank))score_map[chunk_id]score_map.get(chunk_id,0.0)weight*(1.0/(krank))# 只记录第一次遇到的文档实体对象chunk_map.setdefault(chunk_id,item)# 2. 将结果转换为列表并排序merged[]forchunk_id,scoreinscore_map.items():doc_itemchunk_map[chunk_id]merged.append((doc_item,score))# 按分数降序排序 (得分越高越靠前)merged.sort(keylambdax:x[1],reverseTrue)# 3. 截断结果ifmax_resultsisnotNone:mergedmerged[:max_results]returnmerged关键细节排名pos从 1 开始符合实际排序逻辑第 1 名、第 2 名…而非程序默认的 0 索引唯一标识chunk_id作为文档的 “唯一键”实现跨来源的文档匹配不同来源的同一文档通过chunk_id累计得分核心得分公式1.0 / (k pos) * weight基础项1/(kpos)RRF 算法的核心排名越靠前pos 越小该项值越大且通过k平滑排名的影响避免 pos1 时得分过大权重项* weight实现 “带权重融合”重要来源的排序结果对文档最终得分的贡献成比例放大。得分累加逻辑同一文档出现在多个来源中或在单个来源中出现多次极少情况其得分会通过score_map.get(chunk_id, 0.0)持续累加。k** 的核心作用**k是 RRF 算法的核心超参数行业通用默认值 60也可根据业务调整核心作用是对排名的影响做「平滑 / 缓冲」避免排名的微小差异导致得分剧烈波动同时防止极端值如 pos1主导总得分具体解决 2 个关键问题问题 1避免pos1时得分无限制偏高压制其他来源的贡献如果没有k公式会变成1/pos此时pos1得分 1.0pos2得分 0.5pos3得分≈0.33第 1 名的得分是第 2 名的 2 倍、第 3 名的 3 倍单个来源的第 1 名会直接主导总得分其他来源的排名信息几乎被忽略失去 “多来源融合” 的意义。加入k60后公式为1/(60pos)pos1≈0.0164pos2≈0.0162pos3≈0.0160前几名的得分差异被大幅缩小单个来源的排名无法独断总得分必须结合多个来源的排名才能获得高总得 分符合多来源融合的核心目标。问题 2平衡排名的 “边际效应”让排名靠后的文档也有合理贡献没有k时排名靠后的文档得分会快速趋近于 0如pos1001/1000.01几乎没有贡献加入k60后pos100的得分 1/(60100)0.00625与pos50的得分1/110≈0.0091差异更小排名靠后的文档仍能为总得分提供合理贡献避免直接被 “抛弃”。简单总结k的作用让排名对得分的影响更 “温和”保证多来源排名信息都能有效参与融合而非少数高排名文档垄断得分。主流程测试if__name____main__:print(\n*50)print( 启动 node_rrf 本地测试)print(*50)# 1. 构造假数据 (模拟真实数据库字段)# 模拟 Embedding 检索结果mock_embedding_chunks[{id:doc_1,pk:pk_1,file_title:操作手册_v1.pdf,item_name:HAK 180 烫金机,content:内容1打开电源开关...,score:0.9},{id:doc_2,pk:pk_2,file_title:维修指南.pdf,item_name:HAK 180 烫金机,content:内容2遇到故障请联系...,score:0.8},{id:doc_3,pk:pk_3,file_title:参数表.xlsx,item_name:HAK 180 烫金机,content:内容3电压220V...,score:0.7}]# 模拟 HyDE 检索结果 (包含 3 个文档顺序不同且有新文档 doc_4)mock_hyde_chunks[{id:doc_3,pk:pk_3,file_title:参数表.xlsx,item_name:HAK 180 烫金机,content:内容3电压220V...,score:0.85},{id:doc_1,pk:pk_1,file_title:操作手册_v1.pdf,item_name:HAK 180 烫金机,content:内容1打开电源开关...,score:0.82},{id:doc_4,pk:pk_4,file_title:安全须知.docx,item_name:HAK 180 烫金机,content:内容4操作时请佩戴手套...,score:0.75}]# 模拟输入状态mock_state{session_id:test_rrf_session,is_stream:False,embedding_chunks:mock_embedding_chunks,hyde_embedding_chunks:mock_hyde_chunks}try:# 运行节点resultnode_rrf(mock_state)# 验证结果rrf_chunksresult.get(rrf_chunks,[])print(\n*50)print( 测试结果摘要:)print(f输入数量: Embedding{len(mock_embedding_chunks)}, HyDE{len(mock_hyde_chunks)})print(f输出数量:{len(rrf_chunks)})print(-*30)# 打印详细排名print(最终排名:)fori,docinenumerate(rrf_chunks,1):# 注意返回结果中可能没有 chunk_id 字段而是 iddoc_iddoc.get(chunk_id)ordoc.get(id)print(fRank{i}: ID{doc_id}, Title{doc.get(file_title)}, Content{doc.get(content)[:20]}...)# 验证预期逻辑ids[d.get(id)ord.get(chunk_id)fordinrrf_chunks]ifdoc_1inidsanddoc_3inids:print(\n[PASS] 交叉文档 (doc_1, doc_3) 成功融合保留)else:print(\n[FAIL] 交叉文档丢失)iflen(ids)4:print([PASS] 并集数量正确 (33-2重叠4))else:print(f[FAIL] 并集数量错误: 期望4, 实际{len(ids)})print(*50)exceptExceptionase:logger.exception(f测试运行期间发生未捕获异常:{e})