电商推荐系统中的协同过滤算法：从矩阵分解到实时计算的演进

发布时间：2026/7/21 1:08:16

电商推荐系统中的协同过滤算法：从矩阵分解到实时计算的演进

一、从"买了这个的人还买了"到"你应该会喜欢这个"

电商推荐系统经历了三十年的演进。最早的形式是简单的关联规则——"买了 A 的人也买了 B"，本质上是一个频次统计。这个阶段的推荐是粗粒度的：不考虑用户偏好差异，所有看到商品 A 的用户都会看到同一批推荐结果。后来出现了协同过滤（Collaborative Filtering），核心思想是"找到和你品味相似的用户，把他们喜欢的推荐给你"或"找到和你喜欢过的商品相似的商品"。这个阶段的推荐开始有了个性化。

但协同过滤有个致命的问题：当用户和商品数量都达到千万级别时，用户-商品的交互矩阵变得极其稀疏（绝大多数用户只和极少商品有交互），矩阵计算的复杂度也超出了单机的处理能力。于是矩阵分解应运而生，用降维的方式把稀疏的大矩阵压缩成两个稠密的小矩阵——用户隐向量矩阵和商品隐向量矩阵。

二、协同过滤的数学本质与矩阵分解

协同过滤的基本思路可以用一个用户-商品评分矩阵 R 来描述。R 是 m×n 的矩阵，m 是用户数，n 是商品数。绝大多数元素是空的（用户没对商品评过分）。协同过滤的目标就是填充这些空值。

矩阵分解把 R 拆成两个矩阵的乘积：R ≈ P × Q^T，其中 P 是 m×k 的用户隐向量矩阵，Q 是 n×k 的商品隐向量矩阵。k 是隐向量的维度（通常取 50~200）。这个分解的核心直觉是：k 个"隐藏因子"（如价格敏感度、品牌偏好、品类偏好）足以描述用户的偏好和商品的属性。通过这两个小矩阵的乘积，用户 u 对商品 i 的评分可以用 u 的第 k 维向量和 i 的第 k 维向量做点积来近似。

矩阵分解通过梯度下降来优化。损失函数是已知评分和预测评分之间的均方误差，加上防止过拟合的正则化项。每次迭代需要遍历所有已知评分，在大数据集上需要分布式计算框架（如 Spark MLlib）。

""" 协同过滤矩阵分解的最小实现 目标：将 m×n 的评分矩阵 R 分解为 P(m×k) × Q(n×k)^T k 是隐向量维度： - k 太小 → 模型容量不足，无法充分表达用户偏好 - k 太大 → 过拟合，且计算量和存储量都增加 - 经验值：对于百万用户、十万商品的数据，k=50~100 通常足够 """ import numpy as np class MatrixFactorization: def __init__(self, k=50, lr=0.01, reg=0.02, epochs=20): self.k = k # 隐向量维度 self.lr = lr # 学习率 self.reg = reg # 正则化系数（L2） self.epochs = epochs def fit(self, ratings): """ ratings: list of (user_id, item_id, rating) """ # 统计用户和商品数量 self.n_users = max(r[0] for r in ratings) + 1 self.n_items = max(r[1] for r in ratings) + 1 # 随机初始化用户和商品的隐向量 # 用小的随机值初始化，避免对称性问题 # sqrt(1/k) 是 Xavier 初始化的简化版 self.P = np.random.normal(0, np.sqrt(1/self.k), (self.n_users, self.k)) self.Q = np.random.normal(0, np.sqrt(1/self.k), (self.n_items, self.k)) for epoch in range(self.epochs): total_loss = 0 # SGD：每条评分记录作为一个训练样本 # 为什么不用 Mini-batch？因为评分数据太稀疏， # 每个 batch 内样本独立性强，SGD 收敛更快 np.random.shuffle(ratings) # 打散避免有序偏差 for u, i, r in ratings: # 预测评分：用户向量 × 商品向量的点积 pred = np.dot(self.P[u], self.Q[i]) # 误差 error = r - pred total_loss += error ** 2 # 梯度下降更新（含 L2 正则化） # 正则化项防止向量值过大导致过拟合 pu = self.P[u].copy() # 保存旧值供 Q 更新用 self.P[u] += self.lr * (error * self.Q[i] - self.reg * self.P[u]) self.Q[i] += self.lr * (error * pu - self.reg * self.Q[i]) if epoch % 5 == 0: print(f"Epoch {epoch}, Loss: {total_loss:.2f}") def predict(self, user_id, item_id): """预测用户对商品的评分""" if user_id < self.n_users and item_id < self.n_items: return np.dot(self.P[user_id], self.Q[item_id]) return 0