召回 DSSM、YouTubeDNN

• 场景：召回
• 模型：DSSM、YouTubeDNN
• 数据：MovieLens-1M

DSSM 算法原理

• DSSM（Deep Structured Semantic Models，深度结构化语义模型）是微软研究院提出的一种非常经典的深度语义匹配模型。它最早主要用于解决搜索中查询（Query）和文档（Document）的匹配问题，后来也被广泛用于推荐系统、广告投放等场景。
• DSSM的核心目标是：将高维的、稀疏的文本特征（如Query和Doc）映射到一个低维的、稠密的语义空间中，并在该空间中用向量的余弦相似度来表示两者的语义相关性。
• DSSM是一个典型的双塔模型架构。它有两个结构相同（或相似）但参数独立的DNN塔，分别用来处理Query和Document。

1. 输入层：Term Vector+ Word Hashing

• 问题：输入通常是高维的one-hot向量（词袋模型），维度等于词典的大小（可能是几十万甚至百万），直接输入网络会导致参数过多且无法捕捉词之间的相似性。
• 解决方法：Word Hashing（这是DSSM的一个关键亮点，基于词的字母3-gram来压缩输入空间，解决稀疏性和词汇鸿沟问题）。
• 输出：一个相对低维的、基于字母级别的n-gram向量。

2. 表示层：多层DNN

• 这是两个结构相同的深度神经网络（DNN）。
• 输入：经过Word Hashing处理后的Query向量 $x_Q$​ 和 Document向量 $x_D$。
• 计算过程：数据经过若干层全连接层，每一层都使用非线性激活函数（通常是$\tanh$）。
• 公式：

$$l_1 = \tanh (W_1 \cdot x + b_1) \\ l_2 = \tanh (W_2 \cdot x + b_2) \\ ...$$

• 输出：最终得到固定长度的低维语义向量。

假设最后一层维度为 k：

• Query塔输出： $y_Q \in \mathbb{R}^k$（语义向量）
• Document塔输出： $y_D \in \mathbb{R}^k$（语义向量）

3. 匹配层：计算余弦相似度

• 得到Query和Document的语义向量后，在顶层计算它们的相关性分数。
• 公式：

$$R(Q,D)= cosine (y_Q,y_D) = \frac{y_Q^T y_D}{ || y_Q || \cdot || y_D||}$$

• 这个分数 $R(Q,D)$ 代表了Query和Document在语义空间上的匹配程度。

损失函数

• 在训练阶段，通常假设对于一个给定的Query $Q$，我们已经知道它点击过的文档（正样本 $D^+$）和没有点击过的文档（负样本 $D^-$）。
• DSSM使用最大似然估计来优化参数，希望正样本的匹配分数尽可能高，负样本的匹配分数尽可能低。具体来说，它使用softmax函数将匹配分数转化为后验概率：

1. 计算后验概率：

$$P(D^+ | Q) = \frac{exp(\gamma R(Q, D^+))}{\sum_{D' \in \mathbf{D}} R(Q,D')}$$

其中：

• $\gamma$ 是一个平滑因子（通常是一个较大的数，比如5或10），用于放大相似度的影响。
• $\mathbf{D}$ 是候选文档集合，通常包括1个正样本和 N 个随机采样的负样本。

2. 损失函数：
   训练的目标是最大化点击文档的后验概率，也就是最小化负对数似然：

$$L(\Lambda ) = - \log P(D^+ | Q)$$

YouTubeDNN 算法原理

• 它几乎是工业界深度学习推荐系统的教科书级范本，解决了从海量视频中召回（Candidate Generation）到精排序（Ranking）的全链路问题。
• YouTube DNN 将推荐系统明确地拆分为两个独立的阶段：
  • 候选生成（Candidate Generation）：也可以看作召回阶段。从海量视频库（百万级）中选出用户可能感兴趣的几百个视频。
  • 排序（Ranking）：对这几百个视频进行精细打分和排序，选出最终推荐给用户的几十个视频。
• 这种"漏斗形"架构已经成为工业推荐系统的标配。

1. 第一阶段：候选生成（召回）模型

这个阶段的核心是：把推荐问题转化为一个极大规模的"多分类"问题。

1.1 模型架构（也是双塔）

这个阶段的模型也是一个双塔结构，但通常被称为"Query Tower"和"Video Tower"，不过在 Serving 时两者是解耦的。

• 输入特征：
  • 用户侧（Query Tower）：
    • 历史观看视频的 embedding 向量平均（把用户看过的视频 ID 映射成向量，再取平均）。
    • 搜索词（如果有点击日志）。
    • 人口统计学信息（性别、年龄等）。
    • Example Age（样本年龄）：一个非常重要的特征，代表这个视频样本被观看的时间距离现在多久（用于模拟实时兴趣）。
  • 视频侧（Video Tower）：
    • 视频 ID（待分类的目标）。
    • 视频其他特征（频道、上传时间等）。
• 训练目标：Softmax 多分类
  • 模型将预测用户在时刻 t 会观看哪个视频 $w_t$ 作为分类任务。
  • 公式：

$$P(w_t = i|U, C) = \frac{e^{\langle \mathbf{u}, \mathbf{v}_i \rangle}}{\sum_{j \in V} e^{\langle \mathbf{u}, \mathbf{v}_j \rangle}}$$

• 其中：

  • $\mathbf{u}$：是用户塔的输出向量（代表用户的兴趣）。
  • $\mathbf{v}_i$：是视频 $i$ 的向量。
  • $V$：是全部视频的集合（百万甚至千万级）。

• 训练难点与技巧

  • 负采样（Negative Sampling）：分母需要对所有视频计算 Softmax，计算量巨大。YouTube DNN 采用负采样技术，只计算一部分负样本的得分来近似全 Softmax。
  • 上下文（Context）：不仅看用户看了什么，还引入了搜索词作为强特征，捕捉用户当前的主动意图。
  • Example Age 特征：YouTube 视频有很强的时效性（用户喜欢看新视频）。如果不做处理，模型会倾向于推荐训练数据中平均表现好的老视频。加入 Example Age 特征（在 Serving 时设为 0 或很小的负数），可以让模型学到：某个视频刚上传时（Age=0），推荐概率会变高。

1.2 召回阶段的 Serving（线上服务）

训练完成后，召回阶段的服务逻辑是最近邻搜索：

1. 离线计算：把所有视频 $v_j$ 的向量计算好，存入向量数据库（如 FAISS）。
2. 线上实时：用户发起请求，实时计算用户向量 $\mathbf{u}$。
3. ANN 检索：用 $\mathbf{u}$ 去视频向量库中做近似最近邻搜索，快速找到与 $\mathbf{u}$ 最相似的 $N$ 个视频（即得分 $\langle \mathbf{u}, \mathbf{v}_i \rangle$ 最高的视频）。

2. 第二阶段：排序（Ranking）模型

经过召回阶段，候选视频数量从百万级降到了几百个。排序阶段的任务就是对这几百个视频进行精准的点击率预估。

2.1 模型架构

排序模型是一个标准的深度神经网络（DNN）。

• 输入特征：
  • 稠密特征：用户特征（召回的 user vector）、视频特征。
  • 稀疏特征：用户上次与同频道互动的频次、用户最后一次观看本视频的时间等。
  • 类别特征：视频 ID、频道 ID（经过 Embedding 层处理）。
• 网络结构：
  • 将上述特征拼接成一个巨大的输入向量。
  • 经过若干层全连接层（ReLU 激活函数）。
  • 输出层：一个神经元（用于预估点击率）。

2.2 训练目标：加权逻辑回归

与常见的二分类逻辑回归不同，YouTube 排序模型采用了一种特殊的训练方式，以适应视频观看时长的优化目标。

• 传统方法：预测点击率 $pCTR$。
• YouTube 方法：预测观看时长 $E[T]$（期望观看时长）。
• 实现技巧：
  • 模型最后一层使用 逻辑回归（输出 $\frac{1}{1+e^{-x}}$）。
    • 正样本：用户点击并观看了的视频，赋权值为观看时长（如观看 3 分钟，权重就是 3）。
    • 负样本：曝光未点击的视频，赋权值为 1。
    • 损失函数：使用加权逻辑回归。
    • 线上预测：计算 $e^{x}$（逻辑回归的 logits 的指数）。为什么？
      • 在加权逻辑回归中，$Odds = \frac{p}{1-p} = \frac{\sum T_+}{\sum T_-}$。
      • $p$ 通常很小，所以 $e^{x} \approx p \times (\text{正样本平均时长} + 1) $。
        • 这个输出值近似正比于预期观看时长。

2.3 排序阶段的 Serving

将线上请求的几百个视频特征输入模型，进行一次前向传播，得到每个视频的得分（预期观看时长）。按照这个得分从高到低排序，取 Top N 推荐给用户。

代码实现

直接调用

DSSM

import torch
import pandas as pd
import numpy as np
import os
pd.set_option('display.max_rows',500)
pd.set_option('display.max_columns',500)
pd.set_option('display.width',1000)
torch.manual_seed(2026)

# sample中只有两个用户
file_path = '../examples/matching/data/ml-1m/ml-1m_sample.csv'
data = pd.read_csv(file_path)
print(data.head())

# 处理genres特征，取出其第一个作为标签
data["cate_id"] = data["genres"].apply(lambda x: x.split("|")[0])

# 指定用户列和物品列的名字、离散和稠密特征，适配框架的接口
user_col, item_col = "user_id", "movie_id"
sparse_features = ['user_id', 'movie_id', 'gender', 'age', 'occupation', 'zip', "cate_id"]
dense_features = []

save_dir = '../examples/ranking/data/ml-1m/saved/'
if not os.path.exists(save_dir):
    os.makedirs(save_dir)
# 对SparseFeature进行LabelEncoding
from sklearn.preprocessing import LabelEncoder
print(data[sparse_features].head())
feature_max_idx = {}
for feature in sparse_features:
    lbe = LabelEncoder()
    data[feature] = lbe.fit_transform(data[feature]) + 1
    feature_max_idx[feature] = data[feature].max() + 1
    if feature == user_col:
        user_map = {encode_id + 1: raw_id for encode_id, raw_id in enumerate(lbe.classes_)}  #encode user id: raw user id
    if feature == item_col:
        item_map = {encode_id + 1: raw_id for encode_id, raw_id in enumerate(lbe.classes_)}  #encode item id: raw item id
np.save(save_dir+"raw_id_maps.npy", (user_map, item_map))  # evaluation时会用到
print('LabelEncoding后：')
print(data[sparse_features].head())

# 定义两个塔对应哪些特征
user_cols = ["user_id", "gender", "age", "occupation", "zip"]
item_cols = ['movie_id', "cate_id"]

# 从data中取出相应的数据
user_profile = data[user_cols].drop_duplicates('user_id')
item_profile = data[item_cols].drop_duplicates('movie_id')
print(user_profile.head())
print(item_profile.head())

from torch_rechub.utils.match import generate_seq_feature_match, gen_model_input
df_train, df_test = generate_seq_feature_match(data,
                                               user_col,
                                               item_col,
                                               time_col="timestamp",
                                               item_attribute_cols=[],
                                               sample_method=1,
                                               mode=0,
                                               neg_ratio=3,  # 每个正样本对应的负样本数量，如果之后使用in-batch负采样，这个值直接设为0，只生成正样本
                                               min_item=0)
print(df_train.head())
x_train = gen_model_input(df_train, user_profile, user_col, item_profile, item_col, seq_max_len=50)
y_train = x_train["label"]
x_test = gen_model_input(df_test, user_profile, user_col, item_profile, item_col, seq_max_len=50)
y_test = x_test["label"]
print({k: v[:3] for k, v in x_train.items()})

#定义特征类型

from torch_rechub.basic.features import SparseFeature, SequenceFeature
user_features = [
    SparseFeature(feature_name, vocab_size=feature_max_idx[feature_name], embed_dim=16) for feature_name in user_cols
]
user_features += [
    SequenceFeature("hist_movie_id",
                    vocab_size=feature_max_idx["movie_id"],
                    embed_dim=16,
                    pooling="mean",
                    shared_with="movie_id")
]

item_features = [
    SparseFeature(feature_name, vocab_size=feature_max_idx[feature_name], embed_dim=16) for feature_name in item_cols
]
print(user_features)
print(item_features)

# 将dataframe转为dict
from torch_rechub.utils.data import df_to_dict
all_item = df_to_dict(item_profile)
test_user = x_test
print({k: v[:3] for k, v in all_item.items()})
print({k: v[0] for k, v in test_user.items()})

from torch_rechub.models.matching import DSSM
from torch_rechub.trainers import MatchTrainer
from torch_rechub.utils.data import MatchDataGenerator
# 根据之前处理的数据拿到Dataloader
dg = MatchDataGenerator(x=x_train, y=y_train)
train_dl, test_dl, item_dl = dg.generate_dataloader(test_user, all_item, batch_size=256)

# 定义模型
model = DSSM(user_features,
             item_features,
             temperature=0.02,
             user_params={
                 "dims": [256, 128, 64],
                 "activation": 'prelu',  # important!!
             },
             item_params={
                 "dims": [256, 128, 64],
                 "activation": 'prelu',  # important!!
             })

# 模型训练器
trainer = MatchTrainer(model,
                       mode=0,  # 同上面的mode，需保持一致
                       in_batch_neg=True,               # 是否使用batch内负采样
                       in_batch_neg_ratio=3,            # batch内负采样的数量比例
                       optimizer_params={
                           "lr": 1e-4,
                           "weight_decay": 1e-6
                       },
                       n_epoch=1,
                       device='cpu',
                       model_path=save_dir)

# 开始训练
trainer.fit(train_dl)

import collections
import numpy as np
import pandas as pd
from torch_rechub.utils.match import Annoy
from torch_rechub.basic.metric import topk_metrics

def match_evaluation(user_embedding, item_embedding, test_user, all_item, user_col='user_id', item_col='movie_id',
                     raw_id_maps="./raw_id_maps.npy", topk=10):
    print("evaluate embedding matching on test data")
    annoy = Annoy(n_trees=10)
    annoy.fit(item_embedding)

    #for each user of test dataset, get ann search topk result
    print("matching for topk")
    user_map, item_map = np.load(raw_id_maps, allow_pickle=True)
    match_res = collections.defaultdict(dict)  # user id -> predicted item ids
    for user_id, user_emb in zip(test_user[user_col], user_embedding):
        items_idx, items_scores = annoy.query(v=user_emb, n=topk)  #the index of topk match items
        match_res[user_map[user_id]] = np.vectorize(item_map.get)(all_item[item_col][items_idx])

    #get ground truth
    print("generate ground truth")

    data = pd.DataFrame({user_col: test_user[user_col], item_col: test_user[item_col]})
    data[user_col] = data[user_col].map(user_map)
    data[item_col] = data[item_col].map(item_map)
    user_pos_item = data.groupby(user_col).agg(list).reset_index()
    ground_truth = dict(zip(user_pos_item[user_col], user_pos_item[item_col]))  # user id -> ground truth

    print("compute topk metrics")
    out = topk_metrics(y_true=ground_truth, y_pred=match_res, topKs=[topk])
    return out

user_embedding = trainer.inference_embedding(model=model, mode="user", data_loader=test_dl, model_path=save_dir)
item_embedding = trainer.inference_embedding(model=model, mode="item", data_loader=item_dl, model_path=save_dir)
match_evaluation(user_embedding, item_embedding, test_user, all_item, topk=10, raw_id_maps=save_dir+"raw_id_maps.npy")

YouTubeDNN

import os
import numpy as np
import pandas as pd
import torch
from sklearn.preprocessing import LabelEncoder
from torch_rechub.models.matching import YoutubeDNN
from torch_rechub.trainers import MatchTrainer
from torch_rechub.basic.features import SparseFeature, SequenceFeature
from torch_rechub.utils.match import generate_seq_feature_match, gen_model_input
from torch_rechub.utils.data import df_to_dict, MatchDataGenerator


torch.manual_seed(2026)

data = pd.read_csv(file_path)
data["cate_id"] = data["genres"].apply(lambda x: x.split("|")[0])
sparse_features = ['user_id', 'movie_id', 'gender', 'age', 'occupation', 'zip', "cate_id"]
user_col, item_col = "user_id", "movie_id"

feature_max_idx = {}
for feature in sparse_features:
    lbe = LabelEncoder()
    data[feature] = lbe.fit_transform(data[feature]) + 1
    feature_max_idx[feature] = data[feature].max() + 1
    if feature == user_col:
        user_map = {encode_id + 1: raw_id for encode_id, raw_id in enumerate(lbe.classes_)}  #encode user id: raw user id
    if feature == item_col:
        item_map = {encode_id + 1: raw_id for encode_id, raw_id in enumerate(lbe.classes_)}  #encode item id: raw item id
np.save(save_dir+"raw_id_maps.npy", (user_map, item_map))
user_cols = ["user_id", "gender", "age", "occupation", "zip"]
item_cols = ["movie_id", "cate_id"]
user_profile = data[user_cols].drop_duplicates('user_id')
item_profile = data[item_cols].drop_duplicates('movie_id')


#Note: mode=2 means list-wise negative sample generate, saved in last col "neg_items"
df_train, df_test = generate_seq_feature_match(data,
                                               user_col,
                                               item_col,
                                               time_col="timestamp",
                                               item_attribute_cols=[],
                                               sample_method=1,
                                               mode=2,  # [0]训练方式改为List wise
                                               neg_ratio=3,
                                               min_item=0)
x_train = gen_model_input(df_train, user_profile, user_col, item_profile, item_col, seq_max_len=50)
y_train = np.array([0] * df_train.shape[0])  # [1]训练集所有样本的label都取0。因为一个样本的组成是(pos, neg1, neg2, ...)，视为一个多分类任务，正样本的位置永远是0
x_test = gen_model_input(df_test, user_profile, user_col, item_profile, item_col, seq_max_len=50)

user_cols = ['user_id', 'gender', 'age', 'occupation', 'zip']

user_features = [SparseFeature(name, vocab_size=feature_max_idx[name], embed_dim=16) for name in user_cols]
user_features += [
    SequenceFeature("hist_movie_id",
                    vocab_size=feature_max_idx["movie_id"],
                    embed_dim=16,
                    pooling="mean",
                    shared_with="movie_id")
]

item_features = [SparseFeature('movie_id', vocab_size=feature_max_idx['movie_id'], embed_dim=16)]  # [2]物品的特征只有itemID，即movie_id一个
neg_item_feature = [
    SequenceFeature('neg_items',
                    vocab_size=feature_max_idx['movie_id'],
                    embed_dim=16,
                    pooling="concat",
                    shared_with="movie_id")
]  # [3] 多了一个neg item feature，会传入到模型中，在item tower中会用到

all_item = df_to_dict(item_profile)
test_user = x_test

dg = MatchDataGenerator(x=x_train, y=y_train)
train_dl, test_dl, item_dl = dg.generate_dataloader(test_user, all_item, batch_size=256)

model = YoutubeDNN(user_features, item_features, neg_item_feature, user_params={"dims": [128, 64, 16]}, temperature=0.02)  # [4] MLP的最后一层需保持与item embedding一致

#mode=1 means pair-wise learning
trainer = MatchTrainer(model,
                       mode=2,
                       optimizer_params={
                           "lr": 1e-4,
                           "weight_decay": 1e-6
                       },
                       n_epoch=1, #5
                       device='cpu',
                       model_path=save_dir)

trainer.fit(train_dl)

print("inference embedding")
user_embedding = trainer.inference_embedding(model=model, mode="user", data_loader=test_dl, model_path=save_dir)
item_embedding = trainer.inference_embedding(model=model, mode="item", data_loader=item_dl, model_path=save_dir)
match_evaluation(user_embedding, item_embedding, test_user, all_item, topk=10, raw_id_maps="../examples/ranking/data/ml-1m/saved/raw_id_maps.npy")

自定义模型

DSSM

import torch
import torch.nn.functional as F
from torch_rechub.basic.layers import MLP, EmbeddingLayer


class DSSM(torch.nn.Module):
    def __init__(self, user_features, item_features, user_params, item_params, temperature=1.0):
        super().__init__()
        self.user_features = user_features
        self.item_features = item_features
        self.temperature = temperature
        self.user_dims = sum([fea.embed_dim for fea in user_features])
        self.item_dims = sum([fea.embed_dim for fea in item_features])

        self.embedding = EmbeddingLayer(user_features + item_features)
        self.user_mlp = MLP(self.user_dims, output_layer=False, **user_params)
        self.item_mlp = MLP(self.item_dims, output_layer=False, **item_params)
        self.mode = None

    def forward(self, x):
        user_embedding = self.user_tower(x)
        item_embedding = self.item_tower(x)
        if self.mode == "user":
            return user_embedding
        if self.mode == "item":
            return item_embedding


        # calculate cosine score
        y = torch.mul(user_embedding, item_embedding).sum(dim=1)
        # y = y / self.temperature
        return torch.sigmoid(y)

    def user_tower(self, x):
        if self.mode == "item":
            return None
        # [batch_size, num_features*deep_dims]
        input_user = self.embedding(x, self.user_features, squeeze_dim=True)
        # [batch_size, user_params["dims"][-1]]
        user_embedding = self.user_mlp(input_user)
        user_embedding = F.normalize(user_embedding, p=2, dim=1)  # L2 normalize
        return user_embedding

    def item_tower(self, x):
        if self.mode == "user":
            return None
        # [batch_size, num_features*embed_dim]
        input_item = self.embedding(x, self.item_features, squeeze_dim=True)
        # [batch_size, item_params["dims"][-1]]
        item_embedding = self.item_mlp(input_item)
        item_embedding = F.normalize(item_embedding, p=2, dim=1)
        return item_embedding

YouTubeDNN

import torch
import torch.nn.functional as F
from torch_rechub.basic.layers import MLP, EmbeddingLayer

class YoutubeDNN(torch.nn.Module):
    def __init__(self, user_features, item_features, neg_item_feature, user_params, temperature=1.0):
        super().__init__()
        self.user_features = user_features
        self.item_features = item_features
        self.neg_item_feature = neg_item_feature
        self.temperature = temperature
        self.user_dims = sum([fea.embed_dim for fea in user_features])
        self.embedding = EmbeddingLayer(user_features + item_features)
        self.user_mlp = MLP(self.user_dims, output_layer=False, **user_params)
        self.mode = None

    def forward(self, x):
        user_embedding = self.user_tower(x)
        item_embedding = self.item_tower(x)
        if self.mode == "user":
            return user_embedding
        if self.mode == "item":
            return item_embedding

        # calculate cosine score
        y = torch.mul(user_embedding, item_embedding).sum(dim=2)
        y = y / self.temperature
        return y

    def user_tower(self, x):
        if self.mode == "item":
            return None
        # [batch_size, num_features*deep_dims]
        input_user = self.embedding(x, self.user_features, squeeze_dim=True)
        user_embedding = self.user_mlp(input_user).unsqueeze(1)  # [batch_size, 1, embed_dim]
        user_embedding = F.normalize(user_embedding, p=2, dim=2)
        if self.mode == "user":
            # inference embedding mode -> [batch_size, embed_dim]
            return user_embedding.squeeze(1)
        return user_embedding

    def item_tower(self, x):
        if self.mode == "user":
            return None
        pos_embedding = self.embedding(x, self.item_features, squeeze_dim=False)  # [batch_size, 1, embed_dim]
        pos_embedding = F.normalize(pos_embedding, p=2, dim=2)
        if self.mode == "item":  # inference embedding mode
            return pos_embedding.squeeze(1)  # [batch_size, embed_dim]
        neg_embeddings = self.embedding(x, self.neg_item_feature, squeeze_dim=False).squeeze(1)  # [batch_size, n_neg_items, embed_dim]
        neg_embeddings = F.normalize(neg_embeddings, p=2, dim=2)
        # [batch_size, 1+n_neg_items, embed_dim]
        return torch.cat((pos_embedding, neg_embeddings), dim=1)