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

import torchimport torch.nn as nnimport torch.nn.functional as Ffrom torch.utils.data import Dataset, DataLoaderimport pandas as pdimport numpy as npimport ast, warningsimport matplotlib.pyplot as pltwarnings.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

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)ifisinstance(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}')

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}]')

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'].valuestrain_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 = 64model = 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')

NUM_EPOCHS = 30history    = {'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 == 0or 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)  # 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()

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()    ifnot mask.any():        mask = df['title'].str.lower().str.contains(title.lower(), na=False)    ifnot 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()    ifnot mask.any(): return None    q_genres = set(df[mask].iloc[0]['genre_ids_parsed'])  # 原电影类别    recs = recommend(title, top_k, mode)    if recs.empty: return0.0    hits = 0    for t in recs['Title']:        m = df[df['title'] == t]        ifnot 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%}')