2.4 语义搜索与匹配（下）

2. 个性化排序

class PersonalizedSearch:
    def __init__(self, base_searcher, user_profile):
        self.base_searcher = base_searcher
        self.user_profile = user_profile
    
    def search(self, query, user_id, k=10):
        """
        个性化搜索
        """
        # 基础搜索
        base_results = self.base_searcher.search(query, k * 2)  # 获取更多候选
        
        # 个性化重排序
        personalized_results = []
        for result in base_results:
            # 计算个性化得分
            base_score = result['similarity_score']
            personal_score = self.user_profile.get_interest_score(
                result['topics']
            )
            
            # 加权组合
            final_score = 0.7 * base_score + 0.3 * personal_score
            personalized_results.append({
                'result': result,
                'final_score': final_score
            })
        
        # 排序并返回
        personalized_results.sort(key=lambda x: x['final_score'], reverse=True)
        return [r['result'] for r in personalized_results[:k]]
​

性能优化策略

1. 查询缓存

import pickle
import os
from datetime import datetime, timedelta

class QueryCache:
    def __init__(self, cache_dir='./cache', ttl_hours=24):
        self.cache_dir = cache_dir
        self.ttl_hours = ttl_hours
        self.cache = {}
        
        # 创建缓存目录
        os.makedirs(cache_dir, exist_ok=True)
    
    def get_cache_key(self, query, params):
        """
        生成缓存键
        """
        key_str = f"{query}_{params}"
        return hash(key_str)
    
    def get(self, query, params):
        """
        获取缓存结果
        """
        cache_key = self.get_cache_key(query, params)
        
        if cache_key in self.cache:
            cached_data = self.cache[cache_key]
            if datetime.now() - cached_data['timestamp'] < timedelta(hours=self.ttl_hours):
                return cached_data['results']
            else:
                del self.cache[cache_key]
        
        return None
    
    def set(self, query, params, results):
        """
        设置缓存
        """
        cache_key = self.get_cache_key(query, params)
        self.cache[cache_key] = {
            'results': results,
            'timestamp': datetime.now()
        }
    
    def clear_expired(self):
        """
        清理过期缓存
        """
        current_time = datetime.now()
        expired_keys = []
        
        for key, data in self.cache.items():
            if current_time - data['timestamp'] > timedelta(hours=self.ttl_hours):
                expired_keys.append(key)
        
        for key in expired_keys:
            del self.cache[key]
​

2. 异步搜索

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncSearchEngine:
    def __init__(self, search_engines):
        self.search_engines = search_engines
        self.executor = ThreadPoolExecutor(max_workers=4)
    
    async def search_parallel(self, query, k=10):
        """
        并行搜索多个搜索引擎
        """
        tasks = []
        for engine in self.search_engines:
            task = asyncio.get_event_loop().run_in_executor(
                self.executor, 
                engine.search, 
                query, 
                k
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return self._merge_results(results)
    
    def _merge_results(self, results):
        """
        合并多个搜索结果
        """
        merged = {}
        for i, engine_results in enumerate(results):
            for result in engine_results:
                doc_id = result['id']
                if doc_id not in merged:
                    merged[doc_id] = {
                        'id': doc_id,
                        'scores': [],
                        'engines': []
                    }
                
                merged[doc_id]['scores'].append(result['score'])
                merged[doc_id]['engines'].append(i)
        
        # 计算综合得分
        final_results = []
        for doc_id, data in merged.items():
            avg_score = sum(data['scores']) / len(data['scores'])
            final_results.append({
                'id': doc_id,
                'score': avg_score,
                'engines': data['engines']
            })
        
        return sorted(final_results, key=lambda x: x['score'], reverse=True)
​

3. 预计算和索引

class PrecomputedIndex:
    def __init__(self, documents):
        self.documents = documents
        self.inverted_index = self._build_inverted_index()
        self.embeddings_matrix = self._compute_embeddings_matrix()
    
    def _build_inverted_index(self):
        """
        构建倒排索引
        """
        index = {}
        for doc_id, doc in enumerate(self.documents):
            for term in doc['tokens']:
                if term not in index:
                    index[term] = []
                index[term].append(doc_id)
        return index
    
    def _compute_embeddings_matrix(self):
        """
        预计算所有文档的嵌入向量
        """
        embeddings = []
        for doc in self.documents:
            embedding = self._compute_embedding(doc['content'])
            embeddings.append(embedding)
        return np.array(embeddings)
    
    def compute_bm25_scores(self, query_terms):
        """
        计算BM25分数
        """
        scores = np.zeros(len(self.documents))
        
        for term in query_terms:
            if term in self.inverted_index:
                doc_ids = self.inverted_index[term]
                idf = np.log(len(self.documents) / len(doc_ids))
                
                for doc_id in doc_ids:
                    tf = self.documents[doc_id]['term_freq'].get(term, 0)
                    scores[doc_id] += idf * tf
        
        return scores
    
    def compute_semantic_scores(self, query_embedding):
        """
        计算语义相似度分数
        """
        similarities = cosine_similarity([query_embedding], self.embeddings_matrix)[0]
        return similarities
​

实战案例

企业级语义搜索系统

class EnterpriseSemanticSearch:
    def __init__(self, config):
        self.config = config
        self.preprocessor = TextPreprocessor()
        self.embedder = EmbeddingModel(config['embedding_model'])
        self.index = VectorIndex(config['index_type'])
        self.ranker = LearningToRank()
        self.cache = QueryCache()
        
        # 加载文档数据
        self.documents = self._load_documents()
        self._build_index()
    
    def _load_documents(self):
        """
        加载文档数据
        """
        # 实际应用中从数据库或文件系统加载
        return [
            {'id': 1, 'title': 'AI技术发展报告', 'content': '人工智能技术的最新发展...', 'category': '技术'},
            {'id': 2, 'title': '机器学习入门指南', 'content': '机器学习的基础知识和应用...', 'category': '教育'},
            # 更多文档...
        ]
    
    def _build_index(self):
        """
        构建搜索索引
        """
        # 预处理文档
        processed_docs = []
        for doc in self.documents:
            processed_text = self.preprocessor.process(doc['content'])
            embedding = self.embedder.embed(processed_text)
            
            processed_docs.append({
                'id': doc['id'],
                'title': doc['title'],
                'content': processed_text,
                'embedding': embedding,
                'category': doc['category']
            })
        
        # 构建索引
        self.index.build(processed_docs)
    
    def search(self, query, user_profile=None, filters=None):
        """
        执行搜索
        """
        # 检查缓存
        cache_key = self._get_cache_key(query, user_profile, filters)
        cached_result = self.cache.get(query, cache_key)
        if cached_result:
            return cached_result
        
        # 预处理查询
        processed_query = self.preprocessor.process(query)
        query_embedding = self.embedder.embed(processed_query)
        
        # 语义搜索
        semantic_results = self.index.search(query_embedding, k=50)
        
        # 应用过滤条件
        if filters:
            semantic_results = self._apply_filters(semantic_results, filters)
        
        # 个性化排序
        if user_profile:
            semantic_results = self._personalize_results(
                semantic_results, 
                user_profile
            )
        
        # 相关性重排序
        final_results = self.ranker.rank(semantic_results)
        
        # 缓存结果
        self.cache.set(query, cache_key, final_results)
        
        return final_results[:20]  # 返回前20个结果
    
    def _apply_filters(self, results, filters):
        """
        应用过滤条件
        """
        filtered_results = []
        for result in results:
            doc = self.documents[result['doc_id']]
            
            # 检查过滤条件
            match = True
            if 'category' in filters and doc['category'] != filters['category']:
                match = False
            
            if match:
                filtered_results.append(result)
        
        return filtered_results
    
    def _personalize_results(self, results, user_profile):
        """
        个性化结果
        """
        personalized_results = []
        for result in results:
            doc = self.documents[result['doc_id']]
            
            # 计算个性化分数
            personal_score = self._calculate_personal_score(
                doc, 
                user_profile
            )
            
            # 组合分数
            combined_score = 0.7 * result['score'] + 0.3 * personal_score
            
            personalized_results.append({
                **result,
                'personal_score': personal_score,
                'combined_score': combined_score
            })
        
        # 按组合分数排序
        personalized_results.sort(key=lambda x: x['combined_score'], reverse=True)
        return personalized_results
    
    def _calculate_personal_score(self, doc, user_profile):
        """
        计算文档与用户画像的匹配度
        """
        # 简化的个性化计算
        category_match = 1.0 if doc['category'] in user_profile['interested_categories'] else 0.5
        popularity_factor = doc.get('popularity', 0) * 0.1
        
        return category_match + popularity_factor
​

搜索性能监控

class SearchPerformanceMonitor:
    def __init__(self):
        self.metrics = {
            'search_count': 0,
            'avg_response_time': 0,
            'cache_hit_rate': 0,
            'user_satisfaction': 0
        }
        self.search_times = []
        self.cache_hits = 0
        self.cache_misses = 0
    
    def record_search(self, response_time, cache_hit=False):
        """
        记录搜索性能
        """
        self.metrics['search_count'] += 1
        self.search_times.append(response_time)
        
        if cache_hit:
            self.cache_hits += 1
        else:
            self.cache_misses += 1
        
        # 更新平均响应时间
        if len(self.search_times) > 0:
            self.metrics['avg_response_time'] = sum(self.search_times) / len(self.search_times)
        
        # 更新缓存命中率
        total_cache_requests = self.cache_hits + self.cache_misses
        if total_cache_requests > 0:
            self.metrics['cache_hit_rate'] = self.cache_hits / total_cache_requests
    
    def get_metrics(self):
        """
        获取性能指标
        """
        return self.metrics.copy()
    
    def generate_report(self):
        """
        生成性能报告
        """
        report = {
            '总搜索次数': self.metrics['search_count'],
            '平均响应时间': f"{self.metrics['avg_response_time']:.2f}秒",
            '缓存命中率': f"{self.metrics['cache_hit_rate']*100:.1f}%",
            '搜索次数趋势': self._get_trend(),
            '性能建议': self._get_recommendations()
        }
        return report
    
    def _get_trend(self):
        """
        分析搜索趋势
        """
        if len(self.search_times) < 10:
            return '数据不足'
        
        recent_times = self.search_times[-10:]
        avg_recent = sum(recent_times) / len(recent_times)
        avg_overall = self.metrics['avg_response_time']
        
        if avg_recent < avg_overall:
            return '性能改善'
        elif avg_recent > avg_overall:
            return '性能下降'
        else:
            return '性能稳定'
    
    def _get_recommendations(self):
        """
        获取优化建议
        """
        recommendations = []
        
        if self.metrics['avg_response_time'] > 2.0:
            recommendations.append('考虑增加缓存以提高响应速度')
        
        if self.metrics['cache_hit_rate'] < 0.7:
            recommendations.append('优化缓存策略，提高缓存命中率')
        
        if len(self.search_times) > 1000:
            recommendations.append('考虑使用分布式搜索以提高扩展性')
        
        return recommendations
​

本节小结

本节深入探讨了语义搜索与匹配的核心技术，从基础概念到高级实现。我们学习了：

1. 语义搜索的本质：理解语义搜索与传统搜索的区别，掌握其技术栈组成
2. 相似度计算：掌握余弦相似度、欧氏距离等常用方法，理解其适用场景
3. 向量索引优化：学习HNSW、IVF等索引技术，提高检索效率
4. 混合搜索策略：结合语义和关键词搜索，提升结果质量
5. 结果优化：相关性重排序、个性化搜索等高级技术
6. 性能优化：缓存策略、异步搜索、预计算等优化手段
7. 实战案例：企业级语义搜索系统的完整实现

语义搜索是现代AI知识库的核心技术之一，它通过深度语义理解大大提升了搜索的准确性和用户体验。在实际应用中，需要根据具体场景选择合适的搜索策略，并持续优化性能和准确性。

下一节将进入第3章「架构设计」，深入学习如何设计高性能、可扩展的AI知识库系统架构。

延伸阅读

• 官方文档：Faiss向量数据库技术文档
• 相关章节：本教程2.1节向量数据库基础
• 推荐资源：《相似性搜索算法与实践》技术白皮书