抖音推荐算法实战：如何用Python复现一个简易版协同过滤模型

你是否曾经刷着抖音，看着一条条精准推送的视频，心里暗自琢磨：这玩意儿到底是怎么知道我喜欢看什么的？作为一个技术开发者，你可能不满足于仅仅当一个被算法“投喂”的用户，更想亲手揭开这层神秘的面纱，看看推荐系统背后的数学与代码究竟如何运作。

今天，我们就来动手实现一个简易版的协同过滤推荐模型。别担心，你不需要是机器学习专家，只要对Python有基本了解，就能跟着我一步步搭建起来。我们将从最基础的用户-物品交互数据开始，构建一个能计算用户相似度、并据此进行推荐的系统。这个过程不仅能让你理解抖音这类平台推荐算法的核心骨架，更能为你未来探索更复杂的深度学习模型打下坚实的基础。

1. 环境准备与数据理解

在开始敲代码之前，我们需要先搭好台子。推荐系统的核心是数据，没有数据，再精妙的算法也只是空中楼阁。我们首先来准备一个模拟的、但足够有代表性的数据集。

想象一下，我们有一个微型版的“抖音”，里面有5个用户（User1到User5）和6个短视频（Video_A到Video_F）。用户对视频的喜好程度，我们用1到5的评分来表示，5分代表非常喜欢，1分代表不喜欢，没有评分则表示用户还没看过这个视频。这种“用户-物品评分矩阵”是协同过滤算法最经典的输入形式。

import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity

# 模拟用户-视频评分数据
# 行代表用户，列代表视频，NaN表示用户未观看该视频
ratings_data = {
    'User1': [5, 4, np.nan, 2, np.nan, 1],
    'User2': [4, np.nan, 3, 1, 2, np.nan],
    'User3': [np.nan, 5, 4, np.nan, 3, 2],
    'User4': [2, 3, np.nan, 4, 5, np.nan],
    'User5': [np.nan, 2, 5, 3, np.nan, 4]
}
videos = ['Video_A', 'Video_B', 'Video_C', 'Video_D', 'Video_E', 'Video_F']

df_ratings = pd.DataFrame(ratings_data, index=videos).T
print("原始用户-视频评分矩阵：")
print(df_ratings)

运行上面的代码，你会得到一个5x6的矩阵。这个矩阵非常稀疏，充满了NaN值，这恰恰反映了真实场景：一个平台上有海量内容，但单个用户只会与其中极小一部分产生交互。处理这种稀疏性是推荐系统要解决的首要问题。

注意：在实际的抖音系统中，数据量是天文数字，评分也可能隐式地由观看时长、完播率、点赞、评论、分享等复杂行为综合计算得出。我们这里用显式评分是为了简化模型，便于理解核心原理。

2. 协同过滤的核心：相似度计算

协同过滤，顾名思义，就是“协同大家的智慧来过滤信息”。它的基本假设是：如果用户A和用户B在过去对很多物品的喜好一致，那么他们在未来也很可能喜欢相同的东西。因此，第一步就是量化用户之间的“相似度”。

最常用的相似度度量方法是余弦相似度。它把每个用户对所有物品的评分看作一个多维空间中的向量，通过计算两个向量夹角的余弦值来衡量其方向的一致性。余弦值越接近1，说明两个用户的兴趣方向越一致。

def calculate_user_similarity(df):
    """
    计算用户之间的余弦相似度。
    处理NaN值：在计算相似度时，只考虑两个用户都评过分的那部分物品。
    """
    # 创建一个填充了0的副本用于计算，但记录下哪些位置是真实评分
    df_filled = df.fillna(0)
    similarity_matrix = pd.DataFrame(index=df.index, columns=df.index)

    for user_i in df.index:
        for user_j in df.index:
            if user_i == user_j:
                similarity_matrix.loc[user_i, user_j] = 1.0
            else:
                # 获取两个用户的评分向量
                vec_i = df.loc[user_i].values
                vec_j = df.loc[user_j].values

                # 找出两个用户都评过分的位置（即都不为NaN）
                mask = ~(np.isnan(vec_i) | np.isnan(vec_j))
                if np.sum(mask) == 0:
                    # 如果没有共同评分的物品，相似度设为0（或一个很小的数）
                    similarity_matrix.loc[user_i, user_j] = 0.0
                else:
                    vec_i_common = vec_i[mask]
                    vec_j_common = vec_j[mask]
                    # 使用sklearn的cosine_similarity计算
                    sim = cosine_similarity([vec_i_common], [vec_j_common])[0][0]
                    similarity_matrix.loc[user_i, user_j] = sim

    return similarity_matrix.astype(float)

# 计算用户相似度矩阵
user_sim_matrix = calculate_user_similarity(df_ratings)
print("\n用户余弦相似度矩阵：")
print(user_sim_matrix.round(3))

除了余弦相似度，皮尔逊相关系数也是常用的方法，它衡量的是两个变量之间的线性相关程度。在实际的大型系统中，为了计算效率，常常会采用更优化的算法，如将用户和物品映射到低维隐空间的矩阵分解法。

提示：你可以尝试修改上面的函数，实现皮尔逊相关系数的计算，并观察结果与余弦相似度有何不同。皮尔逊相关系数对用户的评分尺度差异更不敏感。

3. 预测评分与生成推荐

有了相似度，我们就可以进行预测了。核心思想是：一个用户对某个他没看过的物品的评分，可以由与他相似的用户对该物品的评分加权平均得到。这就是基于用户的协同过滤（User-Based CF）。

具体公式可以表示为：

预测评分 = (相似用户1的评分 * 相似度1 + 相似用户2的评分 * 相似度2 + ...) / 相似度之和

我们来实现这个预测函数，并为目标用户生成推荐列表。

def predict_ratings_user_cf(df_ratings, user_sim_matrix, target_user, k=2):
    """
    使用基于用户的协同过滤预测目标用户对所有未评分物品的评分。
    :param df_ratings: 评分矩阵
    :param user_sim_matrix: 用户相似度矩阵
    :param target_user: 需要预测的目标用户
    :param k: 选取最相似的k个邻居用户
    :return: 一个包含预测评分的Series，索引为物品名
    """
    # 找出目标用户未评分的物品
    target_ratings = df_ratings.loc[target_user]
    unrated_items = target_ratings[target_ratings.isna()].index.tolist()

    if not unrated_items:
        print(f"用户 {target_user} 已对所有物品评分。")
        return pd.Series()

    # 获取目标用户与其他用户的相似度，并排除自己
    sim_scores = user_sim_matrix[target_user].drop(target_user)
    # 选取最相似的k个用户
    top_k_users = sim_scores.nlargest(k).index.tolist()

    print(f"为 {target_user} 找到的最相似的 {k} 个用户是: {top_k_users}")

    # 初始化一个字典来存储预测评分
    pred_ratings = {}

    for item in unrated_items:
        numerator = 0.0
        denominator = 0.0

        for neighbor in top_k_users:
            neighbor_rating = df_ratings.loc[neighbor, item]
            # 如果邻居用户对这个物品有评分
            if not pd.isna(neighbor_rating):
                sim = user_sim_matrix.loc[target_user, neighbor]
                numerator += sim * neighbor_rating
                denominator += abs(sim)  # 使用相似度的绝对值作为权重

        if denominator > 0:
            pred_rating = numerator / denominator
            # 将预测评分限制在1-5之间
            pred_rating = max(1.0, min(5.0, pred_rating))
        else:
            pred_rating = np.nan  # 如果没有邻居评过分，则无法预测

        pred_ratings[item] = pred_rating

    return pd.Series(pred_ratings, index=unrated_items)

# 以User1为例，预测他对未观看视频的评分
target_user = 'User1'
k_neighbors = 2
predictions = predict_ratings_user_cf(df_ratings, user_sim_matrix, target_user, k_neighbors)

print(f"\n为目标用户 {target_user} 预测的评分：")
print(predictions.round(3))

# 生成推荐：按预测评分从高到低排序
if not predictions.empty:
    recommendations = predictions.sort_values(ascending=False)
    print(f"\n给 {target_user} 的推荐视频列表（按预测兴趣度降序）：")
    for video, score in recommendations.items():
        if not pd.isna(score):
            print(f"  视频 {video}: 预测评分 {score:.2f}")

运行这段代码，你会看到系统为User1找到了最相似的用户，并计算出了他对未观看视频Video_C和Video_E的预测评分，最后给出了一个排序的推荐列表。这就是协同过滤最直观的体现。

4. 另一种视角：基于物品的协同过滤

除了“人以群分”，还有“物以类聚”。基于物品的协同过滤（Item-Based CF）的基本思想是：如果用户喜欢物品A，那么他很可能也喜欢与物品A相似的其他物品。这种方法在物品数量相对稳定、用户数量巨大的场景下（如电商）往往表现更稳定。

它的计算流程与基于用户的CF类似，只是相似度矩阵的计算对象从用户变成了物品。

def calculate_item_similarity(df):
    """
    计算物品之间的余弦相似度。
    注意：这里需要转置矩阵，因为我们要计算的是物品（列）之间的相似度。
    """
    # 转置矩阵，行变成物品，列变成用户
    df_T = df.T
    df_filled_T = df_T.fillna(0)
    # 使用sklearn快速计算所有物品对的余弦相似度
    # 注意：这里为了简化，用0填充NaN，在实际中可能需要更精细的处理
    item_sim_matrix = pd.DataFrame(
        cosine_similarity(df_filled_T),
        index=df_T.index,
        columns=df_T.index
    )
    return item_sim_matrix

def predict_ratings_item_cf(df_ratings, item_sim_matrix, target_user, k=2):
    """
    使用基于物品的协同过滤预测目标用户对所有未评分物品的评分。
    :param k: 选取与目标物品最相似的k个物品
    """
    target_ratings = df_ratings.loc[target_user]
    unrated_items = target_ratings[target_ratings.isna()].index.tolist()
    rated_items = target_ratings[~target_ratings.isna()].index.tolist()

    if not unrated_items:
        print(f"用户 {target_user} 已对所有物品评分。")
        return pd.Series()

    pred_ratings = {}

    for target_item in unrated_items:
        numerator = 0.0
        denominator = 0.0

        # 获取与目标物品最相似的k个物品（从用户已评分的物品中找）
        # 先计算目标物品与所有已评分物品的相似度
        sim_scores = item_sim_matrix[target_item][rated_items]  # 只考虑用户评过分的物品
        top_k_items = sim_scores.nlargest(k).index.tolist()

        for similar_item in top_k_items:
            sim = item_sim_matrix.loc[target_item, similar_item]
            user_rating_for_similar = df_ratings.loc[target_user, similar_item]
            numerator += sim * user_rating_for_similar
            denominator += abs(sim)

        if denominator > 0:
            pred_rating = numerator / denominator
            pred_rating = max(1.0, min(5.0, pred_rating))
        else:
            pred_rating = np.nan

        pred_ratings[target_item] = pred_rating

    return pd.Series(pred_ratings, index=unrated_items)

# 计算物品相似度矩阵
item_sim_matrix = calculate_item_similarity(df_ratings)
print("\n物品余弦相似度矩阵：")
print(item_sim_matrix.round(3))

# 使用基于物品的CF为User1做预测
predictions_item = predict_ratings_item_cf(df_ratings, item_sim_matrix, target_user, k=2)
print(f"\n基于物品的CF为 {target_user} 预测的评分：")
print(predictions_item.round(3))

# 对比两种方法的推荐结果
print("\n--- 推荐结果对比 ---")
print("基于用户的CF推荐排序:", predictions.sort_values(ascending=False).index.tolist())
print("基于物品的CF推荐排序:", predictions_item.sort_values(ascending=False).index.tolist())

你会发现，基于用户和基于物品的方法得出的推荐顺序可能不同。哪种更好？这没有绝对答案，取决于你的数据特性和业务场景。在实际的抖音系统中，这两种思路都可能被融合使用，或者作为更复杂模型的一部分。

5. 模型评估与冷启动问题

任何一个模型，我们都需要知道它效果如何。对于推荐系统，常用的离线评估指标有均方根误差（RMSE）和平均绝对误差（MAE），它们衡量的是预测评分与实际评分（如果有的话）的差距。更业务化的指标则有点击率（CTR）、转化率等。

然而，我们手头的数据没有“实际评分”来验证预测，因为我们模拟的就是用户没看过的视频。在真实项目中，我们通常会采用“留出法”或“交叉验证”，将一部分已知的用户评分隐藏起来，用模型去预测，再与真实值比较。

def evaluate_model(df_ratings, user_sim_matrix, method='user', k=2):
    """
    一个简单的模型评估函数：模拟留出法，随机隐藏一些已知评分，用模型预测，计算误差。
    注意：由于数据量极小，此处仅为演示流程，结果不稳定。
    """
    df = df_ratings.copy()
    non_nan_positions = [(i, j) for i in df.index for j in df.columns if not pd.isna(df.loc[i, j])]
    if len(non_nan_positions) < 5:
        print("数据太少，无法进行有效评估。")
        return None

    # 随机选择20%的已知评分作为测试集
    np.random.seed(42) # 固定随机种子以便复现
    test_size = int(0.2 * len(non_nan_positions))
    test_positions = np.random.choice(len(non_nan_positions), size=test_size, replace=False)

    errors = []
    for idx in test_positions:
        user, item = non_nan_positions[idx]
        true_rating = df.loc[user, item]
        # 暂时隐藏这个评分
        df.loc[user, item] = np.nan

        # 使用指定方法预测
        if method == 'user':
            preds = predict_ratings_user_cf(df, user_sim_matrix, user, k)
        else: # 'item'
            # 注意：这里需要重新计算物品相似度矩阵，因为df改变了
            item_sim = calculate_item_similarity(df)
            preds = predict_ratings_item_cf(df, item_sim, user, k)

        pred_rating = preds.get(item, np.nan)
        if not pd.isna(pred_rating):
            errors.append(true_rating - pred_rating)

        # 恢复隐藏的评分
        df.loc[user, item] = true_rating

    if errors:
        mae = np.mean(np.abs(errors))
        rmse = np.sqrt(np.mean(np.array(errors) ** 2))
        print(f"{method.upper()}-Based CF 评估结果 (k={k}):")
        print(f"  预测了 {len(errors)} 个评分，有 {test_size - len(errors)} 个因无足够邻居而无法预测。")
        print(f"  平均绝对误差 (MAE): {mae:.3f}")
        print(f"  均方根误差 (RMSE): {rmse:.3f}")
        return mae, rmse
    else:
        print("无法计算出任何预测误差。")
        return None

# 尝试评估基于用户的CF模型
print("开始模型评估（数据量小，结果仅供参考）...")
evaluate_model(df_ratings, user_sim_matrix, method='user', k=2)

运行评估代码，你会看到模型预测的误差。在这么小的数据集上，结果可能波动很大，但这演示了模型评估的标准流程。

协同过滤面临的最大挑战之一是“冷启动”问题：

• 新用户问题：一个新用户刚注册，没有任何行为数据，系统无法找到与他相似的用户，也就无法为他做个性化推荐。
• 新物品问题：一个新上传的视频，还没有任何用户与之交互，无法计算它与其它物品的相似度，也就很难被推荐出去。

抖音等平台如何解决？它们通常会混合多种策略：

1. 热门推荐：给新用户推荐当前最热门的视频。
2. 基于内容的推荐：分析视频的标题、标签、封面、音频、视觉特征，将新视频推荐给可能喜欢同类内容的用户。
3. 探索与利用：在推荐中故意加入一定比例的随机内容或小众内容，既能收集新数据，又能帮助用户发现新兴趣，打破“信息茧房”。

6. 从简易模型到工业级系统的思考

我们实现的这个简易模型，距离抖音每天处理千亿级交互的工业级推荐系统，中间隔着巨大的鸿沟。但核心思想是相通的。工业级系统在以下几个维度上进行了极致优化：

1. 数据规模与工程架构 我们的数据存在内存里，而抖音的数据分布在成千上万台服务器上。它们使用如Hadoop、Spark、Flink这样的大数据处理框架进行实时和离线计算，用向量数据库（如Milvus, Faiss）来高效检索数十亿量级的用户和物品向量。

2. 算法模型的演进 现代推荐系统早已超越了简单的协同过滤。以抖音公开的Wide & Deep模型为例，它将“记忆”与“泛化”能力结合：

• Wide部分：类似于我们做的，处理大量的稀疏特征（如用户ID、视频ID），擅长记忆历史数据中明确的特征组合。
• Deep部分：使用深度神经网络，学习用户和物品的稠密向量表示（Embedding），能发掘稀疏特征背后深层次的、非线性的关联。

# 这是一个概念性的代码片段，展示如何使用深度学习框架（如TensorFlow）定义Wide & Deep模型的结构
# 实际工业实现要复杂无数倍

# 伪代码/概念示意
import tensorflow as tf
from tensorflow import keras

def build_wide_deep_model(num_users, num_items, embedding_dim=64):
    # 输入层
    user_input = keras.Input(shape=(1,), name='user_id')
    item_input = keras.Input(shape=(1,), name='item_id')

    # Wide部分：简单线性模型，处理交叉特征
    # 这里简化了，实际会有大量的特征交叉
    wide_branch = keras.layers.Dense(1, activation='linear')(keras.layers.concatenate([user_input, item_input]))

    # Deep部分：Embedding + 多层神经网络
    user_embedding = keras.layers.Embedding(num_users, embedding_dim)(user_input)
    item_embedding = keras.layers.Embedding(num_items, embedding_dim)(item_input)
    user_vec = keras.layers.Flatten()(user_embedding)
    item_vec = keras.layers.Flatten()(item_embedding)
    deep_concat = keras.layers.concatenate([user_vec, item_vec])
    deep_branch = keras.layers.Dense(128, activation='relu')(deep_concat)
    deep_branch = keras.layers.Dense(64, activation='relu')(deep_branch)
    deep_branch = keras.layers.Dense(1, activation='linear')(deep_branch)

    # 合并Wide和Deep
    combined = keras.layers.add([wide_branch, deep_branch])
    output = keras.layers.Activation('sigmoid')(combined)  # 输出点击概率等

    model = keras.Model(inputs=[user_input, item_input], outputs=output)
    model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
    return model

# 模型结构示意结束

3. 多目标优化与实时性 抖音的推荐不是单纯预测“用户会不会点击”，而是综合预估完播率、点赞率、评论率、分享率、关注作者概率等多个目标，并为每个目标赋予不同的权重，最终计算一个“综合价值分”。同时，系统是分钟级甚至秒级更新的，你的每一次滑动、停留、点赞都会实时影响后续的推荐流。

4. 复杂的系统工程 一个完整的推荐系统通常分为多个阶段：

• 召回：从海量内容池（千万级）中快速筛选出几百到几千个候选内容。我们实现的协同过滤，以及双塔模型等，常被用于这一阶段。
• 粗排/精排：对召回的内容进行更精细的打分排序。Wide & Deep等复杂模型在这里发挥作用。
• 重排：考虑多样性、新鲜度、商业规则等，对排序好的列表进行最后调整，然后呈现给用户。

通过今天这个从零搭建的简易模型，你已经摸到了推荐系统的门道。它就像一台精妙机器的简化图纸，虽然省略了无数复杂的齿轮和电路，但让你理解了动力是如何传递的。下次当你再刷抖音时，或许能感受到，每一次流畅的滑动背后，都是数据、算法和工程智慧共同演奏的交响乐。