Python后端AI专题23Rerank 重排为什么召回第一名不一定最会回答向量召回用单个向量近似整段语义关键词召回看精确词RRF 再根据名次投票它们适合快速找候选却没有逐对阅读“问题 文档”。Rerank 的职责是在较小候选集上做更精细的相关性判断而不是替代第一阶段检索全库。RRF 手算与测试答案k60A 1/61 1/63 0.03227 B 1/62 1/61 0.03252 C 1/63 0.01587 D 1/62 0.01613顺序为 B、A、D、C。B 在两路都靠前所以超过向量第一的 A。完整测试还验证输入没有被修改deftest_rrf_hand_calculation_preserves_input_scores()-None:vector[RetrievedChunk(A,A,0.91),RetrievedChunk(B,B,0.82),RetrievedChunk(C,C,0.73),]keyword[RetrievedChunk(B,B,9.0),RetrievedChunk(D,D,7.0),RetrievedChunk(A,A,5.0),]resultreciprocal_rank_fusion(vector,keyword,k60)assert[item.idforiteminresult][B,A,D,C]by_id{item.id:itemforiteminresult}assertby_id[A].diagnostics[vector_rank]1assertby_id[A].diagnostics[keyword_rank]3assertkeyword_ranknotinby_id[C].diagnosticsassert[item.scoreforiteminvector][0.91,0.82,0.73]真实结果4 passed in 0.09s。为什么是两阶段漏斗假设有一百万 chunk。Cross-encoder/Rerank 若对每个候选都联合编码问题与正文计算量远大于向量点积。常见漏斗100 万 → 向量 Top 30 关键词 Top 30 → RRF 去重约 40 条 → Rerank Top 5 → 上下文预算再裁剪召回阶段追求别漏掉重排阶段追求前几名准确。候选太少Rerank 无法救回没被召回的答案候选太多延迟和费用上升。Fake Rerank 如何保证回归确定性课程 Fake 使用问题与正文的字符集合重叠并以原排名稳定打破平分scorelen(query_termsterms)/max(1,len(query_terms))scored.append(RerankResult(item.id,score,rank))scored.sort(keylambdaresult:(-result.score,result.original_rank))它不代表真实重排质量只让管线在无网络时可重复测试。生产适配器应调用真实 reranker并通过评测集比较不能把 Fake 的字符分数写进项目简历说“语义重排”。完整检索管线from__future__importannotationsfromtimeimportperf_counterfromapp.providers.contractsimportEmbeddingProvider,RerankItem,RerankProviderfromapp.providers.vector_storeimportVectorStorefromapp.services.retrieval.fusionimportreciprocal_rank_fusionfromapp.services.retrieval.keywordimportkeyword_searchfromapp.services.retrieval.typesimportRetrievalResultclassRetrievalPipeline:def__init__(self,*,embedder:EmbeddingProvider,vector_store:VectorStore,reranker:RerankProvider)-None:self.embedderembedder self.vector_storevector_store self.rerankerrerankerasyncdefretrieve(self,*,tenant_id:str,knowledge_base_id:str,query:str,limit:int5)-RetrievalResult:timings:dict[str,float]{}startedperf_counter()vector(awaitself.embedder.embed([query]))[0]timings[embedding](perf_counter()-started)*1000startedperf_counter()vector_resultsawaitself.vector_store.search(tenant_idtenant_id,knowledge_base_idknowledge_base_id,vectorvector,limitmax(limit*3,10),)timings[vector](perf_counter()-started)*1000startedperf_counter()recordsawaitself.vector_store.records_for(tenant_idtenant_id,knowledge_base_idknowledge_base_id)keyword_resultskeyword_search(query,records,limitmax(limit*3,10))timings[keyword](perf_counter()-started)*1000fusedreciprocal_rank_fusion(vector_results,keyword_results)startedperf_counter()try:rerankedawaitself.reranker.rerank(query,[RerankItem(item.id,item.text)foriteminfused],top_nlimit,)exceptTimeoutError:chunksfused[:limit]forchunkinchunks:chunk.diagnostics[rerank_degraded]Truetimings[rerank](perf_counter()-started)*1000timings[rerank_degraded]1.0timings[total]sum(valueforkey,valueintimings.items()ifkey!rerank_degraded)returnRetrievalResult(chunks,timings)by_id{item.id:itemforiteminfused}chunks[]forreranked_iteminreranked:chunkby_id[reranked_item.id]chunk.scorereranked_item.score chunk.diagnostics[rerank_score]reranked_item.score chunks.append(chunk)timings[rerank](perf_counter()-started)*1000timings[total]sum(timings.values())returnRetrievalResult(chunks,timings)limitmax(limit*3,10)保证即使最终只要 2 条也给重排至少 10 个向量/关键词候选。真实生产中两个召回大小应独立配置并在评测中寻找成本曲线。diagnostics 不应在重排后丢失管线用 id 找回 RRF 对象只更新最终 score并增加rerank_score。因此输出还能看到vector_rank/keyword_rank/rrf_score。当用户投诉错误答案时可以判断是未召回、融合落后还是重排选错。Rerank 的输入也有安全与长度边界外部 Rerank 服务会看到候选正文仍需遵守租户与数据出境要求。超长 chunk 应按供应商限制截断但截断方式要保留关键段批量请求也应有超时、并发与降级策略。若 Rerank 暂时不可用可降级使用 RRF 顺序但必须在 diagnostics 标记而不是伪造rerank_score。本篇最终完整模块pipeline.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsfromtimeimportperf_counterfromapp.providers.contractsimportEmbeddingProvider,RerankItem,RerankProviderfromapp.core.metricsimportRETRIEVAL_STAGE_SECONDSfromapp.providers.vector_storeimportVectorStorefromapp.services.retrieval.fusionimportreciprocal_rank_fusionfromapp.services.retrieval.keywordimportkeyword_searchfromapp.services.retrieval.typesimportRetrievalResultclassRetrievalPipeline:def__init__(self,*,embedder:EmbeddingProvider,vector_store:VectorStore,reranker:RerankProvider,)-None:self.embedderembedder self.vector_storevector_store self.rerankerrerankerasyncdefretrieve(self,*,tenant_id:str,knowledge_base_id:str,query:str,limit:int5,)-RetrievalResult:timings:dict[str,float]{}startedperf_counter()vector(awaitself.embedder.embed([query]))[0]timings[embedding](perf_counter()-started)*1000RETRIEVAL_STAGE_SECONDS.labels(stageembedding).observe(timings[embedding]/1000)startedperf_counter()vector_resultsawaitself.vector_store.search(tenant_idtenant_id,knowledge_base_idknowledge_base_id,vectorvector,limitmax(limit*3,10),)timings[vector](perf_counter()-started)*1000RETRIEVAL_STAGE_SECONDS.labels(stagevector).observe(timings[vector]/1000)startedperf_counter()recordsawaitself.vector_store.records_for(tenant_idtenant_id,knowledge_base_idknowledge_base_id)keyword_resultskeyword_search(query,records,limitmax(limit*3,10))timings[keyword](perf_counter()-started)*1000RETRIEVAL_STAGE_SECONDS.labels(stagekeyword).observe(timings[keyword]/1000)fusedreciprocal_rank_fusion(vector_results,keyword_results)startedperf_counter()try:rerankedawaitself.reranker.rerank(query,[RerankItem(item.id,item.text)foriteminfused],top_nlimit,)exceptTimeoutError:chunksfused[:limit]forchunkinchunks:chunk.diagnostics[rerank_degraded]Truetimings[rerank](perf_counter()-started)*1000RETRIEVAL_STAGE_SECONDS.labels(stagererank).observe(timings[rerank]/1000)timings[rerank_degraded]1.0timings[total]sum(valueforkey,valueintimings.items()ifkey!rerank_degraded)returnRetrievalResult(chunks,timings)by_id{item.id:itemforiteminfused}chunks[]forreranked_iteminreranked:chunkby_id[reranked_item.id]chunk.scorereranked_item.score chunk.diagnostics[rerank_score]reranked_item.score chunks.append(chunk)timings[rerank](perf_counter()-started)*1000RETRIEVAL_STAGE_SECONDS.labels(stagererank).observe(timings[rerank]/1000)timings[total]sum(timings.values())returnRetrievalResult(chunks,timings)本篇练习给重排做失败降级设计写一个FailingRerankProvider总是抛TimeoutError。为管线设计两种策略并比较A. 整个搜索返回 503B. 返回 RRF Top N并在每个结果 diagnostics 写rerank_degradedTrue。选择一种适合“内部知识库只读搜索”的策略给出完整伪代码和至少两个测试降级顺序正确、非超时的协议错误不能被静默吞掉。下一篇会给出选择与可运行实现然后建立引用校验和无证据拒答让“搜到内容”不再自动等于“答案可信”。
