『簡単・高速』なAmazonっぽいレコメンドcf_recommenderを開発して公開

Pythonで実装されたオープンソースの協調フィルタリング型RealTimeレコメンドエンジンです。Amazonの『この商品を買った人はこんな商品も買っています』機能や、Twitterの『おすすめユーザ』機能と類似した機能を提供します。稼働にはRedisサーバが必要です。

インストール

pip install cf_recommender

https://pypi.python.org/pypi/cf_recommender

特徴

■ 1. Getが早い
レコメンドする商品を平均5msで取得できます。10万Item,10万ユーザ,ユーザが平均50Item購入時のベンチマーク結果です。計算済み結果をRedisのSortedSetに格納していて、クエリ1回でレコメンドする商品を取得することで実現しています。

■ 2. オススメする商品はリアルタイム更新
購入発生時に即時レコメンドする商品が更新され結果に反映されます。

■ 3. インストールが簡単
pip install cf_recommender

■ 4. 仕組みが単純なので汎用性が高い
cf-recommenderはレコメンドする機能のみを提供するエンジンです。EC用途だけでなく、おすすめユーザの紹介、この会社を見ている人は別のこの会社を見ました、等々幅広い用途に利用出来ます。

■ 5. タグ機能でレコメンド商品群を管理可能
タグが異なる商品間でのレコメンドは行われません。たとえばDVDと衣類でレコメンドを行わない場合、異なるタグを登録することで簡単に実現します。

協調フィルタリングとは

wikipediaの協調フィルタリングより引用

協調フィルタリング（きょうちょうフィルタリング、Collaborative Filtering、CF）は、多くのユーザの嗜好情報を蓄積し、あるユーザと嗜好の類似した他のユーザの情報を用いて自動的に推論を行う方法論である。趣味の似た人からの意見を参考にするという口コミの原理に例えられることが多い。

協調フィルタリング - Wikipedia

サンプルコード

cf.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender

cf_settings = {
    # redis
    'expire': 3600 * 24 * 30,
    'redis': {
        'host': 'localhost',
        'port': 6379,
        'db': 0
    },
    # recommendation engine settings
    'recommendation_count': 10,
    'recommendation': {
        'update_interval_sec': 5,
        'search_depth': 100,
        'max_history': 500,
    },
}


# レコメンド商品一覧を取得
item_id = 'Item1'
recommendation = Recommender(cf_settings)
print recommendation.get(item_id, count=3)
#>>> ['Item10', 'Item3', 'Item2']

# 商品購入履歴を登録
user_id = 'user-00001'
tag = 'default'
buy_items = ['Item10', 'Item10', 'Item10', 'Item3', 'Item3', 'Item2', 'Item1']
for item_id in buy_items:
    recommendation.register(item_id, tag=tag)
recommendation.like(user_id, buy_items)

# redis以外はデフォルト設定で動かす
custom_settings = {
    'redis': {
        'host': 'localhost',
        'port': 6379,
        'db': 0
    },
}

recommendation = Recommender(custom_settings)
print recommendation.get(item_id, count=3)
#>>> ['Item10', 'Item3', 'Item2']

アルゴリズムとレコメンド特性

単純共起によりレコメンド対象を決定します。評価数が多いアイテムが高く評価されます。例えばItem1-10の内、100人のユーザの購入履歴の51%がItem10,残り49%をランダムで購入していた場合、Item1-9の最上位レコメンドとして高い確率でItem10が登場します。

具体的にItem-3の場合のレコメンド対象を決定するロジックについて言及します。最新のItem3購入ユーザを100人分メモリ上に展開し、購入ユーザの最新購入履歴100件を参照して商品購入リストを作成します。全履歴を合計し購入数が多い商品から順にレコメンド対象として登録を行います。探索の深さはsettings.recommendation.search_depthで変更可能です。default値は100が設定されています。流行が移りItem1が大量に購入されると履歴が更新されItem1が最上位レコメンドとして登場するようになります。探索の深さはレコメンドする商品の移り変わり速度に影響します。プロダクトが求める適切なレコメンドになるようチューニングしてください。

うまい話ばかりじゃないですよ！向かない用途やデメリット

■ 1. 商品削除の反映が遅い
商品を削除してもレコメンドする商品一覧からは削除されません。商品が新規購入された際にレコメンド商品一覧が更新されるタイミングで削除が実施されます。新規購入が無い場合は残り続けます。問題の解消については『実装例2』で紹介してるupdate_all関数を利用してください。

■ 2. レコメンド精度が低い
高速化するために履歴探索が浅くレコメンド精度が低いです。探索の浅さは時間軸に影響するため最新の情報が反映されやすいとも言い換えられます。プロダクトの特性に沿った最適な値を設定してください。探索の深さはsettings.recommendation.search_depthで変更可能です。古い過去のデータを全て考慮したレコメンドリストを生成する用途には向いていません。

■ 3. Redisのメモリを贅沢に使う
商品数10万,タグ数10件,ユーザ数10万/月,ユーザ平均購入数50品/月の状況でRedisのメモリを6.6GByte消費します。メモリ容量の見積りと負荷試験を行いましょう。

ローカル環境でサンプルコードを稼働するまでの導入手順

■ 1.redisをローカルPCにインストールする。
■ 2.redisを起動する。
■ 3.redis-cliコマンドで疎通確認を行う。

(env)niku > redis-cli
redis 127.0.0.1:6379> set a 1
OK
redis 127.0.0.1:6379> get a
"1"

■ 4.cf-recommenderをインストールする

pip install cf_recommender

■ 5.サンプルコードを書き込んだpyファイルを作成して実行

(env)niku > python cf.py 
[]
['Item10', 'Item3', 'Item2']
(env)niku > python cf.py 
['Item10', 'Item3', 'Item2']
['Item10', 'Item3', 'Item2']

導入方法

■ 1.ユーザと商品に何の値を設定するか決める
たとえばユーザにおすすめユーザを紹介する機能を実装する場合、商品にはユーザIDを設定します。課題毎に適切に設定しましょう。

■ 2.商品のタグ機能を利用するか決める
タグが異なる商品間でのレコメンドはできません。設定しない場合は全てタグ名defaultが設定されます。

■ 3.商品IDについて設計する
商品IDは異なるタグ間でもユニークでなければなりません。コンフリクトによる上書き事故が発生せぬよう適切な商品IDを設計しましょう。

■ 4.導入方法について決定する
データを貯めてから公開するか、データを一括登録してから公開するかを決定しましょう。実装例3-1, 3-2で詳しく紹介しています。

■ 5.負荷試験する
しないとこうなる

設定値について

param	type	初期値	詳細
expire	int	3600 * 24 * 30	Redis上のユーザの購入履歴とレコメンドする商品一覧データの保持期間（秒）
redis.host	str	'localhost'	-
redis.port	int	6379	-
redis.db	int	0	-
recommendation_count	int	10	Recommender().get()で返却されるレコメンドする商品の個数
recommendation.update_interval_sec	int	600	商品毎のレコメンドする商品の更新間隔（秒）。0なら購入時に必ず更新される。5以下は非推奨
recommendation.search_depth	int	100	レコメンドする商品決定時に探索する深さ。商品購入ユーザ履歴とユーザの商品購入履歴の探索個数を決定する。値を増やすと計算時間が増え、レコメンドする商品が変化しにくくなる。
recommendation.max_history	int	500	商品購入ユーザ履歴とユーザの商品購入履歴の最大記録数

RedisDB内のデータ構造

key name	type	ttl	trim
CF:USER:LIKE-HIS:{tag}:{user_id}	LIST	o	o
CF:GOODS:RECO:{tag}:{item_id}	ZSET	o	o
CF:GOODS:TAG:{item_id}	HASH	x	x
CF:GOODS:MUTEX:{tag}:{item_id}	STRING	o	x
CF:INDEX:GOODS-HIS:{tag}:{item_id}	LIST	x	o

実装例

そのうち増やしていきます

実装例1. Djangoでオススメユーザを表示する機能

Djangoで、このギルドを見たプレイヤーはこんなギルドを見ています機能を実装する場合のサンプルコードです。

django-model

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
from django.conf import settings


class GuildRecommendation(object):
    cf = None

    def __init__(self):
        self.cf = Recommender(settings.ANALYTICS_REDIS_SETTINGS)

    def like(self, player_id, guild_ids):
        """
        :param player_id: str
        :param guild_ids: list of int
        """
        for guild_id in guild_ids:
            self.cf.register(guild_id)
        self.cf.like(player_id, guild_ids)

    def gets(self, guild_id, count=5):
        return self.cf.get(guild_id, count=count)

django-view

# ギルドを見たタイミングで登録
GuildRecommendation().like(player.id, [guild_id])

# オススメギルドを取得
GuildRecommendation().gets(guild_id, count=20)
>>> [8, 4, 3]

実装例2. 商品を削除、更新する

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender


r = Recommender(settings={})

user_id = "user1"
goods_id = "Item1"

# ユーザを削除する。
# ユーザの購入履歴とINDEXからユーザの購入情報が削除される。
# ただしユーザの購入履歴が{recommendation.max_history}以上存在して一部履歴が既に存在しない場合
# INDEXにゴミデータとしてユーザの購入履歴が残り続ける。recreate_all_index関数でINDEXを再生成すると消える。
r.remove_user(user_id)

# 商品を削除する
# 商品のタグ情報と商品をKEYとするレコメンドする商品一覧が削除される
# ただし他の商品のレコメンドリストにデータが残り続ける。残データ削除には
# update_allを実行して全レコメンドリスト更新が必要。
r.remove_goods(goods_id)

# 商品タグの変更
# ただし他の商品のレコメンドリストにデータが残り続ける。残データ削除には
# update_allを実行して全レコメンドリスト更新が必要。
r.update_goods_tag(goods_id, "book")

# 全レコメンドリストを更新する。極めて時間が掛かる処理。
# 副次効果として削除した商品が他の商品のレコメンドリストから削除される。
# 商品数 x 100ms〜500ms掛かるため実装例4での分散更新を検討すること
r.update_all()

実装例3-1. 既存システムに導入する[データを貯めてから公開]

まずは登録機能を実装し、十分にデータが蓄積されたらViewを公開してください。

既存システムに導入する[データを貯めてから公開]

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender

# 商品購入履歴を登録
user_id = 'user-00001'
buy_items = ['Item10', 'Item10', 'Item10', 'Item3', 'Item3', 'Item1']
for item_id in buy_items:
    recommendation.register(item_id)
recommendation.like(user_id, buy_items)

実装例3-2. 既存システムに導入する[データを一括で登録]

データを一括で登録

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
import random


# 商品を一括登録する
tags = ['default', 'book', 'computer', 'dvd', 'camera', 'clothes', 'tag7', 'tag8', 'tag9', 'tag10']
settings = {}
r = Recommender(settings=settings)
goods_ids = range(1, 1000)
for goods_id in goods_ids:
    r.register(goods_id, tag=random.choice(tags))

# 商品の購入履歴を一括登録する
users = {
    'player1': [100, 200, 300],
    'player2': [100, 200, 300],
    'player3': [200, 300, 500],
    'player4': [500, 600, 700],
    'player5': [300, 400, 500],
}

ct = 0
for user_id in users:
    like_goods_ids = users.get(user_id)
    # レコメンドする商品一覧を更新せず購入履歴を登録
    r.like(user_id, like_goods_ids, realtime_update=False)
    if ct % 100 == 0:
        print "{}/{}".format(str(ct), str(len(users)))
    ct += 1

# index生成(全ユーザの購入履歴をメモリに展開するため、メモリ大量消費にご注意ください)
r.recreate_all_index()

# 全商品のレコメンドリストを生成(商品数 x 100ms〜500ms)
r.update_all()

実装例4. ワーカーでレコメンドする商品一覧を更新

ワーカーモデルで実装するとレコメンドする商品一覧を分散して更新出来ます。全レコメンドリストの更新には商品数x100〜500msの時間が必要です。削除した商品を他の商品のレコメンドリストから削除するためには全レコメンドリストの再生成が必要なので実装しました。また新規導入時にレコメンドする商品一覧生成を分散して計算したり、商品のタグ情報を一括変更する際に利用することを想定しています。

リアルタイム更新停止時のサンプル実装

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender

# 商品購入履歴を登録
user_id = 'user-00001'
buy_items = ['Item10', 'Item10', 'Item10', 'Item3', 'Item3', 'Item1']
for item_id in buy_items:
    recommendation.register(item_id)
# レコメンドする商品一覧を更新せず購入履歴を登録
recommendation.like(user_id, buy_items, realtime_update=False)

レコメンドする商品を4台のワーカーで更新するサンプル実装ワーカー1

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
Recommender(settings).update_all(scope=(0, 4))

レコメンドする商品を4台のワーカーで更新するサンプル実装ワーカー2

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
Recommender(settings).update_all(scope=(1, 4))

レコメンドする商品を4台のワーカーで更新するサンプル実装ワーカー3

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
Recommender(settings).update_all(scope=(2, 4))

レコメンドする商品を4台のワーカーで更新するサンプル実装ワーカー4

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from cf_recommender.recommender import Recommender
Recommender(settings).update_all(scope=(3, 4))

supervisordでワーカーを動かすと良い感じに動きます。scope=(0, 4)と設定するとソートした全商品リストを4分割し、前半1/4の商品に係るレコメンドリストを更新します。

レコメンドのチューニングを行う

■ 1. リアルタイム更新機能を有効にする
デフォルト設定ではリアルタイム更新機能はOFFになっています。有効にするにはrecommendation.update_interval_secに0を設定してください。ただしスパイク時にAPPサーバが死亡する可能性があるため十分なリソースを確保するか、更新間隔を5秒に設定してください。

■ 2. レコメンドされる商品がすぐ変化する
変化を穏やかにするにはrecommendation.search_depthを大きくして過去方向の履歴検索を強化してください。ただし計算時間が延びるためCPU負荷が大きくなります。

■ 3. レコメンドされる商品がなかなか更新されない
recommendation.update_interval_secを変更してレコメンドされる商品の更新間隔を短く設定してください。デフォルト値は10分です。

■ 4. 昔流行った商品をレコメンド一覧に加えたい
recommendation.search_depthとrecommendation.max_historyを延ばすことで実現できます。計算時間が大きく延びる可能性があるため変更の際は十分に試験を実施してください。計算時間肥大化の対策として実装例4のようにワーカーでレコメンドリストを生成し、リアルタイム更新を停止する方法があります。

# レコメンドする商品一覧を更新せず購入履歴を登録
recommendation.like(user_id, buy_items, realtime_update=False)

■ 5. 特定の商品をレコメンド一覧に加えたい
本エンジンの結果に別のエンジンの結果をマージして利用してください。

導入後に不具合が発生したら

設定変更でなんとかなるかもしれません。
■ 1. サーバのCPU使用率が跳ね上がった
Recommender.like関数内のレコメンドする商品一覧生成処理で時間が掛かっている可能性が高いです。次の設定を見直しましょう。
a. recommendation.update_interval_secの時間を延長して更新間隔を引き上げる。
b. recommendation.search_depthの値を減らし、レコメンドする商品一覧生成時の計算量を削減する。

■ 2. Redisのメモリ使用量が不足
a. expireの値を引き下げる。期間が過ぎると、期間中に一度も読み込まれなかったレコメンドする商品一覧が削除されます。
b. recommendation.max_historyの値を引き下げる。溢れた過去の購入履歴が消失します。

ベンチマーク結果

組み合わせ爆発が起こらないことをベンチマークとって確認してみました。全商品のレコメンドリスト更新に必要な時間を商品数で割り、1商品毎のレコメンドする商品一覧の更新時間を計算しています。MacBookPro2013のローカル環境で検証しています。テストデータ生成用コード

ライセンスとソースコード

https://github.com/subc/cf_recommender

関連記事

完成までの仕様確定や検証を行った記事です。
協調フィルタリング型レコメンドエンジン開発のため仕様について考える
協調フィルタリング（レコメンド）をredisとpythonで実装してみた

参考

• 推薦システムのアルゴリズム Algorithms of Recommender Systems 神嶌 敏弘
• レコメンドにおける類似度計算その傾向と対策 #DSIRNLP 第4回 2013.9.1

1件くらい商用利用されたらいいなー( ･ㅂ･)و
英文のファーストレビューで『やっば笑』『言いたいことはわかるけど、誤字がすごいから信用なくすパターン』と言わていたのでコード内の英文を信用してはいけない。レビューありがとうございました。