从相似推荐到异常检测手把手用PyTorch实现余弦相似度与欧氏距离的实战项目在推荐系统和异常检测领域相似度计算是最基础也最核心的技术之一。想象一下当你在电商平台浏览商品时系统如何精准推荐你可能喜欢的其他商品当工厂的传感器采集到海量数据时又如何快速识别出异常设备状态这些场景背后都离不开相似度算法的支撑。PyTorch作为当前最流行的深度学习框架之一提供了丰富的张量操作和距离计算函数。本文将带你用PyTorch实现两个完整的实战项目一个基于余弦相似度的简易推荐系统和一个基于欧氏距离的异常检测系统。我们不仅会讲解API的使用更会聚焦于如何将这些数学概念转化为实际可用的解决方案。1. 环境准备与数据构建在开始项目之前我们需要准备好开发环境。推荐使用Python 3.8和PyTorch 1.10版本这些版本提供了最稳定的API支持。pip install torch torchvision numpy matplotlib1.1 构建电影推荐数据集为了演示推荐系统我们将创建一个模拟的电影评分数据集。这个数据集包含10个用户对20部电影的评分评分范围1-5分。import torch import numpy as np # 设置随机种子保证可重复性 torch.manual_seed(42) # 生成用户-电影评分矩阵 (10用户 × 20电影) num_users 10 num_movies 20 ratings torch.randint(1, 6, (num_users, num_movies)).float() # 添加一些缺失值(未评分) mask torch.rand(num_users, num_movies) 0.7 ratings[mask] 0 # 0表示未评分 print(f评分矩阵形状: {ratings.shape}) print(ratings[:3, :5]) # 展示前3个用户对前5部电影的评分1.2 准备异常检测数据对于异常检测项目我们将生成包含正常点和异常点的二维数据。正常点来自一个高斯分布异常点则是随机分布的离群点。# 生成正常数据点 normal_data torch.randn(100, 2) * 0.5 torch.tensor([2.0, 2.0]) # 生成异常数据点 anomaly_data torch.rand(20, 2) * 6 - 3 # 范围在[-3,3] # 合并数据集 all_data torch.cat([normal_data, anomaly_data], dim0) labels torch.cat([torch.zeros(100), torch.ones(20)]) # 0正常,1异常 # 可视化数据 import matplotlib.pyplot as plt plt.scatter(normal_data[:,0], normal_data[:,1], label正常) plt.scatter(anomaly_data[:,0], anomaly_data[:,1], colorr, label异常) plt.legend() plt.title(异常检测数据集) plt.show()2. 基于余弦相似度的推荐系统实现余弦相似度衡量的是两个向量在方向上的相似程度而不考虑它们的大小。这在推荐系统中特别有用因为我们更关注用户的偏好模式而非评分绝对值。2.1 计算用户相似度首先我们实现一个基于用户的协同过滤算法找到相似用户进行推荐。from torch.nn.functional import cosine_similarity def user_based_cf(ratings, user_idx, top_k3): 基于用户的协同过滤推荐 :param ratings: 用户-电影评分矩阵 :param user_idx: 目标用户索引 :param top_k: 返回最相似的k个用户 :return: 推荐电影列表 # 计算目标用户与其他用户的相似度 target ratings[user_idx].unsqueeze(0) # 形状变为(1, num_movies) sims cosine_similarity(target, ratings, dim1) # 排除用户自己 sims[user_idx] -1 # 获取最相似的top_k用户 _, similar_users torch.topk(sims, top_k) # 找出这些相似用户喜欢但目标用户未评分的电影 similar_users_ratings ratings[similar_users] avg_ratings similar_users_ratings.mean(dim0) # 目标用户未评分的电影 unrated (ratings[user_idx] 0) # 推荐评分最高的未看电影 recommended torch.argsort(avg_ratings * unrated, descendingTrue) return recommended[:5] # 返回前5个推荐 # 测试推荐系统 user_idx 0 recommendations user_based_cf(ratings, user_idx) print(f为用户{user_idx}推荐的电影索引: {recommendations})2.2 实现物品相似度推荐除了基于用户的推荐我们还可以计算电影之间的相似度实现基于物品的推荐。def item_based_cf(ratings, user_idx, top_k5): 基于物品的协同过滤推荐 :param ratings: 用户-电影评分矩阵 :param user_idx: 目标用户索引 :param top_k: 对每个已评分电影考虑最相似的k个电影 :return: 推荐电影列表 # 转置矩阵得到电影-用户矩阵 movie_user ratings.T # 计算电影之间的相似度矩阵 num_movies movie_user.shape[0] movie_sim torch.zeros(num_movies, num_movies) for i in range(num_movies): for j in range(i, num_movies): # 只计算都评过分的用户 mask (movie_user[i] 0) (movie_user[j] 0) if mask.sum() 0: sim cosine_similarity(movie_user[i][mask].unsqueeze(0), movie_user[j][mask].unsqueeze(0)) movie_sim[i,j] movie_sim[j,i] sim.item() # 获取目标用户已评分的电影 rated_movies torch.where(ratings[user_idx] 0)[0] # 计算推荐得分 scores torch.zeros(num_movies) for movie in rated_movies: # 获取当前电影的top_k相似电影 _, similar_movies torch.topk(movie_sim[movie], top_k) # 累加相似度作为推荐分数 for sim_movie in similar_movies: if ratings[user_idx, sim_movie] 0: # 只考虑未评分的 scores[sim_movie] movie_sim[movie, sim_movie] # 推荐得分最高的电影 recommended torch.argsort(scores, descendingTrue) return recommended[:5] # 测试物品相似度推荐 recommendations item_based_cf(ratings, user_idx) print(f基于物品相似度为用户{user_idx}推荐的电影: {recommendations})2.3 推荐系统评估为了评估推荐系统的效果我们可以使用留出法计算预测评分与实际评分的差异。def evaluate_recommendation(ratings, user_idx, k5): 评估推荐系统效果 :param ratings: 完整评分矩阵 :param user_idx: 目标用户索引 :param k: 隐藏k个评分作为测试集 :return: 预测评分与实际评分的平均绝对误差 # 复制评分矩阵并隐藏k个评分 test_ratings ratings.clone() rated torch.where(ratings[user_idx] 0)[0] test_indices torch.randperm(len(rated))[:k] hidden_ratings ratings[user_idx, rated[test_indices]] test_ratings[user_idx, rated[test_indices]] 0 # 获取推荐 recommended user_based_cf(test_ratings, user_idx, top_k3) # 计算预测评分(使用相似用户的平均评分) target test_ratings[user_idx].unsqueeze(0) sims cosine_similarity(target, test_ratings, dim1) sims[user_idx] -1 _, similar_users torch.topk(sims, 3) # 计算预测评分 pred_ratings test_ratings[similar_users][:, rated[test_indices]].mean(dim0) # 计算MAE mae torch.abs(pred_ratings - hidden_ratings).mean() return mae.item() # 评估推荐系统 mae evaluate_recommendation(ratings, user_idx) print(f推荐系统MAE: {mae:.2f})3. 基于欧氏距离的异常检测系统欧氏距离衡量的是空间中两点之间的直线距离非常适合用于检测偏离正常模式的异常点。3.1 计算欧氏距离PyTorch提供了多种计算欧氏距离的方法我们来比较它们的性能和使用场景。from torch.nn import PairwiseDistance from torch.cdist import cdist # 定义三个计算欧氏距离的方法 def euclidean_dist1(a, b): 使用PairwiseDistance pdist PairwiseDistance(p2) return pdist(a.unsqueeze(0), b.unsqueeze(0)) def euclidean_dist2(a, b): 使用vector_norm return torch.linalg.vector_norm(a - b, ord2) def euclidean_dist3(a, b): 使用cdist return cdist(a.unsqueeze(0), b.unsqueeze(0), p2) # 测试三种方法 a torch.tensor([1.0, 2.0]) b torch.tensor([4.0, 6.0]) print(fPairwiseDistance: {euclidean_dist1(a, b).item():.4f}) print(fvector_norm: {euclidean_dist2(a, b).item():.4f}) print(fcdist: {euclidean_dist3(a, b).item():.4f})3.2 实现简单异常检测器我们将使用欧氏距离来实现一个基于密度的异常检测算法。def density_based_anomaly_detection(data, k5, threshold0.9): 基于密度的异常检测 :param data: 输入数据 (n_samples, n_features) :param k: 考虑的最近邻数量 :param threshold: 异常分数阈值 :return: 异常分数和预测标签 # 计算所有点之间的距离矩阵 dist_matrix cdist(data, data, p2) # 对每个点获取到k个最近邻的距离 topk_dists, _ torch.topk(dist_matrix, k1, largestFalse) # 1因为包含自己 avg_knn_dist topk_dists[:, 1:].mean(dim1) # 排除自身 # 计算异常分数 (标准化到0-1) anomaly_scores (avg_knn_dist - avg_knn_dist.min()) / (avg_knn_dist.max() - avg_knn_dist.min()) # 预测异常 pred_labels (anomaly_scores threshold).long() return anomaly_scores, pred_labels # 运行异常检测 anomaly_scores, pred_labels density_based_anomaly_detection(all_data) # 可视化结果 plt.scatter(all_data[:,0], all_data[:,1], cpred_labels, cmapcoolwarm) plt.title(异常检测结果(红色异常)) plt.show()3.3 评估异常检测性能我们可以计算准确率、召回率等指标来评估异常检测器的性能。def evaluate_anomaly_detection(true_labels, pred_labels): 评估异常检测性能 :param true_labels: 真实标签 (0正常,1异常) :param pred_labels: 预测标签 (0正常,1异常) :return: 准确率,召回率,F1分数 # 计算混淆矩阵 tp ((true_labels 1) (pred_labels 1)).sum().item() fp ((true_labels 0) (pred_labels 1)).sum().item() fn ((true_labels 1) (pred_labels 0)).sum().item() tn ((true_labels 0) (pred_labels 0)).sum().item() # 计算指标 accuracy (tp tn) / (tp fp fn tn) precision tp / (tp fp) if (tp fp) 0 else 0 recall tp / (tp fn) if (tp fn) 0 else 0 f1 2 * (precision * recall) / (precision recall) if (precision recall) 0 else 0 return accuracy, precision, recall, f1 # 评估性能 accuracy, precision, recall, f1 evaluate_anomaly_detection(labels, pred_labels) print(f准确率: {accuracy:.2f}, 精确率: {precision:.2f}, 召回率: {recall:.2f}, F1分数: {f1:.2f})4. 高级应用与性能优化在实际应用中我们需要考虑算法的扩展性和性能优化特别是当数据量很大时。4.1 批量计算与GPU加速PyTorch的一个主要优势是可以利用GPU进行并行计算。我们来看如何优化距离计算。def batch_cosine_similarity(queries, targets, batch_size64, devicecuda): 分批计算余弦相似度以避免内存不足 :param queries: 查询向量 (n_queries, dim) :param targets: 目标向量 (n_targets, dim) :param batch_size: 每批大小 :param device: 计算设备 :return: 相似度矩阵 (n_queries, n_targets) queries queries.to(device) targets targets.to(device) n_queries queries.shape[0] sim_matrix torch.zeros(n_queries, targets.shape[0], devicedevice) for i in range(0, n_queries, batch_size): batch queries[i:ibatch_size] # 计算当前batch与所有目标的相似度 sim_batch cosine_similarity(batch.unsqueeze(1), targets.unsqueeze(0), dim2) sim_matrix[i:ibatch_size] sim_batch return sim_matrix.cpu() # 测试批量计算 large_ratings torch.randn(1000, 100) # 1000用户×100电影 sim_matrix batch_cosine_similarity(large_ratings[:100], large_ratings) print(f批量相似度矩阵形状: {sim_matrix.shape})4.2 近似最近邻搜索当数据量非常大时精确计算所有点对的距离变得不可行。我们可以使用近似最近邻算法来提高效率。def approximate_knn(query, data, k10, n_probes5, hash_size8): 使用局部敏感哈希(LSH)进行近似最近邻搜索 :param query: 查询向量 :param data: 数据集 :param k: 返回的最近邻数量 :param n_probes: 探测的哈希桶数量 :param hash_size: 哈希位数 :return: 最近邻的索引和距离 dim data.shape[1] # 生成随机投影向量 projection torch.randn(dim, hash_size, devicedata.device) # 计算哈希签名 signatures torch.sign(data projection) # 将二进制签名转换为整数哈希值 powers torch.arange(hash_size-1, -1, -1, devicedata.device) powers 2 ** powers hash_values (signatures 0).int() powers.unsqueeze(1) # 查询的哈希值 query_sig torch.sign(query projection) query_hash ((query_sig 0).int() powers).item() # 找到相同哈希桶中的候选点 candidates torch.where(hash_values query_hash)[0] # 如果没有足够候选点扩大搜索范围 if len(candidates) n_probes: # 查找哈希值相近的桶 all_hashes hash_values.unique() hash_diff (all_hashes ^ query_hash).abs() closest_hashes all_hashes[torch.topk(hash_diff, n_probes, largestFalse).indices] for h in closest_hashes: candidates torch.cat([candidates, torch.where(hash_values h)[0]]) # 计算候选点的实际距离 if len(candidates) 0: dists torch.cdist(query.unsqueeze(0), data[candidates], p2).squeeze(0) topk_indices torch.topk(dists, min(k, len(dists)), largestFalse).indices return candidates[topk_indices], dists[topk_indices] else: return torch.tensor([]), torch.tensor([]) # 测试近似最近邻 query torch.randn(100) data torch.randn(10000, 100) indices, dists approximate_knn(query, data) print(f找到的最近邻索引: {indices[:5]}, 距离: {dists[:5]})4.3 混合推荐系统结合用户相似度和物品相似度可以构建更强大的混合推荐系统。class HybridRecommender: def __init__(self, ratings, user_weight0.5): self.ratings ratings self.user_weight user_weight # 用户相似度的权重 self.item_weight 1 - user_weight # 物品相似度的权重 # 预计算物品相似度矩阵 self.item_sim self._compute_item_similarity() def _compute_item_similarity(self): 计算物品相似度矩阵 item_user self.ratings.T num_items item_user.shape[0] item_sim torch.zeros(num_items, num_items) for i in range(num_items): for j in range(i, num_items): mask (item_user[i] 0) (item_user[j] 0) if mask.sum() 0: sim cosine_similarity(item_user[i][mask].unsqueeze(0), item_user[j][mask].unsqueeze(0)) item_sim[i,j] item_sim[j,i] sim.item() return item_sim def recommend(self, user_idx, top_k5): 生成混合推荐 # 用户相似度推荐得分 target self.ratings[user_idx].unsqueeze(0) user_sims cosine_similarity(target, self.ratings, dim1) user_sims[user_idx] -1 # 排除自己 _, similar_users torch.topk(user_sims, 3) user_based_scores self.ratings[similar_users].mean(dim0) # 物品相似度推荐得分 rated_movies torch.where(self.ratings[user_idx] 0)[0] item_based_scores torch.zeros(self.ratings.shape[1]) for movie in rated_movies: _, similar_movies torch.topk(self.item_sim[movie], 3) for sim_movie in similar_movies: if self.ratings[user_idx, sim_movie] 0: item_based_scores[sim_movie] self.item_sim[movie, sim_movie] # 合并得分 combined_scores (self.user_weight * user_based_scores self.item_weight * item_based_scores) # 只推荐未评分的 unrated (self.ratings[user_idx] 0) recommended torch.argsort(combined_scores * unrated, descendingTrue) return recommended[:top_k] # 测试混合推荐 hybrid_rec HybridRecommender(ratings, user_weight0.7) recommendations hybrid_rec.recommend(user_idx) print(f混合推荐结果: {recommendations})