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

从零构建短视频推荐系统：双塔算法架构解析与代码实现

news 2025/9/29 9:06:55

刷短视频本来只想看几分钟，不知不觉一个多小时就没了。每条视频都恰好戳中你的兴趣点，这种精准推送背后其实是一套相当复杂的工程架构。

这种"读心术"般的推荐效果并非偶然。驱动这种短视频页面的核心引擎，正是业内广泛采用的双塔推荐系统（Two-Tower Recommendation System）。

本文将从技术角度剖析：双塔架构的工作原理、为何在短视频场景下表现卓越，以及如何构建一套类似的推荐系统。

推荐系统：注意力经济的核心武器

注意力经济时代，个性化推荐已经成为平台的基本能力。传统的"热门榜单"模式早已过时——因为用户很快就会感到内容乏味，并且推荐内容单一，无法吸引用户最终流失。

而短视频的成功在于能够预判用户需求，这也就是说为什么推荐系统成为当今最具商业价值的 AI 应用之一：

Netflix 通过推荐决定你下一部追的剧，YouTube 用算法填满你的首页和 Shorts 流，Amazon 靠推荐驱动购买决策，Spotify 的 Discover Weekly 帮你发现新音乐。

但论推荐效果，TikTok/抖音应该说是做到了极致。除了响应速度极快，个性化程度还极高。

双塔架构：两个"大脑"的协作机制

双塔系统本质上是两个独立但协调的神经网络模块：

**用户塔（User Tower）**专门建模用户特征——偏好习惯、行为模式、上下文信息等等。**物品塔（Item Tower）**则负责理解内容特征——视频属性、创作者风格、话题标签等。

这种设计的巧妙之处在于将复杂的用户-内容匹配问题，转化为两个向量空间的相似度计算。

技术实现细节拆解

用户塔的特征工程

用户塔的任务是将用户的所有行为数据压缩成一个稠密向量表示，这个向量可以理解为用户的"数字指纹"。

输入特征通常包括：历史观看记录及停留时长、互动行为（点赞、分享、评论、关注），时间和地理位置等上下文信息，以及用户基础属性（年龄、性别等，如果可获取）。

这些原始特征经过 embedding 层和多层神经网络处理，最终输出一个固定维度的向量。这个向量在某种程度上"记住"了用户的兴趣偏好。

物品塔的内容理解

物品塔负责将每条物品转换为相应的向量表示。

以视频为例视频特征相对更加多元化：基础 ID 信息（视频 ID、创作者 ID），内容标签和话题分类，音频特征（BGM、音效等），文本信息（标题、字幕、ASR 转录文本），视觉特征（通常使用预训练的 CNN 模型提取关键帧或缩略图的特征向量）。

同样经过神经网络处理后，每条视频都得到一个与用户向量处于同一语义空间的表示向量。

向量匹配与相似度计算

两个塔产生的向量如何进行匹配？最直接的方法是计算向量间的点积或余弦相似度。相似度越高，表明用户对该视频的潜在兴趣越大。

但是面对百万级别的视频库，逐一计算显然不现实。所以就需要用到近似最近邻搜索（ANN）技术，比如 Facebook 开源的 Faiss 或者 Google 的 ScaNN。这些工具能够在毫秒级时间内从海量向量中找出最相似的候选集合。

从启动到推荐：完整的数据流

当用户打开短视频时，后台的推荐流程大致如下：

首先实时计算或获取用户的最新特征向量。接着通过 ANN 检索从全量视频库中召回几千个潜在候选。然后使用更复杂的排序模型对候选视频进行精排，考虑预期观看时长、互动概率等多个目标。

在内容输出前还会应用各种业务规则：内容去重避免相似视频连续出现，新内容扶持保证流量分配的公平性，安全审核过滤违规内容。最后将排序好的视频推送到用户端，整个过程只需要几十毫秒以内就可以完成。

双塔模型的代码实现

想要实现一个简化版的双塔模型，可以参考以下代码框架。

数据准备阶段

  # Example log data (user_id, video_id, watch_time, liked)  logs = [  ("user1", "video123", 5.2, True),  ("user1", "video456", 1.1, False),  ("user2", "video789", 8.5, True),  ]

特征存储系统

实际生产环境中需要构建实时特征存储系统。用户特征可能包括不同类别内容的观看统计、活跃时段分布等。视频特征则涵盖分类标签、互动指标聚合值等。

特征存储的核心是保证训练和推理阶段特征的一致性，同时满足低延迟的在线服务需求。

模型训练

双塔模型采用对比学习的训练方式。正样本来自真实的用户-视频交互记录，负样本则通过随机采样生成。训练目标是让正样本对应的用户向量和视频向量尽可能接近，负样本对应的向量尽可能远离。

常用的损失函数包括 InfoNCE，配合 in-batch negatives 策略能够有效提升训练效率。

 import tensorflow as tf  
from tensorflow.keras.layers import Input, Dense, Embedding, Dot  # User Tower  
user_input = Input(shape=(1,), name="user_id")  
user_embedding = Embedding(num_users, 64)(user_input)  
user_vec = Dense(32, activation="relu")(user_embedding)  # Item Tower  
item_input = Input(shape=(1,), name="item_id")  
item_embedding = Embedding(num_items, 64)(item_input)  
item_vec = Dense(32, activation="relu")(item_embedding)  # Dot Product (Interaction)  
dot_product = Dot(axes=1)([user_vec, item_vec])  model = tf.keras.Model(inputs=[user_input, item_input], outputs=dot_product)  
model.compile(optimizer="adam", loss="binary_crossentropy")  model.fit([user_ids, item_ids], labels, epochs=10)

向量索引构建

模型训练完成后，需要为所有视频预计算向量表示，并构建 ANN 索引以支持快速检索。

 import faiss  # Store all item embeddings in Faiss for fast search  
item_embeddings = model.get_layer("item_embedding").get_weights()[0]  
index = faiss.IndexFlatIP(32)  # Inner Product search  
index.add(item_embeddings)  # Get recommendations for a user  
user_embedding = user_model.predict(["user123"])  _, recommended_ids = index.search(user_embedding, k=10)  # Top 10 videos

线上服务部署

生产环境的推荐服务需要处理实时用户请求。系统接收用户特征后，调用用户塔计算向量表示，然后查询 ANN 索引获取候选集，最后经过排序模型输出最终推荐结果。

整个链路的延迟控制是关键指标，通常要求在 100 毫秒以内完成。

 import tensorflow as tf  # A simple two-tower model using user IDs and item IDs  
class TwoTowerModel(tf.keras.Model):  def __init__(self, user_vocab_size, item_vocab_size, embed_dim):  super().__init__()  # Embedding layers for users and items (learnable lookup tables)  self.user_embedding = tf.keras.layers.Embedding(input_dim=user_vocab_size, output_dim=embed_dim)  self.item_embedding = tf.keras.layers.Embedding(input_dim=item_vocab_size, output_dim=embed_dim)  def call(self, inputs):  user_id, item_id = inputs  # expecting user and item IDs as inputs  # Get embeddings for user and item  user_vec = self.user_embedding(user_id)    # shape: [batch_size, embed_dim]  item_vec = self.item_embedding(item_id)    # shape: [batch_size, embed_dim]  # Compute dot product similarity (and squeeze to 1D)  score = tf.reduce_sum(user_vec * item_vec, axis=1)  return score  # Example usage:  
model = TwoTowerModel(user_vocab_size=10000, item_vocab_size=50000, embed_dim=32)  
user_ids = tf.constant([123])  # a sample user ID  
item_ids = tf.constant([456])  # a sample item ID  
predicted_score = model((user_ids, item_ids)).numpy()  print("Predicted score:", predicted_score)