Source code for recbole.utils.case_study

# @Time   : 2020/12/25
# @Author : Yushuo Chen
# @Email  : chenyushuo@ruc.edu.cn

# UPDATE
# @Time   : 2020/12/25
# @Author : Yushuo Chen
# @email  : chenyushuo@ruc.edu.cn

"""
recbole.utils.case_study
#####################################
"""

import numpy as np
import torch

from recbole.data.interaction import Interaction


[docs]@torch.no_grad() def full_sort_scores(uid_series, model, test_data, device=None): """Calculate the scores of all items for each user in uid_series. Note: The score of [pad] and history items will be set into -inf. Args: uid_series (numpy.ndarray or list): User id series. model (AbstractRecommender): Model to predict. test_data (FullSortEvalDataLoader): The test_data of model. device (torch.device, optional): The device which model will run on. Defaults to ``None``. Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``. Returns: torch.Tensor: the scores of all items for each user in uid_series. """ device = device or torch.device("cpu") uid_series = torch.tensor(uid_series) uid_field = test_data.dataset.uid_field dataset = test_data.dataset model.eval() if not test_data.is_sequential: input_interaction = dataset.join(Interaction({uid_field: uid_series})) history_item = test_data.uid2history_item[list(uid_series)] history_row = torch.cat( [torch.full_like(hist_iid, i) for i, hist_iid in enumerate(history_item)] ) history_col = torch.cat(list(history_item)) history_index = history_row, history_col else: _, index = (dataset.inter_feat[uid_field] == uid_series[:, None]).nonzero( as_tuple=True ) input_interaction = dataset[index] history_index = None # Get scores of all items input_interaction = input_interaction.to(device) try: scores = model.full_sort_predict(input_interaction) except NotImplementedError: input_interaction = input_interaction.repeat_interleave(dataset.item_num) input_interaction.update( test_data.dataset.get_item_feature().to(device).repeat(len(uid_series)) ) scores = model.predict(input_interaction) scores = scores.view(-1, dataset.item_num) scores[:, 0] = -np.inf # set scores of [pad] to -inf if history_index is not None: scores[history_index] = -np.inf # set scores of history items to -inf return scores
[docs]def full_sort_topk(uid_series, model, test_data, k, device=None): """Calculate the top-k items' scores and ids for each user in uid_series. Note: The score of [pad] and history items will be set into -inf. Args: uid_series (numpy.ndarray): User id series. model (AbstractRecommender): Model to predict. test_data (FullSortEvalDataLoader): The test_data of model. k (int): The top-k items. device (torch.device, optional): The device which model will run on. Defaults to ``None``. Note: ``device=None`` is equivalent to ``device=torch.device('cpu')``. Returns: tuple: - topk_scores (torch.Tensor): The scores of topk items. - topk_index (torch.Tensor): The index of topk items, which is also the internal ids of items. """ scores = full_sort_scores(uid_series, model, test_data, device) return torch.topk(scores, k)