PersonalizeのAdditional objectiveを有効にして、レコメンデーションを試してみた
Personalizeの可能性を広げるAdditional objectiveの紹介
データアナリティクス事業本部機械学習チームの中村です。
今回は、Amazon PersonalizeのAdditional objectiveを有効にして、レコメンデーションを試してみます。
Additional objectiveとは
レコメンデーションは通常、購入などの可能性が高いものをユーザーにレコメンドすることを第一の目的とします。
一方、事例によっては全体の売上も並行して重視したいなどのケースも考えられます。
そういった場合に、Itemのメタデータの特定の部分（例えば価格など）を、追加のターゲットとして設定することで、 通常のレコメンデーションを達成しながら、売り上げを高くするなどの設定が可能です。
マネジメントコンソールでは以下の部分となります。
今回は実際にこれを有効にし、有効にしなかった場合と比較してみます。
実際にやってみる
モジュール
import boto3 from boto3.session import Session import json import pathlib import pandas as pd import numpy as np
定数定義
prefix = 'trial-20220401-xxxxxx'
s3_bucket_name = 'trial-20220401-xxxxxx'
# s3 uri input
s3_uri = f's3://{s3_bucket_name}'
import_interactions_uri = f'{s3_uri}/interactions.csv'
import_items_uri = f'{s3_uri}/items.csv'
batch_input_uri = f'{s3_uri}/batch-input.json'
batch_output_default_uri = f'{s3_uri}/output-default'
batch_output_price_uri = f'{s3_uri}/output-price'
# common
dataset_group_name = f'{prefix}-dataset-group'
iam_role_name = f'{prefix}-personalize-exection-role'
iam_custom_policy_name = f'{prefix}-personalize-execution-policy'
# for interactions
interaction_schema_name = f'{prefix}-interaction-schema'
interaction_dataset_name = f'{prefix}-interaction-dataset'
interaction_import_job_name = f'{prefix}-interaction-import-job'
# for items
item_schema_name = f'{prefix}-item-schema'
item_dataset_name = f'{prefix}-item-dataset'
item_import_job_name = f'{prefix}-item-import-job'
# for default solution
solution_default_name = f'{prefix}-solution-default'
batch_job_default_name = f'{prefix}-batch-job-default'
# for price emphasis solution
solution_price_name = f'{prefix}-solution-price-emphasis'
batch_job_price_name = f'{prefix}-batch-job-price-emphasis'
# client
client_s3 = boto3.client('s3' , region_name=region_name)
client_personalize = boto3.client('personalize', region_name=region_name)
client_iam = boto3.client('iam' , region_name=region_name)
データ準備
ml-test-small.zipのrating.csvを元にinteractions側を作成します。
df_ratings = pd.read_csv("ml-latest-small/ratings.csv")
# columnsの書き換え
df_ratings = df_ratings.rename(
columns={'userId': 'USER_ID', 'movieId': 'ITEM_ID', 'timestamp': 'TIMESTAMP'}
)
# ratingを削除
df_ratings = df_ratings.drop('rating', axis=1)
df_ratings.to_csv("./interactions.csv", index=False)
items側もmovies.csvを元に作成しますが、乱数で価格に相当するPRICEを生成します。
df_movies = pd.read_csv("ml-latest-small/movies.csv")
df_movies = df_movies.rename(columns={'movieId': 'ITEM_ID', 'title': 'TITLE', 'genres': 'GENRES'})
# PRICEを生成して追加
rns = np.random.RandomState(seed=123)
list_price = rns.randint(low=100, high=500, size=len(df_movies))
df_movies['PRICE'] = list_price
df_movies.to_csv("./items-add-price.csv" , index=False)
また後のバッチ推論用に、interactionの一部(10サンプル)のITEM_IDを出力しておきます。
pd.DataFrame(df_ratings_test['USER_ID'].unique(), columns=['userId']).\
sample(n=10, random_state=777).\
to_json('batch-input.json',orient='records',lines=True)
これにより生成された以下のファイルを、定数定義したs3_bucket_nameで指定したS3バケットにアップロードしておきます。
- interactions.csv
- items-add-price.csv
- batch-input.json
dataset group作成
response_create_dataset_group = client_personalize.create_dataset_group(
name=dataset_group_name,
)
schema作成
- interaction側
interaction_schema = {
"type": "record",
"name": "Interactions",
"namespace": "com.amazonaws.personalize.schema",
"fields": [
{
"name": "USER_ID",
"type": "string"
},
{
"name": "ITEM_ID",
"type": "string"
},
{
"name": "TIMESTAMP",
"type": "long"
}
],
"version": "1.0"
}
interaction_schema = json.dumps(interaction_schema)
response_create_interaction_schema = client_personalize.create_schema(
name=interaction_schema_name,
schema=interaction_schema,
)
- item側
PRICEが今回、Additional objectiveとして使用するmetadataとなります。
item_schema = {
"type": "record",
"name": "Items",
"namespace": "com.amazonaws.personalize.schema",
"fields": [
{
"name": "ITEM_ID",
"type": "string"
},
{
"name": "TITLE",
"type": "string"
},
{
"name": "GENRES",
"type": "string",
"categorical": true
},
{
"name": "PRICE",
"type": "int"
}
],
"version": "1.0"
}
item_schema = json.dumps(item_schema)
response_create_item_schema = client_personalize.create_schema(
name=item_schema_name,
schema=item_schema,
)
dataset作成
- interaction側
response_create_interaction_dataset = client_personalize.create_dataset(
name=interaction_dataset_name,
schemaArn=response_create_interaction_schema['schemaArn'],
datasetGroupArn=response_create_dataset_group['datasetGroupArn'],
datasetType='Interactions'
)
- item側
response_create_item_dataset = client_personalize.create_dataset(
name=item_dataset_name,
schemaArn=response_create_item_schema['schemaArn'],
datasetGroupArn=response_create_dataset_group['datasetGroupArn'],
datasetType='Items'
)
IAMポリシー・ロール作成
- IAMポリシー
iam_custom_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Action": [
"s3:ListBucket",
"s3:GetObject",
"s3:PutObject"
],
"Effect": "Allow",
"Resource": [
f"arn:aws:s3:::{bucket_name}",
f"arn:aws:s3:::{bucket_name}/*"
]
},
]
}
response_create_policy = client_iam.create_policy(
PolicyName=iam_custom_policy_name,
PolicyDocument=json.dumps(iam_custom_policy_document),
)
- IAMロール
assume_role_policy_document = {
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "personalize.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
create_role_response = client_iam.create_role(
RoleName = iam_role_name,
AssumeRolePolicyDocument = json.dumps(assume_role_policy_document)
)
client_iam.attach_role_policy(
RoleName = iam_role_name,
PolicyArn = "arn:aws:iam::aws:policy/AmazonS3FullAccess"
)
client_iam.attach_role_policy(
RoleName = iam_role_name,
PolicyArn = response_create_policy['Policy']['Arn']
)
response_get_role = client_iam.get_role(RoleName=iam_role_name)
import job作成
- interaction側
response_create_interaction_dataset_import_job = client_personalize.create_dataset_import_job(
jobName=interaction_import_job_name,
datasetArn=response_create_interaction_dataset['datasetArn'],
dataSource={
'dataLocation': import_interactions_uri
},
roleArn=response_get_role['roleArn']
)
- item側
response_create_item_dataset_import_job = client_personalize.create_dataset_import_job(
jobName=item_import_job_name,
datasetArn=response_create_item_dataset['datasetArn'],
dataSource={
'dataLocation': import_items_uri
},
roleArn=response_get_role['roleArn']
)
solution作成
- default側
response_create_default_solution = client_personalize.create_solution(
name=solution_default_name,
datasetGroupArn=response_create_dataset_group['datasetGroupArn'],
recipeArn='arn:aws:personalize:::recipe/aws-user-personalization',
)
- Additional objective側
solutionConfigがポイントです。itemAttributeで、目的変数とするmetadataのカラム名を指定します。objectiveSensitivityで、目的変数を優先する度合いを設定可能です。
response_create_price_solution = client_personalize.create_solution(
name=solution_price_name,
datasetGroupArn=response_create_dataset_group['datasetGroupArn'],
recipeArn='arn:aws:personalize:::recipe/aws-user-personalization',
solutionConfig={
'optimizationObjective': {
'itemAttribute': 'PRICE',
'objectiveSensitivity': 'MEDIUM',
}
}
)
solution version作成
- 2つ分実行します。
response_create_default_solution_version = client_personalize.create_solution_version(
solutionArn = response_create_default_solution['solutionArn']
)
response_create_price_solution_version = client_personalize.create_solution_version(
solutionArn = response_create_price_solution['solutionArn']
)
バッチ推論実行
- こちらも2つ分実行します。
- 件数は25個に設定します。
- 出力先は変更しないと、結果が上書きされるため注意が必要です。
response_create_default_batch_job = client_personalize.create_batch_inference_job(
jobName = batch_job_default_name,
solutionVersionArn = response_create_default_solution_version['solutionVersionArn'],
numResults = 25,
jobInput = {
"s3DataSource": {
"path": batch_input_uri
}
},
jobOutput = {
"s3DataDestination": {
"path": batch_output_default_uri
}
},
roleArn = response_get_role['roleArn']
)
response_create_price_batch_job = client_personalize.create_batch_inference_job(
jobName = batch_job_price_name,
solutionVersionArn = response_create_price_solution_version['solutionVersionArn'],
numResults = 25,
jobInput = {
"s3DataSource": {
"path": batch_input_uri
}
},
jobOutput = {
"s3DataDestination": {
"path": batch_output_price_uri
}
},
roleArn = response_get_role['roleArn']
)
結果の比較
- 結果は定数定義で指定したS3バケットの以下にそれぞれ格納されます。
- output-default/batch-input.json.out
- output-price/batch-input.json.out
- それぞれローカルに保存し、レコメンドされたアイテムのPRICEを比較してみましょう。
with open("./output-default/batch-input.json.out", 'rt') as f:
results_default = f.readlines()
with open("./output-price/batch-input.json.out", 'rt') as f:
results_price = f.readlines()
- １ユーザーについて比較します。
items_df = pd.read_csv('./items_add_price.csv')
items_dict = dict(zip( [ str(i) for i in items_df['ITEM_ID'] ], items_df['PRICE']))
result_default = json.loads(results_default[0])
default_items_price = [items_dict[i] for i in result_default['output']['recommendedItems']]
result_price = json.loads(results_price[0])
price_items_price = [items_dict[i] for i in result_default['output']['recommendedItems']]
- グラフを生成します。
import matplotlib.pyplot as plt
import matplotlib as mpl
mpl.style.use('ggplot')
fig = plt.figure(figsize=(15, 8), dpi=100)
ax1 = plt.axes()
ax1.plot(price_items_price)
ax1.plot(default_items_price)
ax1.set_ylabel("RECOMMENDATION ORDER")
ax1.set_ylabel("PRICE")
ax1.legend(["default", "price_objective"], fontsize=17, loc="upper left")
plt.tight_layout()
- Additional objectiveにより、よりPRICEの高いITEMが優先してレコメンドされています！！
まとめ
いかがでしたでしょうか？
Additional objectiveを優先しすぎると、買ってもらえないもののレコメンドが増えるリスクもありますが、 バランスを取りながら適用すれば、より効果的なレコメンドを実施することが可能です。
いろんなレコメンドの可能性を秘めている機能だなと個人的には思いました。
この記事がPersonalize活用のきっかけになれば幸いです。
