はじめに
レコメンドエンジンの仕組みってよくわかりませんよね。仕組みを知ろうと思って説明を見てみても複雑な数式とか難しそうな言葉がたくさん出てきたり、そもそも日本語じゃなかったり...
僕もレコメンドエンジンについて何も知らない状態から色々と調べました。それでほんの少しだけレコメンドエンジンについてわかるようになったので、ここに記録を残しておこうと思いました。
この記事の対象者
- レコメンドエンジンの名前くらいしかまだ知らない人
- レコメンドエンジンについて少し興味がある人
レコメンドエンジンは「おすすめシステム」のこと
YoutubeやSpotifyのように、何かコンテンツを提供するサービスで例えます。
コンテンツをユーザーに提供する際に、どのコンテンツが気に入られるのかというのはもちろんユーザーによって異なります。
そのため、ただソートされただけのコンテンツやただ人気度の高いコンテンツを提供するだけでは、それぞれのユーザーにとって自分の欲しいコンテンツが表示される機会が少なくなってしまいます。
そこで最近の大手サービスにはレコメンドエンジンが搭載されています。優秀なエンジニアたちは高性能なレコメンドエンジンをサービスに搭載することで、ユーザーひとりひとりがより興味を持つようなコンテンツを表示してサービスの魅力を向上させています。
レコメンドエンジンの開発自体は難易度が高く、作り方も様々です。今回の記事ではThompson Samplingという手法を使って簡易エンジンの開発をしてみます。
レコメンドの仕組み
典型的なレコメンドの仕組みとして、2つの例を挙げることができます。
- 協調フィルタリング
- コンテンツベースのフィルタリング
もしくは、これらのハイブリッドの手法が用いられたりもします。
協調フィルタリング
協調フィルタリングは、「オススメをしたいユーザーに似たようなユーザーの好みのコンテンツをレコメンドする」仕組みです。
コンテンツベースのフィルタリング
一方コンテンツベースのフィルタリングは、「オススメをしたいユーザーの過去の行動履歴からそのユーザーの好きそうなコンテンツをレコメンドする」仕組みです。
両者の欠点
これら２つともよく出来た仕組みなのですが、両者共通の欠点として「ユーザーの情報が集まらなければレコメンドの効果があまり出せない」ことが挙げられます。
この欠点のことをコールドスタート問題と呼んだりもします。
このコールドスタート問題をなんとかするために、今から私たちは多腕バンディット問題について考えていきます。
多腕バンディット問題
多腕バンディット問題について考えるとき、よくスロットの例が挙げられます。
なので僕もスロットの例を使おうと思います。
多腕バンディット問題
ここに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の列は、ユーザーがそのゲームに対して何かプラスのアクションをしたことにより記録される値とします。
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用クラスの実装
次はレコメンドエンジン本体となるクラスを実装します。
まずは親クラスからです。
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をするためのクラスです。
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のところのディレクトリ構成だけ自分の環境に合わせて変えてください。
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.comhttps://blog.insightdatascience.com/multi-armed-bandits-for-dynamic-movie-recommendations-5eb8f325ed1d