当前位置： 首页 > article >正文

什么是双塔模型？

article 2025/3/1 22:32:08

一、概念

        双塔模型（Two-Tower Model）在深度学习领域，尤其是在推荐系统和信息检索等场景中，是一种常见的模型架构。要了解双塔模型，首先得了解来自2013年被CIKM收录的微软论文《Learning Deep Structured Semantic Models for Web Search using Clickthrough Data》，该论文通过获取搜索引擎中的用户搜索Query与Documents的曝光和日志数据，训练了一个高性能的语义相似模型。而论文中提出的DSSM（Deep Structured Semantic Model）模型后来被简化为各类query和doc/item双塔结构，由此诞生了推荐领域最为著名的双塔系列模型。

0c901d83c6f54b57a4d01fd2d03f8473.png

        双塔模型的核心思想是将输入数据分成两个独立的部分，并通过两个独立的神经网络（即两个塔，塔的结构可以不一样，但应当保证输出向量维度严格一致）分别处理这两个部分，然后在某个层次上将这两个部分数据的语义表示进行融合，以完成最终的任务。例如，在推荐系统中，我们可以用一个塔处理用户数据（如用户特征和行为），另一个塔处理物品数据（如物品特征和属性），这两个塔分别学习输入数据的特征表示，最后通过余弦相似度或点积等方式计算两者之间的匹配分数。这样的机制使得双塔模型可以根据用户的历史信息以及商品的属性实现个性化的商品推荐。

二、原理

        下面以信息推荐任务为例介绍双塔模型的基本原理，下图来自微软2015年的论文《A Multi-View Deep Learning Approach for Cross Domain User Modeling in Recommendation Systems》，讲述了如何将双塔模型应用于新闻推荐。

4358050c1e0246f48a29c93965a41f83.png

  • 输入层：双塔模型接收两组输入，分别是用户特征和新闻特征，这些特征可以包括静态属性和动态行为数据。
  • 特征表示：每个塔将各自的输入特征转化为低维稠密向量（Embedding），这一过程通过深度神经网络来实现。
  • 匹配层：在共享的语义空间中，通过相似性计算（如余弦相似度或点积）得到用户与新闻文本的匹配分数。
  • 损失函数：双塔模型常用对比损失（Contrastive Loss）或交叉熵损失，优化正样本匹配分数高、负样本匹配分数低的目标。

        可以看到，双塔模型的结构及实现非常简单直观，但是在搜广推领域这样的模型架构却非常实用。当然，实际的业务应用中建模要考虑的因素要更为复杂，这其中还有许多可以优化的地方。

三、python实现

        这里，我们构建一个简单的随机数据集，并搭建一个双塔模型。

1、导入必要的库

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader

2、构建数据集类

# 定义用户和物品的特征
class UserItemDataset(Dataset):
    def __init__(self, user_features, item_features, labels):
        # 实际的业务数据需要在此处增加数据处理的过程
        self.user_features = user_features
        self.item_features = item_features
        self.labels = labels

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, index):
        return self.user_features[index], self.item_features[index], self.labels[index]

3、构建模型类

# 定义双塔模型
class TwoTowerModel(nn.Module):
    def __init__(self, user_feature_dim, item_feature_dim, hidden_dim):
        super(TwoTowerModel, self).__init__()
        # 用户塔
        self.user_tower = nn.Sequential(
            nn.Linear(user_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )

        # item塔，这里跟用户塔使用相同的结构，实际可以采用不同的结构，但必须保证输出维度相同
        self.item_tower = nn.Sequential(
            nn.Linear(item_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        self.similarity = nn.CosineSimilarity(dim=1)

    def forward(self, user_features, item_features):
        # 分别提取用户信息和item信息的语义特征
        user_repr = self.user_tower(user_features)
        item_repr = self.item_tower(item_features)
        # 计算语义相似度
        similarity_score = self.similarity(user_repr, item_repr)
        return similarity_score

4、模型初始化及参数设置

# 随机生成的用户和物品特征以及标签，可以替换成实际的数据集
user_features = torch.randn(100, 10)  # 100个用户，每人10个特征
item_features = torch.randn(100, 20)  # 100个物品，每物20个特征
labels = torch.randint(0, 2, (100,))  # 100个标签，0或1，0代表信息不匹配，1代表信息匹配

# 创建数据集和数据加载器
dataset = UserItemDataset(user_features, item_features, labels)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

# 初始化模型、损失函数和优化器
device = torch.device('cuda') if torch.cuda.is_available() else 'cpu'
model = TwoTowerModel(user_feature_dim=10, item_feature_dim=20, hidden_dim=16)
model = model.to(device)
# 这里我们使用二分类交叉熵损失
criterion = nn.BCEWithLogitsLoss()
criterion = criterion.to(device)
optimizer = optim.Adam(model.parameters(), lr=0.001)

5、模型训练

# 训练模型
num_epochs = 5
for epoch in range(num_epochs):
    for user_feature, item_feature, label in dataloader:
        user_feature, item_feature, label = user_feature.to(device), item_feature.to(device), label.to(device)
        optimizer.zero_grad()
        output = model(user_feature, item_feature).squeeze()
        loss = criterion(output, label.float())
        loss.backward()
        optimizer.step()
    print(f'Epoch {epoch+1}, Loss: {loss.item()}')

四、完整代码

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader

# 定义用户和物品的特征
class UserItemDataset(Dataset):
    def __init__(self, user_features, item_features, labels):
        self.user_features = user_features
        self.item_features = item_features
        self.labels = labels

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, index):
        return self.user_features[index], self.item_features[index], self.labels[index]

# 定义双塔模型
class TwoTowerModel(nn.Module):
    def __init__(self, user_feature_dim, item_feature_dim, hidden_dim):
        super(TwoTowerModel, self).__init__()
        self.user_tower = nn.Sequential(
            nn.Linear(user_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        self.item_tower = nn.Sequential(
            nn.Linear(item_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim)
        )
        self.similarity = nn.CosineSimilarity(dim=1)

    def forward(self, user_features, item_features):
        user_repr = self.user_tower(user_features)
        item_repr = self.item_tower(item_features)
        similarity_score = self.similarity(user_repr, item_repr)
        return similarity_score

# 随机生成的用户和物品特征以及标签
user_features = torch.randn(100, 10)  # 100个用户，每人10个特征
item_features = torch.randn(100, 20)  # 100个物品，每物20个特征
labels = torch.randint(0, 2, (100,))  # 100个标签，0或1

# 创建数据集和数据加载器
dataset = UserItemDataset(user_features, item_features, labels)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

# 初始化模型、损失函数和优化器
device = torch.device('cuda') if torch.cuda.is_available() else 'cpu'
model = TwoTowerModel(user_feature_dim=10, item_feature_dim=20, hidden_dim=16)
model = model.to(device)
criterion = nn.BCEWithLogitsLoss()
criterion = criterion.to(device)
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练模型
num_epochs = 5
for epoch in range(num_epochs):
    for user_feature, item_feature, label in dataloader:
        user_feature, item_feature, label = user_feature.to(device), item_feature.to(device), label.to(device)
        optimizer.zero_grad()
        output = model(user_feature, item_feature).squeeze()
        loss = criterion(output, label.float())
        loss.backward()
        optimizer.step()
    print(f'Epoch {epoch+1}, Loss: {loss.item()}')

五、总结

        本文介绍了双塔模型的基本概念，并使用随机构建的数据集搭建了一个Torch版本的双塔模型。双塔模型因其结构简单、易于实现、能够处理大规模稀疏特征等优点，在工业界得到了广泛的应用。但是传统的双塔模型也存在着对新用户或新物品效果较差的冷启动问题，且这种双塔结构无法建模用户和物品之间的深层交叉信息，因此后面也有许多研究针对这些不足不断推出优化的模型。

 


http://www.kler.cn/a/458564.html

相关文章：

  • Linux环境Docker安装Postgres
  • kubernetes学习-集群搭建部署（一）
  • React 中 createContext 和 useContext 的深度应用与优化实战
  • SOME/IP 协议详解——序列化
  • 黑马Java面试教程_P2_MySQL
  • 【Linux】：线程安全 + 死锁问题
  • Cadence学习笔记 13 结构件导入与器件定位
  • 闪极科技，何以抢先进入AI眼镜产业化元年？
  • 聊聊 Mongod 以及 MongoDB 常用命令
  • AI Agent 与 AI Workflow 的区别和深度解析：从自动化到智能化的演进
  • 37 Opencv SIFT 特征检测
  • CTFHUB-web进阶-php
  • 从 ELK Stack 到简单 — Elastic Cloud Serverless 上的 Elastic 可观察性
  • 数学建模两篇小文
  • Unity中列表List使用出类似字典Dictionary的感觉
  • springcloud篇2-feign、gateway
  • 大数据-260 实时数仓 - 项目背景与需求 实时数仓架构 需求分析 技术选型 逻辑架构
  • MicroDiffusion——采用新的掩码方法和改进的 Transformer 架构，实现了低预算的扩散模型
  • 神经网络-VggNet
  • 刷新页面一次，错误地进行了多次重复调用后端服务