DeepFM

• 场景：精排CTR预测
• 数据：Criteo广告数据集
• DeepFM 是为了解决 CTR 预估问题而生的，它的全称是Deep Factorization Machine（深度因子分解机），本质是将FM 层（因子分解机）和DNN 层（深度神经网络） 结合，同时学习特征的低阶交互和高阶交互，最终输出用户点击的概率（预估 CTR）。
  • 传统 LR（逻辑回归）只能学习线性特征。
  • 纯 FM 只能学习二阶特征交互（两个特征的组合）。
  • 纯 DNN 虽然能学高阶交互，但对低阶交互的学习效率低，且需要大量数据。

输入特征 → 嵌入层（Embedding Layer） → 分两路：
          ├─ FM层：学习二阶特征交互 → 输出FM得分
          └─ DNN层：学习高阶特征交互 → 输出DNN得分
→ 拼接FM得分 + DNN得分 → 输出层（Sigmoid）→ 预估CTR（0~1的概率）

算法原理

1. 输入特征与嵌入层

• 输入特征：
  1. 离散特征：若这个离散特征是高维稀疏的，则需要先用一个嵌入层映射到低维稠密。
  2. 连续特征：可以直接归一化后输入；也可以做离散化。

2. FM 学习二阶交互

• 负责捕捉两两特征的交互（比如 “性别=女” 和 “商品类别=口红” 的组合），核心公式如下：

$$y_{FM} = w_0 + \sum_{i=1}^n w_i x_i + \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} x_i)^2 \right)$$

• 第一项是全局偏置；
• 第二项是一阶特征贡献；
• 第三项是二阶特征交互贡献。
  其中的 $v_{i,f}$ 表示特征 $i$ 嵌入向量的第 $f$ 维。

FM 层最终输出一个标量（单个数值），代表所有一阶特征 + 二阶特征交互的总贡献。

3. DNN 学习高阶交互

• DNN 层是 DeepFM 的 “泛化模块”，负责捕捉三阶及以上的特征交互（比如 “性别=女 + 年龄=25 + 消费等级=高 + 商品=口红”）。
• 输入：把嵌入层输出的所有低维嵌入向量拼接成一个大的稠密向量（比如用户 ID 嵌入 (16 维)+ 商品 ID 嵌入 (16 维)+ 性别嵌入 (16 维)=48 维）；
• 结构：多层全连接神经网络（比如隐藏层：256→128→64），每层用 ReLU 激活函数增加非线性；
• 输出：最后一层输出一个标量，代表所有高阶特征交互的总贡献。

4. 输出层：融合预测

• 把 FM 层的输出和 DNN 层的输出拼接，通过一个 Sigmoid 函数，输出 0~1 之间的概率（即预估的 CTR）：

$$y=\sigma (y_{FM}​+y_{DNN​}) \\ 其中\sigma (x)=\frac{1}{1+e^{−x}}​,即映射到0-1的概率空间$$

DeepFM 核心是FM 层 + DNN 层，共享嵌入层，端到端训练，同时学习低阶（二阶）和高阶特征交互；
FM 层负责捕捉两两特征的直接关联（记忆），DNN 层负责捕捉多特征的复杂关联（泛化）；
嵌入层是关键桥梁，将高维稀疏特征转为低维稠密向量，让 FM 和 DNN 都能高效学习。

代码实现

直接调包

import numpy as np
import pandas as pd
import torch
from torch_rechub.models.ranking import WideDeep, DeepFM, DCN
from torch_rechub.trainers import CTRTrainer
from torch_rechub.basic.features import DenseFeature, SparseFeature
from torch_rechub.utils.data import DataGenerator
from tqdm import tqdm
from sklearn.preprocessing import MinMaxScaler, LabelEncoder
torch.manual_seed(2026) #固定随机种子

data_path = '../examples/ranking/data/criteo/criteo_sample.csv'
data = pd.read_csv(data_path)  
#data = pd.read_csv(data_path, compression="gzip") #if the raw_data is .gz file
# data.head()

dense_cols= [f for f in data.columns.tolist() if f[0] == "I"] #以I开头的特征名为dense特征
sparse_cols = [f for f in data.columns.tolist() if f[0] == "C"]  #以C开头的特征名为sparse特征

data[dense_cols] = data[dense_cols].fillna(0) #填充空缺值
data[sparse_cols] = data[sparse_cols].fillna('-996')


#criteo比赛冠军分享的一种离散化思路，不用纠结其原理，大家也可以试试别的离散化手段
def convert_numeric_feature(val):
    v = int(val)
    if v > 2:
        return int(np.log(v)**2)
    else:
        return v - 2
        
for col in tqdm(dense_cols):  #将离散化dense特征列设置为新的sparse特征列
    sparse_cols.append(col + "_sparse")
    data[col + "_sparse"] = data[col].apply(lambda x: convert_numeric_feature(x))

scaler = MinMaxScaler()  #对dense特征列归一化
data[dense_cols] = scaler.fit_transform(data[dense_cols])

for col in tqdm(sparse_cols):  #sparse特征编码
    lbe = LabelEncoder()
    data[col] = lbe.fit_transform(data[col])

#重点：将每个特征定义为torch-rechub所支持的特征基类，dense特征只需指定特征名，sparse特征需指定特征名、特征取值个数(vocab_size)、embedding维度(embed_dim)
dense_features = [DenseFeature(feature_name) for feature_name in dense_cols]
sparse_features = [SparseFeature(feature_name, vocab_size=data[feature_name].nunique(), embed_dim=16) for feature_name in sparse_cols]
y = data["label"]
del data["label"]
x = data

# 构建模型输入所需要的dataloader，区分验证集、测试集，指定batch大小
#split_ratio=[0.7,0.1] 指的是训练集占比70%，验证集占比10%，剩下的全部为测试集
dg = DataGenerator(x, y) 
train_dataloader, val_dataloader, test_dataloader = dg.generate_dataloader(split_ratio=[0.7, 0.1], batch_size=256, num_workers=8)


from torch_rechub.models.ranking import DeepFM
from torch_rechub.trainers import CTRTrainer

#定义模型
model = DeepFM(
        deep_features=dense_features+sparse_features,
        fm_features=sparse_features,
        mlp_params={"dims": [256, 128], "dropout": 0.2, "activation": "relu"},
    )

# 模型训练，需要学习率、设备等一般的参数，此外我们还支持earlystoping策略，及时发现过拟合
ctr_trainer = CTRTrainer(
    model,
    optimizer_params={"lr": 1e-4, "weight_decay": 1e-5},
    regularization_params={"embedding_l1": 1e-6, "embedding_l2": 1e-6, "dense_l1": 0, "dense_l2": 1e-6},
    n_epoch=5,
    earlystop_patience=3,
    device='cpu', #如果有gpu，可设置成cuda:0
    model_path='./', #模型存储路径
)
ctr_trainer.fit(train_dataloader, val_dataloader)

# 查看在测试集上的性能
auc = ctr_trainer.evaluate(ctr_trainer.model, test_dataloader)
print(f'test auc: {auc}')

• 若用其他模型：

#定义相应的模型，用同样的方式训练
model = WideDeep(wide_features=dense_features, deep_features=sparse_features, mlp_params={"dims": [256, 128], "dropout": 0.2, "activation": "relu"})

model = DCN(features=dense_features + sparse_features, n_cross_layers=3, mlp_params={"dims": [256, 128]})

自定义模型

import torch
from torch_rechub.basic.layers import FM, MLP, LR, EmbeddingLayer

class MyDeepFM(torch.nn.Module):
  # Deep和FM为两部分，分别处理不同的特征，因此传入的参数要有两种特征，由此我们得到参数deep_features,fm_features
  # 此外神经网络类的模型中，基本组成原件为MLP多层感知机，多层感知机的参数也需要传进来，即为mlp_params
  def __init__(self, deep_features, fm_features, mlp_params):
    super().__init__()
    self.deep_features = deep_features
    self.fm_features = fm_features
    self.deep_dims = sum([fea.embed_dim for fea in deep_features])
    self.fm_dims = sum([fea.embed_dim for fea in fm_features])
    # 对特征做嵌入表征
    self.embedding = EmbeddingLayer(deep_features + fm_features)
    # LR建模一阶特征交互
    self.linear = LR(self.fm_dims)
    # FM建模二阶特征交互
    self.fm = FM(reduce_sum=True)
    # MLP建模高阶特征交互
    self.mlp = MLP(self.deep_dims, **mlp_params)

  def forward(self, x):
    input_deep = self.embedding(x, self.deep_features, squeeze_dim=True)  #[batch_size, deep_dims]
    input_fm = self.embedding(x, self.fm_features, squeeze_dim=False)  #[batch_size, num_fields, embed_dim]

    y_linear = self.linear(input_fm.flatten(start_dim=1))
    y_fm = self.fm(input_fm)
    y_deep = self.mlp(input_deep)  #[batch_size, 1]
    # 最终的预测值为一阶特征交互，二阶特征交互，以及深层模型的组合
    y = y_linear + y_fm + y_deep
    # 利用sigmoid来将预测得分规整到0,1区间内
    return torch.sigmoid(y.squeeze(1))  # 把[batch_size, 1]改为[batch_size]


model = MyDeepFM(
        deep_features=dense_features+sparse_features,
        fm_features=sparse_features,
        mlp_params={"dims": [256, 128], "dropout": 0.2, "activation": "relu"},
    )
# 模型训练，需要学习率、设备等一般的参数，此外我们还支持earlystoping策略，及时发现过拟合
ctr_trainer = CTRTrainer(
    model,
    optimizer_params={"lr": 1e-4, "weight_decay": 1e-5},
    n_epoch=1,
    earlystop_patience=3,
    device='cpu',
    model_path='./',
)
ctr_trainer.fit(train_dataloader, val_dataloader)

# 查看在测试集上的性能
auc = ctr_trainer.evaluate(ctr_trainer.model, test_dataloader)
print(f'test auc: {auc}')