Tensorflow recommenders 튜토리얼 후기

서두가 매우 길어서, 급하신 분은 텐서플로우 recommenders 소개 항목 부터 보시는것을 추천드립니다.

Tensorflow recommenders 사용 배경

 

랭킹 모델을 생각했을때 가장 쉬운 접근 방법은, 사용자와 아이템 정보를 활용하여 0 / 1 (사용자가 싫어함 / 좋아함) 을 기준으로 binary classificaion 을 해볼 수 있습니다.

이것을 pointwise ranking model 이라고 할 수 있습니다.

user_id	item_id	feature_1	feature_2	group_id (optional)	label
1	1	xx	xxx	1	0
1	2	yy	yyy	1	1
1	3	zz	zzz	1	0

 

위의 데이터 구조처럼 깔끔한 정형 데이터를 사용할 수 있고, 학습 및 추론 속도가 빠르다는 장점이 있습니다.

 

그러나 아이템 간 상대적인 비교를 통해서 정렬하지 않기 때문에 추천의 성능 면에서는 여러 아이템을 비교하는 pairwise, listwise ranking model 보다 낮을 수 있습니다.

 

lightgbm 의 lgbm ranker 라는 모델이 있는데 이를 사용하면 저 데이터 구조에서도 listwise-ranker 모델을 사용해볼 수 있습니다.

어디에서나 적당한 성능을 보여주는 lgbm을 사용하기에는 단점이 있을 수 있는데. 현재 저에게는 범주형 데이터 처리가 아쉬웠습니다.

• 범주형 데이터를 lgbm에 입력 시, 내부적으로 label encoder를 사용한 후 데이터 분포에 따라 부분집합으로 나뉘어서 범주형 데이터를 처리
• 범주형 클래스 내 희소한 데이터들은 같은 값으로 처리될 수 있음을 의미함. → 강력한 bias를 가지게 됨,
• 추천 입장에서는 상당히 난처해지는데, item_id와 같은 정보를 바로 범주형으로 상사용하게 되면 데이터가 많은 item_id에 대해서만 적절히 추천이 이루어짐

이 문제를 해결하기 위해서는 item_id를 임베딩을 통해 처리할 필요가 있었으며 텐서플로우 레이어를 쌓아서 모델을 만들 필요가 있습니다.

 

pointwise two-tower model

 

(draw.io로 간단하게 그려보았습니다)

이 모델은 위의 정형 데이터를 바로 사용할 수 있는 pointwise ranking model 이며.

간단한 실험 후 해당 구조를 사용하여 pairwise, listwise 형태로 변환을 생각하였습니다.

 

pointwise two-tower model

포인트와이즈 투타워 모델은 다음과 같이 데이터를 변경 후 사용할 수 있을것 같습니다..만 실제로 사용해보지는 못했습니다.

1. 먼저 아래처럼 하나의 학습데이터에 두 아이템 아이디를 넣어주고

user_id	item_id_1	item_id_2	features	group_id (optional)	label
1	1	2	xxx	1	0
1	2	3	yyy	1	1
1	1	3	zzz	1	0

 

2. 레이블은 다음과 같이 정할 수 있습니다.

• 1 : if user prefer item_id_1 then item_id_2
• 0 : if user prefer item_id_2 then item_id_1

3. 실제 추론 과정에서는 다음과 같은 수정이 필요합니다.

• N개의 아이템을 정렬 할 때, pointwise 모델을 사용해서 정렬 하려면 n번의 호출이 필요함
• 각 아이템 별 스코어 생성 후, 스코어를 내림차순으로 정렬
• N개의 아이템을 정렬할 때, pairwise 모델을 사용해서 정렬 하려면 최악의 경우로 nC2번의 호출이 필요함. 모든 두 아이템쌍을 대상으로 상대적인 선호도를 고른 후 정렬에 사용
• 귀찮...

 

이런 과정을 겪다가 tensorflow-recommenders를 찾아서 사용 및 정리를 해 보았습니다.

 

텐서플로우 recommenders 소개

TFRS(TensorFlow Recommenders)는 추천자 시스템 모델을 빌드하기 위한 라이브러리.

 

튜토리얼과 공식 문서가 매우 잘 되어 있습니다.

블로그에서는 주요 컴포넌트 정리와, 튜토리얼을 조금 수정하여 pointwise two-twer model 모델에 tensorflow-recommenders를 사용하는 과정을 담아보았습니다.

 

tfrs.models.Model

• tfrs를 사용하여면 해당 클래스를 상속받은 모델을 정의해야 함.

class MovielensModel(tfrs.models.Model):

  def __init__(self):
    super().__init__()
    self.ranking_model: tf.keras.Model = RankingModel()
    self.task: tf.keras.layers.Layer = tfrs.tasks.Ranking(
      loss = tf.keras.losses.MeanSquaredError(),
      metrics=[tf.keras.metrics.RootMeanSquaredError()]
    )

  def call(self, features: Dict[str, tf.Tensor]) -> tf.Tensor:
    return self.ranking_model(
        (features["user_id"], features["movie_title"]))

  def compute_loss(self, features: Dict[Text, tf.Tensor], training=False) -> tf.Tensor:
    labels = features.pop("user_rating")

    rating_predictions = self(features)

    # The task computes the loss and the metrics.
    return self.task(labels=labels, predictions=rating_predictions)

 

복잡한 추천 모델은 일반적인 비지도학습, 지도학습의 페러다임에 맞지 않습니다.

위 클래스를 사용하면 복잡한 모델에 대한 사용자 정의 학습 및 테스트 손실을 쉽게 정의할 수 있습니다.

이는 다음 메소드를 구현하면서 수행됩니다.

• __init__ : 모델 변수, 손실함수, 테스크, 메트릭 초기화
• compute_loss : 손실함수 정의
• [Optional] call : 모델이 예측하는 방법 정의

tfrs.tasks

task = tfrs.tasks.Ranking(
  loss = tf.keras.losses.MeanSquaredError(),
  metrics=[tf.keras.metrics.RootMeanSquaredError()]
)

 

tensorflow racommenders task

추천시스템은 종종 두 컴포넌트의 조합으로 구성됩니다.

• 검색 (retrieval) 모델. 수백만개의 후보자 중에서 천 단위의 후보자를 검색
• 랭킹 (ranker) 모델 : 검색 모델에서 나온 후보자에 점수를 매겨서 후보자를 한번 더 줄임

특정 유형의 작업을 수행하기 위한 클래스 입니다.

추천시스템의 목표를 설정하고, 손실 함수와 지표를 관리하는데 사용할 수 있습니다.

• 손실 함수 정의
• 평가 지표 설정
• 학습 과정 관리

 

tfrs.metrics

• tfrs.metrics.FactorizedTopK

검색 모델을 위한, Top K candidates 들의 메트릭을 계산하는 메트릭 입니다.

positive_scores = tf.reduce_sum(
    query_embeddings * true_candidate_embeddings, axis=1, keepdims=True
)

top_k_predictions, retrieved_ids = self._candidates(
    query_embeddings, k=max(self._ks)
)

 

 

tfrs 를 사용하여 point-wise two tower model을 수정 예시.

1. point-wise two tower model 구현

모듈 import

import numpy as np
import tensorflow as tf
import tensorflow_datasets as tfds
import tensorflow_recommenders as tfrs
tfrs.__version__
>> 'v0.7.3'

 

데이터 정의

ratings = tfds.load("movielens/100k-ratings", split="train")

ratings = ratings.map(lambda x: {
    "movie_title": x["movie_title"],
    "user_id": x["user_id"],
    "user_rating": x["user_rating"]
})

tf.random.set_seed(42)
shuffled = ratings.shuffle(100_000, seed=42, reshuffle_each_iteration=False)

train = shuffled.take(80_000)
test = shuffled.skip(80_000).take(20_000)

movie_titles = ratings.batch(1_000_000).map(lambda x: x["movie_title"])
user_ids = ratings.batch(1_000_000).map(lambda x: x["user_id"])

unique_movie_titles = np.unique(np.concatenate(list(movie_titles)))
unique_user_ids = np.unique(np.concatenate(list(user_ids)))

list(movie_titles.take(1))

>> [<tf.Tensor: shape=(100000,), dtype=string, numpy=
 array([b"One Flew Over the Cuckoo's Nest (1975)",
        b'Strictly Ballroom (1992)', b'Very Brady Sequel, A (1996)', ...,
        b"Wes Craven's New Nightmare (1994)", b'Alien (1979)',
        b'Road to Wellville, The (1994)'], dtype=object)>]

 

모델 정의

class NewRankingModel(tf.keras.Model):
    def __init__(self):
        super().__init__()
        embedding_dimension = 32

        # Compute embeddings for users.
        self.user_embeddings = tf.keras.Sequential([
          tf.keras.layers.StringLookup(
            vocabulary=unique_user_ids, mask_token=None),
          tf.keras.layers.Embedding(len(unique_user_ids) + 1, embedding_dimension)
        ])

        # Compute embeddings for movies.
        self.movie_embeddings = tf.keras.Sequential([
          tf.keras.layers.StringLookup(
            vocabulary=unique_movie_titles, mask_token=None),
          tf.keras.layers.Embedding(len(unique_movie_titles) + 1, embedding_dimension)
        ])

        # Compute predictions.
        self.ratings = tf.keras.Sequential([
          # Learn multiple dense layers.
          tf.keras.layers.Dense(256, activation="relu"),
          tf.keras.layers.Dense(64, activation="relu"),
          # Make rating predictions in the final layer.
          tf.keras.layers.Dense(1)
      ])

        
    def call(self, inputs):
        # user_id, movie_title = inputs
        user_id = inputs["user_id"] # Fixed
        movie_title = inputs["movie_title"]


        user_embedding = self.user_embeddings(user_id)
        movie_embedding = self.movie_embeddings(movie_title)

        return self.ratings(tf.concat([user_embedding, movie_embedding], axis=1))

 

 

모델 테스트

NewRankingModel()(
    {
        "user_id": ["42", "43"], 
        "movie_title": ["One Flew Over the Cuckoo's Nest (1975)", "Wes Craven's New Nightmare (1994)"]
    }
)

>> <tf.Tensor: shape=(2, 1), dtype=float32, numpy=
array([[-0.01615258],
       [-0.01816231]], dtype=float32)>

 

학습 진행

# Thanks to chat GPT
# TensorFlow 데이터셋으로 변환
def preprocess(user_id, movie_title, rating):
    return {"user_id": user_id, "movie_title": movie_title}, rating

# 데이터를 TensorFlow Dataset으로 변환
dataset = tf.data.Dataset.from_tensor_slices((
    list(ratings.take(1_000).map(lambda x: x["user_id"]).as_numpy_iterator()),
    list(ratings.take(1_000).map(lambda x: x["movie_title"]).as_numpy_iterator()),
    list(ratings.take(1_000).map(lambda x: x["user_rating"]).as_numpy_iterator())
))
dataset = dataset.map(preprocess)

# 학습 및 평가 데이터셋 분할
train_dataset = dataset.shuffle(buffer_size=1_000).batch(32).prefetch(tf.data.AUTOTUNE)


tf_model = NewRankingModel()

tf_model.compile(
    optimizer=tf.keras.optimizers.Adagrad(learning_rate=0.1),
    loss=tf.keras.losses.MeanSquaredError(),
    metrics=[tf.keras.metrics.RootMeanSquaredError()]
)

tf_model.fit(
    train_dataset
)

>> 32/32 [==============================] - 0s 1ms/step - loss: 2.7392 - root_mean_squared_error: 1.6550
<keras.src.callbacks.History at 0x169a0d7b820>

 

2. tfrs ranking model 적용

class MovielensModel(tfrs.models.Model):

  def __init__(self):
    super().__init__()
    self.ranking_model: tf.keras.Model = NewRankingModel()
    self.task: tf.keras.layers.Layer = tfrs.tasks.Ranking(
      loss = tf.keras.losses.MeanSquaredError(),
      metrics=[tf.keras.metrics.RootMeanSquaredError()]
    )

  def call(self, features: Dict[str, tf.Tensor]) -> tf.Tensor:
    return self.ranking_model({ # Changed
        "user_id": features["user_id"],
        "movie_title": features["movie_title"]
    })

  def compute_loss(self, features: Dict[Text, tf.Tensor], training=False) -> tf.Tensor:
    labels = features.pop("user_rating")

    rating_predictions = self(features)

    # The task computes the loss and the metrics.
    return self.task(labels=labels, predictions=rating_predictions)

 

 

학습 진행

model = MovielensModel()
model.compile(optimizer=tf.keras.optimizers.Adagrad(learning_rate=0.1))

cached_train = train.shuffle(100_000).batch(8192).cache()
cached_test = test.batch(4096).cache()

model.fit(cached_train, epochs=3)

>> Epoch 1/3
10/10 [==============================] - 1s 21ms/step - root_mean_squared_error: 2.0387 - loss: 4.0000 - regularization_loss: 0.0000e+00 - total_loss: 4.0000
Epoch 2/3
10/10 [==============================] - 0s 14ms/step - root_mean_squared_error: 1.1865 - loss: 1.4001 - regularization_loss: 0.0000e+00 - total_loss: 1.4001
Epoch 3/3
10/10 [==============================] - 0s 13ms/step - root_mean_squared_error: 1.1183 - loss: 1.2474 - regularization_loss: 0.0000e+00 - total_loss: 1.2474
<keras.src.callbacks.History at 0x169a0da7f70>

 

 

읽고 난 후 생각..

- tensorflow recommenders를 사용하지 않아도 기존의 랭킹모델을 만들 수 있습니다.

- 그렇다면 tfrs를 사용하면 무엇이 더 좋을까요???

  이것은 추후 다른 문서에서 다뤄 보도록 하겠습니다.