深度学习实战-基于ContentBranch和CFBranch的混合电影推荐模型

🤵‍♂️ 个人主页：@艾派森的个人主页

✍🏻作者简介：Python学习者
🐋 希望大家多多支持，我们一起进步！😄
如果文章对你有帮助的话，
欢迎评论 💬点赞👍🏻 收藏 📂加关注+

目录

1.项目背景

2.数据集介绍

3.技术工具

4.实验过程

4.1导入数据

4.2数据预处理

4.3特征工程

4.4构建模型

4.5模型训练

4.6模型评估

4.7模型预测

5.总结

文末推荐

源代码

1.项目背景

随着在线视频平台与流媒体服务的快速发展，电影内容的供给规模呈现出爆发式增长，用户在海量信息中进行有效选择的难度显著提升，推荐系统逐渐成为提升用户体验与平台留存的核心技术支撑。在实际应用中，传统的协同过滤方法依赖用户行为数据，虽在刻画群体偏好方面具有优势，但容易受到冷启动问题与数据稀疏性的制约；而基于内容的推荐方法能够利用文本、类型等结构化与非结构化信息进行建模，却往往难以捕捉用户隐含兴趣与跨内容的潜在关联。因此，如何在两类方法之间实现有效融合，构建兼顾表达能力与泛化能力的推荐模型，成为当前推荐系统研究中的关键问题。在这一背景下，结合电影文本描述、类型标签及评分等多源信息，通过深度学习方法构建内容表示，并引入协同信号进行联合建模，不仅能够提升推荐结果的相关性与多样性，也为解决新内容冷启动与兴趣迁移问题提供了可行路径。本研究正是在这一现实需求与技术发展背景下展开，旨在探索一种结构清晰、可扩展的混合推荐建模方案，以更好地适应复杂多变的内容分发场景。

2.数据集介绍

本实验数据集来源于Kaggle，该数据集全面展现了截至2026年初电影数据库（TMDB）评分最高的10000部电影。它旨在帮助数据分析师和电影爱好者探索跨越数十年和多种语言的“顶级”电影的特征。这对于推荐系统来说非常有用。

3.技术工具

Python版本:3.9

代码编辑器：jupyter notebook

4.实验过程

4.1导入数据

这一部分主要完成实验所需环境与数据的准备工作。这里使用的是TMDB电影数据集，包含电影的基本信息、文本描述以及用户评分等内容，为后续构建基于内容和协同过滤的混合推荐模型提供数据基础。在代码层面，主要引入了PyTorch用于模型构建与训练，pandas和numpy用于数据处理，同时加载了一些常用的特征工程工具，如TF-IDF、SVD以及多标签编码等，为后续特征提取和降维做准备。

import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import Dataset, DataLoader import pandas as pd import numpy as np import ast, warnings import matplotlib.pyplot as plt warnings.filterwarnings('ignore') # 忽略不必要的警告信息 # 文本特征与数据预处理相关工具 from sklearn.feature_extraction.text import TfidfVectorizer # 文本向量化（用于内容分支） from sklearn.preprocessing import MinMaxScaler, MultiLabelBinarizer # 归一化、多标签编码 from sklearn.decomposition import TruncatedSVD # 降维（缓解高维稀疏问题） from sklearn.model_selection import train_test_split # 数据集划分 # 设置设备（优先使用GPU） device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # 固定随机种子，保证实验可复现 torch.manual_seed(42) np.random.seed(42) # 读取电影数据集（包含电影内容信息与评分等） df = pd.read_csv('/kaggle/input/datasets/dhritisisodia/tmdb-top-10000-movies-dataset-2026/tmdb_top_10k_movies_2026.csv.csv')

4.2数据预处理

这一部分对原始电影数据进行清洗与结构化处理，核心目的是把原本较为杂乱的字段转化为模型可以直接使用的特征形式。包括去除关键字段缺失的数据、解析类别标签、提取上映年份以及补全数值字段等，同时为每部电影生成唯一索引，方便后续构建推荐模型时进行映射和训练。

print(f'Raw shape: {df.shape}') # 查看原始数据规模 # 删除关键字段缺失的数据（简介、标题、评分是核心信息） df = df.dropna(subset=['overview', 'title', 'vote_average']).reset_index(drop=True) # 安全解析字符串形式的列表（如 genre_ids） def safe_parse(x): try: return ast.literal_eval(x) if isinstance(x, str) else [] except: return [] df['genre_ids_parsed'] = df['genre_ids'].apply(safe_parse) # 解析电影类别 # 提取上映年份（时间特征） df['release_year'] = pd.to_datetime(df['release_date'], errors='coerce').dt.year.fillna(0).astype(int) # 对数值字段做缺失值填充 df['vote_count'] = df['vote_count'].fillna(0) df['popularity'] = df['popularity'].fillna(0) # 为每部电影创建唯一索引（后续Embedding或索引映射会用到） df['movie_idx'] = df.index # 数据规模信息 N_MOVIES = len(df) print(f'Clean shape : {df.shape}') print(f'Total movies: {N_MOVIES}')

4.3特征工程

这一部分主要完成内容侧特征的构建以及训练目标的定义。思路是把电影的文本信息、类别信息和结构化数值信息统一编码成一个向量表示，用于内容分支建模；同时基于评分和投票数构造一个隐式偏好分数，作为后续模型的学习目标，使模型既能理解内容相似性，又能捕捉受欢迎程度。

tfidf = TfidfVectorizer(max_features=5000, stop_words='english', ngram_range=(1, 2)) # 文本向量化（考虑1-2gram） tfidf_matrix = tfidf.fit_transform(df['overview'].fillna('')) # 将电影简介转为稀疏向量 svd = TruncatedSVD(n_components=100, random_state=42) # 降维，缓解高维稀疏问题 text_features = svd.fit_transform(tfidf_matrix) # (N, 100) 文本语义特征 mlb = MultiLabelBinarizer() genre_features = mlb.fit_transform(df['genre_ids_parsed']) # 多标签编码（每个类别一个维度） N_GENRES = genre_features.shape[1] scaler = MinMaxScaler() numeric_features = scaler.fit_transform( df[['vote_average', 'vote_count', 'popularity', 'release_year']] # 数值特征归一化 ) # (N, 4) content_features = np.hstack([text_features, genre_features, numeric_features]).astype(np.float32) # 拼接所有特征 CONTENT_DIM = content_features.shape[1] # 打印各部分特征维度 print(f'Text : {text_features.shape}') print(f'Genre : {genre_features.shape}') print(f'Numeric: {numeric_features.shape}') print(f'Total content dim: {CONTENT_DIM}')

# 构造隐式评分：综合评分和投票数（避免只看高分但样本少的情况） df['implicit_score'] = df['vote_average'] * np.log1p(df['vote_count']) # Normalize to [0, 1] 作为模型训练目标 s_min = df['implicit_score'].min() s_max = df['implicit_score'].max() df['target'] = (df['implicit_score'] - s_min) / (s_max - s_min) # 划分训练集和验证集 train_idx, val_idx = train_test_split(df.index.tolist(), test_size=0.2, random_state=42) # 数据分布信息 print(f'Train: {len(train_idx)} | Val: {len(val_idx)}') print(f'Target range: [{df["target"].min():.3f}, {df["target"].max():.3f}]')

4.4构建模型

这一部分完成混合推荐模型的核心构建，包括数据集封装、内容分支（ContentBranch）、协同过滤分支（CFBranch）以及二者的融合结构。整体思路是让模型一方面从电影内容特征中学习语义表示，另一方面通过Embedding学习电影之间的隐式关系，再通过融合层输出最终评分，同时引入对齐损失，使两个分支的表示空间更加一致。

class HybridMovieDataset(Dataset): def __init__(self, indices, content_features, targets): self.indices = indices self.content = torch.tensor(content_features, dtype=torch.float32) # 内容特征矩阵 self.targets = torch.tensor(targets, dtype=torch.float32) # 目标评分 def __len__(self): return len(self.indices) def __getitem__(self, i): idx = self.indices[i] return ( self.content[idx], # 内容特征向量 torch.tensor(idx, dtype=torch.long), # 电影ID（用于CF分支Embedding） self.targets[idx] # 目标值 ) targets = df['target'].values train_dataset = HybridMovieDataset(train_idx, content_features, targets) val_dataset = HybridMovieDataset(val_idx, content_features, targets) # 构建数据加载器 train_loader = DataLoader(train_dataset, batch_size=256, shuffle=True, drop_last=True) val_loader = DataLoader(val_dataset, batch_size=256, shuffle=False) # 查看一个batch数据结构 cv, ids, tgt = next(iter(train_loader)) class ContentBranch(nn.Module): def __init__(self, input_dim, embed_dim=64, dropout=0.3): super().__init__() # 多层全连接网络，将内容特征映射到低维embedding空间 self.encoder = nn.Sequential( nn.Linear(input_dim, 256), nn.LayerNorm(256), nn.GELU(), nn.Dropout(dropout), nn.Linear(256, 128), nn.LayerNorm(128), nn.GELU(), nn.Dropout(dropout), nn.Linear(128, embed_dim) ) def forward(self, x): return self.encoder(x) class CFBranch(nn.Module): def __init__(self, n_movies, embed_dim=64): super().__init__() # 为每个电影学习一个Embedding向量（协同过滤思想） self.item_emb = nn.Embedding(n_movies, embed_dim) nn.init.normal_(self.item_emb.weight, mean=0, std=0.01) # 初始化Embedding def forward(self, ids): return self.item_emb(ids) class HybridRecommender(nn.Module): def __init__(self, content_dim, n_movies, embed_dim=64, dropout=0.3): super().__init__() # 两个分支 self.content_branch = ContentBranch(content_dim, embed_dim, dropout) self.cf_branch = CFBranch(n_movies, embed_dim) # 融合层：拼接两个Embedding后进行预测 self.fusion = nn.Sequential( nn.Linear(embed_dim * 2, 128), nn.LayerNorm(128), nn.GELU(), nn.Dropout(dropout), nn.Linear(128, 64), nn.GELU(), nn.Linear(64, 1), nn.Sigmoid() # 输出归一化评分 ) def forward(self, content_vec, movie_ids): ce = self.content_branch(content_vec) # 内容Embedding cfe = self.cf_branch(movie_ids) # CF Embedding score = self.fusion(torch.cat([ce, cfe], dim=1)).squeeze(1) # 拼接后预测 return score, ce, cfe def get_content_emb(self, x): return self.content_branch(x) def get_cf_emb(self, ids): return self.cf_branch(ids) EMBED_DIM = 64 model = HybridRecommender(CONTENT_DIM, N_MOVIES, EMBED_DIM).to(device) # 统计可训练参数量 params = sum(p.numel() for p in model.parameters() if p.requires_grad) mse_loss = nn.MSELoss() # 主损失：回归误差 def alignment_loss(ce, cfe): # 计算两个分支Embedding的余弦相似度，鼓励表示一致 cn = F.normalize(ce, p=2, dim=1) cfn = F.normalize(cfe, p=2, dim=1) return (1 - (cn * cfn).sum(dim=1)).mean() def hybrid_loss(pred, target, ce, cfe, alpha=0.1): # 总损失 = MSE + 对齐损失 main = mse_loss(pred, target) align = alignment_loss(ce, cfe) return main + alpha * align, main.item(), align.item() # 优化器与学习率调度 optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30, eta_min=1e-5) print('Loss : MSE + 0.1 × alignment') print('Optimizer : AdamW lr=3e-4, wd=1e-4') print('Scheduler : CosineAnnealingLR T_max=30')

4.5模型训练

这一部分进入模型训练阶段，整体流程包括前向传播、损失计算、反向传播以及验证集评估，同时记录训练过程中的指标变化，并在验证集表现最优时保存模型参数。这里采用的是混合损失函数，一方面优化评分预测误差，另一方面约束两个分支的表示一致性，从而提升模型整体效果。

NUM_EPOCHS = 30 history = {'train': [], 'val': [], 'align': []} # 记录训练、验证和对齐损失 best_val = float('inf') # 用于保存最优验证损失 for epoch in range(1, NUM_EPOCHS + 1): # ── train ────────────────────────────────────────── model.train() # 训练模式 t_loss, a_loss = [], [] for cv, ids, tgt in train_loader: cv, ids, tgt = cv.to(device), ids.to(device), tgt.to(device) # 数据送入设备 pred, ce, cfe = model(cv, ids) # 前向传播 loss, ml, al = hybrid_loss(pred, tgt, ce, cfe) # 计算总损失（主损失+对齐损失） optimizer.zero_grad() # 梯度清零 loss.backward() # 反向传播 torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) # 梯度裁剪，防止梯度爆炸 optimizer.step() # 更新参数 t_loss.append(ml) # 记录主损失 a_loss.append(al) # 记录对齐损失 # ── validate ─────────────────────────────────────── model.eval() # 验证模式 v_loss = [] with torch.no_grad(): # 关闭梯度计算 for cv, ids, tgt in val_loader: cv, ids, tgt = cv.to(device), ids.to(device), tgt.to(device) pred, _, _ = model(cv, ids) # 仅关注预测结果 v_loss.append(mse_loss(pred, tgt).item()) # 验证集MSE # ── epoch stats ──────────────────────────────────── at, av, aa = np.mean(t_loss), np.mean(v_loss), np.mean(a_loss) # 计算平均损失 history['train'].append(at) history['val'].append(av) history['align'].append(aa) scheduler.step() # 更新学习率 # 保存验证集表现最好的模型 if av < best_val: best_val = av torch.save(model.state_dict(), 'best_hybrid.pt') # 每隔5轮输出一次训练信息 if epoch % 5 == 0 or epoch == 1: print(f'Ep {epoch:02d} | train={at:.5f} val={av:.5f} align={aa:.4f}') print(f'\nBest val loss: {best_val:.5f}')

4.6模型评估

这一部分主要用于对模型训练过程进行可视化分析，通过绘制训练集与验证集的损失曲线，以及两个分支之间的对齐损失变化情况，可以直观判断模型是否收敛、是否存在过拟合，以及内容分支与协同过滤分支之间的融合效果。

fig, axes = plt.subplots(1, 2, figsize=(14, 4)) e = range(1, NUM_EPOCHS + 1) # epoch范围 # 左图：训练集与验证集的MSE损失曲线 axes[0].plot(e, history['train'], label='Train MSE', color='steelblue', lw=2) axes[0].plot(e, history['val'], label='Val MSE', color='coral', lw=2, linestyle='--') axes[0].set_title('Prediction Loss (MSE)') axes[0].legend() axes[0].grid(alpha=0.3) # 右图：两个分支的对齐损失（余弦距离） axes[1].plot(e, history['align'], color='mediumseagreen', lw=2) axes[1].set_title('Branch Alignment Loss (1 − cos)') axes[1].grid(alpha=0.3) plt.tight_layout() plt.show()

4.7模型预测

这一部分主要是将训练好的模型用于实际推荐任务中，包括提取三种不同表示（内容向量、协同过滤向量、融合向量），并基于向量相似度实现电影推荐。同时通过简单的类别重合指标，对不同推荐模式的效果进行对比分析，从而验证混合模型的优势。

model.load_state_dict(torch.load('best_hybrid.pt', map_location=device)) # 加载最优模型 model.eval() all_ce, all_cfe = [], [] feat_t = torch.tensor(content_features, dtype=torch.float32) # 内容特征张量 ids_t = torch.arange(N_MOVIES, dtype=torch.long) # 电影ID序列 # 分批提取embedding，避免显存不足 with torch.no_grad(): for i in range(0, N_MOVIES, 512): cv = feat_t[i:i+512].to(device) ids = ids_t[i:i+512].to(device) all_ce.append(model.get_content_emb(cv).cpu()) # 内容向量 all_cfe.append(model.get_cf_emb(ids).cpu()) # CF向量 # 拼接并归一化 content_embs = F.normalize(torch.cat(all_ce), p=2, dim=1) # (N, 64) cf_embs = F.normalize(torch.cat(all_cfe), p=2, dim=1) # (N, 64) hybrid_embs = F.normalize(torch.cat([content_embs, cf_embs], dim=1), p=2, dim=1) # (N, 128) # 不同模式下的embedding映射 EMB_MAP = {'content': content_embs, 'cf': cf_embs, 'hybrid': hybrid_embs} def recommend(title, top_k=10, mode='hybrid'): embs = EMB_MAP[mode] # 选择embedding类型 # 精确匹配或模糊匹配电影标题 mask = df['title'].str.lower() == title.lower() if not mask.any(): mask = df['title'].str.lower().str.contains(title.lower(), na=False) if not mask.any(): print(f'Not found: {title}'); return pd.DataFrame() q_idx = df[mask].index[0] # 查询电影索引 # 计算余弦相似度（向量点积） sims = torch.mm(embs[q_idx].unsqueeze(0), embs.T).squeeze(0) scores, indices = sims.topk(top_k + 1) # 取top-k相似电影 rows = [] for s, i in zip(scores.tolist(), indices.tolist()): if i == q_idx: continue # 跳过自身 m = df.iloc[i] rows.append({ 'Title': m['title'], 'Year': int(m['release_year']), 'Lang': m['original_language'], 'Vote': round(m['vote_average'], 2), 'Sim': round(s, 4) }) if len(rows) == top_k: break return pd.DataFrame(rows) # 示例：不同模式下的推荐结果 for mode in ['content', 'cf', 'hybrid']: print(f'\n── {mode.upper()} ──') recs = recommend('Inception', top_k=5, mode=mode) print(recs[['Title', 'Year', 'Vote', 'Sim']].to_string(index=False))

def genre_overlap(title, top_k=10, mode='hybrid'): # 计算推荐结果中与目标电影类别的重合比例 mask = df['title'].str.lower() == title.lower() if not mask.any(): return None q_genres = set(df[mask].iloc[0]['genre_ids_parsed']) # 原电影类别 recs = recommend(title, top_k, mode) if recs.empty: return 0.0 hits = 0 for t in recs['Title']: m = df[df['title'] == t] if not m.empty and q_genres & set(m.iloc[0]['genre_ids_parsed']): hits += 1 return hits / top_k # 重合比例 # 测试不同模式推荐效果 test_movies = ['Inception', 'The Dark Knight', 'Parasite', 'Spirited Away'] print(f'{"Movie":<30} {"Mode":<10} {"Genre Overlap":>15}') print('-' * 58) for movie in test_movies: for mode in ['content', 'cf', 'hybrid']: go = genre_overlap(movie, top_k=10, mode=mode) if go is not None: print(f'{movie:<30} {mode:<10} {go:>14.1%}')

5.总结

本研究基于TMDB高评分电影数据构建了一个融合内容信息与协同信号的混合推荐模型，通过将文本语义特征、类型结构特征与影片隐式评分信息进行统一建模，有效缓解了单一推荐范式在信息利用上的局限。实验结果表明，模型在训练过程中收敛稳定，预测误差持续下降，同时通过引入分支对齐机制，使内容向量与协同向量在嵌入空间中实现一致性约束，从而提升了整体表征能力。在实际推荐效果上，混合模式相较于单一的Content或CF方式，在相似电影检索中表现出更好的语义相关性与类型一致性，能够在保证多样性的同时维持较高的匹配精度。整体来看，该模型在兼顾冷启动问题与用户偏好建模方面展现出较强的实用价值，为复杂场景下的个性化推荐提供了一种具有可扩展性的实现思路。

文末推荐

《AI绘画教程：AI绘画工具使用方法与技巧从入门到精通》

内容简介

本书以AI绘画技术提升现代设计效率为核心，通过对AI绘画现状的分析与讨论，详细地讲解了即时设计、Stable Diffusion、Photoshop AI、Liblib AI、海艺AI等主流AI绘画工具的特色、操作方法与绘画技巧。全书共分为7章，第1章为新手入门篇，介绍了AIGC的行业现状与相关知识；第2章为引导篇，介绍了当下AI绘画的主流工具；第3～7章为应用篇，为新生代设计师、AI画师、老牌设计师、轻办公体验人群推荐了不同的AI绘画工具，并且在介绍时辅以案例来讲解相关操作方法。本书语言通俗易懂，最大程度地将软件和平台的特色功能进行展示。本书适合艺术专业人士和AI绘画爱好者阅读。无论是否具备专业背景，读者都能从本书中得到关于AIGC时代下AI绘画应用的全面指导和创意灵感的启发。

京东购买链接：https://item.jd.com/14569327.html

源代码

import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import Dataset, DataLoader import pandas as pd import numpy as np import ast, warnings import matplotlib.pyplot as plt warnings.filterwarnings('ignore') from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.preprocessing import MinMaxScaler, MultiLabelBinarizer from sklearn.decomposition import TruncatedSVD from sklearn.model_selection import train_test_split device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') torch.manual_seed(42) np.random.seed(42) df = pd.read_csv('/kaggle/input/datasets/dhritisisodia/tmdb-top-10000-movies-dataset-2026/tmdb_top_10k_movies_2026.csv.csv') print(f'Raw shape: {df.shape}') df = df.dropna(subset=['overview', 'title', 'vote_average']).reset_index(drop=True) def safe_parse(x): try: return ast.literal_eval(x) if isinstance(x, str) else [] except: return [] df['genre_ids_parsed'] = df['genre_ids'].apply(safe_parse) df['release_year'] = pd.to_datetime(df['release_date'], errors='coerce').dt.year.fillna(0).astype(int) df['vote_count'] = df['vote_count'].fillna(0) df['popularity'] = df['popularity'].fillna(0) df['movie_idx'] = df.index N_MOVIES = len(df) print(f'Clean shape : {df.shape}') print(f'Total movies: {N_MOVIES}') # TF-IDF on overviews → reduced with LSA (Latent Semantic Analysis) tfidf = TfidfVectorizer(max_features=5000, stop_words='english', ngram_range=(1, 2)) tfidf_matrix = tfidf.fit_transform(df['overview'].fillna('')) svd = TruncatedSVD(n_components=100, random_state=42) text_features = svd.fit_transform(tfidf_matrix) # (N, 100) # Multi-hot genre encoding mlb = MultiLabelBinarizer() genre_features = mlb.fit_transform(df['genre_ids_parsed']) # (N, n_genres) N_GENRES = genre_features.shape[1] # Numeric features scaled to [0, 1] scaler = MinMaxScaler() numeric_features = scaler.fit_transform( df[['vote_average', 'vote_count', 'popularity', 'release_year']] ) # (N, 4) # Final concatenated content matrix content_features = np.hstack([text_features, genre_features, numeric_features]).astype(np.float32) CONTENT_DIM = content_features.shape[1] print(f'Text : {text_features.shape}') print(f'Genre : {genre_features.shape}') print(f'Numeric: {numeric_features.shape}') print(f'Total content dim: {CONTENT_DIM}') df['implicit_score'] = df['vote_average'] * np.log1p(df['vote_count']) # Normalize to [0, 1] s_min = df['implicit_score'].min() s_max = df['implicit_score'].max() df['target'] = (df['implicit_score'] - s_min) / (s_max - s_min) train_idx, val_idx = train_test_split(df.index.tolist(), test_size=0.2, random_state=42) print(f'Train: {len(train_idx)} | Val: {len(val_idx)}') print(f'Target range: [{df["target"].min():.3f}, {df["target"].max():.3f}]') class HybridMovieDataset(Dataset): def __init__(self, indices, content_features, targets): self.indices = indices self.content = torch.tensor(content_features, dtype=torch.float32) self.targets = torch.tensor(targets, dtype=torch.float32) def __len__(self): return len(self.indices) def __getitem__(self, i): idx = self.indices[i] return ( self.content[idx], torch.tensor(idx, dtype=torch.long), self.targets[idx] ) targets = df['target'].values train_dataset = HybridMovieDataset(train_idx, content_features, targets) val_dataset = HybridMovieDataset(val_idx, content_features, targets) train_loader = DataLoader(train_dataset, batch_size=256, shuffle=True, drop_last=True) val_loader = DataLoader(val_dataset, batch_size=256, shuffle=False) cv, ids, tgt = next(iter(train_loader)) class ContentBranch(nn.Module): def __init__(self, input_dim, embed_dim=64, dropout=0.3): super().__init__() self.encoder = nn.Sequential( nn.Linear(input_dim, 256), nn.LayerNorm(256), nn.GELU(), nn.Dropout(dropout), nn.Linear(256, 128), nn.LayerNorm(128), nn.GELU(), nn.Dropout(dropout), nn.Linear(128, embed_dim) ) def forward(self, x): return self.encoder(x) class CFBranch(nn.Module): def __init__(self, n_movies, embed_dim=64): super().__init__() self.item_emb = nn.Embedding(n_movies, embed_dim) nn.init.normal_(self.item_emb.weight, mean=0, std=0.01) def forward(self, ids): return self.item_emb(ids) class HybridRecommender(nn.Module): def __init__(self, content_dim, n_movies, embed_dim=64, dropout=0.3): super().__init__() self.content_branch = ContentBranch(content_dim, embed_dim, dropout) self.cf_branch = CFBranch(n_movies, embed_dim) self.fusion = nn.Sequential( nn.Linear(embed_dim * 2, 128), nn.LayerNorm(128), nn.GELU(), nn.Dropout(dropout), nn.Linear(128, 64), nn.GELU(), nn.Linear(64, 1), nn.Sigmoid() ) def forward(self, content_vec, movie_ids): ce = self.content_branch(content_vec) cfe = self.cf_branch(movie_ids) score = self.fusion(torch.cat([ce, cfe], dim=1)).squeeze(1) return score, ce, cfe def get_content_emb(self, x): return self.content_branch(x) def get_cf_emb(self, ids): return self.cf_branch(ids) EMBED_DIM = 64 model = HybridRecommender(CONTENT_DIM, N_MOVIES, EMBED_DIM).to(device) params = sum(p.numel() for p in model.parameters() if p.requires_grad) mse_loss = nn.MSELoss() def alignment_loss(ce, cfe): # Cosine similarity between the two branches — pushes them to align cn = F.normalize(ce, p=2, dim=1) cfn = F.normalize(cfe, p=2, dim=1) return (1 - (cn * cfn).sum(dim=1)).mean() def hybrid_loss(pred, target, ce, cfe, alpha=0.1): main = mse_loss(pred, target) align = alignment_loss(ce, cfe) return main + alpha * align, main.item(), align.item() optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4, weight_decay=1e-4) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30, eta_min=1e-5) print('Loss : MSE + 0.1 × alignment') print('Optimizer : AdamW lr=3e-4, wd=1e-4') print('Scheduler : CosineAnnealingLR T_max=30') NUM_EPOCHS = 30 history = {'train': [], 'val': [], 'align': []} best_val = float('inf') for epoch in range(1, NUM_EPOCHS + 1): # ── train ────────────────────────────────────────── model.train() t_loss, a_loss = [], [] for cv, ids, tgt in train_loader: cv, ids, tgt = cv.to(device), ids.to(device), tgt.to(device) pred, ce, cfe = model(cv, ids) loss, ml, al = hybrid_loss(pred, tgt, ce, cfe) optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() t_loss.append(ml) a_loss.append(al) # ── validate ─────────────────────────────────────── model.eval() v_loss = [] with torch.no_grad(): for cv, ids, tgt in val_loader: cv, ids, tgt = cv.to(device), ids.to(device), tgt.to(device) pred, _, _ = model(cv, ids) v_loss.append(mse_loss(pred, tgt).item()) # ── epoch stats ──────────────────────────────────── at, av, aa = np.mean(t_loss), np.mean(v_loss), np.mean(a_loss) history['train'].append(at) history['val'].append(av) history['align'].append(aa) scheduler.step() if av < best_val: best_val = av torch.save(model.state_dict(), 'best_hybrid.pt') if epoch % 5 == 0 or epoch == 1: print(f'Ep {epoch:02d} | train={at:.5f} val={av:.5f} align={aa:.4f}') print(f'\nBest val loss: {best_val:.5f}') fig, axes = plt.subplots(1, 2, figsize=(14, 4)) e = range(1, NUM_EPOCHS + 1) axes[0].plot(e, history['train'], label='Train MSE', color='steelblue', lw=2) axes[0].plot(e, history['val'], label='Val MSE', color='coral', lw=2, linestyle='--') axes[0].set_title('Prediction Loss (MSE)'); axes[0].legend(); axes[0].grid(alpha=0.3) axes[1].plot(e, history['align'], color='mediumseagreen', lw=2) axes[1].set_title('Branch Alignment Loss (1 − cos)'); axes[1].grid(alpha=0.3) plt.tight_layout() plt.show() model.load_state_dict(torch.load('best_hybrid.pt', map_location=device)) model.eval() all_ce, all_cfe = [], [] feat_t = torch.tensor(content_features, dtype=torch.float32) ids_t = torch.arange(N_MOVIES, dtype=torch.long) with torch.no_grad(): for i in range(0, N_MOVIES, 512): cv = feat_t[i:i+512].to(device) ids = ids_t[i:i+512].to(device) all_ce.append(model.get_content_emb(cv).cpu()) all_cfe.append(model.get_cf_emb(ids).cpu()) content_embs = F.normalize(torch.cat(all_ce), p=2, dim=1) # (N, 64) cf_embs = F.normalize(torch.cat(all_cfe), p=2, dim=1) # (N, 64) hybrid_embs = F.normalize(torch.cat([content_embs, cf_embs], dim=1), p=2, dim=1) # (N, 128) EMB_MAP = {'content': content_embs, 'cf': cf_embs, 'hybrid': hybrid_embs} def recommend(title, top_k=10, mode='hybrid'): embs = EMB_MAP[mode] mask = df['title'].str.lower() == title.lower() if not mask.any(): mask = df['title'].str.lower().str.contains(title.lower(), na=False) if not mask.any(): print(f'Not found: {title}'); return pd.DataFrame() q_idx = df[mask].index[0] sims = torch.mm(embs[q_idx].unsqueeze(0), embs.T).squeeze(0) scores, indices = sims.topk(top_k + 1) rows = [] for s, i in zip(scores.tolist(), indices.tolist()): if i == q_idx: continue m = df.iloc[i] rows.append({'Title': m['title'], 'Year': int(m['release_year']), 'Lang': m['original_language'], 'Vote': round(m['vote_average'], 2), 'Sim': round(s, 4)}) if len(rows) == top_k: break return pd.DataFrame(rows) for mode in ['content', 'cf', 'hybrid']: print(f'\n── {mode.upper()} ──') recs = recommend('Inception', top_k=5, mode=mode) print(recs[['Title', 'Year', 'Vote', 'Sim']].to_string(index=False)) def genre_overlap(title, top_k=10, mode='hybrid'): mask = df['title'].str.lower() == title.lower() if not mask.any(): return None q_genres = set(df[mask].iloc[0]['genre_ids_parsed']) recs = recommend(title, top_k, mode) if recs.empty: return 0.0 hits = 0 for t in recs['Title']: m = df[df['title'] == t] if not m.empty and q_genres & set(m.iloc[0]['genre_ids_parsed']): hits += 1 return hits / top_k test_movies = ['Inception', 'The Dark Knight', 'Parasite', 'Spirited Away'] print(f'{"Movie":<30} {"Mode":<10} {"Genre Overlap":>15}') print('-' * 58) for movie in test_movies: for mode in ['content', 'cf', 'hybrid']: go = genre_overlap(movie, top_k=10, mode=mode) if go is not None: print(f'{movie:<30} {mode:<10} {go:>14.1%}')

资料获取，更多粉丝福利，关注下方公众号获取