006、检索篇：相似度算法、混合检索与重排序（Rerank）技术详解

G_RE_RE

233人浏览 · 2026-03-29 16:46:43

G_RE_RE · 2026-03-29 16:46:43 发布

006、检索篇：相似度算法、混合检索与重排序（Rerank）技术详解

从一次深夜调试说起

线上问答系统的召回结果突然出现严重偏差。用户问“如何配置Redis集群”，系统返回的却是“Redis单机安装教程”和“MySQL主从配置指南”。打开监控一看，Embedding相似度分数都在0.85以上——从数值看似乎很匹配，但实际业务场景中，这种“表面相似但实质偏离”的问题正在伤害用户体验。

这个问题暴露了单纯依赖向量相似度的脆弱性。今晚我们就深入聊聊：当相似度算法遇上真实业务时，如何通过混合检索与重排序构建更健壮的召回系统。

相似度算法：不只是点积那么简单

很多人一提到向量相似度就只想到余弦相似度，其实这里有个选择陷阱。不同的相似度计算方式，在特定场景下表现差异巨大。

# 常见相似度计算实现（注释是踩坑记录）
def calculate_similarity(vec1, vec2, method='cosine'):
    """
    vec1, vec2: 归一化后的向量
    method: cosine/dot_product/euclidean
    """
    if method == 'cosine':
        # 最常用的方式，对向量长度不敏感
        # 适合文本语义匹配，但注意输入向量必须归一化！
        return np.dot(vec1, vec2)  # 已经归一化所以等价于余弦
    
    elif method == 'dot_product':
        # 点积：向量长度会影响分数
        # 在BM25+Embedding混合场景有用，可以体现“重要性权重”
        # 但别单独用，容易受embedding模长影响出问题
        return np.dot(vec1, vec2)
    
    elif method == 'euclidean':
        # 欧氏距离：越小越相似
        # 需要转换成相似度分数时：1/(1+distance)
        # 在视觉搜索里更常见，文本场景慎用
        dist = np.linalg.norm(vec1 - vec2)
        return 1 / (1 + dist)  # 转换到0-1范围
    
    # 生产环境加个兜底
    raise ValueError(f"Unknown method: {method}")

实际项目中，我习惯用余弦相似度打底，但在以下情况会调整策略：

当文档长度差异大且长度信息重要时，点积可能更合适
做A/B测试时发现某些query类型在特定相似度下表现更好
混合检索时，不同检索路径可能需要不同的相似度计算

混合检索：让多路召回协同作战

单一检索方式总有局限。混合检索的核心思想是：让不同检索器各司其职，然后融合结果。

class HybridRetriever:
    def __init__(self):
        self.vector_retriever = VectorRetriever()  # 向量检索
        self.keyword_retriever = KeywordRetriever()  # 关键词检索（如BM25）
        self.sparse_retriever = SparseRetriever()   # 稀疏向量检索
        
    def hybrid_search(self, query, top_k=10):
        # 并行多路召回（生产环境建议用异步）
        vector_results = self.vector_retriever.search(query, top_k*2)
        keyword_results = self.keyword_retriever.search(query, top_k*2)
        
        # 分数归一化：关键步骤！
        # 不同检索器的分数尺度不同，必须归一化到同一量纲
        vector_results = self._normalize_scores(vector_results)
        keyword_results = self._normalize_scores(keyword_results)
        
        # 融合策略：加权平均
        fused_results = {}
        for doc_id, score in vector_results.items():
            fused_results[doc_id] = score * 0.7  # 向量权重
            
        for doc_id, score in keyword_results.items():
            if doc_id in fused_results:
                fused_results[doc_id] += score * 0.3
            else:
                fused_results[doc_id] = score * 0.3
        
        # 按最终分数排序
        sorted_results = sorted(fused_results.items(), 
                               key=lambda x: x[1], reverse=True)
        return sorted_results[:top_k]
    
    def _normalize_scores(self, results):
        """Min-Max归一化，简单但有效"""
        if not results:
            return {}
        scores = [s for _, s in results]
        min_s, max_s = min(scores), max(scores)
        if max_s == min_s:
            return {doc_id: 1.0 for doc_id, _ in results}
        
        return {
            doc_id: (score - min_s) / (max_s - min_s)
            for doc_id, score in results
        }

混合比例需要根据业务调整。我们的经验是：技术文档场景，向量权重高些（0.6-0.8）；FAQ问答场景，关键词权重可以适当提高。

重排序（Rerank）：最后的精调关口

召回阶段追求“全面”，重排序阶段追求“精准”。这是两个不同的优化目标。

class Reranker:
    def __init__(self):
        # 小型交叉编码器，比双塔模型更精准但更慢
        self.cross_encoder = load_cross_encoder()
        
        # 业务规则处理器（硬约束）
        self.rule_engine = RuleEngine()
        
    def rerank(self, query, candidates):
        """
        candidates: [(doc_id, text, raw_score), ...]
        返回重排序后的列表
        """
        reranked = []
        
        for doc_id, text, raw_score in candidates:
            # 1. 交叉编码器计算精细相关度
            ce_score = self.cross_encoder.predict([[query, text]])
            
            # 2. 业务规则调整（比如时效性、权威性）
            rule_adjust = self.rule_engine.apply_rules(query, text)
            
            # 3. 综合打分（这个公式调了两个月...）
            final_score = (
                ce_score * 0.6 + 
                raw_score * 0.2 + 
                rule_adjust * 0.2
            )
            
            reranked.append({
                'doc_id': doc_id,
                'text': text,
                'score': final_score,
                'components': {  # 保留各维度分数，调试用
                    'ce': ce_score,
                    'raw': raw_score,
                    'rule': rule_adjust
                }
            })
        
        # 按最终分数排序
        reranked.sort(key=lambda x: x['score'], reverse=True)
        return reranked