in src/recommendations/src/recommendations-service/app.py [0:0]
def get_ranking(user_id, items, feature,
default_campaign_arn_param_name='retaildemostore-personalized-ranking-campaign-arn',
top_n=None, context=None):
"""
Re-ranks a list of items using personalized reranking.
Or delegates to experiment manager if there is an active experiment.
Args:
user_id (int):
items (list[dict]): e.g. [{"itemId":"33", "url":"path_to_product33"},
{"itemId":"22", "url":"path_to_product22"}]
feature: Used to lookup the currently active experiment.
default_campaign_arn_param_name: For discounts this would be different.
top_n (Optional[int]): Only return the top N ranked if not None.
context (Optional[dict]): If available, passed to the reranking Personalization recipe.
Returns:
Items as passed in, but ordered according to reranker - also might have experimentation metadata added.
"""
app.logger.info(f"Items given for ranking: {items}")
# Extract item IDs from items supplied by caller. Note that unranked items
# can be specified as a list of objects with just an 'itemId' key or as a
# list of fully defined items/products (i.e. with an 'id' key).
item_map = {}
unranked_items = []
for item in items:
item_id = item.get('itemId') if item.get('itemId') else item.get('id')
item_map[item_id] = item
unranked_items.append(item_id)
app.logger.info(f"Unranked items: {unranked_items}")
resp_headers = {}
experiment = None
exp_manager = None
# Get active experiment if one is setup for feature.
if feature:
exp_manager = ExperimentManager()
experiment = exp_manager.get_active(feature)
if experiment:
app.logger.info('Using experiment: ' + experiment.name)
# Get ranked items from experiment.
tracker = exp_manager.default_tracker()
ranked_items = experiment.get_items(
user_id=user_id,
item_list=unranked_items,
tracker=tracker,
context=context
)
app.logger.debug(f"Experiment ranking resolver gave us this ranking: {ranked_items}")
resp_headers['X-Experiment-Name'] = experiment.name
resp_headers['X-Experiment-Type'] = experiment.type
resp_headers['X-Experiment-Id'] = experiment.id
else:
# Fallback to default behavior of checking for campaign ARN parameter and
# then the default product resolver.
values = get_parameter_values([default_campaign_arn_param_name, filter_purchased_param_name])
app.logger.info(f'Falling back to Personalize: {values}')
campaign_arn = values[0]
filter_arn = values[1]
if campaign_arn:
resolver = PersonalizeRankingResolver(campaign_arn=campaign_arn, filter_arn=filter_arn)
resp_headers['X-Personalize-Recipe'] = get_recipe(campaign_arn)
else:
app.logger.info(f'Falling back to No-op: {values}')
resolver = RankingProductsNoOpResolver()
ranked_items = resolver.get_items(
user_id=user_id,
product_list=unranked_items,
context=context
)
response_items = []
if top_n is not None:
# We may not want to return them all - for example in a "pick the top N" scenario.
ranked_items = ranked_items[:top_n]
for ranked_item in ranked_items:
# Unlike with /recommendations and /related we are not hitting the products API to get product info back
# The caller may have left that info in there so in case they have we want to leave it in.
item = item_map.get(ranked_item.get('itemId'))
if 'experiment' in ranked_item:
item['experiment'] = ranked_item['experiment']
if 'url' in item:
# Append the experiment correlation ID to the product URL so it gets tracked if used by client.
product_url = item.get('url')
if '?' in product_url:
product_url += '&'
else:
product_url += '?'
product_url += 'exp=' + ranked_item['experiment']['correlationId']
item['url'] = product_url
response_items.append(item)
return response_items, resp_headers