in src/recommendations/src/recommendations-service/app.py [0:0]
def coupon_offer():
"""
Returns an offer recommendation for a given user.
Hits the offers endpoint to find what offers are available, get their preferences for adjusting scores.
Uses Amazon Personalize if available to score them.
Returns the highest scoring offer.
Experimentation is disabled because we are sending the offers through Pinpoint emails and for this
demonstration we would need to add some more complexity to track those within the current framework.
Pinpoint also has A/B experimentation built in which can be used.
"""
user_id = request.args.get('userID')
if not user_id:
raise BadRequest('userID is required')
resp_headers = {}
try:
campaign_arn = get_parameter_values(offers_arn_param_name)[0]
offers_service_host, offers_service_port = get_offers_service()
url = f'http://{offers_service_host}:{offers_service_port}/offers'
app.logger.debug(f"Asking for offers info from {url}")
offers_response = requests.get(url)
app.logger.debug(f"Got offer info: {offers_response}")
if not offers_response.ok:
app.logger.exception('Offers service did not return happily', offers_response.reason())
raise BadRequest(message='Cannot obtain offers', status_code=500)
else:
offers = offers_response.json()['tasks']
offers_by_id = {str(offer['id']): offer for offer in offers}
offer_ids = sorted(list(offers_by_id.keys()))
if not campaign_arn:
app.logger.warning('No campaign Arn set for offers - returning arbitrary')
# We deterministically choose an offer
# - random approach would have been chosen_offer_id = random.choice(offer_ids)
chosen_offer_id = offer_ids[int(user_id) % len(offer_ids)]
chosen_score = None
else:
resp_headers['X-Personalize-Recipe'] = get_recipe(campaign_arn)
logger.info(f"Input to Personalized Ranking for offers: userId: {user_id}({type(user_id)}) "
f"inputList: {offer_ids}")
get_recommendations_response = personalize_runtime.get_recommendations(
campaignArn=campaign_arn,
userId=user_id,
numResults=len(offer_ids)
)
logger.info(f'Recommendations returned: {json.dumps(get_recommendations_response)}')
return_key = 'itemList'
# Here is where might want to incorporate some business logic
# for more information on how these scores are used see
# https://aws.amazon.com/blogs/machine-learning/introducing-recommendation-scores-in-amazon-personalize/
# An alternative approach would be to train Personalize to produce recommendations based on objectives
# we specify rather than the default which is to maximise the target event. For more information, see
# https://docs.aws.amazon.com/personalize/latest/dg/optimizing-solution-for-objective.html
user_scores = {item['itemId']: float(item['score']) for item in
get_recommendations_response[return_key]}
# We assume we have pre-calculated the adjusting factor, can be a mix of probability when applicable,
# calculation of expected return per offer, etc.
adjusted_scores = {offer_id: score * offers_by_id[offer_id]['preference']
for offer_id, score in user_scores.items()}
logger.info(f"Scores after adjusting for preference parameters: {adjusted_scores}")
# Normalise these - makes it easier to do further adjustments
score_sum = sum(adjusted_scores.values())
adjusted_scores = {offer_id: score / score_sum
for offer_id, score in adjusted_scores.items()}
logger.info(f"Scores after normalising adjusted scores: {adjusted_scores}")
# Just one way we could add some randomness - adds serendipity though removes personalization a bit
# because we have the scores though we retain a lot of the personalization
random_factor = 0.0
adjusted_scores = {offer_id: score*(1-random_factor) + random_factor*random.random()
for offer_id, score in adjusted_scores.items()}
logger.info(f"Scores after adding randomness: {adjusted_scores}")
# We can do many other things here, like randomisation, normalisation in different dimensions,
# tracking per-user, offer, quotas, etc. Here, we just select the most promising adjusted score
chosen_offer_id = max(adjusted_scores, key=adjusted_scores.get)
chosen_score = user_scores[chosen_offer_id]
chosen_adjusted_score = adjusted_scores[chosen_offer_id]
chosen_offer = offers_by_id[chosen_offer_id]
if chosen_score is not None:
chosen_offer['score'] = chosen_score
chosen_offer['adjusted_score'] = chosen_adjusted_score
resp = Response(json.dumps({'offer': chosen_offer}, cls=CompatEncoder),
content_type='application/json', headers=resp_headers)
app.logger.debug(f"Recommendations response to be returned for offers: {resp}")
return resp
except Exception as e:
app.logger.exception('Unexpected error generating recommendations', e)
raise BadRequest(message='Unhandled error', status_code=500)