Go to Qiita Advent Calendar 2024 Top
LoginSignup
44
37

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@Papillon6814(Papillon6814)in株式会社Papillon

Pythonではじめての簡易レコメンドエンジン開発

Last updated at Posted at 2021-05-16

はじめに

レコメンドエンジンの仕組みってよくわかりませんよね。仕組みを知ろうと思って説明を見てみても複雑な数式とか難しそうな言葉がたくさん出てきたり、そもそも日本語じゃなかったり...
僕もレコメンドエンジンについて何も知らない状態から色々と調べました。それでほんの少しだけレコメンドエンジンについてわかるようになったので、ここに記録を残しておこうと思いました。

この記事の対象者

  • レコメンドエンジンの名前くらいしかまだ知らない人
  • レコメンドエンジンについて少し興味がある人

レコメンドエンジンは「おすすめシステム」のこと

YoutubeやSpotifyのように、何かコンテンツを提供するサービスで例えます。
コンテンツをユーザーに提供する際に、どのコンテンツが気に入られるのかというのはもちろんユーザーによって異なります。
そのため、ただソートされただけのコンテンツやただ人気度の高いコンテンツを提供するだけでは、それぞれのユーザーにとって自分の欲しいコンテンツが表示される機会が少なくなってしまいます。

そこで最近の大手サービスにはレコメンドエンジンが搭載されています。優秀なエンジニアたちは高性能なレコメンドエンジンをサービスに搭載することで、ユーザーひとりひとりがより興味を持つようなコンテンツを表示してサービスの魅力を向上させています。

レコメンドエンジンの開発自体は難易度が高く、作り方も様々です。今回の記事ではThompson Samplingという手法を使って簡易エンジンの開発をしてみます。

レコメンドの仕組み

典型的なレコメンドの仕組みとして、2つの例を挙げることができます。

  • 協調フィルタリング
  • コンテンツベースのフィルタリング

もしくは、これらのハイブリッドの手法が用いられたりもします。

協調フィルタリング

協調フィルタリングは、「オススメをしたいユーザーに似たようなユーザーの好みのコンテンツをレコメンドする」仕組みです。

コンテンツベースのフィルタリング

一方コンテンツベースのフィルタリングは、「オススメをしたいユーザーの過去の行動履歴からそのユーザーの好きそうなコンテンツをレコメンドする」仕組みです。

両者の欠点

これら２つともよく出来た仕組みなのですが、両者共通の欠点として「ユーザーの情報が集まらなければレコメンドの効果があまり出せない」ことが挙げられます。
この欠点のことをコールドスタート問題と呼んだりもします。

このコールドスタート問題をなんとかするために、今から私たちは多腕バンディット問題について考えていきます。

多腕バンディット問題

多腕バンディット問題について考えるとき、よくスロットの例が挙げられます。
なので僕もスロットの例を使おうと思います。
image.png

多腕バンディット問題

ここに4つのスロットマシンがあります。このスロットマシンを100回まで引けるとき、どのようにすれば報酬を最大化できるでしょうか。

この問題を解く方法は複数提案されています。
ε-greedyという手法は比較的イメージがしやすいと思います。「決められた数だけスロットマシンを試してみて、もっとも結果の良かったマシンを残りの回数分ずっと引き続ける」というものです。よって最も結果の良かったマシンを引ける回数は100-4n回となります。

しかしこの解法にはジレンマがあります。最初に検証のために4n回スロットマシンを回すのですが、検証の回数が多いと結果の良いマシンを回す回数が減ってしまい、逆に検証の回数が少ないと結果の良いマシンを選ぶことができる確率が下がってしまいます。

このように難しい多腕バンディット問題ですが、今回は先程も名前が出てきたThompson Samplingという方法をPythonで実行していきます。

Thompson Sampling

Thompson Samplingでは3つの段階に分けて処理が行われます。

  • それぞれのスロットマシンが当たる確率を確率分布から推定する
  • 確率分布の値が最大であるスロットマシンを引く
  • 結果から、スロットマシンの当たる確率を推定する

このThompson Samplingは昔からある手法なのですが、当たる確率が高いことでも知られています。

実装

Thompson Samplingを用いて、今回はゲームを推薦するシステムを試しに実装してみます。
※この記事を非常に参考にしています！

前提

Python3系がインストールされているものとします。
ライブラリは必要に応じてinstallしてもらえれば大丈夫です。

動作確認用CSVの自動生成

id ユーザーid ゲーム名 ユーザーのアクションから得られたreward

というCSVデータが仮の行動履歴として保存されるようにします。
rewardの列は、ユーザーがそのゲームに対して何かプラスのアクションをしたことにより記録される値とします。

makecsv.py
import csv
import random

records = 10000
users = 250
print("Making %d records...", records)

fieldnames = ['id', 'user_id', 'game_name', 'gain']
writer = csv.DictWriter(open("data.csv", "w"), fieldnames=fieldnames)

user_id_list = range(1, users)
game_names = ["Fortnite", "Valorant", "PUBG", "Clash Royale", "Apex Legends", "FIFA", "Shadowverse", "CSGO", "CoD"]
gains = [1, 5, 7]

writer.writerow(dict(zip(fieldnames, fieldnames)))
for i in range(0, records):
    writer.writerow(dict([
        ('id', i),
        ('user_id', random.choice(user_id_list)),
        ('game_name', random.choice(game_names)),
        ('gain', random.choice(gains))
    ]))
  • 250人のユーザー
  • 9種類のゲーム
  • 10000行生成

という設定のプログラムです。このプログラムを実行すると10000行の行動履歴CSVデータが生成されます。

Thompson Sampling用クラスの実装

次はレコメンドエンジン本体となるクラスを実装します。

まずは親クラスからです。

thompson_sampling_replayer.py
import numpy as np
from tqdm import tqdm

class ReplaySimulator(object):
    '''
    Class to provide base functionality for simulatoing the replayer method for online algorithms.
    '''
    
    def __init__(self, n_visits, reward_history, item_col_name, visitor_col_name, reward_col_name, n_iterations=1, random_seed=1):
        
        np.random.seed(random_seed)
        
        self.reward_history = reward_history
        self.item_col_name = item_col_name
        self.visitor_col_name = visitor_col_name
        self.reward_col_name = reward_col_name
        
        # number of visits to replay/simulate
        self.n_visits = n_visits
        
        # number of runs to average over
        self.n_iterations = n_iterations
        
        # items under test
        self.items = self.reward_history[self.item_col_name].unique()
        self.n_items = len(self.items)
        
        # visitors in the historical reward_history (e.g., from gain df)
        self.visitors = self.reward_history[self.visitor_col_name].unique()
        self.n_visitors = len(self.visitors)
        
    def reset(self):
        # number of times each item has been sampled
        self.n_item_samples = np.zeros(self.n_items)
        
        # fraction of time each item has resulted in a reward
        self.n_item_rewards = np.zeros(self.n_items)
        
    def replay(self):
        
        results = []
        
        for iteration in tqdm(range(0, self.n_iterations)):
            
            self.reset()
            
            total_rewards = 0
            fraction_relevant = np.zeros(self.n_visits)
            
            for visit in range(0, self.n_visits):
                
                found_match = False
                while not found_match:
                    
                    # choose a random visitor
                    visitor_idx = np.random.randint(self.n_visitors)
                    visitor_id = self.visitors[visitor_idx]
                    
                    # select an item to offer the visitor
                    item_idx = self.select_item()
                    item_id = self.items[item_idx]

                    reward = self.reward_history.query('{} == @item_id and {} == @visitor_id'.format(self.item_col_name, self.visitor_col_name))[self.reward_col_name]
                    
                    found_match = reward.shape[0] > 0
                    
                reward_value = reward.iloc[0]
                
                self.record_result(visit, item_idx, reward_value)
                
                ## record metrics
                total_rewards += reward_value
                fraction_relevant[visit] = total_rewards * 1. / (visit + 1)
                
                result = {}
                result['iteration'] = iteration
                result['visit'] = visit
                result['item_id'] = item_id
                result['visitor_id'] = visitor_id
                result['reward'] = reward_value
                result['total_reward'] = total_rewards
                result['fraction_relevant'] = total_rewards * 1. / (visit + 1)
                
                results.append(result)
                
            return results
        
        def select_item(self):
            game_names = ["Fortnite", "Valorant", "PUBG", "Clash Royale", "Apex Legends", "FIFA", "Shadowverse", "CSGO", "CoD"]
            return game_names[np.random.randint(self.n_items)]
        
        def record_result(self, visit, item_idx, reward):
            self.n_item_samples[item_idx] += 1
            
            alpha = 1. / self.n_item_samples[item_idx]
            self.n_item_rewards[item_idx] += alpha * (reward - self.n_item_rewards[item_idx])

今回不必要な処理もありますが、僕はこのクラスを親クラスとして使っています。
こちらはThompson Samplingをするためのクラスです。

thompson_sampling_replayer.py
class ThompsonSamplingReplayer(ReplaySimulator):
    '''
    Class to provide functionality for simulating the replayer method on a 
    Thompson Sampling bandit algorithm.
    '''
    
    def reset(self):
        self.alphas = np.ones(self.n_items)
        self.betas = np.ones(self.n_items)
        
    def select_item(self):
        samples = [np.random.beta(a, b) for a, b in zip(self.alphas, self.betas)]
        
        return np.argmax(samples)
    
    def record_result(self, visit, item_idx, reward):
        
        ## update value estimate
        if reward == 1:
            self.alphas[item_idx] += 1
        else:
            self.betas[item_idx] += 1

先程の親クラスを継承し、Thompson Samplingの処理を入れてあります。

実行

ではこのクラスを実行してレコメンドをさせてみます。
from appのところのディレクトリ構成だけ自分の環境に合わせて変えてください。

recommender.py
import numpy as np
import pandas as pd

from app import ThompsonSamplingReplayer

gain_df = pd.read_csv('./output.csv')

reward_threshold = 5
gain_df['reward'] = gain_df.eval('gain >= @reward_threshold').astype(int)

n_visits = 250
n_iterations = 20

reward_history = gain_df
item_col_name = 'game_name'
visitor_col_name = 'user_id'
reward_col_name = 'reward'

thompson_results = ThompsonSamplingReplayer(n_visits, reward_history,
                                           item_col_name, visitor_col_name, reward_col_name,
                                           n_iterations=n_iterations).replay()

thompson_results_df = pd.DataFrame(thompson_results)
thompson_results_df.head()

thompson_results_df.to_csv('result/thompson_sampling.csv')

このファイルを実行すると、thompson_sampling.csvという名前のファイルが生成されています。
そしてそこにはこんな感じのデータが載っています。

,iteration,visit,item_id,visitor_id,reward,total_reward,fraction_relevant
0,0,0,DOTA2,888,1,1,1.0
1,0,1,CoD,436,1,2,1.0
2,0,2,Pokemon,541,1,3,1.0
3,0,3,DOTA2,906,0,3,0.75
4,0,4,Clash Royale,1096,0,3,0.6

ループ数が少ないのでそこまで信憑性は高くないのですが、それぞれのユーザーに対するレコメンドの結果が出ていると思います。

さいごに

レコメンドエンジンは概念も難しく理解しきれていないところもありますが、学習を進めていこうと思います。
見てくださった方ありがとうございました。

参考

https://developers.cyberagent.co.jp/blog/archives/25099/

Icons made by Nikita Golubev from www.flaticon.com
https://blog.insightdatascience.com/multi-armed-bandits-for-dynamic-movie-recommendations-5eb8f325ed1d
44
37
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
44
37

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?