推荐系统是互联网平台的核心能力，从协同过滤到深度模型，技术演进持续推动个性化体验升级。本文覆盖经典算法、深度学习推荐模型与工业级推荐架构。
1. 推荐系统架构
用户请求 → 召回层（千级）→ 粗排（百级）→ 精排（十级）→ 重排（展示）
↑ ↑ ↑
多路召回 轻量模型 复杂模型 多样性/规则
CF/向量检索 GBDT/LR DeepFM/DIN 曝光去重
1.1 各阶段职责
| 阶段 | 数量级 | 延迟 | 模型 | 目标 |
|---|---|---|---|---|
| 召回 | 百万→千 | < 50ms | 轻量、多路并行 | 高召回率 |
| 粗排 | 千→百 | < 10ms | 简单模型 | 快速筛选 |
| 精排 | 百→十 | < 50ms | 深度模型 | 精准排序 |
| 重排 | 十→展示 | < 5ms | 规则/控制 | 产品策略 |
2. 协同过滤 (Collaborative Filtering)
2.1 用户协同过滤 (User-CF)
找到与用户兴趣相似的其他用户，推荐他们喜欢的物品。
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
# 用户-物品评分矩阵 (5用户 x 6物品)
ratings = np.array([
[5, 3, 4, 0, 0, 0], # 用户 0
[3, 1, 2, 0, 0, 0], # 用户 1
[4, 3, 4, 0, 0, 0], # 用户 2
[0, 0, 0, 4, 5, 4], # 用户 3
[0, 0, 0, 5, 4, 5], # 用户 4
])
# 计算用户相似度
user_sim = cosine_similarity(ratings)
print("用户相似度矩阵:")
print(user_sim)
# 为用户 0 推荐（基于 Top-K 相似用户）
K = 2
target_user = 0
similar_users = np.argsort(user_sim[target_user])[-(K+1):-1][::-1]
# 相似用户评分过但目标用户未评分的物品
unrated = np.where(ratings[target_user] == 0)[0]
predictions = []
for item in unrated:
score = 0
weights = 0
for u in similar_users:
if ratings[u, item] > 0:
score += user_sim[target_user, u] * ratings[u, item]
weights += abs(user_sim[target_user, u])
if weights > 0:
predictions.append((item, score / weights))
predictions.sort(key=lambda x: x[1], reverse=True)
print(f"为用户 {target_user} 推荐: {predictions}")
2.2 物品协同过滤 (Item-CF)
基于物品间的相似度推荐。计算量更小，更适合物品数远少于用户的场景（如电商）。
# 物品相似度
item_sim = cosine_similarity(ratings.T)
# 为用户推荐（基于已有物品的相似物品）
target_user = 0
rated_items = np.where(ratings[target_user] > 0)[0]
unrated_items = np.where(ratings[target_user] == 0)[0]
scores = {}
for unrated_item in unrated_items:
score = 0
for rated_item in rated_items:
score += item_sim[unrated_item, rated_item] * ratings[target_user, rated_item]
scores[unrated_item] = score
recommendations = sorted(scores.items(), key=lambda x: x[1], reverse=True)
User-CF vs Item-CF：
- User-CF：发现社交圈兴趣，适合新闻/音乐（口味多变）
- Item-CF：发现相似物品，适合电商/电影（物品属性稳定）
2.3 协同过滤的问题
| 问题 | 原因 | 解决 |
|---|---|---|
| 冷启动 | 新用户/新物品无历史 | 内容推荐、热门兜底 |
| 稀疏性 | 用户-物品矩阵极稀疏 | 矩阵分解、隐向量 |
| 可扩展性 | 用户/物品量大 | 近似最近邻 (ANN) |
| 头部效应 | 热门物品过度推荐 | 长尾挖掘、探索策略 |
3. 矩阵分解
3.1 SVD / FunkSVD
将用户-物品评分矩阵分解为用户隐向量和物品隐向量：
$$R \approx P \times Q^T = \hat{R}$$
$$\min_{P,Q} \sum_{(u,i) \in K} (r_{ui} - p_u^T q_i)^2 + \lambda(|p_u|^2 + |q_i|^2)$$
from sklearn.decomposition import TruncatedSVD
# SVD 降维
svd = TruncatedSVD(n_components=2)
user_factors = svd.fit_transform(ratings) # 用户隐向量
item_factors = svd.components_.T # 物品隐向量
# 预测评分
predicted = user_factors @ item_factors.T
print(predicted)
3.2 交替最小二乘 (ALS)
from implicit.als import AlternatingLeastSquares
# 隐式反馈 (点击/购买，无显式评分)
als = AlternatingLeastSquares(
factors=50, # 隐向量维度
regularization=0.01,
iterations=20,
use_cg=True
)
# 训练 (用户-物品交互矩阵)
user_items = ratings.T # 转置为 (items, users)
als.fit(user_items)
# 为用户推荐
user_id = 0
recommendations = als.recommend(user_id, user_items[user_id], N=10)
4. 因子分解机 (FM/FFM)
4.1 FM (Factorization Machines)
自动学习特征间二阶交叉，解决稀疏数据下的特征组合问题。
$$\hat{y}(x) = w_0 + \sum_{i=1}^{n}w_i x_i + \sum_{i=1}^{n}\sum_{j=i+1}^{n}x_i x_j \sum_{f=1}^{k}v_{i,f} v_{j,f}$$
二阶项复杂度从 $O(n^2k)$ 降至 $O(nk)$：
$$\sum_{i=1}^{n}\sum_{j=i+1}^{n}x_i x_j \sum_{f=1}^{k}v_{i,f} v_{j,f} = \frac{1}{2}\sum_{f=1}^{k}\left[\left(\sum_{i=1}^{n}v_{i,f}x_i\right)^2 - \sum_{i=1}^{n}v_{i,f}^2 x_i^2\right]$$
import torch.nn as nn
class FactorizationMachine(nn.Module):
def __init__(self, num_features, embed_dim):
super().__init__()
self.linear = nn.Linear(num_features, 1)
self.embedding = nn.Embedding(num_features, embed_dim)
def forward(self, x):
# x: (batch, num_features)
linear_part = self.linear(x)
# 二阶交叉
embed = self.embedding.weight # (num_features, embed_dim)
square_of_sum = torch.pow(torch.matmul(x, embed), 2)
sum_of_square = torch.matmul(torch.pow(x, 2), torch.pow(embed, 2))
fm = 0.5 * torch.sum(square_of_sum - sum_of_square, dim=1, keepdim=True)
return linear_part + fm
4.2 FFM (Field-aware FM)
每个特征对每个 field 学习独立的隐向量，表达能力更强但参数量更大。
5. 深度学习推荐模型
5.1 Wide & Deep (Google, 2016)
import torch.nn as nn
class WideAndDeep(nn.Module):
def __init__(self, num_features, embed_dim, hidden_dims):
super().__init__()
# Wide: 线性部分（记忆能力）
self.wide = nn.Linear(num_features, 1)
# Deep: DNN 部分（泛化能力）
self.embeddings = nn.ModuleList([
nn.Embedding(vocab_size, embed_dim)
for vocab_size in feature_dims
])
layers = []
input_dim = len(feature_dims) * embed_dim
for h_dim in hidden_dims:
layers.extend([
nn.Linear(input_dim, h_dim),
nn.ReLU(),
nn.Dropout(0.2)
])
input_dim = h_dim
layers.append(nn.Linear(input_dim, 1))
self.deep = nn.Sequential(*layers)
def forward(self, dense_features, sparse_features):
# Wide
wide_out = self.wide(dense_features)
# Deep: 稀疏特征嵌入后拼接
embeds = []
for i, emb in enumerate(self.embeddings):
embeds.append(emb(sparse_features[:, i]))
deep_input = torch.cat(embeds, dim=1)
deep_out = self.deep(deep_input)
return torch.sigmoid(wide_out + deep_out)
Wide 组件：记忆历史规则（用户买过 A 也会买 B）
Deep 组件：泛化新组合（与 A 类似的物品）
5.2 DeepFM
将 Wide 组件替换为 FM，同时学习低阶和高阶特征交互。
class DeepFM(nn.Module):
def __init__(self, feature_dims, embed_dim, hidden_dims):
super().__init__()
self.field_size = len(feature_dims)
self.embed_dim = embed_dim
# FM 部分
self.fm_first = nn.Embedding(sum(feature_dims), 1)
self.fm_second = nn.Embedding(sum(feature_dims), embed_dim)
# Deep 部分
self.offsets = torch.cumsum(torch.tensor([0] + feature_dims[:-1]), dim=0)
deep_input_dim = self.field_size * embed_dim
layers = []
for h_dim in hidden_dims:
layers.extend([nn.Linear(deep_input_dim, h_dim), nn.ReLU(), nn.Dropout(0.3)])
deep_input_dim = h_dim
layers.append(nn.Linear(deep_input_dim, 1))
self.deep = nn.Sequential(*layers)
def forward(self, features):
# features: (batch, field_size)，每个值是全局索引
features = features + self.offsets.to(features.device)
# FM 一阶
fm_first = torch.sum(self.fm_first(features).squeeze(-1), dim=1, keepdim=True)
# FM 二阶
fm_second = self.fm_second(features) # (batch, field_size, embed_dim)
square_of_sum = torch.pow(torch.sum(fm_second, dim=1), 2)
sum_of_square = torch.sum(torch.pow(fm_second, 2), dim=1)
fm_second_order = 0.5 * torch.sum(square_of_sum - sum_of_square, dim=1, keepdim=True)
# Deep 部分
deep_input = fm_second.view(features.size(0), -1)
deep_out = self.deep(deep_input)
return torch.sigmoid(fm_first + fm_second_order + deep_out)
5.3 DIN：深度兴趣网络 (Alibaba, 2018)
引入注意力机制，对用户历史行为按候选物品进行加权。
class DINAttention(nn.Module):
def __init__(self, embed_dim):
super().__init__()
self.fc = nn.Sequential(
nn.Linear(embed_dim * 4, 200),
nn.ReLU(),
nn.Linear(200, 80),
nn.ReLU(),
nn.Linear(80, 1)
)
def forward(self, query, keys, keys_length):
# query: (batch, embed_dim) 候选物品
# keys: (batch, max_len, embed_dim) 用户历史
# keys_length: (batch) 实际历史长度
batch_size, max_len, embed_dim = keys.shape
# 扩展 query 匹配 keys 长度
queries = query.unsqueeze(1).expand(-1, max_len, -1)
# 输入特征：query, key, query-key, query*key
din_input = torch.cat([
queries,
keys,
queries - keys,
queries * keys
], dim=-1)
# 计算注意力权重
attention = self.fc(din_input.view(-1, embed_dim * 4))
attention = attention.view(batch_size, max_len)
# Mask padding
mask = torch.arange(max_len).to(keys.device).unsqueeze(0) < keys_length.unsqueeze(1)
attention = attention.masked_fill(~mask, float('-inf'))
attention = F.softmax(attention, dim=1)
# 加权求和
output = torch.bmm(attention.unsqueeze(1), keys).squeeze(1)
return output
6. 向量召回
6.1 双塔模型
class TwoTowerModel(nn.Module):
def __init__(self, user_features, item_features, embed_dim):
super().__init__()
self.user_tower = nn.Sequential(
nn.Linear(user_features, 256),
nn.ReLU(),
nn.Linear(256, embed_dim)
)
self.item_tower = nn.Sequential(
nn.Linear(item_features, 256),
nn.ReLU(),
nn.Linear(256, embed_dim)
)
def forward(self, user_features, item_features):
user_vec = F.normalize(self.user_tower(user_features), dim=1)
item_vec = F.normalize(self.item_tower(item_features), dim=1)
score = torch.sum(user_vec * item_vec, dim=1) # 余弦相似度
return score
def get_user_embedding(self, user_features):
return F.normalize(self.user_tower(user_features), dim=1)
def get_item_embedding(self, item_features):
return F.normalize(self.item_tower(item_features), dim=1)
6.2 向量检索 (ANN)
import faiss
# 构建 FAISS 索引
item_embeddings = model.get_item_embedding(all_item_features).detach().cpu().numpy()
dimension = item_embeddings.shape[1]
# IVF 索引（适合百万级）
quantizer = faiss.IndexFlatIP(dimension) # 内积 = 余弦相似度（归一化后）
index = faiss.IndexIVFFlat(quantizer, dimension, nlist=100)
index.train(item_embeddings)
index.add(item_embeddings)
# 查询
user_vec = model.get_user_embedding(user_feature).detach().cpu().numpy()
D, I = index.search(user_vec, k=100) # 召回 Top-100
# HNSW（适合千万级以上）
index = faiss.IndexHNSWFlat(dimension, 32)
index.hnsw.efConstruction = 200
index.add(item_embeddings)
index.hnsw.efSearch = 128
7. 特征工程
7.1 用户/物品/上下文特征
user_features = {
'user_id': 12345,
'age': 28,
'gender': 'M',
'city_level': 1,
'device_type': 'iOS',
'history_click': ['item_1', 'item_3', 'item_7'],
'history_category': ['电子产品', '图书', '服装'],
'avg_price': 150.0,
'purchase_freq': 3.5
}
item_features = {
'item_id': 'item_999',
'category': '电子产品',
'brand': 'Apple',
'price': 8999.0,
'ctr_7d': 0.05,
'ctr_30d': 0.03,
'score': 4.8,
'stock': 100
}
context_features = {
'hour': 14,
'day_of_week': 6,
'is_weekend': True,
'is_holiday': False,
'position': 3
}
7.2 交叉特征
# 统计类交叉特征
user_category_click = user.click_count_by_category[category]
user_brand_click = user.click_count_by_brand[brand]
# 比率特征
user_price_preference = user.avg_order_price / item.price # 价格偏好
user_category_ctr = user_category_click / user.total_click
8. 评估指标
| 指标 | 定义 | 适用 |
|---|---|---|
| HR@K (Hit Ratio) | Top-K 中有真实交互的比例 | 评估召回 |
| NDCG@K | 归一化折损累计增益 | 评估排序 |
| MAP@K | 平均精度均值 | 评估排序 |
| AUC | 正负样本排序正确率 | 评估模型 |
| LogLoss | 对数损失 | 训练损失 |
| Coverage | 推荐物品覆盖全部物品的比例 | 多样性 |
| Diversity | 推荐列表中物品的不相似度 | 多样性 |
| Novelty | 推荐物品的平均流行度倒数 | 新颖性 |
from sklearn.metrics import roc_auc_score, log_loss
# AUC
auc = roc_auc_score(y_true, y_pred)
# NDCG
def ndcg_at_k(y_true, y_pred, k=10):
order = np.argsort(y_pred)[::-1][:k]
y_true_sorted = np.take(y_true, order)
dcg = np.sum(y_true_sorted / np.log2(np.arange(2, k+2)))
ideal = np.sum(np.sort(y_true)[::-1][:k] / np.log2(np.arange(2, k+2)))
return dcg / ideal if ideal > 0 else 0
9. 工业级实践
9.1 多路召回
class MultiChannelRecall:
def __init__(self):
self.channels = {
'cf_user': UserCFRecall(topk=100),
'cf_item': ItemCFRecall(topk=100),
'embedding': EmbeddingRecall(topk=100),
'hot': HotRecall(topk=50),
'new': NewItemRecall(topk=30)
}
def recall(self, user_id):
results = []
for name, channel in self.channels.items():
items = channel.get_candidates(user_id)
for item_id, score in items:
results.append((item_id, score, name))
# 去重 + 合并
seen = set()
final = []
for item_id, score, channel in results:
if item_id not in seen:
seen.add(item_id)
final.append((item_id, score, channel))
return final
9.2 实时特征流
# Flink 实时特征更新
class RealtimeFeature:
def on_click(self, user_id, item_id):
# 更新用户实时点击序列
redis.lpush(f"user:{user_id}:clicks", item_id)
redis.ltrim(f"user:{user_id}:clicks", 0, 99) # 保留最近 100
# 更新物品实时 CTR
redis.hincrby(f"item:{item_id}:stats", "clicks", 1)
总结
| 模型/方法 | 核心思想 | 适用场景 |
|---|---|---|
| User/Item CF | 相似用户/物品的协同 | 快速基线、解释性强 |
| 矩阵分解 | 隐向量学习 | 稀疏数据处理 |
| FM | 自动二阶交叉 | 特征组合 |
| Wide & Deep | 记忆 + 泛化 | 通用推荐 |
| DeepFM | FM + DNN | CTR 预估 |
| DIN | 注意力兴趣 | 序列行为丰富的场景 |
| 双塔模型 | 向量召回 | 大规模实时召回 |
推荐系统要点：
- 召回优先保证相关性，精排追求点击率/转化率最大化
- 冷启动用内容相似（物品维度）+ 热门兜底（用户维度）
- 数据量决定模型复杂度，小数据用 FM，大数据用 DeepFM/DIN
- 实时性对 CTR 影响大，用户行为特征需秒级更新
- 多目标优化：点击、收藏、加购、购买联合优化
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。