Summary
AWS Lex has a full SDK for model building and run time execution. In this post, I'll demonstrate use of that SDK in Python. I'll demonstrate how create a simple/toy chatbot, integrate with a Lambda validation function, do a real time test, then finally - delete the bot. Use of the AWS console will not be necessary at all. All provisioning will be done in code.
AWS Lex Architecture
Below is a diagram of the overall architecture. The AWS Python SDK is used here for provisioning. Lambda is used for real time validation of data. In this particular bot, I'm using a 3rd party (SmartyStreets) for validating street addresses. That consists of a web service call from Lambda itself.
Bot-specific Architecture
Below is a diagram of how Lex bots are constructed. Lex bot is goal-oriented sort of chatbot. Goals are called 'Intents'. Items necessary to fulfill an Intent are called Slots. A bot consists of a bot definition in which Intent definitions are referenced. Intents can include references to custom Slot Type definitions. Intents are also where Lambda function calls for validation of slot input and overall fulfillment are made.
Bot Provisioning Architecture
Below is an architectural diagram of my particular provisioning application. The Python application itself is composed of generic AWS SDK function calls. All of the bot-specific provisioning configuration exists in JSON files.
Lex Provisioning Code
My Lex bot provisioning code consists of a single Python class. That class gets its configuration info from an external config file.
if __name__ == '__main__':
bot = AWSBot('awsbot.cfg')
bot.build()
bot.test('I want to order 2 cords of split firewood to be delivered at 1 pm on tomorrow to 900 Tamarac Pkwy 80863')
bot.destroy()
Lines 1-5: The AWSBot class exposes a simple interface to build, test, and destroy a bot on Lex.As mentioned, all configuration is driven by a single config file and multiple JSON files.
class AWSBot(object):
def __init__(self, config):
logger.debug('Entering')
self.bot, self.slots, self.intents, self._lambda, self.permission = self.__loadResources(config)
self.buildClient = boto3.client('lex-models')
self.testClient = boto3.client('lex-runtime')
self.lambdaClient = boto3.client('lambda')
logger.debug('Exiting')
def __loadResources(self, config):
logger.debug('Entering')
cfgParser = configparser.ConfigParser()
cfgParser.optionxform = str
cfgParser.read(config)
filename = cfgParser.get('AWSBot', 'botJsonFile')
with open(filename, 'r') as file:
bot = json.load(file)
slotsDir = cfgParser.get('AWSBot', 'slotsDir')
slots = []
for root,_,filenames in os.walk(slotsDir):
for filename in filenames:
with open(os.path.join(root,filename), 'r') as file:
jobj = json.load(file)
slots.append(jobj)
logger.debug(json.dumps(jobj, indent=4, sort_keys=True))
intentsDir = cfgParser.get('AWSBot', 'intentsDir')
intents = []
for root,_,filenames in os.walk(intentsDir):
for filename in filenames:
with open(os.path.join(root,filename), 'r') as file:
jobj = json.load(file)
intents.append(jobj)
logger.debug(json.dumps(jobj, indent=4, sort_keys=True))
filename = cfgParser.get('AWSBot', 'lambdaJsonFile')
dirname = os.path.dirname(filename)
with open(filename, 'r') as file:
_lambda = json.load(file)
with open(os.path.join(dirname,_lambda['Code']['ZipFile']), 'rb') as zipFile:
zipBytes = zipFile.read()
_lambda['Code']['ZipFile'] = zipBytes
filename = cfgParser.get('AWSBot', 'permissionJsonFile')
with open(filename, 'r') as file:
permission = json.load(file)
return bot, slots, intents, _lambda, permission
Lines 1-8: Load up dict objects with the Lex config and instantiate the AWS SDK objects.Lines 10-14: Read a config file that holds directory paths to the JSON files used to provision Lex.
Lines 16-18: Load a dict object with the a JSON Lex Bot definition file.
Lines 20-27: Load a dict object with a custom slot type JSON definition.
Lines 29-35: Load a dict object with the Intent JSON definition.
Lines 38-44: Load a dict object with Lambda code hook definition. Read the bytes of a zip file containing the Python code hook along with all non-AWS-standard libraries it references.
Lines 46-48: Load a dict object with the attributes necessary to add permission for the Lambda code hook to be called from Lex.
The public build interface consists calls to private methods to build the various Lex-related objects: Lambda code hook, slot types, intents, and finally the bot itself.
def build(self):
logger.debug('Entering')
self.__buildLambda()
self.__buildSlotTypes()
self.__buildIntents()
self.__buildBot()
logger.debug('Exiting')
Below is the code for the private build methods: def __buildLambda(self):
logger.debug('Entering')
resp = self.lambdaClient.create_function(**self._lambda)
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
resp = self.lambdaClient.add_permission(**self.permission)
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
logger.debug('Exiting')
def __buildSlotTypes(self):
logger.debug('Entering')
for slot in self.slots:
resp = self.buildClient.put_slot_type(**slot)
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
logger.debug('Exiting')
def __buildIntents(self):
logger.debug('Entering')
for intent in self.intents:
resp = self.buildClient.put_intent(**intent)
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
logger.debug('Exiting')
def __buildBot(self):
logger.debug('Entering')
self.buildClient.put_bot(**self.bot)
complete = False
for _ in range(20):
time.sleep(20)
resp = self.buildClient.get_bot(name=self.bot['name'], versionOrAlias='$LATEST')
logger.debug(resp['status'])
if resp['status'] == 'FAILED':
logger.debug('***Bot Build Failed***')
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
complete = True
break
elif resp['status'] == 'READY':
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
complete = True
break
if not complete:
logger.debug('***Bot Build Timed Out***')
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
logger.debug('Exiting')
Lines 1-7: Call the AWS Lambda SDK client to create the function with the previously-loaded JSON definition. Add the permission for the Intent to call that Lambda function.Lines 9-14: Loop through the slots JSON definitions and build each via AWS SDK call.
Lines 16-21: Same thing, but with the Intents.
Lines 23-44: Build the bot with its JSON definition. Although this SDK call is synchronous and returns almost immediately, the Bot will not be completed upon return from the call. It takes around 1-2 minutes. The for loop here is checking AWS's progress on the Bot build every 20 sec.
After the Bot is complete, the Lex runtime SDK can be used to test it with a sample utterance.
def test(self, msg):
logger.debug('Entering')
params = {
'botAlias': '$LATEST',
'botName': self.bot['name'],
'inputText': msg,
'userId': 'fred',
}
resp = self.testClient.post_text(**params)
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
logger.debug('Exiting')
Deleting/cleaning-up the Lex objects on AWS is a simple matter of making the corresponding delete function calls from the SDK. Similar to the build process, deleting an object takes time on AWS, even though the function may return immediately. You can mitigate the issues delays associated with deletion and corresponding dependencies by putting artificial delays in the code, such as below.def __destroyBot(self):
logger.debug('Entering')
try:
resp = self.buildClient.delete_bot(name=self.bot['name'])
logger.debug(json.dumps(resp, indent=4, sort_keys=True, default=self.__dateSerializer))
except Exception as err:
logger.debug(err)
time.sleep(5) #artificial delay to allow the operation to be completed on AWS
logger.debug('Exiting')
Lambda Code Hook
If you require any logic for data validation or fulfillment (and you will for any real bot implementation), there is no choice but to use AWS Lambda for that function. That Lambda function needs a single entry point where the Lex event (Dialog validation and/or Fulfillment) is passed. Below is entry point for the function I developed. All the validation logic is contained in single Python class - LexHandler.
def lambda_handler(event, context):
handler = LexHandler(event)
os.environ['TZ'] = 'America/Denver'
time.tzset()
return handler.respond()
I'm not going to post all the code for the handler class as it's pretty straight-forward (full source will be on github as well), but here's one snippet of the Address validation. It actually makes a call to an external webservice (SmartyStreets) to perform the validation function. def __isValidDeliveryStreet(self, deliveryStreet, deliveryZip):
if deliveryStreet and deliveryZip:
credentials = StaticCredentials(AUTH_ID, AUTH_TOKEN)
client = ClientBuilder(credentials).build_us_street_api_client()
lookup = Lookup()
lookup.street = deliveryStreet
lookup.zipcode = deliveryZip
try:
client.send_lookup(lookup)
except exceptions.SmartyException:
return False
if lookup.result:
return True
else:
return False
else:
return False
Full source here: https://github.com/joeywhelan/AWSBot
Copyright ©1993-2024 Joey E Whelan, All rights reserved.
