搜索架构：倒排索引与相关性排序

搜索系统架构概览

搜索系统是信息检索的核心，需要处理海量数据的实时查询。主要模块包括：爬虫/数据采集、索引构建、查询处理、相关性排序、搜索质量监控。

搜索系统架构：

数据层：数据采集、数据清洗、数据存储
索引层：倒排索引构建、索引更新、索引存储
查询层：查询理解、查询改写、查询执行
排序层：召回、粗排、精排、重排
展示层：搜索结果展示、摘要生成、高亮
监控层：搜索质量、性能监控、用户行为分析

倒排索引原理

倒排索引是搜索引擎的核心数据结构，将文档中的词项映射到包含该词项的文档列表。

# 倒排索引构建示例
class InvertedIndex:
    def __init__(self):
        self.index = defaultdict(list)  # term -> [doc_id, ...]
        self.doc_store = {}             # doc_id -> document
    
    def add_document(self, doc_id, content):
        """添加文档到索引"""
        self.doc_store[doc_id] = content
        
        # 分词
        tokens = self.tokenize(content)
        
        # 构建倒排索引
        for token in tokens:
            if doc_id not in self.index[token]:
                self.index[token].append(doc_id)
    
    def search(self, query):
        """搜索文档"""
        # 查询分词
        query_tokens = self.tokenize(query)
        
        # 查找包含所有查询词的文档（AND查询）
        result_docs = None
        for token in query_tokens:
            if token in self.index:
                doc_set = set(self.index[token])
                if result_docs is None:
                    result_docs = doc_set
                else:
                    result_docs = result_docs.intersection(doc_set)
            else:
                return []
        
        return list(result_docs) if result_docs else []
    
    def tokenize(self, text):
        """简单分词"""
        # 实际应用中会使用jieba等分词工具
        return text.lower().split()

查询理解与改写

查询理解是搜索质量的关键，包括分词、纠错、同义词扩展、意图识别等。

// 查询理解服务
@Service
public class QueryUnderstandingService {
    @Autowired
    private Tokenizer tokenizer;
    @Autowired
    private SpellChecker spellChecker;
    @Autowired
    private SynonymExpander synonymExpander;
    
    public ParsedQuery parseQuery(String rawQuery) {
        // 1. 查询清洗
        String cleanedQuery = cleanQuery(rawQuery);
        
        // 2. 分词
        List<String> tokens = tokenizer.tokenize(cleanedQuery);
        
        // 3. 拼写纠错
        List<String> correctedTokens = tokens.stream()
            .map(token -> spellChecker.correct(token))
            .collect(Collectors.toList());
        
        // 4. 同义词扩展
        List<String> expandedTokens = correctedTokens.stream()
            .flatMap(token -> synonymExpander.expand(token).stream())
            .collect(Collectors.toList());
        
        // 5. 意图识别
        QueryIntent intent = classifyIntent(cleanedQuery);
        
        return ParsedQuery.builder()
            .originalQuery(rawQuery)
            .tokens(expandedTokens)
            .intent(intent)
            .build();
    }
}

相关性排序

相关性排序是搜索引擎的核心，需要综合考虑文本相关性、文档质量、用户行为等因素。

# 排序模型
class RankingModel:
    def __init__(self):
        self.tfidf = TFIDFModel()
        self.bm25 = BM25Model()
        self.learning_to_rank = LTRModel()
    
    def rank(self, query, candidates):
        """多阶段排序"""
        # 1. 召回阶段：使用倒排索引快速召回
        recalled = self.retrieve(query, candidates)
        
        # 2. 粗排阶段：使用BM25等简单模型
        coarse_ranked = self.coarse_rank(query, recalled)
        
        # 3. 精排阶段：使用机器学习模型
        fine_ranked = self.fine_rank(query, coarse_ranked)
        
        # 4. 重排阶段：考虑多样性、时效性等
        final_ranked = self.re_rank(query, fine_ranked)
        
        return final_ranked
    
    def coarse_rank(self, query, documents):
        """粗排：使用BM25"""
        scores = []
        for doc in documents:
            score = self.bm25.score(query, doc)
            scores.append((doc, score))
        
        # 按分数排序，取top N
        scores.sort(key=lambda x: x[1], reverse=True)
        return [doc for doc, score in scores[:1000]]
    
    def fine_rank(self, query, documents):
        """精排：使用Learning to Rank"""
        features = self.extract_features(query, documents)
        scores = self.learning_to_rank.predict(features)
        
        ranked = list(zip(documents, scores))
        ranked.sort(key=lambda x: x[1], reverse=True)
        return [doc for doc, score in ranked[:100]]

BM25算法详解

BM25是经典的文本相关性算法，广泛应用于搜索引擎。

import math
from collections import Counter

class BM25:
    def __init__(self, k1=1.5, b=0.75):
        self.k1 = k1
        self.b = b
        self.doc_count = 0
        self.avg_doc_length = 0
        self.doc_lengths = {}
        self.doc_freqs = {}  # term -> doc_count
        self.term_freqs = {}  # doc_id -> {term -> freq}
    
    def index_document(self, doc_id, content):
        """索引文档"""
        terms = content.lower().split()
        self.doc_lengths[doc_id] = len(terms)
        self.doc_count += 1
        
        # 计算平均文档长度
        total_length = sum(self.doc_lengths.values())
        self.avg_doc_length = total_length / self.doc_count
        
        # 统计词频
        term_freq = Counter(terms)
        self.term_freqs[doc_id] = term_freq
        
        # 更新文档频率
        for term in set(terms):
            self.doc_freqs[term] = self.doc_freqs.get(term, 0) + 1
    
    def score(self, query, doc_id):
        """计算BM25分数"""
        terms = query.lower().split()
        score = 0
        
        for term in terms:
            if term not in self.term_freqs.get(doc_id, {}):
                continue
            
            # 词频
            tf = self.term_freqs[doc_id][term]
            
            # 文档频率
            df = self.doc_freqs.get(term, 0)
            
            # IDF
            idf = math.log((self.doc_count - df + 0.5) / (df + 0.5) + 1)
            
            # BM25公式
            doc_length = self.doc_lengths[doc_id]
            tf_norm = (tf * (self.k1 + 1)) / (
                tf + self.k1 * (1 - self.b + self.b * doc_length / self.avg_doc_length)
            )
            
            score += idf * tf_norm
        
        return score

搜索质量优化

搜索质量优化需要从多个维度入手，包括查询理解、召回、排序、展示等。

搜索质量优化策略：

查询优化：
  - 查询纠错（拼写纠错、语音纠错）
  - 查询扩展（同义词、上位词）
  - 查询推荐（搜索建议、相关搜索）

召回优化：
  - 多路召回（文本召回、语义召回、个性化召回）
  - 召回率提升（扩展索引、模糊匹配）

排序优化：
  - 特征工程（文本特征、用户特征、上下文特征）
  - 模型优化（Learning to Rank、深度学习排序）
  - 实时性优化（时效性加权、新鲜度排序）

展示优化：
  - 摘要生成（关键词高亮、智能摘要）
  - 结果多样化（类型分散、来源分散）
  - 个性化展示（用户画像、历史行为）