引言

推荐系统是互联网产品中提升用户体验的核心技术之一，其核心目标是“猜你喜欢”。协同过滤（Collaborative Filtering, CF）作为最早的推荐算法之一，通过分析用户的历史行为（如评分、点击）来发现用户间的相似性，进而为目标用户推荐相似用户喜欢的物品。本文将详细介绍基于用户的协同过滤（UserCF）​的实现原理，并通过Python代码演示其完整流程。

---

一、UserCF核心思想

UserCF的核心逻辑可概括为“人以群分”：

1. 找到与目标用户兴趣相似的“邻居用户”（近邻）；
2. 将这些邻居用户喜欢但目标用户未接触过的物品推荐给目标用户。

关键步骤包括：数据预处理、用户相似度计算、近邻选择、推荐列表生成。

---

二、环境准备与数据读取

1. 依赖库

代码依赖以下Python库，需提前安装：

pip install tqdm collections  # tqdm用于进度条，collections提供高效数据结构

2. 数据格式与读取

假设我们有一份用户-物品评分数据（如rating_index.tsv），格式为三列：用户ID\t物品ID\t评分（或空格分隔）。例如：

user1	item1	1
user1	item2	0
user2	item1	1
...

代码中通过readTriple函数逐行读取数据，生成三元组（用户ID, 物品ID, 评分）；readRecData函数进一步处理数据：

• 提取所有唯一用户和物品ID；
• 按比例（如10%）随机划分训练集和测试集（测试集用于后续评估模型效果）。

​注意​：代码假设评分r=1表示用户“喜欢”该物品（如收藏、购买），r=0或其他值表示不感兴趣。实际场景中需根据业务定义调整。

---

三、核心算法实现

1. 提取用户偏好集合

通过getSet函数，从训练集中筛选出用户“喜欢”的物品（评分=1），存储为用户ID→物品集合的字典。例如：

user_items = {
    "user1": {"item1", "item3", "item5"},  # user1喜欢这3个物品
    "user2": {"item1", "item2", "item4"},  # user2喜欢这3个物品
    ...
}

2. 计算用户相似度

相似度计算是UserCF的关键。代码中使用余弦相似度衡量用户兴趣的重叠程度，公式为：
\text{cos相似度} = \frac{|A \cap B|}{\sqrt{|A|} \times \sqrt{|B|}}
其中，A和B是两个用户喜欢的物品集合，|A ∩ B|是共同喜欢的物品数，分母是两个集合大小的几何平均。

函数cos4set实现了这一计算。若两个用户无共同喜欢的物品（交集为空），则相似度为0（无法推荐）。

3. 寻找近邻用户

通过knn4set函数，为每个用户找到相似度最高的k个邻居（近邻）。例如，当k=5时，每个用户会得到5个最相似的用户列表。

4. 生成推荐列表

根据近邻用户的喜好，为目标用户推荐未被自己喜欢过的物品。具体来说：

• 遍历目标用户的每个近邻；
• 收集近邻喜欢但目标用户未喜欢的物品；
• 去重后形成最终推荐列表。

---

四、完整代码

import os
import random
from tqdm import tqdm  # 用于显示进度条的库
import collections  # 提供高效的数据结构，如defaultdict


def readTriple(path, sep=None):
    """
    读取三元组数据文件，逐行生成分割后的三元组

    参数:
        path (str): 数据文件路径
        sep (str): 分隔符，默认None（使用空白符分割）

    返回:
        generator: 生成器，每次产出一个三元组（列表形式）
    """
    with open(path, 'r', encoding='utf-8') as f:
        for line in f.readlines():  # 逐行读取文件
            stripped_line = line.strip()  # 去除首尾空白
            if not stripped_line:  # 跳过空行
                continue
            # 根据分隔符分割行（默认使用空白符分割）
            if sep:
                lines = stripped_line.split(sep)
            else:
                lines = stripped_line.split()
            # 只保留长度为3的有效三元组
            if len(lines) == 3:
                yield lines  # 以生成器形式返回，节省内存


def readRecData(path, test_ratio=0.1):
    """
    读取推荐系统数据，划分用户集合、物品集合、训练集和测试集

    参数:
        path (str): 数据文件路径
        test_ratio (float): 测试集比例（0~1之间）

    返回:
        tuple: (用户集合列表, 物品集合列表, 训练集三元组列表, 测试集三元组列表)
    """
    print('读取用户评分三元组...')
    user_set, item_set = set(), set()  # 存储所有用户和物品的唯一标识
    triples = []  # 存储所有有效三元组（用户ID, 物品ID, 评分）

    # 使用tqdm显示读取进度
    for u, i, r in tqdm(readTriple(path), desc="加载数据"):
        # 将字符串转换为整数（假设数据中的ID是整数类型）
        user_set.add(int(u))  # 添加用户ID到集合（自动去重）
        item_set.add(int(i))  # 添加物品ID到集合（自动去重）
        triples.append((int(u), int(i), int(r)))  # 存储三元组

    # 划分测试集：从所有三元组中随机采样test_ratio比例的样本
    test_size = int(len(triples) * test_ratio)
    test_set = random.sample(triples, test_size)  # 随机选择测试集

    # 训练集 = 所有三元组 - 测试集（转换为集合去重后再转回列表）
    # 注意：这里可能存在问题，因为集合是无序的，且三元组是唯一的，所以直接集合相减可能没问题
    train_set = list(set(triples) - set(test_set))

    # 返回用户集合列表、物品集合列表、训练集、测试集
    return list(user_set), list(item_set), train_set, test_set


def getSet(triples):
    """
    从三元组数据中提取用户喜欢的物品集合（假设评分r=1表示喜欢）

    参数:
        triples (list): 三元组列表，每个元素为(用户ID, 物品ID, 评分)

    返回:
        defaultdict(set): 键为用户ID，值为该用户喜欢的物品集合（评分=1的物品）
    """
    user_items = collections.defaultdict(set)  # 初始化默认字典（键不存在时返回空集合）
    for u, i, r in triples:
        if r == 1:  # 假设评分1表示用户喜欢该物品
            user_items[u].add(i)  # 将物品ID添加到对应用户的集合中
    return user_items


# 协同过滤（UserCF）核心算法
def knn4set(trainset, k, sim_method):
    """
    基于用户的k近邻算法（User-based KNN）

    参数:
        trainset (dict): 训练集，格式为{用户ID: 该用户喜欢的物品集合}
        k (int): 选取的近邻数量
        sim_method (function): 相似度计算方法（如cos4set）

    返回:
        dict: 格式为{用户ID: [近邻用户ID1, 近邻用户ID2, ...]}（按相似度从高到低排序）
    """
    sims = {}  # 存储每个用户的近邻列表
    # 遍历所有用户（训练集中的每个用户作为目标用户e1）
    for e1 in tqdm(trainset, desc="计算用户相似度"):
        ulist = []  # 存储e1与其他用户的相似度及用户ID
        # 遍历所有可能的邻居用户（训练集中的每个用户作为候选邻居e2）
        for e2 in trainset:
            # 跳过自身（e1和e2是同一个用户）
            if e1 == e2:
                continue
            # 获取e1和e2喜欢的物品集合
            e1_items = trainset[e1]
            e2_items = trainset[e2]
            # 如果两个用户没有共同喜欢的物品，跳过（无法计算相似度）
            if len(e1_items & e2_items) == 0:
                continue
            # 计算e1和e2的相似度（调用传入的相似度方法）
            sim = sim_method(e1_items, e2_items)
            ulist.append((e2, sim))  # 记录邻居用户及其相似度

        # 按相似度从高到低排序，取前k个邻居（如果不足k个则取全部）
        # sorted(ulist, key=lambda x: x[1], reverse=True) 按相似度降序排列
        # [:k] 取前k个元素，[i[0] for i in ...] 提取邻居用户ID
        sims[e1] = [i[0] for i in sorted(ulist, key=lambda x: x[1], reverse=True)[:k]]

    return sims


def get_recomedations_by_usrCF(user_sims, user_o_set):
    """
    根据用户相似度和用户历史行为生成推荐列表

    参数:
        user_sims (dict): 用户近邻字典，格式为{用户ID: [近邻用户ID列表]}
        user_o_set (dict): 用户历史喜欢的物品集合，格式为{用户ID: 物品集合}

    返回:
        dict: 推荐结果，格式为{用户ID: [推荐物品ID列表]}（去重后的物品）
    """
    recommendations = collections.defaultdict(set)  # 存储推荐结果（集合自动去重）
    # 遍历每个目标用户u
    for u in user_sims:
        # 遍历u的每个近邻用户sim_u
        for sim_u in user_sims[u]:
            # 近邻用户sim_u喜欢的物品集合 减去 u已经喜欢的物品集合（去重）
            # 然后将这些物品添加到u的推荐列表中
            recommendations[u] |= (user_o_set[sim_u] - user_o_set[u])

    return recommendations


def trainUserCF(user_items, sim_method, k=5):
    """
    训练UserCF模型，生成推荐结果

    参数:
        user_items (dict): 用户-物品集合字典，格式为{用户ID: 物品集合}
        sim_method (function): 相似度计算方法（如cos4set）
        k (int): 近邻数量

    返回:
        dict: 推荐结果，格式为{用户ID: [推荐物品ID列表]}
    """
    # 步骤1：计算用户近邻（k近邻）
    user_sims = knn4set(user_items, k, sim_method)
    # 步骤2：根据近邻生成推荐列表
    recommendations = get_recomedations_by_usrCF(user_sims, user_items)
    return recommendations


def cos4set(set1, set2):
    """
    计算两个集合的余弦相似度

    公式: cos相似度 = |A∩B| / (√|A| * √|B|)

    参数:
        set1 (set): 集合1
        set2 (set): 集合2

    返回:
        float: 余弦相似度值（范围0~1）
    """
    intersection = len(set1 & set2)  # 交集大小
    norm1 = len(set1) ** 0.5  # 集合1的模长（L2范数）
    norm2 = len(set2) ** 0.5  # 集合2的模长（L2范数）
    # 避免除零错误（如果其中一个集合为空，相似度为0）
    if norm1 == 0 or norm2 == 0:
        return 0.0
    return intersection / (norm1 * norm2)


if __name__ == '__main__':
    # 获取当前脚本所在目录路径
    ROOT = os.path.split(os.path.realpath(__file__))[0]
    # 构建数据文件路径（假设数据文件与脚本同目录，名为rating_index.tsv）
    data_path = os.path.join(ROOT, 'rating_index.tsv')

    # 读取数据并划分训练集、测试集
    # user_list: 所有用户ID列表
    # item_list: 所有物品ID列表
    # train_set: 训练集三元组列表（用户ID, 物品ID, 评分）
    # test_set: 测试集三元组列表
    _, _, train_set, test_set = readRecData(data_path, test_ratio=0.1)

    # 从训练集中提取用户喜欢的物品集合（仅评分=1的物品）
    user_items = getSet(train_set)

    # 训练UserCF模型（使用余弦相似度，k=5）
    # recomedations_by_userCF: 推荐结果，格式为{用户ID: 推荐物品ID集合}
    recomedations_by_userCF = trainUserCF(user_items, cos4set, k=5)

    # 打印推荐结果（示例输出）
    print("推荐结果示例（用户ID: 推荐物品集合）:")
    for user, items in list(recomedations_by_userCF.items())[:3]:  # 打印前3个用户的推荐结果
        print(f"用户{user}: {items}")

---

五、总结

本文通过代码演示了UserCF推荐算法的完整流程，从数据读取到推荐生成，核心在于利用用户间的相似性挖掘潜在兴趣。实际应用中需根据业务场景调整参数（如k值、相似度计算方式），并结合评估指标（如准确率、召回率）优化模型。