Baysian Personalized RankingとMatrix Factorizationの比較(実装編)

こんにちは、Wantedlyでデータサイエンティストをしている21新卒の樋口です。

www.wantedly.com

本記事はWantedly 21新卒 Advent Calendar 2021の2日目の記事です!
5人で25記事書くの半端なく大変ですが、頑張っていきたいです...!

qiita.com

先日Bayesian Personalized Ranking(BPR)という推薦システムの有名な論文(引用数4000程度)を読みました。 [WIP] BPR: Bayesian Personalized Ranking from Implicit Feedback · Issue #51 · zerebom/paper-books · GitHub

アイテムの評価値の予測誤差を最小化するのではなく、評価値の順序を最適化することで、精度高くアイテムを推薦しようという趣旨の論文です。しかし、読んだだけでは本当にうまく行くのか?実装はどうなるのか?と疑問がつきませんでした。

そこでより理解を深めるために、人の実装を参考にしつつ実際にPythonで動かしてみました。本記事では、pytorch-lightingを使った、BPRとMatrixFacotorization(MF)を実装方法と、簡単な評価について紹介します。

github.com

(※免責: 本記事は解釈や実装に間違いがあるかもしれないです、ご了承ください🙇)

BPRとMFの簡単な解説

まず今回実装したMFとBPRについて解説します。

Matrix Factorization(MF)

MFは協調フィルタリングの一種です。評価値行列をユーザ・アイテムそれぞれを表す行列に分解したのち、その2つの行列積と元の評価値行列の誤差を小さくすることで、未評価アイテムの評価値を算出します。

2006年にNetflix Prizeの参加者のSimon Funkさんが考案したものが元祖で、行列積の二乗誤差と正則化項の合計を最小化します。

更新式は以下の通りです。

数式は: Matrix Factorizationとは - Qiita から引用させていただきました🙇‍♂️

詳しくは下記リンクを参考にしてください。

qiita.com

sifter.org

Baysian Personalized Ranking(BPR)

MFは評価値の誤差の最小化を測る推薦システムです。しかし、以下のように評価値の誤差が小さくなっても実際にアイテムがユーザにとって良い並び順になるとは限りません。 (label 1: 興味あり, 0:なし なので, label 1のアイテムが上位に来てほしい)

そこで、BPRは誤差の最小化ではなく、興味の(ある/ない)アイテムの評価値を大小関係の最適化をします。具体的にはユーザの興味度は、評価したアイテム(pos_item) > 評価していないアイテム(neg_item)となるという仮定を置いて、 p(pos_item > neg_item) となる確率が最大化するようにします。

以下の式変形のように、理想的な並び順を出力する確率が最も高いモデルパラメータを算出するため、 pos_item > neg_itemとなる確率の積和が最大になるようにします。

(i: pos_item, j: neg_item, Ds: u,i,jの組み合わせ, x_uij: p(i) - p(j) )

詳しくは下記リンクを参考にしてください。

www.smartbowwow.com

実装の解説

次に比較実験で実装したコードを解説します。実装はhttps://github.com/EthanRosenthal/torchmf を元に、pytorch-lightningで学習し、wandbで結果の記録をするように書き換えています。

main.py

python main.py の run() にて実装された実験が開始されます。

def run():
    train_data, test_data = utils.get_movielens_train_test_split(implicit=True)
    params = Params(batch_size=1024, num_workers=0, n_epochs=5)

    mf_module = train(
        train_data,
        test_data,
        model=MFModule,
        datasets_cls=Interactions,
        loss=nn.MSELoss(reduction="sum"),
        params=params,
    )

    bpr_module = train(
        train_data,
        test_data,
        model=BPRModule,
        datasets_cls=PairwiseInteractions,
        loss=bpr_loss,
        params=params,
    )

    print(evaluate(mf_module.model, test_data.tocsr()))
    print(evaluate(bpr_module.model, test_data.tocsr()))

dataset

今回はmovie lens-100kを2値に変換し、sparce_matrix形式にしたものを実験で使います。( case score > 4 then 1 else 0 )

def get_movielens_train_test_split(implicit=False):
    interactions = get_movielens_interactions()
    if implicit:
        interactions = (interactions >= 4).astype(np.float32)
    train, test = train_test_split(interactions)
    train = sp.coo_matrix(train)
    test = sp.coo_matrix(test)
    return train, test

MF用のDataset classの実装は以下の通りです。

class Interactions(data.Dataset):
    """
    Hold data in the form of an interactions matrix.
    Typical use-case is like a ratings matrix:
    - Users are the rows
    - Items are the columns
    - Elements of the matrix are the ratings given by a user for an item.
    """

    def __init__(self, mat: sp.coo_matrix):
        self.mat = mat.astype(np.float32).tocoo()
        self.mat_csr = self.mat.tocsr()
        self.n_users = self.mat.shape[0]
        self.n_items = self.mat.shape[1]

    def __getitem__(self, index: int) -> Tuple[Tuple[int, int], int]:
        row = self.mat.row[index]
        col = self.mat.col[index]
        val = self.mat.data[index]
        return (row, col), val

BPR用のDataset classの実装は以下の通りです。

class PairwiseInteractions(data.Dataset):
    """
    Sample data from an interactions matrix in a pairwise fashion. The row is
    treated as the main dimension, and the columns are sampled pairwise.
    """

    def __init__(self, mat: sp.coo_matrix):
        self.mat = mat.astype(np.float32).tocoo()

        self.n_users = self.mat.shape[0]
        self.n_items = self.mat.shape[1]

        self.mat_csr = self.mat.tocsr()
        if not self.mat_csr.has_sorted_indices:
            self.mat_csr.sort_indices()

    def __getitem__(self, index) -> Tuple[Tuple[int, Tuple[int, int]], int]:
        row = self.mat.row[index]
        found = False

        while not found:
            neg_col = np.random.randint(self.n_items)
            if self.not_rated(row, neg_col, self.mat_csr.indptr, self.mat_csr.indices):
                found = True

        pos_col = self.mat.col[index]
        val = self.mat.data[index]

        return (row, (pos_col, neg_col)), val

同じデータセットを利用しますが、出力形式を変えることでそれぞれの学習スキーマに対応します。

models

次にモデルの実装を解説します。

MFmodule

class MFModule(nn.Module):
    """
    Base module for explicit matrix factorization.
    """

    def __init__(
        self,
        n_users: int,
        n_items: int,
        n_factors: int = 40,
        dropout_p: float = 0,
        sparse: bool = False,
    ):

        super().__init__()
        self.n_users = n_users
        self.n_items = n_items
        self.n_factors = n_factors
        self.user_biases = nn.Embedding(n_users, 1, sparse=sparse)
        self.item_biases = nn.Embedding(n_items, 1, sparse=sparse)
        self.user_embeddings = nn.Embedding(n_users, n_factors, sparse=sparse)
        self.item_embeddings = nn.Embedding(n_items, n_factors, sparse=sparse)
        self.dropout_p = dropout_p
        self.dropout = nn.Dropout(p=self.dropout_p)
        self.sparse = sparse

    def forward(self, users: np.ndarray, items: np.ndarray) -> np.ndarray:
        ues = self.user_embeddings(users)
        uis = self.item_embeddings(items)

        preds = self.user_biases(users)
        preds += self.item_biases(items)
        preds += (self.dropout(ues) * self.dropout(uis)).sum(dim=1, keepdim=True)

nn.Embeddingを使って、batchで入力されるusers, itemsそれぞれの表現ベクトルの積を算出します。この実装ではSimonさんが考案したMFのように、正例だけで最適化できていないので注意してください。

元実装ではdropoutを挟んでいたため、自分の実装でも利用しています。余談ですが、MFにおけるdropoutの理論・特性はこの論文にまとまっていました。気になる方は読んでみてください。

arxiv.org

BPR module

class BPRModule(nn.Module):
    def __init__(
        self,
        n_users: int,
        n_items: int,
        n_factors: int = 40,
        dropout_p: float = 0,
        sparse: bool = False,
        model=MFModule,
    ):
        super().__init__()

        self.n_users = n_users
        self.n_items = n_items
        self.n_factors = n_factors
        self.dropout_p = dropout_p
        self.sparse = sparse
        self.pred_model = model(
            self.n_users,
            self.n_items,
            n_factors=n_factors,
            dropout_p=dropout_p,
            sparse=sparse,
        )

    def forward(self, users: np.ndarray, items: np.ndarray) -> np.ndarray:
        # assert isinstance(items, tuple), "Must pass in items as (pos_items, neg_items)"
        # Unpack
        (pos_items, neg_items) = items
        pos_preds = self.pred_model(users, pos_items)
        neg_preds = self.pred_model(users, neg_items)
        return pos_preds - neg_preds

pl_module

学習サイクルは、pytorch-lightningのLightningModuleで定義します。(data_loader → model → lossと値渡しているだけ)。各エポック終了時に、評価値を算出したかったのですが、sparse matrix形式で上手に計算する実装を組み立てられず、処理が遅くなってしまったのでコメントアウトしています🙈

class MFPLModule(pl.LightningModule):
    def __init__(
        self,
        csr_mat,
        n_factors=10,
        lr=0.02,
        dropout_p=0.02,
        weight_decay=0.1,
        model=MFModule,
        loss=bpr_loss,
    ):
        super().__init__()
        self.csr_mat = csr_mat
        self.n_users = csr_mat.shape[0]
        self.n_items = csr_mat.shape[1]

        self.n_factors = n_factors
        self.dropout_p = dropout_p
        self.lr = lr
        self.loss = loss
        self.weight_decay = weight_decay
        self.model = model(
            self.n_users,
            self.n_items,
            n_factors=self.n_factors,
            dropout_p=self.dropout_p,
            sparse=False,
        )

    def forward(self, users: np.ndarray, pairs: np.ndarray) -> np.ndarray:
        return self.model(users, pairs)

    def training_step(self, batch, batch_idx):
        (users, paris), vals = batch
        preds = self.model(users, paris)
        loss = self.loss(preds, vals)
        self.log("train_loss", loss, prog_bar=False)
        return loss

    def validation_step(self, batch, batch_idx):
        (users, paris), vals = batch
        preds = self.model(users, paris)
        loss = self.loss(preds, vals)
        self.log("val_loss", loss, prog_bar=False)
        return {"users": users, "preds": preds, "loss": loss}

    # def validation_epoch_end(self, outputs):
        # TODO: batch_auc is too slow. We should use a faster metric.
        # aucs = []
        # for output in outputs:
        #     aucs.append(batch_auc(output["users"], self.csr_mat, self.model))
        # self.log("val_roc", torch.Tensor([np.mean(aucs)]), prog_bar=True)

    def configure_optimizers(self):
        optimizer = torch.optim.Adam(
            self.parameters(), lr=self.lr, weight_decay=self.weight_decay
        )
        return optimizer

結果

結果は以下の通りです。

wandb.ai

BPRの方がMRR,Precisionどちらも高い値になっていることが確認できました。 ただ、今回の実験では、本来Explicit feedback用のアルゴリズムであるMFに対して、評価値を2値に変換し、0,1どちらも重み付けせず入力しているなど不利な条件の比較になっています。
今後、より実験条件を揃えた比較などもやっていきたいです。

次やりたいこと

今回は簡単に実装と評価を行いましたが、まだまだ実運用できるレベルではないと感じています。余力があれば、以下を改善して、またブログにしたいと思います...!

• MFの実装を改良したい
  • 正例だけで最適化する
  • 二値分類に向いたアルゴリズムがあるのか調べ、実装する
• もっと細かく評価したい
  • diversity, serendipityなどの指標を見てみる
  • item, userのembeddingを可視化する
  • valid_epoch_endでmetricsを測る
    • 今は全ユーザ x アイテムに対して推論する必要があるので高速化が必要
• 大型のデータセットで動かしたい
  • GPU対応にする
  • sparce_matrix形式を保ったまま、評価値・行列計算をする
  • 今回の実装では、疎行列を高速に扱うために sparce_matrix形式( scipy.sparse.matrix) を利用していますが、評価値を計算など一部対応できていない

おわり

実装して、ブログにしようとすると意外とわからないことがいっぱい出てきました。

• MFにdropoutって使うのか?
• Adamで最適化していいのか?
• sparse_matrix形式を保って、pytorchに値を渡すにはどうしたら良いのか?

これらの疑問はそもそも、読んでるだけでは思いつかないものなので、実装して理解が深まったと思います💪

本や論文を読んでなんとなく理解した気になっていたのですが、実装して、運用可能な精度に持っていくのは思ったより難しく、いかに普段の理解が浅いかということを痛感させられました。今後論文を読むときは、ブログで説明できるくらい理解できたか?と意識したいです💪

最後まで読んでいただきありがとうございました〜