Source code for recoder.recommender

from recoder.embedding import EmbeddingsIndex

import numpy as np

import recoder.utils as utils

[docs]class Recommender(object): """ Base Recommender that provide recommendations given users history of interactions. All Recommenders should implement the ``recommend`` function. """
[docs] def recommend(self, users_hist): """ Recommends a list of items for each user list of :class:``. Args: users_hist (list): list of users list of :class:``. Returns: list: items recommended for each user """ raise NotImplementedError
[docs]class SimilarityRecommender(Recommender): """ Recommends items based on similarity search of the items in the user list of :class:``. Implementation based on [1]. Note: This still needs improvement and optimization, and its implementation might change. Args: embeddings_index (EmbeddingsIndex): the embeddings index used to fetch embeddings and do nearest neighbor search. num_recommendations (int): number of recommendations to generate for each user. Note: the number of recommendations requirement is not necessarily satisfied. n (int, optional): number of similar items to retrieve for every item in user interactions. scale (int, optional): how much to scale the similarity between two items [1]: Fabio Aiolli. 2013. Efficient top-n recommendation for very large scale binary rated datasets. In Proceedings of the 7th ACM conference on Recommender systems (RecSys '13). ACM, New York, NY, USA, 273-280. DOI= """ def __init__(self, embeddings_index: EmbeddingsIndex, num_recommendations, n=1, scale=1): self.embeddings_index = embeddings_index self.scale = scale self.num_recommendations = num_recommendations self.n = n def __recommend_single(self, user_hist): user_items = np.array(user_hist.items) items_pool = [self.embeddings_index.get_nns_by_id(item_id, self.n) for item_id in user_items] items_pool = np.unique(items_pool) filtered_items = items_pool[np.isin(items_pool, user_items, invert=True)] items_scores = self.__compute_scores(filtered_items, user_items) if len(items_scores) > self.num_recommendations: top_ind_not_sorted = np.argpartition(-items_scores, self.num_recommendations) top_ind_not_sorted = top_ind_not_sorted[:self.num_recommendations] else: top_ind_not_sorted = np.arange(len(items_scores)) top_sorted_reset_ind = np.argsort(-items_scores[top_ind_not_sorted]) top_ind_sorted = top_ind_not_sorted[top_sorted_reset_ind] top_items = filtered_items[top_ind_sorted] return top_items def __compute_scores(self, items_pool, user_items): pool_embeddings = np.array([self.embeddings_index.get_embedding(item_id) for item_id in items_pool]) user_embeddings = np.array([self.embeddings_index.get_embedding(item_id) for item_id in user_items]) pool_embeddings = utils.normalize(pool_embeddings, axis=1) user_embeddings = utils.normalize(user_embeddings, axis=1) scores =, np.transpose(user_embeddings)) # range: -1 to 1 scores = (scores + 1) / 2 # range: 0 to 1 scaled_scores = np.power(scores, self.scale) agg_scores = np.sum(scaled_scores, axis=1) return agg_scores
[docs] def recommend(self, users_hist): recommendations = [self.__recommend_single(user_hist) for user_hist in users_hist] return recommendations
[docs]class InferenceRecommender(Recommender): """ Recommends items based on the predictions by a :class:`recoder.model.Recoder` model. Args: model (Recoder): model used to predict recommendations num_recommendations (int): number of recommendations to generate for each user. """ def __init__(self, model, num_recommendations): self.model = model self.num_recommendations = num_recommendations
[docs] def recommend(self, users_hist): return self.model.recommend(users_hist, self.num_recommendations)