问FactorizedTopK在TensorFlow推荐程序中的应用

EN

import os
import pprint
import tempfile
from typing import Dict, Text
import numpy as np
import tensorflow as tf
import tensorflow_datasets as tfds
import tensorflow_recommenders as tfrs
import pandas as pd
import matplotlib.pyplot as plt

# Variables
seed = 42
test_percentage = 20
train_percentage = 100-test_percentage
embedding_dimension = 32 # 64 ?
metrics_batchsize = 16
train_batchsize = 128
test_batchsize = 64
learning_rate = 0.1 # 0.5 ?
epochs = 3
index_batchsize = 100

views_df = pd.read_csv('filepath')

views_df = views_df[['user_id','content_id']]
users_df = views_df['user_id'].unique()
contents_df = views_df['content_id'].unique()

views_ds = tf.data.Dataset.from_tensor_slices(dict(views_df))
contents_ds = tf.data.Dataset.from_tensor_slices(contents_df)

view_size = len(views_df)
train_size = round(view_size/100*train_percentage)
test_size = view_size-train_size

tf.random.set_seed(seed)

views_ds_shuffled = views_ds.shuffle(len(views_df), seed=seed, reshuffle_each_iteration=False)

train = views_ds_shuffled.take(train_size)
test = views_ds_shuffled.skip(train_size).take(test_size)

user_model = tf.keras.Sequential([
  tf.keras.layers.experimental.preprocessing.IntegerLookup(vocabulary=users_df, mask_value=None),
  tf.keras.layers.Embedding(input_dim=len(users_df) + 1, output_dim=embedding_dimension)
])

content_model = tf.keras.Sequential([
  tf.keras.layers.experimental.preprocessing.IntegerLookup(vocabulary=contents_df, mask_value=None),
  tf.keras.layers.Embedding(input_dim=len(contents_df) + 1, output_dim=embedding_dimension)
])

candidates=contents_ds.batch(metrics_batchsize).map(content_model)

metrics = tfrs.metrics.FactorizedTopK(
  candidates=candidates
)

task = tfrs.tasks.Retrieval(
  metrics=metrics
)

class ContentModel(tfrs.Model):

  def __init__(self, user_model, content_model):
    super().__init__()
    self.content_model: tf.keras.Model = content_model
    self.user_model: tf.keras.Model = user_model
    self.task: tf.keras.layers.Layer = task

  def compute_loss(self, features: Dict[Text, tf.Tensor], training=False) -> tf.Tensor:
    content_embeddings = self.content_model(features["content_id"])
    user_embeddings = self.user_model(features["user_id"])

    return self.task(user_embeddings, content_embeddings)

model = ContentModel(user_model, content_model)
model.compile(optimizer=tf.keras.optimizers.Adagrad(learning_rate=learning_rate))

cached_train = train.shuffle(view_size).batch(train_batchsize).cache()
cached_test = test.batch(test_batchsize).cache()

model.fit(cached_train, epochs=epochs)

model.evaluate(cached_test, return_dict=True)

index = tfrs.layers.factorized_top_k.BruteForce(model.user_model)
index.index(contents_ds.batch(index_batchsize).map(model.content_model), contents_ds)

user_id = 2
_, contents = index(tf.constant([user_id]))
print(f"Recommendations for user {user_id}: {contents}")