推荐系统实战¶
概述¶
推荐系统是AI最成功的商业应用之一，本教程从零构建推荐系统，涵盖协同过滤、深度学习等方法。
推荐系统类型¶
# 主流推荐算法
1. 基于内容（Content-based）
- 推荐相似物品
- 适合冷启动
2. 协同过滤（Collaborative Filtering）
- 用户-物品交互
- 用户协同过滤
- 物品协同过滤
3. 深度学习
- 神经网络推荐
- 序列推荐
4. 混合推荐
- 结合多种方法
协同过滤¶
基于用户的协同过滤¶
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class UserCF:
def __init__(self, n_neighbors=10):
self.n_neighbors = n_neighbors
def fit(self, ratings_matrix):
"""ratings_matrix: 用户-物品评分矩阵"""
self.ratings = ratings_matrix
self.user_similarity = cosine_similarity(ratings_matrix)
def predict(self, user_id, item_id):
# 找到相似用户
sim_scores = self.user_similarity[user_id]
top_users = np.argsort(sim_scores)[-self.n_neighbors:]
# 加权平均
weighted_sum = 0
sim_sum = 0
for u in top_users:
if u != user_id and self.ratings[u, item_id] > 0:
weighted_sum += sim_scores[u] * self.ratings[u, item_id]
sim_sum += abs(sim_scores[u])
if sim_sum == 0:
return np.mean(self.ratings[user_id][self.ratings[user_id] > 0])
return weighted_sum / sim_sum
def recommend(self, user_id, n_items=10):
# 预测未评分物品
unrated = np.where(self.ratings[user_id] == 0)[0]
predictions = [(item, self.predict(user_id, item)) for item in unrated]
predictions.sort(key=lambda x: x[1], reverse=True)
return predictions[:n_items]
基于物品的协同过滤¶
class ItemCF:
def __init__(self, n_neighbors=10):
self.n_neighbors = n_neighbors
def fit(self, ratings_matrix):
self.ratings = ratings_matrix
self.item_similarity = cosine_similarity(ratings_matrix.T)
def predict(self, user_id, item_id):
# 用户评过的物品
rated_items = np.where(self.ratings[user_id] > 0)[0]
weighted_sum = 0
sim_sum = 0
for i in rated_items:
sim = self.item_similarity[item_id, i]
weighted_sum += sim * self.ratings[user_id, i]
sim_sum += abs(sim)
if sim_sum == 0:
return 0
return weighted_sum / sim_sum
矩阵分解（SVD）¶
class MatrixFactorization:
def __init__(self, n_factors=50, lr=0.01, reg=0.02, epochs=100):
self.n_factors = n_factors
self.lr = lr
self.reg = reg
self.epochs = epochs
def fit(self, ratings_matrix):
n_users, n_items = ratings_matrix.shape
# 初始化用户和物品矩阵
self.P = np.random.normal(0, 0.1, (n_users, self.n_factors))
self.Q = np.random.normal(0, 0.1, (n_items, self.n_factors))
# 记录非零位置
self.samples = [(i, j, ratings_matrix[i, j])
for i in range(n_users)
for j in range(n_items)
if ratings_matrix[i, j] > 0]
# 训练
for _ in range(self.epochs):
np.random.shuffle(self.samples)
total_loss = 0
for user, item, rating in self.samples:
# 预测
pred = self.P[user] @ self.Q[item]
error = rating - pred
# 更新
self.P[user] += self.lr * (error * self.Q[item] - self.reg * self.P[user])
self.Q[item] += self.lr * (error * self.P[user] - self.reg * self.Q[item])
total_loss += error ** 2
def predict(self, user_id, item_id):
return self.P[user_id] @ self.Q[item_id]
def recommend(self, user_id, n_items=10):
scores = self.P[user_id] @ self.Q.T
top_items = np.argsort(scores)[-n_items:][::-1]
return [(item, scores[item]) for item in top_items]
深度学习推荐¶
NCF（神经协同过滤）¶
import torch
import torch.nn as nn
class NCF(nn.Module):
def __init__(self, n_users, n_items, embed_dim=64, hidden_dims=[128, 64, 32]):
super().__init__()
# 嵌入层
self.user_embed = nn.Embedding(n_users, embed_dim)
self.item_embed = nn.Embedding(n_items, embed_dim)
# MLP层
mlp_layers = []
input_dim = embed_dim * 2
for dim in hidden_dims:
mlp_layers.extend([
nn.Linear(input_dim, dim),
nn.ReLU(),
nn.Dropout(0.2)
])
input_dim = dim
self.mlp = nn.Sequential(*mlp_layers)
# 输出层
self.output = nn.Linear(hidden_dims[-1], 1)
def forward(self, user_ids, item_ids):
user_embed = self.user_embed(user_ids)
item_embed = self.item_embed(item_ids)
# 拼接
x = torch.cat([user_embed, item_embed], dim=1)
# MLP
x = self.mlp(x)
# 输出
return torch.sigmoid(self.output(x)).squeeze()
双塔模型¶
class TwoTower(nn.Module):
def __init__(self, n_users, n_items, embed_dim=64):
super().__init__()
# 用户塔
self.user_tower = nn.Sequential(
nn.Embedding(n_users, embed_dim),
nn.Linear(embed_dim, 128),
nn.ReLU(),
nn.Linear(128, 64)
)
# 物品塔
self.item_tower = nn.Sequential(
nn.Embedding(n_items, embed_dim),
nn.Linear(embed_dim, 128),
nn.ReLU(),
nn.Linear(128, 64)
)
def forward(self, user_ids, item_ids):
user_vec = self.user_tower(user_ids)
item_vec = self.item_tower(item_ids)
# 内积作为相似度
return (user_vec * item_vec).sum(dim=1)
def get_user_embedding(self, user_ids):
return self.user_tower(user_ids)
def get_item_embedding(self, item_ids):
return self.item_tower(item_ids)
实战：电影推荐系统¶
数据准备¶
import pandas as pd
from sklearn.model_selection import train_test_split
# 加载MovieLens数据
ratings = pd.read_csv('ratings.csv')
movies = pd.read_csv('movies.csv')
# 构建用户-物品矩阵
n_users = ratings['userId'].nunique()
n_items = ratings['movieId'].nunique()
# 创建映射
user_map = {id: i for i, id in enumerate(ratings['userId'].unique())}
item_map = {id: i for i, id in enumerate(ratings['movieId'].unique())}
# 训练测试集
train_data, test_data = train_test_split(ratings, test_size=0.2, random_state=42)
训练模型¶
from torch.utils.data import Dataset, DataLoader
class RatingsDataset(Dataset):
def __init__(self, users, items, ratings):
self.users = torch.LongTensor(users)
self.items = torch.LongTensor(items)
self.ratings = torch.FloatTensor(ratings)
def __len__(self):
return len(self.ratings)
def __getitem__(self, idx):
return self.users[idx], self.items[idx], self.ratings[idx]
# 创建数据集
train_dataset = RatingsDataset(
train_data['userId'].map(user_map),
train_data['movieId'].map(item_map),
train_data['rating']
)
train_loader = DataLoader(train_dataset, batch_size=256, shuffle=True)
# 训练
model = NCF(n_users, n_items)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()
for epoch in range(10):
model.train()
total_loss = 0
for users, items, ratings in train_loader:
pred = model(users, items)
loss = criterion(pred, ratings)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
print(f"Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.4f}")
评估指标¶
from sklearn.metrics import mean_squared_error, mean_absolute_error
import numpy as np
def evaluate(model, test_data, user_map, item_map):
model.eval()
with torch.no_grad():
users = torch.LongTensor(test_data['userId'].map(user_map))
items = torch.LongTensor(test_data['movieId'].map(item_map))
true_ratings = test_data['rating'].values
pred_ratings = model(users, items).numpy()
rmse = np.sqrt(mean_squared_error(true_ratings, pred_ratings))
mae = mean_absolute_error(true_ratings, pred_ratings)
return rmse, mae
rmse, mae = evaluate(model, test_data, user_map, item_map)
print(f"RMSE: {rmse:.4f}, MAE: {mae:.4f}")