倒排索引与TF-IDF的关系及搜索相关性实践

import re
from collections import defaultdict
 
class DocumentProcessor:
    def __init__(self):
        self.stop_words = {'的', '了', '在', '是', '和', '与', '或'}
    
    def preprocess(self, text):
        # 分词（简化版）
        words = re.findall(r'\w+', text.lower())
        # 去除停用词
        return [word for word in words if word not in self.stop_words]

class InvertedIndex:
    def __init__(self):
        self.index = defaultdict(set)  # 词项 -> 文档ID集合
        self.documents = {}  # 文档ID -> 文档内容
        self.doc_freq = defaultdict(int)  # 词项的文档频率
    
    def add_document(self, doc_id, content):
        words = DocumentProcessor().preprocess(content)
        self.documents[doc_id] = content
        
        for word in set(words):  # 使用set避免重复计算
            self.index[word].add(doc_id)
            self.doc_freq[word] += 1
    
    def search(self, query):
        query_words = DocumentProcessor().preprocess(query)
        if not query_words:
            return []
        
        # 取第一个词的文档集合作为基础
        result_docs = self.index[query_words[0]].copy()
        
        # 与其他词的文档集合求交集
        for word in query_words[1:]:
            result_docs &= self.index[word]
        
        return list(result_docs)

# 使用跳表（Skip List）优化长倒排列表
class SkipList:
    def __init__(self, postings, skip_length=4):
        self.postings = sorted(postings)
        self.skip_length = skip_length
        self.skip_pointers = self._build_skip_pointers()
    
    def _build_skip_pointers(self):
        skips = {}
        for i in range(0, len(self.postings), self.skip_length):
            if i + self.skip_length < len(self.postings):
                skips[i] = i + self.skip_length
        return skips
    
    def intersect(self, other):
        # 优化的交集算法
        result = []
        i = j = 0
        
        while i < len(self.postings) and j < len(other.postings):
            if self.postings[i] == other.postings[j]:
                result.append(self.postings[i])
                i += 1
                j += 1
            elif self.postings[i] < other.postings[j]:
                # 使用跳指针加速
                if i in self.skip_pointers and self.postings[self.skip_pointers[i]] <= other.postings[j]:
                    i = self.skip_pointers[i]
                else:
                    i += 1
            else:
                j += 1
        
        return result

import math
from collections import Counter
 
class TFIDFCalculator:
    def __init__(self):
        self.documents = {}
        self.corpus_size = 0
        self.doc_freq = defaultdict(int)
    
    def add_document(self, doc_id, content):
        words = DocumentProcessor().preprocess(content)
        word_count = Counter(words)
        
        self.documents[doc_id] = {
            'word_count': word_count,
            'total_words': len(words)
        }
        
        # 更新文档频率
        for word in word_count.keys():
            self.doc_freq[word] += 1
        
        self.corpus_size += 1
    
    def calculate_tf(self, word, doc_id):
        if doc_id not in self.documents:
            return 0
        
        doc_info = self.documents[doc_id]
        word_count = doc_info['word_count'].get(word, 0)
        total_words = doc_info['total_words']
        
        # 使用对数缩放
        if word_count > 0:
            return 1 + math.log10(word_count)
        return 0
    
    def calculate_idf(self, word):
        if word not in self.doc_freq or self.doc_freq[word] == 0:
            return 0
        
        return math.log10(self.corpus_size / self.doc_freq[word])
    
    def calculate_tfidf(self, word, doc_id):
        tf = self.calculate_tf(word, doc_id)
        idf = self.calculate_idf(word)
        return tf * idf
    
    def get_document_vector(self, doc_id):
        if doc_id not in self.documents:
            return {}
        
        vector = {}
        for word in self.documents[doc_id]['word_count'].keys():
            vector[word] = self.calculate_tfidf(word, doc_id)
        
        return vector

class SearchEngine:
    def __init__(self):
        self.inverted_index = InvertedIndex()
        self.tfidf_calculator = TFIDFCalculator()
    
    def build_index(self, documents):
        for doc_id, content in documents.items():
            self.inverted_index.add_document(doc_id, content)
            self.tfidf_calculator.add_document(doc_id, content)
    
    def calculate_cosine_similarity(self, query_vector, doc_vector):
        # 计算余弦相似度
        intersection = set(query_vector.keys()) & set(doc_vector.keys())
        
        if not intersection:
            return 0.0
        
        dot_product = sum(query_vector[word] * doc_vector[word] for word in intersection)
        
        query_norm = math.sqrt(sum(val ** 2 for val in query_vector.values()))
        doc_norm = math.sqrt(sum(val ** 2 for val in doc_vector.values()))
        
        if query_norm == 0 or doc_norm == 0:
            return 0.0
        
        return dot_product / (query_norm * doc_norm)
    
    def search(self, query, top_k=10):
        # 1. 获取候选文档
        candidate_docs = self.inverted_index.search(query)
        
        if not candidate_docs:
            return []
        
        # 2. 构建查询向量
        query_words = DocumentProcessor().preprocess(query)
        query_vector = {}
        for word in set(query_words):
            # 查询词的TF-IDF计算
            tf = 1 + math.log10(query_words.count(word))
            idf = self.tfidf_calculator.calculate_idf(word)
            query_vector[word] = tf * idf
        
        # 3. 计算相关性得分
        scores = []
        for doc_id in candidate_docs:
            doc_vector = self.tfidf_calculator.get_document_vector(doc_id)
            similarity = self.calculate_cosine_similarity(query_vector, doc_vector)
            scores.append((doc_id, similarity))
        
        # 4. 排序并返回结果
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_k]

# TRAE IDE会自动提示优化方案
def calculate_tfidf_optimized(self, word, doc_id):
    """
    TRAE IDE提示：考虑使用缓存机制避免重复计算
    """
    cache_key = f"{word}_{doc_id}"
    if cache_key in self.cache:
        return self.cache[cache_key]
    
    tf = self.calculate_tf(word, doc_id)
    idf = self.calculate_idf(word)
    result = tf * idf
    
    # 缓存结果
    self.cache[cache_key] = result
    return result

# TRAE IDE性能监控示例
@trae_performance_monitor
def search_with_performance_tracking(self, query):
    """
    该函数的执行时间、内存使用将被TRAE IDE实时监控
    """
    start_time = time.time()
    
    # 搜索逻辑
    results = self.search(query)
    
    # TRAE IDE会自动记录并展示性能指标
    execution_time = time.time() - start_time
    return results

# 在TRAE IDE中设置断点，可视化索引构建过程
def debug_index_building(self):
    documents = {
        'doc1': '机器学习算法原理',
        'doc2': '深度学习神经网络',
        'doc3': '强化学习应用场景'
    }
    
    for doc_id, content in documents.items():
        # TRAE IDE断点：查看每个文档的索引构建过程
        self.inverted_index.add_document(doc_id, content)
        self.tfidf_calculator.add_document(doc_id, content)
        
        # 可视化当前索引状态
        print(f"文档 {doc_id} 索引构建完成")
        print(f"倒排索引大小: {len(self.inverted_index.index)}")
        print(f"TF-IDF向量维度: {len(self.tfidf_calculator.doc_freq)}")

class CompressedIndex:
    def __init__(self):
        self.compressed_index = {}
    
    def compress_postings_list(self, doc_ids):
        """使用Variable Byte编码压缩倒排列表"""
        sorted_ids = sorted(doc_ids)
        compressed = []
        
        # 差值编码
        gaps = []
        prev = 0
        for doc_id in sorted_ids:
            gaps.append(doc_id - prev)
            prev = doc_id
        
        # Variable Byte编码
        for gap in gaps:
            compressed.extend(self.variable_byte_encode(gap))
        
        return bytes(compressed)
    
    def variable_byte_encode(self, number):
        """Variable Byte编码实现"""
        bytes_list = []
        while number >= 128:
            bytes_list.append((number & 0x7F) | 0x80)
            number >>= 7
        bytes_list.append(number)
        return bytes_list

import concurrent.futures
from multiprocessing import cpu_count
 
class ParallelSearchEngine(SearchEngine):
    def parallel_tfidf_calculation(self, doc_ids, query_vector):
        """并行计算TF-IDF相似度"""
        def calculate_similarity(doc_id):
            doc_vector = self.tfidf_calculator.get_document_vector(doc_id)
            return (doc_id, self.calculate_cosine_similarity(query_vector, doc_vector))
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=cpu_count()) as executor:
            futures = [executor.submit(calculate_similarity, doc_id) for doc_id in doc_ids]
            results = [future.result() for future in concurrent.futures.as_completed(futures)]
        
        return results

from functools import lru_cache
import hashlib
 
class CachedSearchEngine(SearchEngine):
    def __init__(self, cache_size=1000):
        super().__init__()
        self.cache_size = cache_size
        self.query_cache = {}
    
    def get_query_hash(self, query):
        """生成查询的哈希值"""
        return hashlib.md5(query.encode()).hexdigest()
    
    def cached_search(self, query, top_k=10):
        """带缓存的搜索功能"""
        query_hash = self.get_query_hash(query)
        
        if query_hash in self.query_cache:
            print(f"缓存命中: {query}")
            return self.query_cache[query_hash][:top_k]
        
        # 执行搜索
        results = self.search(query, top_k)
        
        # 缓存结果
        if len(self.query_cache) >= self.cache_size:
            # LRU淘汰策略
            oldest_key = next(iter(self.query_cache))
            del self.query_cache[oldest_key]
        
        self.query_cache[query_hash] = results
        return results

class EcommerceSearchEngine(CachedSearchEngine):
    def __init__(self):
        super().__init__()
        self.product_features = {}  # 存储商品特征
    
    def add_product(self, product_id, title, description, category, attributes):
        """添加商品到搜索索引"""
        # 构建商品文本内容
        content = f"{title} {description} {category} {' '.join(attributes.values())}"
        
        # 添加到基础索引
        self.build_index({product_id: content})
        
        # 存储商品特征用于个性化排序
        self.product_features[product_id] = {
            'category': category,
            'attributes': attributes,
            'popularity': 0,  # 商品热度
            'sales': 0        # 销量
        }
    
    def personalized_search(self, query, user_profile=None, top_k=10):
        """个性化商品搜索"""
        # 基础搜索
        base_results = self.cached_search(query, top_k * 2)  # 获取更多结果用于重排序
        
        if not user_profile:
            return base_results[:top_k]
        
        # 个性化重排序
        personalized_scores = []
        for product_id, relevance_score in base_results:
            # 计算个性化得分
            personalization_score = self.calculate_personalization_score(
                product_id, user_profile
            )
            
            # 综合得分 = 相关性得分 * 个性化权重
            final_score = relevance_score * (1 + personalization_score)
            personalized_scores.append((product_id, final_score))
        
        # 重新排序
        personalized_scores.sort(key=lambda x: x[1], reverse=True)
        return personalized_scores[:top_k]
    
    def calculate_personalization_score(self, product_id, user_profile):
        """计算商品与用户画像的匹配度"""
        product_info = self.product_features.get(product_id, {})
        
        score = 0.0
        
        # 类别偏好匹配
        if 'preferred_categories' in user_profile:
            if product_info.get('category') in user_profile['preferred_categories']:
                score += 0.3
        
        # 价格范围匹配
        if 'price_range' in user_profile and 'price' in product_info:
            min_price, max_price = user_profile['price_range']
            if min_price <= product_info['price'] <= max_price:
                score += 0.2
        
        # 销量权重（热门商品加权）
        if 'sales' in product_info:
            sales = product_info['sales']
            if sales > 1000:
                score += 0.1
            elif sales > 100:
                score += 0.05
        
        return score

# TRAE IDE性能分析示例
@trae_performance_profile
def optimize_search_performance(self):
    """
    TRAE IDE会自动分析此函数的性能瓶颈
    """
    # 模拟大量查询
    test_queries = ['机器学习', '深度学习', '自然语言处理'] * 100
    
    for query in test_queries:
        results = self.personalized_search(query)
        
        # TRAE IDE会显示：
        # 1. 每个查询的平均响应时间
        # 2. 内存使用峰值
        # 3. CPU使用率
        # 4. 缓存命中率