短视频推荐算法揭秘：双塔模型、用户行为建模与冷启动策略实战

# 短视频推荐算法揭秘：双塔模型、用户行为建模与冷启动策略实战

## 摘要

抖音、快手、视频号的推荐算法决定了内容的生死。本文揭秘工业级推荐系统的核心技术，附带PyTorch实现双塔模型代码。

## 一、工业推荐系统架构

```

用户请求

↓

┌────────────────┐

│ 召回层（Recall） │ ← 从百万候选中选千级

│ 双塔模型 / 多路召回 │

└────────┬───────────────┘

↓ 千级候选

┌────────────────┐

│ 粗排层（Pre-Ranking） │ ← 快速打分，筛选百级

│ 轻量模型（LR/小DNN） │

└────────┬───────────────┘

↓ 百级候选

┌────────────────┐

│ 精排层（Ranking） │ ← 多目标打分，选几十条

│ 多目标DNN（点击/点赞/完播）│

└────────┬───────────────┘

↓ 几十条

┌────────────────┐

│ 重排层（Re-Ranking） │ ← 多样性、去重、策略调整

│ MMR / DPP / 业务规则 │

└────────┬───────────────┘

↓ 几至几十条

推送给用户

```

## 二、双塔模型（Two-Tower Model）

### 2.1 原理

双塔模型将用户和物品分别编码为向量，通过向量相似度（内积/余弦）预测匹配度：

```

用户塔（User Tower） 物品塔（Item Tower）

输入：用户特征 ─→ DNN ─→ 用户向量u ∈ R^d

输入：物品特征 ─→ DNN ─→ 物品向量v ∈ R^d

预测分数 = u · v / (||u|| ||v||) [余弦相似度]

```

**优势**：物品向量可预计算，线上只需计算用户向量与候选物品向量的内积。

### 2.2 PyTorch实现

```python

import torch

import torch.nn as nn

import torch.nn.functional as F

class TwoTowerModel(nn.Module):

def __init__(self, user_feat_dim: int, item_feat_dim: int, hidden_dims: list[int], embed_dim: int):

super().__init__()

# 用户塔

user_layers = []

prev_dim = user_feat_dim

for h in hidden_dims:

user_layers.extend([nn.Linear(prev_dim, h), nn.BatchNorm1d(h), nn.ReLU()])

prev_dim = h

user_layers.append(nn.Linear(prev_dim, embed_dim))

self.user_tower = nn.Sequential(*user_layers)

# 物品塔

item_layers = []

prev_dim = item_feat_dim

for h in hidden_dims:

item_layers.extend([nn.Linear(prev_dim, h), nn.BatchNorm1d(h), nn.ReLU()])

prev_dim = h

item_layers.append(nn.Linear(prev_dim, embed_dim))

self.item_tower = nn.Sequential(*item_layers)

def forward(self, user_feats: torch.Tensor, item_feats: torch.Tensor):

user_emb = self.user_tower(user_feats)

item_emb = self.item_tower(item_feats)

# L2归一化

user_emb = F.normalize(user_emb, dim=-1)

item_emb = F.normalize(item_emb, dim=-1)

# 余弦相似度

logits = torch.sum(user_emb * item_emb, dim=-1)

return logits

def encode_items(self, item_feats: torch.Tensor) -> torch.Tensor:

"""预计算物品向量（离线）"""

with torch.no_grad():

item_emb = self.item_tower(item_feats)

item_emb = F.normalize(item_emb, dim=-1)

return item_emb

def encode_user(self, user_feats: torch.Tensor) -> torch.Tensor:

"""计算用户向量（在线）"""

with torch.no_grad():

user_emb = self.user_tower(user_feats)

user_emb = F.normalize(user_emb, dim=-1)

return user_emb

```

### 2.3 负采样策略

```python

def negative_sampling(batch_users, batch_pos_items, all_items, num_neg=4):

"""每个正样本配4个负样本"""

batch_size = batch_users.size(0)

neg_items = []

for i in range(batch_size):

# 简单负采样：随机采样非正样本

neg = torch.randint(0, len(all_items), (num_neg,))

while neg == batch_pos_items[i]: # 避免采样到正样本

neg = torch.randint(0, len(all_items), (num_neg,))

neg_items.append(neg)

return torch.stack(neg_items, dim=0)

```

## 三、用户行为序列建模

### 3.1 SIM（Sparse Interest Model）

短视频场景用户行为序列极长（可能数千步），SIM通过两阶段处理：

```

阶段1（离线）：长序列兴趣提取

用户行为序列（最长5000步）

→ 用搜索单元（Search Unit）提取与目标物品相关的历史行为

→ 输出：百级浓缩行为序列

阶段2（在线）：短期+长期兴趣融合

短期兴趣：最近50步行为 → GRU编码

长期兴趣：SIM提取的百级行为 → Attention编码

→ 拼接 → 多目标预测

```

### 3.2 MIMN（Multi-Channel User Interest Memory Network）

```python

class MIMN(nn.Module):

"""多通道用户兴趣记忆网络"""

def __init__(self, memory_size=128, memory_dim=64):

super().__init__()

# 记忆单元（类NTM记忆网络）

self.memory = nn.Parameter(torch.randn(memory_size, memory_dim))

self.write_gate = nn.GRUCell(memory_dim, memory_dim)

self.read_attention = nn.MultiheadAttention(memory_dim, num_heads=4)

def forward(self, behavior_seq: torch.Tensor) -> torch.Tensor:

batch_size, seq_len, feat_dim = behavior_seq.shape

memory = self.memory.unsqueeze(0).expand(batch_size, -1, -1)

for t in range(seq_len):

# 读操作：用当前行为查询记忆

query = behavior_seq[:, t:t+1, :] # [B, 1, D]

attended, _ = self.read_attention(query, memory, memory)

# 写操作：更新记忆

memory = self.write_gate(attended.squeeze(1), memory)

memory = F.normalize(memory, dim=-1)

# 返回压缩后的用户兴趣表示

user_interest = memory.mean(dim=1) # [B, D]

return user_interest

```

## 四、多目标精排

### 4.1 为什么需要多目标

短视频平台需要同时优化多个指标：

| 目标 | 说明 | 权重示例 |

|------|------|----------|

| 点击率（CTR） | 用户是否点击 | 0.3 |

| 点赞率 | 用户是否点赞 | 0.2 |

| 评论率 | 用户是否评论 | 0.15 |

| 完播率 | 视频是否看完 | 0.25 |

| 分享率 | 用户是否分享 | 0.1 |

### 4.2 MMoE（Multi-gate Mixture-of-Experts）

```python

class MMoE(nn.Module):

"""多目标专家混合模型"""

def __init__(self, input_dim: int, num_experts: int, expert_dim: int, num_tasks: int):

super().__init__()

# 共享专家

self.experts = nn.ModuleList([

nn.Sequential(

nn.Linear(input_dim, expert_dim),

nn.ReLU(),

nn.Linear(expert_dim, expert_dim),

nn.ReLU(),

) for _ in range(num_experts)

])

# 每个任务一个门控网络

self.gates = nn.ModuleList([

nn.Linear(input_dim, num_experts) for _ in range(num_tasks)

])

# 每个任务一个塔

self.task_towers = nn.ModuleList([

nn.Sequential(

nn.Linear(expert_dim, 64),

nn.ReLU(),

nn.Linear(64, 1),

nn.Sigmoid(),

) for _ in range(num_tasks)

])

def forward(self, x: torch.Tensor) -> list[torch.Tensor]:

# 专家输出

expert_outputs = [expert(x) for expert in self.experts]

expert_outputs = torch.stack(expert_outputs, dim=1) # [B, num_experts, D]

task_outputs = []

for gate, tower in zip(self.gates, self.task_towers):

# 门控权重

gate_weights = F.softmax(gate(x), dim=-1).unsqueeze(-1) # [B, num_experts, 1]

# 加权专家输出

task_input = (gate_weights * expert_outputs).sum(dim=1) # [B, D]

task_output = tower(task_input)

task_outputs.append(task_output)

return task_outputs # [CTR, 点赞率, 评论率, 完播率, 分享率]

```

## 五、冷启动策略

### 5.1 新用户冷启动

| 策略 | 方法 | 效果 |

|------|------|------|

| 兴趣标签引导 | 注册时选择兴趣标签 | 解决50% |

| 热门内容兜底 | 无行为时推热门 | 解决30% |

| 探索策略（ε-greedy） | 10%流量推未知内容 | 解决15% |

| 社交关系引入 | 通讯录/关注列表 | 解决5% |

### 5.2 新内容冷启动

```python

def cold_start_boost(new_item_score: float, item_age_hours: float) -> float:

"""新内容冷启动加分（时间衰减）"""

# 新内容在24小时内获得2倍曝光机会

if item_age_hours < 24:

boost = 2.0 * (1 - item_age_hours / 24)

return new_item_score * (1 + boost)

return new_item_score

class ExplorationPolicy:

"""Bandit-based探索策略"""

def __init__(self, num_items: int):

self.counts = torch.zeros(num_items) # 每个物品的展示次数

self.values = torch.zeros(num_items) # 每个物品的平均回报

def select(self, user_emb: torch.Tensor, item_embs: torch.Tensor) -> int:

"""UCB算法选择物品"""

# UCB = 预期回报 + 探索奖励

ucb_scores = self.values + torch.sqrt(2 * torch.log(self.counts.sum()) / (self.counts + 1e-5))

return ucb_scores.argmax().item()

def update(self, item_id: int, reward: float):

"""更新Bandit统计"""

self.counts[item_id] += 1

n = self.counts[item_id]

old_value = self.values[item_id]

self.values[item_id] = old_value + (reward - old_value) / n

```

## 六、A/B测试与评估

### 6.1 核心指标

```

线上A/B测试指标

├── 留存指标

│ ├── 次日留存率

│ ├── 7日留存率

│ └── 30日留存率

├── 互动指标

│ ├── 人均播放时长

│ ├── 点赞率

│ ├── 评论率

│ └── 分享率

└── 系统指标

├── 召回率（Recall@100）

├── 精排AUC

└── 多样性指标（ILAD）

```

### 6.2 离线评估代码

```python

def evaluate_recall(model, test_loader, item_embs, k=100):

"""评估召回率@K"""

model.eval()

total_recall = 0

total_samples = 0

with torch.no_grad():

for user_feats, pos_items in test_loader:

user_emb = model.encode_user(user_feats) # [B, D]

# 计算所有物品的分数

scores = torch.matmul(user_emb, item_embs.T) # [B, num_items]

_, topk = torch.topk(scores, k, dim=-1) # [B, K]

for i, pos in enumerate(pos_items):

if pos in topk[i]:

total_recall += 1

total_samples += 1

return total_recall / total_samples

```

## 总结

短视频推荐算法是一个多阶段漏斗系统：双塔召回从百万候选选千级，多目标精排选几十至百条，重排保证多样性。行为序列建模（SIM/MIMN）是长期兴趣捕捉的关键，冷启动策略决定新内容和新用户的生存空间。完整系统需离线训练和线上serving协同设计。

---

*本文由北科信息日采集系统自动生成，发布日期：2026-05-05*