2.4 语义搜索与匹配 — AI知识库搭建全攻略 本节导读:深入理解语义搜索的核心原理、算法实现和优化策略,掌握如何构建高精度的智能检索系统,解决传统关键词搜索的局限性。 学习目标 掌握语义搜索的基本原理和核心概念 理解向量相似度计算的不同方法 学会实现高效的语义搜索和匹配算法 掌握搜索结果排序和优化的关键技术 了解语义搜索在企业级应用中的最佳实践 核心概念 语义搜索的本质 语义搜索是一种基于内容理解的智能检索技术,它超越了传统的关键词匹配,通过深度语义分析来理解用户的真实需求。
本节导读:深入理解语义搜索的核心原理、算法实现和优化策略,掌握如何构建高精度的智能检索系统,解决传统关键词搜索的局限性。
语义搜索是一种基于内容理解的智能检索技术,它超越了传统的关键词匹配,通过深度语义分析来理解用户的真实需求。
一个完整的语义搜索系统包含以下核心组件:
查询理解模块
向量表示模块
相似度计算模块
结果排序模块
反馈优化模块
余弦相似度计算两个向量之间的夹角余弦值,适用于稀疏向量和文本相似度计算。
import numpy as np from sklearn.metrics.pairwise import cosine_similarity def cosine_sim(vec1, vec2): """ 计算两个向量的余弦相似度 """ # 归一化向量 vec1_norm = vec1 / np.linalg.norm(vec1) vec2_norm = vec2 / np.linalg.norm(vec2) # 计算余弦相似度 similarity = np.dot(vec1_norm, vec2_norm) return similarity # 使用示例 query_embedding = np.array([0.1, 0.3, 0.5, 0.1]) doc_embedding = np.array([0.2, 0.4, 0.3, 0.1]) similarity = cosine_sim(query_embedding, doc_embedding) print(f"余弦相似度: {similarity:.4f}")
特点:
欧氏距离计算两个向量之间的直线距离,常用于密集向量空间。
from scipy.spatial.distance import euclidean def euclidean_dist(vec1, vec2): """ 计算欧氏距离(转换为相似度) """ distance = euclidean(vec1, vec2) # 转换为相似度(距离越小,相似度越高) similarity = 1 / (1 + distance) return similarity # 使用示例 query_embedding = np.array([0.1, 0.3, 0.5, 0.1]) doc_embedding = np.array([0.2, 0.4, 0.3, 0.1]) similarity = euclidean_dist(query_embedding, doc_embedding) print(f"欧氏距离相似度: {similarity:.4f}")
特点:
余弦距离是余弦相似度的补值,常用于距离计算。
def cosine_distance(vec1, vec2): """ 计算余弦距离 """ similarity = cosine_similarity([vec1], [vec2])[0][0] distance = 1 - similarity return distance
特点:
Jaccard 相似度用于计算两个集合的相似程度,适合关键词匹配场景。
def jaccard_similarity(set1, set2): """ 计算 Jaccard 相似度 """ intersection = len(set1.intersection(set2)) union = len(set1.union(set2)) return intersection / union if union != 0 else 0
特点:
大规模语义搜索需要高效的索引技术来提高检索速度。
实现代码:
from sklearn.neighbors import NearestNeighbors import numpy as np class HNSWSearch: def __init__(self, vectors, n_neighbors=10): self.vectors = vectors self.n_neighbors = n_neighbors self.model = NearestNeighbors( n_neighbors=n_neighbors, algorithm='auto', metric='cosine' ) self.model.fit(vectors) def search(self, query_vector, k=5): """ 搜索最相似的k个向量 """ distances, indices = self.model.kneighbors( [query_vector], n_neighbors=k ) return distances[0], indices[0] def batch_search(self, query_vectors, k=5): """ 批量搜索 """ distances, indices = self.model.kneighbors( query_vectors, n_neighbors=k ) return distances, indices
IVF 是一种基于聚类的索引方法,特别适合大规模向量检索。
from sklearn.cluster import KMeans import numpy as np class IVFSearch: def __init__(self, vectors, n_clusters=100): self.vectors = vectors self.n_clusters = n_clusters self.kmeans = KMeans(n_clusters=n_clusters, random_state=42) self.cluster_centers = self.kmeans.fit(vectors).cluster_centers_ self.index = self._build_index() def _build_index(self): """ 构建倒排索引 """ index = {} for i, vector in enumerate(self.vectors): cluster_id = self.kmeans.predict([vector])[0] if cluster_id not in index: index[cluster_id] = [] index[cluster_id].append(i) return index def search(self, query_vector, n_probe=10, k=5): """ 搜索最相似的k个向量 """ # 找到最近的几个聚类中心 distances = np.linalg.norm( self.cluster_centers - query_vector, axis=1 ) nearest_clusters = np.argsort(distances)[:n_probe] # 在这些聚类中搜索 candidates = [] for cluster_id in nearest_clusters: if cluster_id in self.index: candidates.extend(self.index[cluster_id]) # 计算相似度并排序 results = [] for idx in candidates: sim = cosine_similarity([query_vector], [self.vectors[idx]])[0][0] results.append((sim, idx)) # 返回前k个结果 results.sort(reverse=True) return [idx for _, idx in results[:k]]
现代语义搜索系统通常采用混合搜索策略,结合多种搜索方法。
实现代码:
class HybridSearch: def __init__(self, semantic_index, keyword_index): self.semantic_index = semantic_index self.keyword_index = keyword_index def search(self, query, semantic_weight=0.7, keyword_weight=0.3, k=10): """ 混合搜索 """ # 语义搜索 semantic_results = self.semantic_index.search(query, k) # 关键词搜索 keyword_results = self.keyword_index.search(query, k) # 合并和重排序 combined_results = self._reorder_results( semantic_results, keyword_results, semantic_weight, keyword_weight ) return combined_results[:k] def _reorder_results(self, semantic_results, keyword_results, semantic_weight, keyword_weight): """ 结果重排序 """ # 计算综合得分 combined_scores = {} # 语义得分 for sim, idx in semantic_results: if idx not in combined_scores: combined_scores[idx] = 0 combined_scores[idx] += sim * semantic_weight # 关键词得分 for score, idx in keyword_results: if idx not in combined_scores: combined_scores[idx] = 0 combined_scores[idx] += score * keyword_weight # 排序并返回结果 sorted_results = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True) return sorted_results
语义搜索的结果需要通过重排序来提高相关性。
from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split class LearningToRank: def __init__(self): self.model = RandomForestRegressor(n_estimators=100, random_state=42) def train(self, training_data): """ 训练排序模型 """ X = training_data[['semantic_sim', 'keyword_match', 'recency', 'popularity', 'user_rating']] y = training_data['relevance_score'] X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) self.model.fit(X_train, y_train) return self.model.score(X_test, y_test) def rank(self, results): """ 对结果进行排序 """ X = [] for result in results: features = [ result['semantic_similarity'], result['keyword_match_score'], result['recency_score'], result['popularity_score'], result['user_rating'] ] X.append(features) scores = self.model.predict(X) ranked_results = [] for i, (result, score) in enumerate(zip(results, scores)): ranked_results.append((result, score)) ranked_results.sort(key=lambda x: x[1], reverse=True) return [result for result, score in ranked_results]
class FeatureEngineering: @staticmethod def extract_cross_features(semantic_sim, keyword_match, query_length, doc_length): """ 提取交叉特征 """ features = { 'semantic_keyword_ratio': semantic_sim / (keyword_match + 0.001), 'query_doc_ratio': query_length / (doc_length + 1), 'semantic_keyword_interaction': semantic_sim * keyword_match, 'harmonic_mean': 2 * semantic_sim * keyword_match / (semantic_sim + keyword_match + 0.001) } return features
个性化搜索根据用户的历史行为和偏好调整搜索结果。
class UserProfile: def __init__(self): self.interests = {} self.search_history = [] self.click_patterns = {} def update_from_search(self, query, results, clicked_docs): """ 根据搜索行为更新用户画像 """ # 记录搜索历史 self.search_history.append({ 'query': query, 'results': results, 'clicked': clicked_docs, 'timestamp': datetime.now() }) # 更新兴趣标签 for doc_id in clicked_docs: # 提取文档主题(简化示例) topics = self._extract_topics(results[doc_id]) for topic in topics: self.interests[topic] = self.interests.get(topic, 0) + 1 def get_interest_score(self, doc_topics): """ 计算文档与用户兴趣的匹配度 """ score = 0 for topic in doc_topics: score += self.interests.get(topic, 0) return score def _extract_topics(self, doc_content): """ 从文档内容提取主题(简化示例) """ # 实际应用中可以使用更复杂的主题模型 simple_topics = ['技术', '产品', '设计', '开发', '分析'] return simple_topics[:2] # 返回前两个主题