双塔模型实战：从零构建一个工业级推荐召回引擎

最近和几个做后端和算法的朋友聊天，发现一个挺有意思的现象：很多工程师对推荐系统的排序模型如数家珍，各种复杂的CTR、CVR模型都能聊上几句，但一提到召回，特别是如何从零搭建一个高效、可落地的召回模块，大家往往就有点犯怵。要么觉得“不就是算个相似度嘛”，要么被各种论文里的数学公式吓退。其实，双塔模型作为召回阶段的基石，其工程实现远比想象中更有挑战，也更有趣。

今天，我就结合自己最近在一个内容推荐项目中的实战经验，抛开那些冗长的理论推导，用TensorFlow 2.x带你一步步搭建一个真正能跑起来、能上线的双塔召回模型。我们会重点关注那些论文里不会写、但实践中能让你少踩80%坑的细节：从特征工程怎么处理稀疏特征和序列特征，到模型架构如何平衡效果与性能，再到负采样策略到底该怎么选，最后还会给出一个完整的、可直接在Colab或本地运行的代码框架。无论你是想深入了解推荐系统内部机制的后端开发，还是希望将算法模型真正产品化的算法工程师，这篇文章都能给你带来一些直接的启发。

1. 工程化思维：重新理解双塔模型的价值与挑战

在开始写代码之前，我们得先统一思想：为什么在深度学习模型层出不穷的今天，结构看似“简单”的双塔模型依然是工业界召回系统的首选？

核心优势在于极致的线上效率。召回阶段面对的是百万甚至亿级的候选物品库，必须在几十毫秒内完成筛选。双塔模型巧妙地将计算压力“离线化”：物品塔（Item Tower）可以预先计算好所有物品的向量并存入向量数据库（如Faiss、Milvus）；线上请求到来时，只需要实时计算用户塔（User Tower）的向量，然后进行一次快速的近似最近邻搜索（ANN）即可。这种“离线算物，在线算人”的模式，是它能支撑海量并发请求的关键。

但“简单”只是表象，其工程挑战暗藏于细节之中：

• 特征表达瓶颈：用户和物品的特征在塔内完全独立，无法进行精细的特征交叉（如“年轻女性”与“美妆教程”的关联），这限制了模型的上限。
• 负样本的“艺术”：召回的本质是从海量未曝光物品中找出用户可能喜欢的，如何构造负样本，直接影响模型是“精准”还是“偏颇”。
• 服务化与更新：如何设计模型服务架构，以支持物品向量的实时/近实时更新，应对新闻、短视频等强时效性场景？

理解了这些，我们的实战就不再是机械地搭网络，而是带着问题去寻找解决方案。接下来，我们从最基础的环境与数据准备开始。

2. 实战第一步：构建可复现的数据管道与特征工程

很多教程把重点放在模型代码上，但我的经验是，数据管道和特征工程决定了项目50%的成功率。一个混乱的数据流会让后续的模型迭代和调试变成噩梦。

2.1 环境与数据模拟

我们使用TensorFlow 2.x和TensorFlow Recommenders (TFRS)库，后者封装了许多推荐系统的标准组件，能极大提升开发效率。

# 建议的依赖环境
pip install tensorflow==2.10.0
pip install tensorflow-recommenders
pip install pandas numpy

对于实战演示，我们模拟一个视频推荐场景的数据集。这里的关键是理解数据应有的结构。

import tensorflow as tf
import pandas as pd
import numpy as np

# 模拟生成用户和物品数据
def generate_mock_data(num_users=10000, num_items=50000, num_interactions=200000):
    np.random.seed(42)
    # 用户基础特征
    user_ids = [f"user_{i}" for i in range(num_users)]
    user_genders = np.random.choice(['M', 'F'], num_users)
    user_ages = np.random.randint(18, 60, num_users)

    # 物品基础特征
    item_ids = [f"item_{i}" for i in range(num_items)]
    item_categories = np.random.choice(['科技', '娱乐', '体育', '教育', '生活'], num_items)
    item_duration = np.random.exponential(300, num_items).astype(int)  # 视频时长

    # 生成交互记录（正样本）
    interactions = []
    for _ in range(num_interactions):
        u = np.random.randint(0, num_users)
        i = np.random.randint(0, num_items)
        # 让交互有一定模式，而非完全随机
        if (user_genders[u] == 'F' and item_categories[i] == '娱乐') or (user_ages[u] < 30 and item_categories[i] == '科技'):
            watch_ratio = np.random.beta(5, 2)  # 观看完成度，Beta分布模拟用户行为
        else:
            watch_ratio = np.random.beta(2, 5)
        interactions.append((user_ids[u], item_ids[i], float(watch_ratio > 0.7)))  # 观看超过70%记为点击

    return pd.DataFrame(interactions, columns=['user_id', 'item_id', 'clicked']), user_ids, item_ids

# 生成并查看数据
interactions_df, user_list, item_list = generate_mock_data()
print(interactions_df.head())
print(f"总交互数: {len(interactions_df)}， 点击率: {interactions_df['clicked'].mean():.3f}")

2.2 特征工程：超越One-Hot的稀疏特征处理

原始ID类特征（user_id, item_id）通常是高维稀疏的。直接One-Hot会导致维度爆炸，因此Embedding是标准操作。但实践中，我们常遇到更多类型的特征：

特征类型	示例	处理方法	TFRS中的实现要点
类别型特征	用户性别、物品分类	Embedding	使用tf.keras.layers.StringLookup进行词汇表映射，再接Embedding层。
数值型特征	用户年龄、物品时长	直接输入或分桶	对于分布不均匀的特征（如时长），建议先进行对数变换或分桶，再作为类别特征处理。
序列型特征	用户最近点击的10个物品ID	序列Embedding + Pooling	使用tf.keras.layers.Embedding后，接tf.keras.layers.GlobalAveragePooling1D或注意力池化。
多值类别特征	物品标签（如[“Python”, “算法”, “深度学习”]）	多值Embedding + Pooling	处理为RaggedTensor，Embedding后同样进行池化操作。

注意：对于用户侧和物品侧都有的特征（例如“分类”），务必使用不同的Embedding层。用户塔和物品塔的Embedding层参数是独立学习的，即使特征名相同，其语义空间也不同。

下面是一个更贴近实战的特征处理模块示例：

import tensorflow as tf
import tensorflow_recommenders as tfrs

def create_feature_engineering_layers(user_ids, item_ids, item_categories):
    """创建特征处理的预处理层"""
    # 用户ID处理层
    user_id_vocab = tf.keras.layers.StringLookup(vocabulary=user_ids, mask_token=None)
    user_id_embedding = tf.keras.layers.Embedding(input_dim=user_id_vocab.vocabulary_size(), output_dim=32)

    # 物品ID处理层
    item_id_vocab = tf.keras.layers.StringLookup(vocabulary=item_ids, mask_token=None)
    item_id_embedding = tf.keras.layers.Embedding(input_dim=item_id_vocab.vocabulary_size(), output_dim=32)

    # 物品分类处理层（示例）
    item_category_vocab = tf.keras.layers.StringLookup(vocabulary=np.unique(item_categories), mask_token=None)
    item_category_embedding = tf.keras.layers.Embedding(input_dim=item_category_vocab.vocabulary_size(), output_dim=16)

    return {
        "user_id_vocab": user_id_vocab,
        "user_id_embedding": user_id_embedding,
        "item_id_vocab": item_id_vocab,
        "item_id_embedding": item_id_embedding,
        "item_category_vocab": item_category_vocab,
        "item_category_embedding": item_category_embedding,
    }

# 假设我们有这些数据
feature_layers = create_feature_engineering_layers(user_list, item_list, ['科技', '娱乐', '体育', '教育', '生活'])

3. 模型架构设计：构建灵活高效的双塔

有了特征处理层，我们就可以搭建模型本体了。这里的目标是构建一个模块化、易扩展的双塔模型。

3.1 用户塔与物品塔的实现

每个塔本质上是一个多层感知机，但输入是经过预处理和Embedding后的特征拼接。

class UserModel(tf.keras.Model):
    """用户塔模型"""
    def __init__(self, feature_layers):
        super().__init__()
        self.user_id_vocab = feature_layers["user_id_vocab"]
        self.user_id_embedding = feature_layers["user_id_embedding"]

        # 假设我们还有用户年龄（数值型）和性别（类别型）特征
        self.gender_vocab = tf.keras.layers.StringLookup(vocabulary=['M', 'F'], mask_token=None)
        self.gender_embedding = tf.keras.layers.Embedding(input_dim=self.gender_vocab.vocabulary_size(), output_dim=8)

        # 定义塔身结构
        self.dense_layers = tf.keras.Sequential([
            tf.keras.layers.Dense(128, activation="relu"),
            tf.keras.layers.Dropout(0.2),  # 防止过拟合
            tf.keras.layers.Dense(64, activation="relu"),
            # 输出层不需要激活函数，因为后面会计算相似度
            tf.keras.layers.Dense(32)
        ])

    def call(self, inputs):
        # inputs 是一个字典，例如 {'user_id': ['user_1', 'user_2'], 'gender': ['M', 'F'], 'age': [25, 30]}
        user_id_encoded = self.user_id_vocab(inputs["user_id"])
        user_embedding = self.user_id_embedding(user_id_encoded)

        gender_encoded = self.gender_vocab(inputs["gender"])
        gender_embedding = self.gender_embedding(gender_encoded)

        # 处理数值特征：年龄，这里简单做归一化（假设年龄范围18-60）
        age_normalized = (tf.cast(inputs["age"], tf.float32) - 18.0) / (60.0 - 18.0)
        age_embedding = tf.expand_dims(age_normalized, axis=1)  # 变成 [batch_size, 1]

        # 拼接所有特征向量
        concatenated = tf.concat([user_embedding, gender_embedding, age_embedding], axis=1)

        # 通过DNN塔身
        return self.dense_layers(concatenated)

class ItemModel(tf.keras.Model):
    """物品塔模型，结构与用户塔类似但独立"""
    def __init__(self, feature_layers):
        super().__init__()
        self.item_id_vocab = feature_layers["item_id_vocab"]
        self.item_id_embedding = feature_layers["item_id_embedding"]
        self.item_category_vocab = feature_layers["item_category_vocab"]
        self.item_category_embedding = feature_layers["item_category_embedding"]

        self.dense_layers = tf.keras.Sequential([
            tf.keras.layers.Dense(128, activation="relu"),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(64, activation="relu"),
            tf.keras.layers.Dense(32)
        ])

    def call(self, inputs):
        # inputs 例如 {'item_id': ['item_1', 'item_2'], 'category': ['科技', '娱乐']}
        item_id_encoded = self.item_id_vocab(inputs["item_id"])
        item_embedding = self.item_id_embedding(item_id_encoded)

        category_encoded = self.item_category_vocab(inputs["category"])
        category_embedding = self.item_category_embedding(category_encoded)

        concatenated = tf.concat([item_embedding, category_embedding], axis=1)
        return self.dense_layers(concatenated)

3.2 组合成完整的双塔模型

使用TFRS可以非常方便地定义任务和模型。

class TwoTowerModel(tfrs.models.Model):
    """完整的双塔召回模型"""
    def __init__(self, user_model, item_model, task):
        super().__init__()
        self.user_model = user_model
        self.item_model = item_model
        self.task = task  # 检索任务，定义了损失函数和评估指标

    def compute_loss(self, features, training=False):
        user_embeddings = self.user_model(features)
        item_embeddings = self.item_model(features)

        # 计算损失和指标
        return self.task(user_embeddings, item_embeddings)

    def call(self, inputs):
        # 此方法用于推理，返回用户向量和物品向量
        user_embeddings = self.user_model(inputs)
        item_embeddings = self.item_model(inputs)
        return user_embeddings, item_embeddings

4. 模型训练的核心：负采样策略深度解析

这是双塔模型训练中最关键、也最容易被忽视的一环。不同的负采样策略，直接导向不同的模型行为。

4.1 常见负采样策略对比

我们不再使用简单的“曝光未点击”，而是探讨更适用于召回场景的策略。

策略	描述	优点	缺点	适用场景
随机负采样	从全量物品池中随机抽取物品作为负样本。	实现简单，能让模型看到全量分布。	负样本太“简单”，模型学不到困难样本的区分度，容易偏向流行物品。	物品池较小，分布相对均匀的初期阶段。
Batch内负采样	将一个batch内其他用户的正样本，作为当前用户的负样本。	训练效率极高，无需额外负样本存储和采样。	会引入“流行度偏差”，热门物品被过度当作负样本，导致模型不公平地打压热门物品。	需要快速迭代实验、对热门物品打压不敏感的场景。
流行度校正负采样	在随机或Batch内采样基础上，根据物品出现频率对损失进行加权校正。	能缓解流行度偏差，使模型更公平地对待长尾物品。	需要估计物品频率，增加复杂度。	大多数推荐场景的优选方案，平衡效果与公平性。
困难负采样	选择模型当前预测分数较高（但用户未交互）的物品作为负样本。	能主动挖掘困难样本，提升模型区分边界的能力。	需要动态更新“困难样本池”，工程复杂，可能引入噪声。	模型效果进入平台期后，用于进一步提升精度。

4.2 实现流行度校正的Batch内负采样

TFRS的tfrs.tasks.Retrieval任务默认支持Batch内负采样，并可以方便地加入流行度校正。关键在于自定义candidate_sampler和计算损失时的权重。

# 假设我们有一个物品频率字典 item_freq
item_freq = {item_id: count for item_id, count in zip(item_list, np.random.poisson(100, len(item_list)))}
# 计算采样概率的倒数作为权重（频率越高，权重越低）
sampling_prob = np.array([item_freq.get(i, 1) for i in item_list], dtype=np.float32)
sampling_prob = sampling_prob / sampling_prob.sum()
# 为避免数值下溢，取对数并做平滑
inverse_prob_weight = 1.0 / (sampling_prob + 1e-8)
inverse_prob_weight = np.log(inverse_prob_weight + 1.0)

# 创建TF Dataset时，可以将权重作为特征加入
def add_weights(features):
    features["item_weight"] = tf.gather(tf.constant(inverse_prob_weight), features["item_index"]) # 假设有item_index
    return features

# 在定义Task时，使用自定义的损失函数来融入权重
class WeightedRetrievalTask(tfrs.tasks.Retrieval):
    def __init__(self, temperature=0.1, **kwargs):
        super().__init__(temperature=temperature, **kwargs)

    def call(self, user_embeddings, item_embeddings, item_weights=None):
        scores = tf.matmul(user_embeddings, item_embeddings, transpose_b=True) / self.temperature

        num_queries = tf.shape(scores)[0]
        num_items = tf.shape(scores)[1]

        # 正样本位置（假设对角线为正样本对）
        positive_idx = tf.range(num_queries)
        labels = tf.one_hot(positive_idx, num_items)

        # 计算加权交叉熵损失
        if item_weights is not None:
            # 这里简化演示：将权重应用到负样本的logits上
            # 更精细的实现需要对损失函数项进行加权
            weights_matrix = tf.tile(tf.expand_dims(item_weights, 0), [num_queries, 1])
            # 将正样本位置的权重设为1
            weights_matrix = tf.tensor_scatter_nd_update(weights_matrix, tf.stack([positive_idx, positive_idx], axis=1), tf.ones(num_queries))
            # 加权softmax交叉熵（这是一个简化示例，实际需根据论文实现）
            unweighted_loss = tf.nn.softmax_cross_entropy_with_logits(labels, scores)
            loss = tf.reduce_mean(unweighted_loss * tf.reduce_sum(weights_matrix * (1 - labels), axis=1))
        else:
            loss = self._compute_loss(scores, labels)

        self._factor_losses(loss)
        return loss

提示：上述加权损失是一个原理性示例。工业级实现通常会参考Google的《Sampling-Bias Corrected Neural Modeling》论文，精确计算校正后的softmax概率。

4.3 训练循环与评估

将数据转换为tf.data.Dataset，并配置训练流程。

# 1. 准备数据集（示例）
def df_to_dataset(df, user_features_df, item_features_df, batch_size=1024):
    # 将DataFrame合并，构建特征字典
    # ... 此处省略具体的合并与转换代码 ...
    dataset = tf.data.Dataset.from_tensor_slices(features_dict)
    dataset = dataset.shuffle(10000).batch(batch_size).prefetch(tf.data.AUTOTUNE)
    return dataset

train_dataset = df_to_dataset(train_df, user_df, item_df)
test_dataset = df_to_dataset(test_df, user_df, item_df)

# 2. 初始化模型、任务和优化器
user_model = UserModel(feature_layers)
item_model = ItemModel(feature_layers)
task = WeightedRetrievalTask(temperature=0.1, metrics=tfrs.metrics.FactorizedTopK(
    candidates=item_embeddings_dataset.batch(128).map(item_model), # 用于评估的候选集
))
model = TwoTowerModel(user_model, item_model, task)

# 3. 编译和训练
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.001))
history = model.fit(train_dataset, epochs=10, validation_data=test_dataset)

# 4. 评估
eval_metrics = model.evaluate(test_dataset, return_dict=True)
print(f"Top-10召回率: {eval_metrics['factorized_top_k/top_10_categorical_accuracy']:.4f}")

5. 从模型到服务：部署与线上推理全链路

模型训练完成只是第一步，如何将它部署成可服务、高性能的召回模块，是工程落地的关键。

5.1 物品向量离线计算与索引构建

这是双塔模型线上效率的基石。流程必须自动化。

# 离线批量计算所有物品向量
def export_item_embeddings(model, item_features_df, save_path):
    """计算并保存所有物品的向量"""
    item_dataset = tf.data.Dataset.from_tensor_slices(dict(item_features_df)).batch(512)
    all_item_embeddings = []
    all_item_ids = []

    for batch in item_dataset:
        batch_ids = batch["item_id"].numpy()
        batch_embeddings = model.item_model(batch).numpy()
        all_item_ids.extend(batch_ids)
        all_item_embeddings.append(batch_embeddings)

    all_item_embeddings = np.vstack(all_item_embeddings)
    # 保存为.npy文件或存入向量数据库
    np.save(f"{save_path}/item_ids.npy", np.array(all_item_ids))
    np.save(f"{save_path}/item_embeddings.npy", all_item_embeddings)
    print(f"已导出 {len(all_item_ids)} 个物品向量。")
    return all_item_ids, all_item_embeddings

# 使用Faiss建立向量索引
import faiss

def build_faiss_index(embeddings, index_type="IVF"):
    """构建近似最近邻索引"""
    d = embeddings.shape[1]  # 向量维度
    if index_type == "IVF":
        quantizer = faiss.IndexFlatL2(d)
        index = faiss.IndexIVFFlat(quantizer, d, 100)  # 100个聚类中心
        index.train(embeddings.astype('float32'))
        index.add(embeddings.astype('float32'))
        index.nprobe = 10  # 搜索时探查的聚类数
    else:
        index = faiss.IndexFlatL2(d)
        index.add(embeddings.astype('float32'))
    return index

# 执行导出和索引构建
item_ids, item_embeddings = export_item_embeddings(model, item_features_df, "./embedding_data")
faiss_index = build_faiss_index(item_embeddings, "IVF")
faiss.write_index(faiss_index, "./embedding_data/item_faiss_index.index")

5.2 模型服务化与线上推理

线上服务需要拆分为两个部分：用户向量实时计算服务和物品向量检索服务。

用户向量服务（实时）：使用TensorFlow Serving或Triton Inference Server加载用户塔模型。接收用户实时特征（如ID、当前上下文），返回用户向量。

物品向量检索服务（近实时）：部署一个轻量级服务（如Flask + Faiss），加载预构建的Faiss索引和物品ID映射。接收用户向量，返回Top-K个最相似的物品ID。

# 一个简化的检索服务示例（伪代码）
class RetrievalService:
    def __init__(self, index_path, id_map_path):
        self.index = faiss.read_index(index_path)
        self.item_ids = np.load(id_map_path, allow_pickle=True)

    def search(self, user_embedding, k=100):
        """检索Top-K物品"""
        user_embedding = user_embedding.astype('float32').reshape(1, -1)
        distances, indices = self.index.search(user_embedding, k)
        retrieved_ids = self.item_ids[indices[0]]
        return retrieved_ids.tolist(), distances[0].tolist()

# 线上推理流程
# 1. 用户请求到来，提取特征。
# 2. 调用用户向量服务，得到用户向量。
# 3. 调用检索服务，传入用户向量，得到Top-K物品ID列表。
# 4. 将物品ID列表返回给上游的排序阶段。

5.3 更新与监控策略

一个健壮的召回系统需要完善的更新和监控机制。

• 物品向量更新：
  • 全量更新：每天或每周，用最新模型重新计算全量物品向量，重建索引。这是主流程。
  • 增量更新：对于新上架的物品，实时或准实时（分钟级）计算其向量，并增量添加到索引中。Faiss支持add操作，但频繁增量添加后需要定期重新平衡索引。
• 用户模型更新：用户模型可以以天或周为单位进行全量训练更新。对于实时性要求极高的场景（如信息流），可以考虑使用在线学习框架进行分钟级模型更新，但这会极大增加系统复杂度。
• 监控指标：
  • 服务性能：用户向量计算P99延迟、ANN检索延迟、QPS。
  • 业务指标：召回率（Recall@K）、覆盖率、新颖性、物品新鲜度。
  • 数据健康度：特征缺失率、向量分布变化（如余弦相似度分布漂移）。

6. 进阶优化与踩坑指南

在实际项目中，还有一些细节决定了模型的最终效果。

特征交叉的弥补：既然双塔结构无法做特征交叉，我们可以在塔内做文章。例如，在将特征送入DNN之前，可以加入特征重要性加权层（如SENet结构），让模型自动学习哪些特征组合更重要。也可以尝试在Embedding层后加入Cross Network或FM Layer来显式地学习低阶特征交互。

温度系数（Temperature）的调优：在计算相似度时，score = (user_emb * item_emb) / temperature。这个temperature参数至关重要。它控制着预测得分的“尖锐”程度。值越小，模型对高分物品的置信度越高，但可能过于激进；值越大，得分分布越平缓。通常需要通过验证集仔细调整。

多目标学习：召回不仅可以优化点击率。我们可以尝试多任务学习，让模型同时预测点击、点赞、完播等多个目标。这可以通过在双塔的顶层输出多个头，并设计加权损失函数来实现，从而召回更符合综合用户体验的物品。

冷启动问题：对于新用户或新物品，ID特征无效。解决方案包括：

• 新用户：更多地依赖画像特征（如地域、设备）和行为上下文特征。
• 新物品：使用内容特征（如标题Embedding、分类、标签）和上传者信息，在模型训练时就要加入这些内容特征，确保模型即使在没有ID的情况下也能产生合理的向量。

在我负责的项目里，我们最初忽略了温度系数的影响，默认用了1.0，结果线上召回的结果总是那么几个最热门的视频，多样性极差。后来将其调整到0.2左右，并加入了流行度校正，召回列表的丰富度和长尾覆盖才有了明显改善。另一个坑是，我们一开始把所有物品向量都存在内存字典里做暴力检索，当物品量超过百万后，内存和延迟都爆了。迁移到Faiss的IVF索引后，检索速度提升了近百倍，内存占用也降了下来。这些看似不大的调整，往往是线上效果提升的关键。