协同过滤算法深入：BPR 与矩阵分解的工程实现

什么是协同过滤#场景你在选电影传统方式基于内容 你喜欢科幻片 → 推荐科幻片 协同过滤 和你口味相似的人喜欢X → 推荐X给你协同的含义利用群体智慧alice 喜欢A, B, C bob 喜欢 A, B, D charlie 喜欢A, C, E 观察 - alice 和 bob 都喜欢 A, B → 口味相似 - 推荐D 给 alicebob 喜欢但 alice 没看过 - 推荐C 给 bobalice 喜欢但 bob 没看过两种协同过滤#User-Based基于用户1. 找到和你相似的用户 2. 看他们喜欢什么 3. 推荐给你 问题 - 用户数量大时计算慢100万用户 → 100万²次比较 - 用户兴趣变化快Item-Based基于物品1. 找到你喜欢物品的相似物品 2. 推荐相似物品 优点 - 物品数量相对稳定 - 可以预计算Matrix Factorization矩阵分解最现代的方法 - 用向量表示用户和物品 - 通过机器学习找到最佳向量 - 这就是我们要讲的重点矩阵分解降维的艺术#问题引入#用户-物品交互矩阵movie1 movie2 movie3 movie4 movie5 alice 1 0 1 0 0 bob 0 1 0 1 0 charlie 1 0 0 0 1 1 喜欢0 未交互 问题 1. 矩阵很稀疏99% 是 0 2. 无法预测未交互的矩阵分解的想法#核心思想用低维向量表示用户和物品原始矩阵100万用户 × 100万物品 1万亿个数 分解后 - 用户向量100万 × 50维 5000万个数 - 物品向量100万 × 50维 5000万个数 - 总计1亿个数 压缩率1万亿 / 1亿 10000 倍数学表示简化版#评分预测 r̂ 用户向量 · 物品向量 具体例子 alice 的向量[0.8, 0.2, 0.5] 3维 movie1 的向量[0.9, 0.1, 0.6] 预测 alice 对 movie1 的评分 r̂ 0.8×0.9 0.2×0.1 0.5×0.6 0.72 0.02 0.3 1.04 归一化后接近 1表示喜欢向量的含义可解释性#假设用 3 维向量表示维度1: 科幻程度 维度2: 动作程度 维度3: 文艺程度 alice 的向量[0.9, 0.2, 0.1] → 非常喜欢科幻不太喜欢动作不喜欢文艺 电影《星际穿越》的向量[0.95, 0.1, 0.3] → 科幻片少量动作有点文艺 预测 alice 对《星际穿越》的评分 0.9×0.95 0.2×0.1 0.1×0.3 0.905很高注意实际中向量维度更高50-200维含义不一定可解释。BPR算法成对学习的智慧#为什么需要 BPR#传统方法的问题问题预测评分1-5星 数据alice 给 movie1 打了 5 星 传统方法 目标 让预测值接近 5 问题 - 推荐系统中大部分是隐式反馈点击、浏览 - 没有明确的评分 - 只知道用户喜欢什么不知道具体多喜欢BPR 的创新不预测绝对评分而是预测相对偏好 数据 - alice 看了 movie1正样本 - alice 没看 movie2负样本 目标 让 score(alice, movie1) score(alice, movie2) 这就是成对学习Pairwise LearningBPR 的核心思想## 伪代码 for each user: positive_item 用户交互过的物品 negative_item 用户没交互过的物品随机采样 score_pos predict(user, positive_item) score_neg predict(user, negative_item) # 目标正样本分数 负样本分数 loss -log(sigmoid(score_pos - score_neg)) # 梯度下降更新参数 update_parameters()为什么用 sigmoidscore_pos - score_neg 的范围(-∞, ∞) sigmoid(x) 1 / (1 e^(-x)) - x 0 时sigmoid(x) → 1正样本分数高好 - x 0 时sigmoid(x) → 0负样本分数高不好 - x 0 时sigmoid(x) 0.5分不清 -log(sigmoid(x)) - x 0 时loss → 0已经很好了 - x 0 时loss → ∞很差需要优化 - x 0 时loss 0.69中等负采样策略#随机负采样func sampleNegative(user User, items []Item, interacted Set) Item { for { idx : rand.Int() % len(items) item : items[idx] if !interacted.Contains(item.ID) { return item // 找到一个未交互的 } } }问题热门物品被采样概率高简单负样本明显不相关学习效果差改进热度采样// 按热度的平方根采样降低热门物品权重 func sampleNegativeByPopularity(items []Item, popularity []int) Item { weights : make([]float64, len(items)) for i, pop : range popularity { weights[i] math.Sqrt(float64(pop)) // 平方根 } return weightedRandomSample(items, weights) }源码剖析Gorse 的 BPR 实现#数据结构#// model/cf/model.go type BPR struct { BaseMatrixFactorization // 超参数 nFactors int // 向量维度默认50 nEpochs int // 训练轮数默认100 lr float32 // 学习率默认0.05 reg float32 // 正则化系数默认0.01 // 模型参数 UserFactor [][]float32 // 用户向量 [n_users × n_factors] ItemFactor [][]float32 // 物品向量 [n_items × n_factors] }初始化#func NewBPR(params Params) *BPR { bpr : BPR{ nFactors: params.GetInt(n_factors, 50), nEpochs: params.GetInt(n_epochs, 100), lr: params.GetFloat(lr, 0.05), reg: params.GetFloat(reg, 0.01), } return bpr } // 初始化向量小随机数 func (bpr *BPR) Init(trainSet dataset.CFSplit) { nUsers : trainSet.CountUsers() nItems : trainSet.CountItems() // 用户向量 bpr.UserFactor make([][]float32, nUsers) for i : range bpr.UserFactor { bpr.UserFactor[i] make([]float32, bpr.nFactors) for j : range bpr.UserFactor[i] { // 小随机数初始化-0.01 到 0.01 bpr.UserFactor[i][j] (rand.Float32() - 0.5) * 0.02 } } // 物品向量同样方式 // ... }为什么用小随机数初始化如果全部初始化为 0 - 所有梯度相同 - 所有参数以相同方式更新 - 无法学到不同的特征 小随机数 - 打破对称性 - 不同特征独立演化核心训练循环#// 源码model/cf/model.go 第 442-487 行简化 func (bpr *BPR) Fit(ctx context.Context, trainSet dataset.CFSplit, config *FitConfig) Score { // 每个 worker 独立的随机数生成器 rng : make([]*rand.Rand, config.Jobs) for i : range rng { rng[i] rand.New(rand.NewSource(time.Now().UnixNano())) } // 训练循环 for epoch : 1; epoch bpr.nEpochs; epoch { // 并行训练 parallel.Parallel(trainSet.CountFeedback(), config.Jobs, func(workerId, _ int) error { // 1. 随机选择一个用户 userIndex : rng[workerId].Int31n(trainSet.CountUsers()) // 2. 选择一个正样本用户交互过的 userFeedback : trainSet.GetUserFeedback()[userIndex] if len(userFeedback) 0 { return nil // 跳过没有反馈的用户 } posIndex : userFeedback[rng[workerId].Intn(len(userFeedback))] // 3. 负采样用户未交互的 negIndex : int32(-1) for { temp : rng[workerId].Int31n(trainSet.CountItems()) if !userFeedback.Contains(temp) { negIndex temp break } } // 4. 计算预测分数 scorePosi : bpr.internalPredict(userIndex, posIndex) scoreNeg : bpr.internalPredict(userIndex, negIndex) // 5. 计算损失和梯度 diff : scorePosi - scoreNeg // loss -log(sigmoid(diff)) // grad sigmoid(-diff) e^(-diff) / (1 e^(-diff)) grad : math32.Exp(-diff) / (1.0 math32.Exp(-diff)) // 6. 梯度更新 bpr.updateGradient(userIndex, posIndex, negIndex, grad) return nil }) } }预测函数#func (bpr *BPR) internalPredict(userIndex, itemIndex int32) float32 { if userIndex 0 || itemIndex 0 { return 0 } // 向量点积 score : float32(0) for f : 0; f bpr.nFactors; f { score bpr.UserFactor[userIndex][f] * bpr.ItemFactor[itemIndex][f] } return score }优化版向量化func (bpr *BPR) internalPredict(userIndex, itemIndex int32) float32 { // 使用 SIMD 加速的点积 return floats.Dot( bpr.UserFactor[userIndex], bpr.ItemFactor[itemIndex], ) } // 性能对比 // 普通循环100 ns // SIMD 优化10 ns // 快 10 倍梯度更新#func (bpr *BPR) updateGradient(userIndex, posIndex, negIndex int32, grad float32) { // 更新正样本物品向量 for f : 0; f bpr.nFactors; f { // 梯度grad × 用户向量 - reg × 物品向量 delta : grad * bpr.UserFactor[userIndex][f] - bpr.reg * bpr.ItemFactor[posIndex][f] // 更新向量 学习率 × 梯度 bpr.ItemFactor[posIndex][f] bpr.lr * delta } // 更新负样本物品向量符号相反 for f : 0; f bpr.nFactors; f { delta : -grad * bpr.UserFactor[userIndex][f] - bpr.reg * bpr.ItemFactor[negIndex][f] bpr.ItemFactor[negIndex][f] bpr.lr * delta } // 更新用户向量 for f : 0; f bpr.nFactors; f { delta : grad * (bpr.ItemFactor[posIndex][f] - bpr.ItemFactor[negIndex][f]) - bpr.reg * bpr.UserFactor[userIndex][f] bpr.UserFactor[userIndex][f] bpr.lr * delta } }为什么要减去正则化项没有正则化 - 参数可能变得很大 - 模型过拟合训练数据 - 泛化能力差 加入正则化L2 - 惩罚过大的参数 - 参数保持在合理范围 - 提高泛化能力 loss -log(sigmoid(diff)) λ × (||user||² ||item||²) ↑ 正则化项训练优化技巧#技巧1学习率调度#// 初始学习率0.05 // 随训练进行逐渐降低 func (bpr *BPR) getLearningRate(epoch int) float32 { // 方案1线性衰减 return bpr.lr * (1.0 - float32(epoch)/float32(bpr.nEpochs)) // 方案2指数衰减 return bpr.lr * math32.Pow(0.95, float32(epoch)) // 方案3余弦退火 return bpr.lr * 0.5 * (1 math32.Cos(float32(epoch) * math32.Pi / float32(bpr.nEpochs))) }效果对比固定学习率 - NDCG10 0.32 线性衰减 - NDCG10 0.359% 余弦退火 - NDCG10 0.3716%技巧2Early Stopping#type EarlyStopping struct { patience int // 容忍的 epoch 数 bestScore float32 counter int shouldStop bool } func (es *EarlyStopping) Update(score float32) { if score es.bestScore { es.bestScore score es.counter 0 // 重置计数器 } else { es.counter if es.counter es.patience { es.shouldStop true } } } // 使用 earlyStopping : EarlyStopping{patience: 10} for epoch : 1; epoch maxEpochs; epoch { // 训练... score : evaluate(model, testSet) earlyStopping.Update(score) if earlyStopping.shouldStop { break // 提前停止 } }好处不使用 Early Stopping - 训练 100 epoch - 时间10 分钟 - 最佳 epoch: 60 使用 Early Stoppingpatience10 - 训练 70 epoch 后停止 - 时间7 分钟 - 节省 30% 时间技巧3批量负采样#// ❌ 慢方式每次采样一个 for i : 0; i nSamples; i { negative : sampleNegative() train(positive, negative) } // ✅ 快方式批量采样 negatives : sampleNegativeBatch(batchSize) for i : 0; i batchSize; i { train(positive, negatives[i]) } // 性能提升 // 慢方式1000 采样/秒 // 快方式10000 采样/秒 // 快 10 倍技巧4向量归一化#// 训练后归一化向量 func (bpr *BPR) Normalize() { for i : range bpr.UserFactor { norm : float32(0) for j : range bpr.UserFactor[i] { norm bpr.UserFactor[i][j] * bpr.UserFactor[i][j] } norm math32.Sqrt(norm) for j : range bpr.UserFactor[i] { bpr.UserFactor[i][j] / norm } } // 物品向量同样处理 }效果不归一化 - 向量长度不一致 - 热门物品向量很大 - 影响推荐公平性 归一化后 - 向量长度为 1 - 只比较方向 - 提高推荐多样性实战手写简化版 BPR#package main import ( fmt math math/rand ) type SimpleBPR struct { nFactors int nEpochs int lr float64 reg float64 UserFactor [][]float64 ItemFactor [][]float64 } func NewSimpleBPR(nUsers, nItems, nFactors int) *SimpleBPR { bpr : SimpleBPR{ nFactors: nFactors, nEpochs: 100, lr: 0.05, reg: 0.01, } // 初始化 bpr.UserFactor make([][]float64, nUsers) for i : range bpr.UserFactor { bpr.UserFactor[i] make([]float64, nFactors) for j : range bpr.UserFactor[i] { bpr.UserFactor[i][j] (rand.Float64() - 0.5) * 0.02 } } bpr.ItemFactor make([][]float64, nItems) for i : range bpr.ItemFactor { bpr.ItemFactor[i] make([]float64, nFactors) for j : range bpr.ItemFactor[i] { bpr.ItemFactor[i][j] (rand.Float64() - 0.5) * 0.02 } } return bpr } func (bpr *SimpleBPR) Predict(userIdx, itemIdx int) float64 { score : 0.0 for f : 0; f bpr.nFactors; f { score bpr.UserFactor[userIdx][f] * bpr.ItemFactor[itemIdx][f] } return score } func (bpr *SimpleBPR) Fit(userItems [][]int) { nUsers : len(userItems) nItems : len(bpr.ItemFactor) for epoch : 0; epoch bpr.nEpochs; epoch { totalLoss : 0.0 // 遍历所有用户 for userIdx : 0; userIdx nUsers; userIdx { if len(userItems[userIdx]) 0 { continue } // 正样本 posIdx : userItems[userIdx][rand.Intn(len(userItems[userIdx]))] // 负采样 negIdx : rand.Intn(nItems) for contains(userItems[userIdx], negIdx) { negIdx rand.Intn(nItems) }