Skip to content

Instantly share code, notes, and snippets.

@alexcasalboni alexcasalboni/lex.py
Created May 29, 2019

Embed
What would you like to do?
Amazon Lex fulfillment function - Lambda handler (Python) + utilities
import logging
from lex_utils import elicit_slot, delegate, close, ElicitAction, DelegateAction
from utils import validate_dialog, init_or_load_session, finalize_session, actually_book_the_hotel
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
def lambda_handler(event, context):
logger.debug('event.bot.name=%s', event['bot']['name'])
logger.debug('userId=%s, intentName=%s', event['userId'], event['currentIntent']['name'])
intent_name = event['currentIntent']['name']
if intent_name == 'BookHotel':
return book_hotel(event)
# elif (add more intents here)
else:
raise Exception('Intent with name %s not supported' % intent_name)
def book_hotel(event):
session_attributes = init_or_load_session(event)
slots = event['currentIntent']['slots']
name = event['currentIntent']['name']
try:
if event['invocationSource'] == 'DialogCodeHook':
validate_dialog(slots, session_attributes)
except ElicitAction as ea:
# request a new value
return elicit_slot(
session_attributes,
name,
slots,
ea.invalid_slot, # elicit this invalid slot
ea.message, # with this custom message
)
except DelegateAction as da:
# tell Lex to move on with the next slot
return delegate(session_attributes, slots)
# ok, we have all the slots and we can finally book the hotel!
actually_book_the_hotel(session_attributes)
finalize_session(session_attributes)
return close(
session_attributes,
'Thanks, I have placed your hotel reservation. What shall we do next?',
)
""" Some serialization utilities for Amazon Lex """
class ElicitAction(Exception):
def __init__(self, invalid_slot, message):
super(ElicitAction, self).__init__()
self.invalid_slot = invalid_slot
self.message = message
class DelegateAction(Exception):
pass
def elicit_slot(session_attributes, intent_name, slots, slot_to_elicit, message):
return {
'sessionAttributes': session_attributes,
'dialogAction': {
'type': 'ElicitSlot',
'intentName': intent_name,
'slots': slots,
'slotToElicit': slot_to_elicit,
'message': message
}
}
def close(session_attributes, message, fulfillment_state='Fulfilled'):
response = {
'sessionAttributes': session_attributes,
'dialogAction': {
'type': 'Close',
'fulfillmentState': fulfillment_state,
'message': {
'contentType': 'PlainText',
'content': message,
}
}
}
return response
def delegate(session_attributes, slots):
return {
'sessionAttributes': session_attributes,
'dialogAction': {
'type': 'Delegate',
'slots': slots
}
}
def build_validation_result(isvalid, violated_slot, message_content):
return {
'isValid': isvalid,
'violatedSlot': violated_slot,
'message': {
'contentType': 'PlainText',
'content': message_content,
}
}
""" Some validation utilities """
import logging
import json
import datetime
import dateutil.parser
from lex_utils import build_validation_result, ElicitAction, DelegateAction
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
def generate_hotel_price(location, nights, room_type):
"""
Generates a number within a reasonable range that might be expected for a hotel.
The price is fixed for a pair of location and roomType.
"""
room_types = ['queen', 'king', 'deluxe']
cost_of_living = 0
for i in range(len(location)):
cost_of_living += ord(location.lower()[i]) - 97
return nights * (100 + cost_of_living + (100 + room_types.index(room_type.lower())))
def isvalid_city(city):
valid_cities = ['new york', 'los angeles', 'chicago', 'houston', 'philadelphia', 'phoenix', 'san antonio',
'san diego', 'dallas', 'san jose', 'austin', 'jacksonville', 'san francisco', 'indianapolis',
'columbus', 'fort worth', 'charlotte', 'detroit', 'el paso', 'seattle', 'denver', 'washington dc',
'memphis', 'boston', 'nashville', 'baltimore', 'portland']
return city.lower() in valid_cities
def isvalid_room_type(room_type):
room_types = ['queen', 'king', 'deluxe']
return room_type.lower() in room_types
def isvalid_date(date):
try:
dateutil.parser.parse(date)
return True
except ValueError:
return False
def validate_slots(slots):
location = slots.get('Location')
checkin_date = slots.get('CheckInDate')
nights = int(slots.get('Nights') or 0)
room_type = slots.get('RoomType')
if location and not isvalid_city(location):
return build_validation_result(False, 'Location', 'We currently do not support {} as a valid destination. Can you try a different city?'.format(location))
if checkin_date:
if not isvalid_date(checkin_date):
return build_validation_result(False, 'CheckInDate', 'I did not understand your check in date. When would you like to check in?')
if datetime.datetime.strptime(checkin_date, '%Y-%m-%d').date() <= datetime.date.today():
return build_validation_result(False, 'CheckInDate', 'Reservations must be scheduled at least one day in advance. Can you try a different date?')
if not 0 < nights < 30:
return build_validation_result(False, 'Nights', 'You can make a reservations for from one to thirty nights. How many nights would you like to stay for?')
if room_type and not isvalid_room_type(room_type):
return build_validation_result(False, 'RoomType', 'I did not recognize that room type. Would you like to stay in a queen, king, or deluxe room?')
return {
'isValid': True,
}
def init_or_load_session(event):
current = event['currentIntent']
location = current['slots'].get('Location')
checkin_date = current['slots'].get('CheckInDate')
nights = int(current['slots'].get('Nights') or 0)
room_type = current['slots'].get('RoomType')
# serialize new session data
reservation = {
'ReservationType': 'Hotel',
'Location': location,
'RoomType': room_type,
'CheckInDate': checkin_date,
'Nights': nights
}
# fetch or initialize session
session_attributes = event.get('sessionAttributes') or {}
# update session data
session_attributes['currentReservation'] = json.dumps(reservation)
return session_attributes
def finalize_session(session_attributes):
# clear/update session with final attributes
reservation = session_attributes.pop('currentReservation', None)
session_attributes['lastConfirmedReservation'] = reservation
def actually_book_the_hotel(session_attributes):
reservation = session_attributes['lastConfirmedReservation']
logger.debug('bookHotel under=%s', json.dumps(reservation))
# TODO actually book the hotel :)
def validate_dialog(slots, session_attributes):
# Validate any slots which have been specified.
# If any are invalid, re-elicit for their value
validation_result = validate_slots(slots)
if not validation_result['isValid']:
invalid_slot = validation_result['violatedSlot']
slots[invalid_slot] = None
raise ElicitAction(
invalid_slot=invalid_slot,
message=validation_result['message'],
)
reservation = json.loads(session_attributes['currentReservation'])
if reservation['Location'] and reservation['Nights'] and reservation['RoomType']:
# ok, slots are valid, we can generate the total price
reservation['currentReservationPrice'] = generate_hotel_price(
location=reservation['Location'],
nights=reservation['Nights'],
room_type=reservation['RoomType'],
)
# re-serialize reservation into session (with new price)
session_attributes['currentReservation'] = json.dumps(reservation)
raise DelegateAction()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.