★ FEATURED ARTICLE 简介本资源是一份基于PyTorch实现DeepFM模型的完整推荐系统实践项目面向具备Python与深度学习基础的算法工程师、推荐系统初学者及高校相关方向研究者聚焦点击率CTR预测这一核心业务场景。压缩包共37个文件涵盖5个核心Python模块含dataPreprocess.py数据预处理、DeepFM.py模型定义、main.py训练主逻辑、11个样本数据文件、4个XML配置与IDE配置文件以及README.md说明文档和.git版本控制元信息结构清晰、开箱即用总大小仅40KB轻量易部署。已有666人学习下载适合快速理解FM与DNN双路结构设计、掌握特征嵌入、二阶交互建模与端到端训练全流程。读者可直接复现论文级CTR模型获取可调试的工程化代码骨架、模块化目录组织范式及典型数据预处理逻辑为构建工业级推荐系统打下扎实实践基础。1. DeepFM with PyTorch不是调包跑通就完事而是把特征交叉建模、梯度传播路径、稀疏输入对齐这三关全打穿的推荐系统实战资源你手头有一份用户行为日志和商品侧信息想快速验证 DeepFM 在点击率预估CTR任务上的效果——但发现 GitHub 上多数 PyTorch 实现要么只跑通了 MovieLens 这类玩具数据一换到真实场景的稀疏 ID 特征就报RuntimeError: expected scalar type Long but found Float要么 embedding 层维度硬编码死改个 field 数就得重写整个模型结构更常见的是训练 loss 不降反升debug 半天才发现是 FM 部分的二阶交互项没做 proper normalization梯度爆炸直接把 embedding 向量炸成 NaN。这份DeepFM_with_PyTorch资源不是教学 Demo而是一线推荐系统工程师在某跨平台系统中落地 CTR 模型时拆解出的完整可复现包它包含适配工业级稀疏特征的 DataLoader 封装、支持动态 field 数的 DeepFM 模块、带梯度裁剪与 warmup 的训练脚本、以及一份能直接喂进torchrec或自研特征平台的特征 schema 映射表。适合正在从 LR/GBDT 过渡到深度模型、需要在两周内交付 baseline 并支撑 AB 实验的算法工程师也适合想搞懂“为什么 PyTorch 版 DeepFM 的 FM 部分必须用torch.einsum而不是简单matmul”的进阶学习者。2. DeepFM 原理再拆解为什么 PyTorch 实现必须重写 FM 二阶项计算逻辑DeepFM 的核心价值在于用 FM 模块显式建模低阶特征交互解决稀疏场景下高阶交叉难以学习的问题同时用 DNN 模块隐式学习高阶非线性关系弥补 FM 线性假设的局限。但很多初学者直接照搬 TensorFlow/Keras 实现把 FM 的二阶项写成sum(sum(embedding * embedding))这在 PyTorch 中会引发两个致命问题一是无法正确处理 batch 内不同样本的 field 数差异真实场景中用户历史行为长度不一二是梯度回传时 embedding 更新方向混乱——因为 FM 公式中的 $\sum_{i1}^{n}\sum_{ji1}^{n} \langle v_i, v_j \rangle x_i x_j$ 要求每个 $(i,j)$ 对只参与一次计算而 naive 实现会让 $v_i$ 同时被多个 $j$ 拉扯。2.1 FM 二阶项的 PyTorch 正确实现torch.einsum是唯一可靠解法标准 FM 的二阶交互项数学表达为 $$ \sum_{i1}^{n}\sum_{ji1}^{n} \langle v_i, v_j \rangle x_i x_j \frac{1}{2} \left[ \left( \sum_{i1}^n v_i x_i \right)^2 - \sum_{i1}^n (v_i x_i)^2 \right] $$ 这个恒等变形是 PyTorch 实现的关键跳板。它把 $O(n^2)$ 的双重循环压缩为两次 $O(n)$ 操作且天然支持 batch 维度。下面这段代码就是本资源中FMBlock的核心import torch import torch.nn as nn class FMBlock(nn.Module): def __init__(self, embed_dim: int): super().__init__() self.embed_dim embed_dim def forward(self, x: torch.Tensor, embeddings: torch.Tensor) - torch.Tensor: x: [B, num_fields] # 稠密化后的 field indicator0/1 mask embeddings: [B, num_fields, embed_dim] # 已经 gather 好的 embedding 矩阵 返回 FM 二阶项标量输出 [B, 1] # Step 1: 计算 sum(v_i * x_i) - [B, embed_dim] summed_emb torch.sum(embeddings * x.unsqueeze(-1), dim1) # [B, D] # Step 2: 计算 sum((v_i * x_i)^2) - [B, embed_dim] squared_emb torch.sum((embeddings * x.unsqueeze(-1)) ** 2, dim1) # [B, D] # Step 3: 应用恒等式0.5 * (sum^2 - sum_of_squares) # 使用 einsum 避免 matmul 引入的维度错位风险 sum_sq torch.einsum(bd,bd-b, summed_emb, summed_emb) # [B] sum_sq_term torch.sum(squared_emb, dim1) # [B] fm_second_order 0.5 * (sum_sq - sum_sq_term) # [B] return fm_second_order.unsqueeze(-1) # [B, 1]参数说明x是 field-level mask不是原始 IDembeddings必须是已通过nn.EmbeddingBag或torch.gather对齐后的张量。einsum(bd,bd-b)显式声明了 batch 维度b和 embedding 维度d的收缩规则比torch.bmm(summed_emb.unsqueeze(1), summed_emb.unsqueeze(2)).squeeze()更安全——后者在summed_emb为[B,1]时会意外触发广播错误。2.2 DeepFM 整体结构DNN 与 FM 的特征对齐才是难点很多开源实现把 DNN 输入直接设为torch.cat([linear_out, fm_first, fm_second], dim1)这是严重错误。FM 的一阶项sum(w_i * x_i)和二阶项都依赖原始特征x_i而 DNN 的输入必须是 embedding 向量拼接即torch.cat([emb_1, emb_2, ..., emb_n], dim1)否则模型根本学不到 embedding 表征能力。本资源采用如下结构class DeepFM(nn.Module): def __init__(self, field_dims: list, # e.g., [10000, 500, 2000, ...] 每个 field 的 vocab size embed_dim: int 10, mlp_dims: tuple (64, 32), dropout: float 0.2): super().__init__() self.field_dims field_dims self.num_fields len(field_dims) self.embed_dim embed_dim # 1. Embedding 层每个 field 独立 embedding table self.embedding_tables nn.ModuleList([ nn.Embedding(num_embeddingsdim, embedding_dimembed_dim) for dim in field_dims ]) # 2. FM 部分Linear FMBlock self.linear nn.Linear(self.num_fields, 1) # 一阶项 self.fm_block FMBlock(embed_dim) # 3. DNN 部分输入是所有 embedding 拼接 dnn_input_dim self.num_fields * embed_dim self.dnn MLP(dnn_input_dim, mlp_dims, dropout) # 4. 输出层FM DNN 拼接后过 sigmoid self.output nn.Sequential( nn.Linear(mlp_dims[-1] 2, 1), # 2: fm_first fm_second nn.Sigmoid() ) def forward(self, x: torch.Tensor) - torch.Tensor: x: [B, num_fields] 整数型 ID tensor每个位置是该 field 的原始 ID # Step 1: 获取每个 field 的 embedding - [B, num_fields, embed_dim] embs [] for i in range(self.num_fields): emb_i self.embedding_tables[i](x[:, i]) # [B, embed_dim] embs.append(emb_i) embs torch.stack(embs, dim1) # [B, num_fields, embed_dim] # Step 2: FM 一阶项直接查表加权求和注意这里 linear 是对 field index 加权不是对 embedding fm_first self.linear(x.float()) # [B, 1] # Step 3: FM 二阶项使用上节定义的 FMBlock # 构造 maskx ! 0 表示该 field 有有效值真实场景中 padding 用 0 mask (x ! 0).float() # [B, num_fields] fm_second self.fm_block(mask, embs) # [B, 1] # Step 4: DNN 输入flatten 所有 embedding dnn_input embs.view(embs.size(0), -1) # [B, num_fields * embed_dim] dnn_out self.dnn(dnn_input) # [B, mlp_dims[-1]] # Step 5: 拼接并输出 concat torch.cat([dnn_out, fm_first, fm_second], dim1) # [B, mlp_dims[-1] 2] return self.output(concat)关键设计点mask (x ! 0).float()是工业级必备操作。真实数据中用户历史行为字段常被 zero-padded 到固定长度若直接用x.float()作为 FM 一阶项输入padding 的 0 会被当成有效 field 参与加权导致 bias 偏移。而mask只标记“该位置是否有值”确保 FM 仅对有效 field 建模。3. 数据加载与特征工程如何把原始 CSV 转成 DeepFM 可吃的field_dims和torch.TensorDeepFM 的成败70% 在数据准备。本资源不提供“一键清洗”脚本而是给出一套可审计、可复现、可对接生产 pipeline 的特征处理链路。核心是两点field 边界必须显式声明不能靠 pandas 自动 inferID 映射必须全局一致避免 train/val/test 出现未登录词。3.1 构建field_dims用collections.Counter统计每个 field 的唯一值数量假设原始数据train.csv结构如下逗号分隔无 header12345,2,101,5.2,1 67890,1,205,4.8,0 ...其中第 0 列是 user_id第 1 列是 gender1/2第 2 列是 item_id第 3 列是 rating此处为连续特征DeepFM 中需离散化第 4 列是 labelclick1。import csv from collections import defaultdict, Counter def build_field_vocab(csv_path: str, field_indices: list [0, 1, 2], # 只对这些列做 ID 映射 min_freq: int 5) - dict: 统计每个 field 的 token 频次过滤低频词返回 {field_idx: {token: idx}} counters {idx: Counter() for idx in field_indices} with open(csv_path, r) as f: reader csv.reader(f) for row in reader: for idx in field_indices: if idx len(row): token row[idx].strip() if token: # 跳过空值 counters[idx][token] 1 # 构建 vocab dict每个 field 保留 freq min_freq 的 token并加入 UNK vocabs {} for idx in field_indices: vocab {PAD: 0, UNK: 1} idx_counter 2 for token, freq in counters[idx].most_common(): if freq min_freq: vocab[token] idx_counter idx_counter 1 vocabs[idx] vocab return vocabs # 示例调用 vocabs build_field_vocab(train.csv, field_indices[0,1,2]) print(User ID field size:, len(vocabs[0])) # e.g., 12500 print(Gender field size:, len(vocabs[1])) # e.g., 3 (1,2,UNK) print(Item ID field size:, len(vocabs[2])) # e.g., 89200为什么不用LabelEncoder因为LabelEncoder是 fit-transform 分离的train/val/test 必须用同一套 encoder。而build_field_vocab一次性扫描全量数据或采样足够大的训练集生成确定性 vocab后续所有 split 都共享该字典彻底规避 OOV 问题。3.2 构造 DataLoader用torch.utils.data.Dataset封装稀疏特征PyTorch 的DataLoader默认按 batch 维度堆叠 tensor但 DeepFM 要求每个样本是num_fields个整数 ID。我们定义一个CTRDatasetimport torch from torch.utils.data import Dataset class CTRDataset(Dataset): def __init__(self, csv_path: str, vocabs: dict, field_indices: list [0,1,2], label_index: int 4): self.csv_path csv_path self.vocabs vocabs self.field_indices field_indices self.label_index label_index # 预加载所有样本避免 __getitem__ 中反复 IO self.samples [] with open(csv_path, r) as f: reader csv.reader(f) for row in reader: if len(row) max(self.field_indices [label_index]): continue # 提取 field ID field_ids [] for idx in self.field_indices: token row[idx].strip() # 查 vocab未登录词用 UNK field_id self.vocabs[idx].get(token, self.vocabs[idx][UNK]) field_ids.append(field_id) # 提取 label label int(row[self.label_index].strip()) self.samples.append((field_ids, label)) def __len__(self): return len(self.samples) def __getitem__(self, idx): field_ids, label self.samples[idx] # 转为 tensor x torch.tensor(field_ids, dtypetorch.long) # [num_fields] y torch.tensor([label], dtypetorch.float32) # [1] return x, y # 使用示例 dataset CTRDataset(train.csv, vocabs, field_indices[0,1,2]) dataloader torch.utils.data.DataLoader( dataset, batch_size1024, shuffleTrue, num_workers4, collate_fnlambda batch: tuple(torch.stack(tensors, dim0) for tensors in zip(*batch)) ) # 输出x_batch: [B, num_fields], y_batch: [B, 1]collate_fn 关键点默认DataLoader会尝试torch.stack但__getitem__返回的是(x,y)tuple所以必须自定义collate_fn把 batch 内所有xstack 成[B, num_fields]所有ystack 成[B, 1]。zip(*batch)是 Python 解包技巧把[(x1,y1), (x2,y2), ...]变成([x1,x2,...], [y1,y2,...])。4. 训练与避坑为什么你的 DeepFM loss 不降五个血泪经验总结DeepFM 在 PyTorch 下训练失败90% 不是模型写错了而是数据、初始化、优化器配置这三个环节出了隐蔽问题。以下是我在线上环境踩过的坑每一条都附带print定位语句和修复命令。4.1 常见问题embedding 初始化不当导致梯度爆炸现象训练前 10 个 steploss 从 0.69 飙升到inftorch.isnan(model.embedding_tables[0].weight).any()返回True。原因nn.Embedding默认用uniform(-sqrt(1/embed_dim), sqrt(1/embed_dim))初始化当embed_dim10时范围是(-0.316, 0.316)但 FM 的二阶项含平方运算小数值平方后更小导致梯度回传时 embedding 更新幅度过大。解决改用nn.init.xavier_uniform_并设置gain1.0for table in model.embedding_tables: nn.init.xavier_uniform_(table.weight, gain1.0) # 验证print(table.weight.std().item()) 应在 0.1~0.3 之间4.2 常见问题field mask 未对齐FM 二阶项计算失效现象fm_second输出全是 0或fm_second.mean()远小于fm_first.mean()正常应接近。原因mask构造错误。例如用了mask (x 0).float()但 gender 字段中0是合法值表示未知导致 mask 把有效值判为 0。解决严格按业务定义 valid token。在build_field_vocab中记录每个 field 的valid_tokens并在CTRDataset中生成 mask# 在 CTRDataset.__getitem__ 中 field_ids [] mask_flags [] # 记录该 field 是否有效 for idx in self.field_indices: token row[idx].strip() field_id self.vocabs[idx].get(token, self.vocabs[idx][UNK]) field_ids.append(field_id) # 只有 token 在 vocab 且不为 PAD/UNK 才算有效 is_valid token and token not in [PAD, UNK] mask_flags.append(1.0 if is_valid else 0.0) x torch.tensor(field_ids, dtypetorch.long) mask torch.tensor(mask_flags, dtypetorch.float32) # [num_fields]4.3 常见问题DNN 输入维度错位matmul报 dimension mismatch现象RuntimeError: mat1 and mat2 shapes cannot be multiplied提示mat1: [1024 x 120], mat2: [100 x 64]。原因dnn_input_dim self.num_fields * embed_dim计算错误。例如num_fields12,embed_dim10但实际 embedding stack 后是[B, 12, 10]view(-1)得到[B, 120]而 DNN 第一层权重是[120, 64]没问题。报错说明num_fields或embed_dim传参不一致。解决在__init__中强制校验assert len(field_dims) self.num_fields, ffield_dims length {len(field_dims)} ! num_fields {self.num_fields} assert all(isinstance(d, int) and d 0 for d in field_dims), field_dims must be positive integers print(fModel initialized: {self.num_fields} fields, {embed_dim} dim each - DNN input {self.num_fields * embed_dim})4.4 常见问题label 未归一化sigmoid 输出饱和现象model(x).mean()接近 0.5但BCELoss输出巨大10且梯度grad.norm()超过 1000。原因label 是int类型0/1但BCELoss要求输入是float且model(x)输出是[0,1]区间若 label 是longPyTorch 会隐式转换但某些版本存在精度丢失。解决显式 cast labelcriterion nn.BCELoss() for x_batch, y_batch in dataloader: x_batch x_batch.to(device) y_batch y_batch.to(device).float() # 关键 pred model(x_batch) loss criterion(pred, y_batch)4.5 常见问题学习率过大FM 部分收敛慢于 DNN现象loss 曲线震荡剧烈val AUC 波动超过 0.05且 FM 模块的参数更新幅度远小于 DNN。原因FM 的二阶项对 learning rate 更敏感而 DNN 可以用较大 lr。统一 lr 导致 FM 学习不足。解决为 FM 和 DNN 设置不同 lroptimizer torch.optim.Adam([ {params: model.linear.parameters(), lr: 0.01}, {params: model.fm_block.parameters(), lr: 0.01}, {params: model.dnn.parameters(), lr: 0.001}, {params: model.embedding_tables.parameters(), lr: 0.001}, ])5. 模型验证与线上部署用 AUC 和 KS 指标卡住效果底线用 TorchScript 封装推理黑匣子训练完模型只是开始。真正决定 DeepFM 落地价值的是它能否在真实流量中稳定提升 CTR以及能否无缝接入现有 serving 系统。本资源提供一套轻量但完整的验证方案不依赖 Spark 或 Flink纯 PyTorch 即可完成。5.1 离线评估用sklearn.metrics计算 AUC/KS拒绝“看 loss 下降就上线”AUC 是 CTR 模型的核心指标但它只反映排序能力。KSKolmogorov-Smirnov统计量则衡量模型对正负样本的区分度更能暴露“模型把所有样本都打高分”的灾难性 failure。以下函数封装了二者计算from sklearn.metrics import roc_auc_score, roc_curve import numpy as np def evaluate_model(model: nn.Module, dataloader: torch.utils.data.DataLoader, device: torch.device) - dict: model.eval() all_preds [] all_labels [] with torch.no_grad(): for x_batch, y_batch in dataloader: x_batch x_batch.to(device) y_batch y_batch.to(device).float() pred model(x_batch).cpu().numpy().flatten() label y_batch.cpu().numpy().flatten() all_preds.extend(pred) all_labels.extend(label) # AUC auc roc_auc_score(all_labels, all_preds) # KS: 计算正负样本的累积分布差最大值 fpr, tpr, _ roc_curve(all_labels, all_preds) ks max(tpr - fpr) # 分桶校准检查预测分是否均匀分布 bins np.linspace(0, 1, 11) digitized np.digitize(all_preds, bins) - 1 bin_counts np.array([np.sum(digitized i) for i in range(len(bins)-1)]) bin_ratio bin_counts / len(all_preds) return { auc: round(auc, 4), ks: round(ks, 4), calibration: [round(r, 3) for r in bin_ratio.tolist()] } # 使用示例 val_loader torch.utils.data.DataLoader(val_dataset, batch_size2048) metrics evaluate_model(model, val_loader, device) print(fAUC: {metrics[auc]}, KS: {metrics[ks]}) print(fScore distribution: {metrics[calibration]}) # 理想 calibration: [0.1, 0.1, ..., 0.1]若首尾两桶占比超 0.3 则需重新校准为什么看 calibration如果calibration[0]预测分 0~0.1占比 0.4说明模型极度保守大量真实 click 被压到低分段即使 AUC0.8 也不可信。这是线上最常被忽略的“玄学”指标。5.2 TorchScript 导出把模型变成无 Python 依赖的.pt文件直连 C servingPyTorch 模型要上生产必须脱离 Python 解释器。torch.jit.trace是最稳妥的方式比script更少报错# 准备 dummy inputshape 必须和实际 serving 一致 dummy_x torch.randint(0, 1000, (1, len(field_dims)), dtypetorch.long) model.eval() # trace 导出 traced_model torch.jit.trace(model, dummy_x) traced_model.save(deepfm_traced.pt) # 验证导出结果 loaded torch.jit.load(deepfm_traced.pt) pred loaded(dummy_x) print(Traced model output shape:, pred.shape) # 应为 [1, 1]关键约束dummy_x的num_fields必须和训练时完全一致且每个 field 的 vocab size 要覆盖线上最大 ID。建议在build_field_vocab中保存max_id_per_field导出时用torch.randint(0, max_id1, ...)。5.3 特征 schema 映射表让算法和工程不再吵架最后但最重要——一份feature_schema.json明确告诉工程同学“你们传来的第 3 个字段对应我模型的第 2 个 embedding table取值范围是 0~89199”。本资源生成该文件import json def generate_schema_json(vocabs: dict, field_names: list [user_id, gender, item_id]) - None: schema [] for i, (idx, vocab) in enumerate(vocabs.items()): schema.append({ field_index: i, field_name: field_names[i] if i len(field_names) else ffield_{i}, vocab_size: len(vocab), valid_range: [0, len(vocab)-1], pad_token: 0, unk_token: 1 }) with open(feature_schema.json, w) as f: json.dump(schema, f, indent2) generate_schema_json(vocabs)feature_schema.json示例[ { field_index: 0, field_name: user_id, vocab_size: 12500, valid_range: [0, 12499], pad_token: 0, unk_token: 1 } ]从那以后我每次交付模型都强制走一遍generate_schema_jsontorch.jit.traceevaluate_model三步验证。不是为了炫技而是因为某次漏了 schema 文档工程同学把gender字段的1/2当成0/1传进来模型把所有男性样本打成 0 分AB 实验全盘作废。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
