search

aws-powertools / powertools-lambda-python

A developer toolkit to implement Serverless best practices and increase developer velocity.
https://docs.powertools.aws.dev/lambda/python/latest/
MIT No Attribution
2.89k stars 398 forks source link

feat(data-classes): add Lex and Lex V2 #1014

Closed michaelbrewer closed 2 years ago

michaelbrewer commented 2 years ago

Is your feature request related to a problem? Please describe.

Python has no event source data classes for Lex V1 and Lex V2 event sources. Example documentation include mistakes and invoke copy and pasting handler functions for response. Blueprints are not pythonic or include v2 examples.

Describe the solution you'd like

Event Source Data class like (Go, Java, Typescript).

Features:

Describe alternatives you've considered

Building own duplicating effort

Additional context

Lex V1 docs:

Lex V2 docs:

Lex v1 schema:

{
    "messageVersion": "1.0",
    "invocationSource": "FulfillmentCodeHook or DialogCodeHook",
    "userId": "User ID specified in the POST request to Amazon Lex.",
    "sessionAttributes": {
      "key1": "value1",
      "key2": "value2"
    },
    "bot": {
      "name": "bot name",
      "alias": "bot alias",
      "version": "bot version"
    },
    "outputDialogMode": "Text or Voice, based on ContentType request header in runtime API request",
    "currentIntent": {
      "name": "intent-name",
      "slots": {
        "slot name1": "value1",
        "slot name2": "value2"
      },
      "slotDetails": {
        "slot name1": {
          "resolutions": [
            { "value1": "resolved value1" },
            { "value2": "resolved value2" }
          ],
          "originalValue": "original text"
        },
        "slot name2": {
          "resolutions": [
            { "value1": "resolved value1" },
            { "value2": "resolved value2" }
          ],
          "originalValue": "original text"
        }
      },
      "confirmationStatus": "None, Confirmed, or Denied (intent confirmation, if configured)"
    },
    "inputTranscript": "Text used to process the request",
    "requestAttributes": {
      "key1": "value1",
      "key2": "value2"
    }
  }

Lex V2 event structure:

{
    "messageVersion": "1.0",
    "invocationSource": "DialogCodeHook | FulfillmentCodeHook",
    "inputMode": "DTMF | Speech | Text",
    "responseContentType": "CustomPayload | ImageResponseCard | PlainText | SSML",
    "sessionId": "string",
    "inputTranscript": "string",
    "bot": {
        "id": "string",

        "name": "string",
        "aliasId": "string",
        "localeId": "string",
        "version": "string"
    },
    "interpretations": [
        {
            "intent": {

                "confirmationState": "Confirmed | Denied | None",
                "name": "string",
                "slots": {
                    "string": {
                        "value": {
                            "interpretedValue": "string",
                            "originalValue": "string",
                            "resolvedValues": [
                                "string"
                            ]
                        }
                    },
                    "string": {
                        "shape": "List",
                        "value": {
                            "interpretedValue": "string",
                            "originalValue": "string",
                            "resolvedValues": [
                                "string"
                            ]
                        },
                        "values": [
                            {

                                "shape": "Scalar",
                                "value": {
                                    "originalValue": "string",
                                    "interpretedValue": "string",
                                    "resolvedValues": [
                                        "string"
                                    ]
                                }
                            },
                            {

                                "shape": "Scalar",
                                "value": {

                                    "originalValue": "string",
                                    "interpretedValue": "string",
                                    "resolvedValues": [
                                        "string"
                                    ]
                                }
                            }
                        ]
                    }
                },
                "state": "Failed | Fulfilled | FulfillmentInProgress | InProgress | ReadyForFulfillment | Waiting",
                "kendraResponse": {
                    // Only present when intent is KendraSearchIntent. For details, see 
                    // https://docs.aws.amazon.com/kendra/latest/dg/API_Query.html#API_Query_ResponseSyntax
                }
            },
            "nluConfidence": {
                "score": number
            },
            "sentimentResponse": {

                "sentiment": "string",
                "sentimentScore": {
                    "mixed": number,
                    "negative": number,
                    "neutral": number,
                    "positive": number
                }
            }
        }
    ],
    "proposedNextState": {
        "dialogAction": {
            "slotToElicit": "string",
            "type": "Close | ConfirmIntent | Delegate | ElicitIntent | ElicitSlot"
        },
        "intent": {
            "name": "string",
            "confirmationState": "Confirmed | Denied | None",
            "slots": {},
            "state": "Failed | Fulfilled | InProgress | ReadyForFulfillment | Waiting"
        }
    },
    "requestAttributes": {
        "string": "string"

    },
    "sessionState": {
        "activeContexts": [
            {
                "name": "string",
                "contextAttributes": {
                    "string": "string"
                },
                "timeToLive": {
                    "timeToLiveInSeconds": number,
                    "turnsToLive": number
                }
            }
        ],
        "sessionAttributes": {
            "string": "string"
        },
        "runtimeHints": {
            "slotHints": {
                "string": {
                    "string": {
                        "runtimeHintValues": [
                            {
                                "phrase": "string"
                            },
                            {
                                "phrase": "string"
                            }
                        ]
                    }
                }
            }
        },
        "dialogAction": {
            "slotToElicit": "string",
            "type": "Close | ConfirmIntent | Delegate | ElicitIntent | ElicitSlot"
        },
        "intent": {
            "confirmationState": "Confirmed | Denied | None",
            "name": "string",
            "slots": {
                "string": {
                    "value": {
                        "interpretedValue": "string",
                        "originalValue": "string",
                        "resolvedValues": [
                            "string"
                        ]
                    }
                },
                "string": {
                    "shape": "List",
                    "value": {
                        "interpretedValue": "string",
                        "originalValue": "string",

                        "resolvedValues": [
                            "string"

                        ]
                    },
                    "values": [
                        {
                            "shape": "Scalar",
                            "value": {
                                "originalValue": "string",
                                "interpretedValue": "string",
                                "resolvedValues": [
                                    "string"

                                ]
                            }
                        },
                        {
                            "shape": "Scalar",
                            "value": {
                                "originalValue": "string",
                                "interpretedValue": "string",
                                "resolvedValues": [
                                    "string"
                                ]
                            }
                        }
                    ]
                }
            },
            "state": "Failed | Fulfilled | FulfillmentInProgress | InProgress | ReadyForFulfillment | Waiting",
            "kendraResponse": {
                // Only present when intent is KendraSearchIntent. For details, see
                // https://docs.aws.amazon.com/kendra/latest/dg/API_Query.html#API_Query_ResponseSyntax                     }
            },
            "originatingRequestId": "string"
        }
    },
    "transcriptions": [
        {
            "transcription": "string",
            "transcriptionConfidence": {
                "score": "number"
            },
            "resolvedContext": {
                "intent": "string"
            },
            "resolvedSlots": {
                "string": {
                    "shape": "List",
                    "value": {
                        "originalValue": "string",
                        "resolvedValues": [
                            "string"
                        ]
                    },
                    "values": [
                        {
                            "shape": "Scalar",
                            "value": {
                                "originalValue": "string",
                                "resolvedValues": [
                                    "string"
                                ]
                            }
                        },
                        {
                            "shape": "Scalar",
                            "value": {
                                "originalValue": "string",
                                "resolvedValues": [
                                    "string"
                                ]
                            }
                        }
                    ]
                }
            }
        }
    ]
}

Example Lex V2 utils:

"""
Standard util methods to manage dialog state
"""

import traceback
import json

def remove_inactive_context(context_list):
    if not context_list:
        return context_list
    new_context = []
    for context in context_list:
        time_to_live = context.get('timeToLive')
        if  time_to_live and time_to_live.get('turnsToLive') != 0:
            new_context.append(context)
    return new_context

def close(active_contexts, session_attributes, intent, messages):
    active_contexts = remove_inactive_context(active_contexts)
    intent['state'] = 'Fulfilled'
    return {
        'sessionState': {
            'activeContexts': active_contexts,
            'sessionAttributes': session_attributes,
            'dialogAction': {
                'type': 'Close'
            },
            'intent': intent
        },
        'requestAttributes': {},
        'messages': messages
    }

def elicit_intent(active_contexts, session_attributes, intent, messages):
    intent['state'] = 'Fulfilled'
    active_contexts = remove_inactive_context(active_contexts)
    if not session_attributes:
        session_attributes = {}
    session_attributes['previous_message'] = json.dumps(messages)
    session_attributes['previous_dialog_action_type'] = 'ElicitIntent'
    session_attributes['previous_slot_to_elicit'] = None
    session_attributes['previous_intent'] = intent['name']

    return {
        'sessionState': {
            'sessionAttributes': session_attributes,
            'activeContexts': active_contexts,
            'dialogAction': {
                'type': 'ElicitIntent'
            },
            "state": "Fulfilled"
        },
        'requestAttributes': {},
        'messages': messages
    }

def elicit_slot(slotToElicit, active_contexts, session_attributes, intent, messages):
    intent['state'] = 'InProgress'
    active_contexts = remove_inactive_context(active_contexts)
    if not session_attributes:
        session_attributes = {}
    session_attributes['previous_message'] = json.dumps(messages)
    session_attributes['previous_dialog_action_type'] = 'ElicitSlot'
    session_attributes['previous_slot_to_elicit'] = slotToElicit

    return {
        'sessionState': {
            'sessionAttributes': session_attributes,
            'activeContexts': active_contexts,
            'dialogAction': {
                'type': 'ElicitSlot',
                'slotToElicit': slotToElicit
            },
            'intent': intent
        },
        'requestAttributes': {},
        'messages': messages
    }

def confirm_intent(active_contexts, session_attributes, intent, messages, **previous_state):
    active_contexts = remove_inactive_context(active_contexts)
    del intent['state']
    if not session_attributes:
        session_attributes = {}
    session_attributes['previous_message'] = json.dumps(messages)
    session_attributes['previous_dialog_action_type'] = 'ConfirmIntent'
    session_attributes['previous_slot_to_elicit'] = None
    if previous_state:
        session_attributes['previous_dialog_action_type'] = previous_state.get('previous_dialog_action_type')
        session_attributes['previous_slot_to_elicit'] = previous_state.get('previous_slot_to_elicit')
    return {
            'sessionState': {
                'activeContexts': active_contexts,
                'sessionAttributes': session_attributes,
                'dialogAction': {
                    'type': 'ConfirmIntent'
                },
                'intent': intent
            },
            'requestAttributes': {},
            'messages': messages
        }

def delegate(active_contexts, session_attributes, intent):
    active_contexts = remove_inactive_context(active_contexts)
    return {
        'sessionState': {
            'activeContexts': active_contexts,
            'sessionAttributes': session_attributes,
            'dialogAction': {
                'type': 'Delegate'
            },
            'intent': intent,
            'state': 'ReadyForFulfillment'
        },
        'requestAttributes': {}
    }

def get_intent(intent_request):
    interpretations = intent_request['interpretations'];
    if len(interpretations) > 0:
        return interpretations[0]['intent']
    else:
        return None;

# Look for interpretedValue first. If not go for originalValue
def get_slot(slotname, intent, **kwargs):
    try:
        slot = intent['slots'].get(slotname)
        if not slot:
            return None
        slotvalue = slot.get('value')
        if slotvalue:
            interpretedValue = slotvalue.get('interpretedValue')
            originalValue = slotvalue.get('originalValue')
            if kwargs.get('preference') == 'interpretedValue':
                return interpretedValue
            elif kwargs.get('preference') == 'originalValue':
                return originalValue
            # where there is no preference
            elif interpretedValue:
                return interpretedValue
            else:
                return originalValue
        else:
            return None
    except:
        return None

def set_slot(slotname, slotvalue, intent):
    if slotvalue == None:
        intent['slots'][slotname] = None
    else:
        intent['slots'][slotname] = {
                "value": {
                "interpretedValue": slotvalue,
                "originalValue": slotvalue,
                "resolvedValues": [
                    slotvalue
                ]
            }
        }

def get_multi_valued_slot(slotname, intent):
    try:
        values = intent['slots'][slotname]['values']
        if not values:
            return None
        slot_values = [{'interpretedValue':item['value']['interpretedValue'],
                        'originalValue':item['value']['originalValue']}  for item in values]
        return slot_values
    except:
        try:
            return intent['slots'][slotname]['value']['originalValue']
        except:
            return None

def get_multi_valued_slot_originalvalue(slotname, intent):
    try:
        values = intent['slots'][slotname]['values']
        if not values:
            return None
        original_slot_values = [item['value']['originalValue'] for item in values]
        return original_slot_values
    except:
        try:
            return intent.get['slots'][slotname]['value']['originalValue']
        except:
            return None

def get_active_contexts(intent_request):
    try:
        return intent_request['sessionState'].get('activeContexts')
    except:
        return []

def get_context_attribute(active_contexts, context_name, attribute_name):
    try:
        context = list(filter(lambda x: x.get('name') == context_name, active_contexts))
        return context[0]['contextAttributes'].get(attribute_name)
    except Exception:
        return None

def get_session_attributes(intent_request):
    try:
        return intent_request['sessionState']['sessionAttributes']
    except:
        return {}

def get_session_attribute(intent_request, session_attribute):
    try:
        return intent_request['sessionState']['sessionAttributes'].get(session_attribute)
    except:
        return None

def set_session_attribute(intent_request, session_attribute, value):
    try:
        if intent_request['sessionState'].get('sessionAttributes'):
            intent_request['sessionState']['sessionAttributes'][session_attribute] = value
        else:
            intent_request['sessionState']['sessionAttributes'] = {}
            intent_request['sessionState']['sessionAttributes'][session_attribute] = value
        return intent_request

    except:    
        return intent_request

def set_active_contexts(intent_request, context_name, context_attributes, time_to_live, turns_to_live):
    try:
        active_contexts = intent_request.get('sessionState')['activeContexts']
        if not active_contexts:
            intent_request.get('sessionState')['activeContexts'] = []
    except KeyError:
        intent_request.get('sessionState')['activeContexts'] = []
        active_contexts = intent_request.get('sessionState')['activeContexts']
    except Exception:
        return []
    finally:
        active_contexts.append({
            'name': context_name,
            'contextAttributes': context_attributes,
            "timeToLive": {
                "timeToLiveInSeconds": time_to_live,
                "turnsToLive": turns_to_live
            }
        })
        intent_request.get('sessionState')['activeContexts'] = active_contexts

def get_interpreted_intents(intent_request):
    try:
        intents = [{'name':intents_list['intent']['name'], 'nluConfidence':intents_list.get('nluConfidence')}  
                        for intents_list in intent_request['interpretations']]
        return intents
    except:
        return []

def get_previous_slot_to_elicit(intent_request):
    session_attributes = get_session_attributes(intent_request)
    if session_attributes: return session_attributes.get('previous_slot_to_elicit')
    return None

Example events:

Lex V1 sample event:

{
    "messageVersion": "1.0",
    "invocationSource": "DialogCodeHook",
    "userId": "John",
    "sessionAttributes": {
      "key": "value"
    },
    "bot": {
      "name": "BookTrip",
      "alias": "$LATEST",
      "version": "$LATEST"
    },
    "outputDialogMode": "Text",
    "currentIntent": {
      "name": "BookHotel",
      "slots": {
        "Location": "Chicago",
        "CheckInDate": "2030-11-08",
        "Nights": 4,
        "RoomType": "queen"
      },
      "confirmationStatus": "None"
    }
}

Lex V2: Captured event from Banking Bot example app

{
    "sessionId": "254688924456798",
    "inputTranscript": "01/01/1990",
    "interpretations": [
        {
            "intent": {
                "slots": {
                    "dateofBirth": {
                        "shape": "Scalar",
                        "value": {
                            "originalValue": "01/01/1990",
                            "resolvedValues": [
                                "1990-01-01"
                            ],
                            "interpretedValue": "1990-01-01"
                        }
                    },
                    "accountType": {
                        "shape": "Scalar",
                        "value": {
                            "originalValue": "savings",
                            "resolvedValues": [
                                "Savings"
                            ],
                            "interpretedValue": "Savings"
                        }
                    }
                },
                "confirmationState": "None",
                "name": "CheckBalance",
                "state": "ReadyForFulfillment"
            },
            "nluConfidence": 1
        },
        {
            "intent": {
                "slots": {},
                "confirmationState": "None",
                "name": "FallbackIntent",
                "state": "ReadyForFulfillment"
            }
        },
        {
            "intent": {
                "slots": {},
                "confirmationState": "None",
                "name": "Welcome",
                "state": "ReadyForFulfillment"
            },
            "nluConfidence": 0.23
        }
    ],
    "responseContentType": "text/plain; charset=utf-8",
    "sessionState": {
        "sessionAttributes": {},
        "activeContexts": [],
        "intent": {
            "slots": {
                "dateofBirth": {
                    "shape": "Scalar",
                    "value": {
                        "originalValue": "01/01/1990",
                        "resolvedValues": [
                            "1990-01-01"
                        ],
                        "interpretedValue": "1990-01-01"
                    }
                },
                "accountType": {
                    "shape": "Scalar",
                    "value": {
                        "originalValue": "savings",
                        "resolvedValues": [
                            "Savings"
                        ],
                        "interpretedValue": "Savings"
                    }
                }
            },
            "confirmationState": "None",
            "name": "CheckBalance",
            "state": "ReadyForFulfillment"
        },
        "originatingRequestId": "f57dfc3f-44be-4df9-ae72-9681fc14e67f"
    },
    "messageVersion": "1.0",
    "invocationSource": "FulfillmentCodeHook",
    "transcriptions": [
        {
            "transcription": "01/01/1990",
            "transcriptionConfidence": 1,
            "resolvedSlots": {
                "dateofBirth": {
                    "shape": "Scalar",
                    "value": {
                        "originalValue": "01/01/1990",
                        "resolvedValues": [
                            "1990-01-01"
                        ]
                    }
                }
            },
            "resolvedContext": {
                "intent": "CheckBalance"
            }
        }
    ],
    "inputMode": "Text",
    "bot": {
        "aliasName": "TestBotAlias",
        "aliasId": "TSTALIASID",
        "name": "BankingBot",
        "version": "DRAFT",
        "localeId": "en_US",
        "id": "J866BA0UQC"
    }
}
heitorlessa commented 2 years ago

Thanks for the feature request, Michael! Adding a label to hear from more customers looking to use it.

github-actions[bot] commented 2 years ago

Comments on closed issues are hard for our team to see.