从零构建cherry搜索结果压缩与重排模型：新手避坑指南

1次阅读

共计 3227 个字符，预计需要花费 9 分钟才能阅读完成。

最近在做一个电商搜索项目时，用户反馈搜索 ” 樱桃手机壳 ” 会出现大量重复商品（仅颜色不同），且前两页结果中真正匹配需求的不足 30%。原始 BM25 排序将包含关键词但实际无关的文档排到前列，而真正相关的长尾结果被埋没。这种冗余和低效问题会直接影响用户体验和转化率。

TF-IDF/BM25 的局限性
仅考虑词频统计，无法理解语义相似性（如 ”cherry” 和 ” 樱桃 ”）
对长文档过拟合，短文本匹配效果不稳定
无法处理一词多义问题（如 ” 苹果 ” 指水果还是手机）
神经排序模型优势
Sentence-BERT 等模型可生成语义向量，捕获深层关联
通过余弦相似度计算比单纯关键词匹配更精准
支持跨语言匹配等复杂场景

# 安装关键库
pip install sentence-transformers pandas numpy psutil

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict
import psutil

class CherrySearchOptimizer:
    """
    搜索结果压缩与重排处理器
    功能：1. 使用 SBERT 提取查询 - 文档语义向量
    2. 计算余弦相似度并去重
    3. 按相关性重新排序
    """def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2'):
        # 加载轻量级预训练模型（平衡效果与性能）self.model = SentenceTransformer(model_name)
        self.mem_monitor = []

    def _log_memory(self):
        """记录内存使用情况"""
        self.mem_monitor.append(psutil.virtual_memory().percent)

    def extract_features(self, queries: List[str], docs: List[str]) -> np.ndarray:
        """
        批量生成文本向量
        :param queries: 搜索请求列表
        :param docs: 待处理文档列表
        :return: (n_samples, 384)维向量矩阵
        """
        # 合并查询和文档减少模型调用次数
        all_texts = queries + docs
        self._log_memory()

        # 启用多线程加速（适合批量处理）return self.model.encode(all_texts, 
                               show_progress_bar=True,
                               convert_to_numpy=True,
                               device='cuda')

    def compute_similarity(self, query_vec: np.ndarray, doc_vecs: np.ndarray) -> np.ndarray:
        """计算余弦相似度矩阵"""
        # 归一化向量提升计算效率
        query_vec = query_vec / np.linalg.norm(query_vec)
        doc_vecs = doc_vecs / np.linalg.norm(doc_vecs, axis=1)[:, None]
        return np.dot(doc_vecs, query_vec.T)

    def deduplicate(self, docs: List[str], threshold=0.95) -> List[int]:
        """
        基于语义相似度去重
        :param threshold: 相似度高于此值视为重复
        :return: 保留文档的索引列表
        """
        doc_vecs = self.model.encode(docs)
        sim_matrix = np.inner(doc_vecs, doc_vecs)

        keep_indices = []
        seen = set()
        for i in range(len(docs)):
            if i not in seen:
                keep_indices.append(i)
                # 找到所有相似文档标记为已处理
                duplicates = np.where(sim_matrix[i] > threshold)[0]
                seen.update(duplicates)
        return keep_indices

    def rerank(self, query: str, docs: List[str], top_k=10) -> List[Dict]:
        """完整重排流程"""
        # 1. 特征提取
        vecs = self.extract_features([query], docs)
        query_vec, doc_vecs = vecs[0], vecs[1:]

        # 2. 相似度计算
        scores = self.compute_similarity(query_vec, doc_vecs)

        # 3. 结果包装
        ranked = sorted(zip(range(len(docs)), docs, scores.flatten()), 
                       key=lambda x: -x[2])

        return [{"doc": doc, "score": float(score), "rank": i+1} 
               for i, (idx, doc, score) in enumerate(ranked[:top_k])]

向量化加速
使用 np.einsum 替代循环计算（速度提升 5 - 8 倍）
对短文本启用 padding='max_length' 减少动态计算开销
内存管理

监控代码片段：

def check_memory():
    import psutil
    return {'total': psutil.virtual_memory().total / (1024**3),
        'used': psutil.virtual_memory().used / (1024**3),
        'process': psutil.Process().memory_info().rss / (1024**2)
    }

批量处理时每 1000 条主动调用gc.collect()

多线程方案

from concurrent.futures import ThreadPoolExecutor

def parallel_encode(texts, workers=4):
    """多线程加速向量化"""
    chunk_size = len(texts) // workers
    with ThreadPoolExecutor(max_workers=workers) as executor:
        return list(executor.map(lambda x: model.encode(x), 
            [texts[i*chunk_size:(i+1)*chunk_size] for i in range(workers)]
        ))