Skip to content

Core API

The core module contains the main functionality for handling Amazon Lex requests, managing dialog state, and processing session attributes. This is where you'll find the primary LexHelper class and essential utilities.

Main Handler

LexHelper

lex_helper.core.handler.LexHelper 

LexHelper(config: Config[T])
Source code in lex_helper/core/handler.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(self, config: Config[T]):
    self.config = config

    # Initialize disambiguation components if enabled
    self.disambiguation_handler = None
    self.disambiguation_analyzer = None

    if self.config.enable_disambiguation and disambiguation_available:
        disambiguation_config = self.config.disambiguation_config or DisambiguationConfig()  # type: ignore
        self.disambiguation_handler = DisambiguationHandler(disambiguation_config)  # type: ignore
        self.disambiguation_analyzer = DisambiguationAnalyzer(disambiguation_config)  # type: ignore
        logger.debug("Disambiguation components initialized")
    elif self.config.enable_disambiguation and not disambiguation_available:
        logger.warning("Disambiguation requested but components not available - falling back to regular behavior")

Functions

handler 

handler(event: dict[str, Any], context: Any) -> dict[str, Any]

Primary entry point for the lex_helper library.

This function is designed to handle AWS Lambda events triggered by Amazon Lex. It processes the incoming event and context, and utilizes custom session attributes if provided. The function orchestrates the entire flow, including parsing the Lex request, handling intents, managing session state, and formatting the response for the channel.

Parameters:

Name Type Description Default
event dict[str, Any]

The event data from AWS Lambda, typically containing the Lex request.

 required
context Any

The context object provided by AWS Lambda, containing runtime information.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: A formatted response ready to be sent back to Amazon Lex.
Source code in lex_helper/core/handler.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def handler(self, event: dict[str, Any], context: Any) -> dict[str, Any]:
    """
    Primary entry point for the lex_helper library.

    This function is designed to handle AWS Lambda events triggered by Amazon Lex. It processes the incoming
    event and context, and utilizes custom session attributes if provided. The function orchestrates the
    entire flow, including parsing the Lex request, handling intents, managing session state, and formatting
    the response for the channel.

    Args:
        event (dict[str, Any]): The event data from AWS Lambda, typically containing the Lex request.
        context (Any): The context object provided by AWS Lambda, containing runtime information.

    Returns:
        dict[str, Any]: A formatted response ready to be sent back to Amazon Lex.
    """
    if self.config.auto_handle_exceptions:
        try:
            return self._handle_request_with_auto_exception_handling(event, context)
        except Exception as e:
            logger.exception("Error processing request")
            # Create a basic LexRequest for error handling
            try:
                lex_request = LexRequest(**event)
            except Exception:
                # If we can't even parse the event, create a minimal error response
                return self._create_minimal_error_response()

            # Use handle_exceptions and format the response
            error_response = handle_exceptions(e, lex_request, error_message=self.config.error_message)
            return format_for_channel(response=error_response, channel_string="lex")
    else:
        return self._handle_request_with_auto_exception_handling(event, context)

disambiguation_intent_handler 

disambiguation_intent_handler(lex_payload: LexRequest[T]) -> LexResponse[T] | None

Handle disambiguation responses and trigger disambiguation when needed.

This handler processes user responses to disambiguation questions and triggers new disambiguation when confidence is low.

Source code in lex_helper/core/handler.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def disambiguation_intent_handler(self, lex_payload: LexRequest[T]) -> LexResponse[T] | None:
    """
    Handle disambiguation responses and trigger disambiguation when needed.

    This handler processes user responses to disambiguation questions and
    triggers new disambiguation when confidence is low.
    """
    if not self.disambiguation_handler:
        return None

    # First, check if this is a response to a previous disambiguation
    disambiguation_response = self.disambiguation_handler.process_disambiguation_response(lex_payload)
    if disambiguation_response is not None:
        return disambiguation_response

    # If not a disambiguation response, check if we need to trigger disambiguation
    if self.disambiguation_analyzer:
        # Analyze the request for disambiguation
        analysis_result = self.disambiguation_analyzer.analyze_request(cast(Any, lex_payload))

        if analysis_result.should_disambiguate and analysis_result.candidates:
            logger.info("Triggering disambiguation with %d candidates", len(analysis_result.candidates))
            return self.disambiguation_handler.handle_disambiguation(lex_payload, analysis_result.candidates)

    # No disambiguation needed, let regular handler process
    return None

regular_intent_handler 

regular_intent_handler(lex_payload: LexRequest[T]) -> LexResponse[T] | None

Route the incoming request based on intent. The JSON body of the request is provided in the event slot.

Source code in lex_helper/core/handler.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def regular_intent_handler(self, lex_payload: LexRequest[T]) -> LexResponse[T] | None:
    """
    Route the incoming request based on intent.
    The JSON body of the request is provided in the event slot.
    """
    logger.debug("Payload from Lex: %s", lex_payload)
    intent = get_intent(lex_payload)
    intent_name = intent.name
    lex_payload.sessionState.sessionAttributes.lex_intent = intent.name

    response: LexResponse[T] | None = None

    if not intent_name.__contains__("Common_Exit_Feedback"):
        lex_payload.sessionState.activeContexts = [
            {
                "name": "transition_to_exit",
                "contextAttributes": {},
                "timeToLive": {"timeToLiveInSeconds": 900, "turnsToLive": 20},
            }
        ]

    if any_unknown_slot_choices(lex_request=lex_payload):
        response = handle_any_unknown_slot_choice(lex_request=lex_payload)

    else:
        intent_name = lex_payload.sessionState.intent.name
        logger.debug("Calling handler for intent: %s", intent_name)
        response = call_handler_for_file(
            intent_name=intent_name, lex_request=lex_payload, package_name=self.config.package_name
        )
    return response

Configuration

lex_helper.core.handler.Config

Bases: BaseModel

Dialog Utilities

The dialog module provides essential functions for managing conversation flow and dialog state.

Standard util methods to manage dialog state

Functions

get_sentiment 

get_sentiment(lex_request: LexRequest[T]) -> str | None

Extracts the sentiment from the first interpretation in a LexRequest.

This function checks if the 'interpretations' attribute exists in the LexRequest and retrieves the sentiment from the first interpretation if available.

Parameters: lex_request (LexRequest): The LexRequest object containing interpretations.

Returns: Optional[str]: The sentiment string if available, otherwise None.

Raises: AttributeError: If the 'interpretations' attribute is missing or not a list.

Source code in lex_helper/core/dialog.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_sentiment[T: SessionAttributes](lex_request: LexRequest[T]) -> str | None:
    """
    Extracts the sentiment from the first interpretation in a LexRequest.

    This function checks if the 'interpretations' attribute exists in the LexRequest
    and retrieves the sentiment from the first interpretation if available.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing interpretations.

    Returns:
    Optional[str]: The sentiment string if available, otherwise None.

    Raises:
    AttributeError: If the 'interpretations' attribute is missing or not a list.
    """
    # Check if 'interpretations' attribute exists and it's a list
    if not hasattr(lex_request, "interpretations"):
        raise AttributeError("Invalid LexRequest: 'interpretations' attribute missing or not a list")

    interpretations = lex_request.interpretations
    if len(interpretations) > 0:
        element = interpretations[0]

        # Check if 'sentimentResponse' attribute exists in 'element'
        if not hasattr(element, "sentimentResponse"):
            return None

        if element.sentimentResponse:
            sentiment = element.sentimentResponse.sentiment
            logger.debug("We have a sentiment: %s", sentiment)
            return sentiment
    return None

remove_context 

remove_context(context_list: ActiveContexts, context_name: str) -> ActiveContexts

Removes a specific context from the active contexts list.

Parameters: context_list (ActiveContexts): The list of active contexts. context_name (str): The name of the context to remove.

Returns: ActiveContexts: The updated list of active contexts without the specified context.

Source code in lex_helper/core/dialog.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def remove_context(context_list: ActiveContexts, context_name: str) -> ActiveContexts:
    """
    Removes a specific context from the active contexts list.

    Parameters:
    context_list (ActiveContexts): The list of active contexts.
    context_name (str): The name of the context to remove.

    Returns:
    ActiveContexts: The updated list of active contexts without the specified context.
    """
    if not context_list:
        return context_list
    new_context: ActiveContexts = []
    for context in context_list:
        if not context_name == context["name"]:
            new_context.append(context)
    return new_context

remove_inactive_context 

remove_inactive_context(lex_request: LexRequest[T]) -> ActiveContexts

Removes inactive contexts from the active contexts list in a LexRequest.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: ActiveContexts: The updated list of active contexts with inactive ones removed.

Source code in lex_helper/core/dialog.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def remove_inactive_context[T: SessionAttributes](lex_request: LexRequest[T]) -> ActiveContexts:
    """
    Removes inactive contexts from the active contexts list in a LexRequest.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    ActiveContexts: The updated list of active contexts with inactive ones removed.
    """
    context_list = lex_request.sessionState.activeContexts
    if not context_list:
        return context_list
    new_context: ActiveContexts = []
    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

close 

close(lex_request: LexRequest[T], messages: LexMessages) -> LexResponse[T]

Closes the dialog with the user by setting the intent state to 'Fulfilled'.

Parameters: lex_request (LexRequest): The LexRequest object containing session state. messages (LexMessages): The messages to send to the user.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def close[T: SessionAttributes](lex_request: LexRequest[T], messages: LexMessages) -> LexResponse[T]:
    """
    Closes the dialog with the user by setting the intent state to 'Fulfilled'.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.
    messages (LexMessages): The messages to send to the user.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    intent, active_contexts, session_attributes, _ = get_request_components(lex_request)

    lex_request.sessionState.activeContexts = remove_inactive_context(lex_request)
    lex_request.sessionState.intent.state = "Fulfilled"
    session_attributes.previous_dialog_action_type = "Close"
    response = LexResponse(
        sessionState=SessionState(
            activeContexts=active_contexts,
            sessionAttributes=session_attributes,
            intent=intent,
            dialogAction=DialogAction(type="Close"),
        ),
        requestAttributes={},
        messages=messages,
    )

    logger.debug("Dialog closed")

    return response

elicit_intent 

elicit_intent(messages: LexMessages, lex_request: LexRequest[T]) -> LexResponse[T]

Elicits the user's intent by sending a message and updating session attributes.

Parameters: messages (LexMessages): The messages to send to the user. lex_request (LexRequest): The LexRequest object containing session state.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def elicit_intent[T: SessionAttributes](messages: LexMessages, lex_request: LexRequest[T]) -> LexResponse[T]:
    """
    Elicits the user's intent by sending a message and updating session attributes.

    Parameters:
    messages (LexMessages): The messages to send to the user.
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    intent, active_contexts, session_attributes, _ = get_request_components(lex_request=lex_request)

    active_contexts = remove_inactive_context(lex_request)
    intent.state = "Fulfilled"

    session_attributes.previous_dialog_action_type = "ElicitIntent"
    session_attributes.previous_slot_to_elicit = ""
    session_attributes.options_provided = get_provided_options(messages)

    logger.debug("Elicit-Intent")

    return LexResponse(
        sessionState=SessionState(
            activeContexts=active_contexts,
            sessionAttributes=session_attributes,
            intent=intent,
            dialogAction=DialogAction(type="ElicitIntent"),
        ),
        requestAttributes={},
        messages=messages,
    )

elicit_slot 

elicit_slot(slot_to_elicit: LexSlot | str, messages: LexMessages, lex_request: LexRequest[T]) -> LexResponse[T]

Elicits a specific slot from the user by sending a message and updating session attributes.

Parameters: slot_to_elicit (LexSlot | str): The slot to elicit from the user. messages (LexMessages): The messages to send to the user. lex_request (LexRequest): The LexRequest object containing session state.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def elicit_slot[T: SessionAttributes](
    slot_to_elicit: LexSlot | str, messages: LexMessages, lex_request: LexRequest[T]
) -> LexResponse[T]:
    """
    Elicits a specific slot from the user by sending a message and updating session attributes.

    Parameters:
    slot_to_elicit (LexSlot | str): The slot to elicit from the user.
    messages (LexMessages): The messages to send to the user.
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    intent, active_contexts, session_attributes, _ = get_request_components(lex_request=lex_request)
    active_contexts = remove_inactive_context(lex_request)
    intent.state = "InProgress"

    # options_provided are only used by the service lambda to lookup the user's response in the event that they have only
    # provided a single character answer, i.e. A, or 1, etc.
    session_attributes.options_provided = get_provided_options(messages)
    session_attributes.previous_intent = intent.name
    session_attributes.previous_message = json.dumps(messages, cls=PydanticEncoder)
    # session_attributes.previous_dialog_action_type = "ElicitSlot"
    slot_name = slot_to_elicit

    if not isinstance(slot_to_elicit, str):
        slot_name = slot_to_elicit.value

    session_attributes.previous_slot_to_elicit = (intent.name.replace("_", "") + "Slot") + "." + str(slot_name).upper()

    if "." in str(slot_name):
        raise Exception("SLOT PARSED INCORRECTLY")

    response = LexResponse(
        sessionState=SessionState(
            activeContexts=active_contexts,
            sessionAttributes=session_attributes,
            intent=intent,
            dialogAction=DialogAction(type="ElicitSlot", slotToElicit=str(slot_name)),
        ),
        requestAttributes={},
        messages=messages,
    )

    return response

delegate 

delegate(lex_request: LexRequest[T]) -> LexResponse[T]

Delegates the dialog to Lex by updating the session state and returning a response.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def delegate[T: SessionAttributes](lex_request: LexRequest[T]) -> LexResponse[T]:
    """
    Delegates the dialog to Lex by updating the session state and returning a response.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    logger.debug("Delegating")

    updated_active_contexts = remove_inactive_context(lex_request)
    lex_request.sessionState.intent.state = "ReadyForFulfillment"
    lex_request.sessionState.sessionAttributes.previous_dialog_action_type = "Delegate"

    updated_session_state = SessionState(
        activeContexts=updated_active_contexts,
        sessionAttributes=lex_request.sessionState.sessionAttributes,
        intent=lex_request.sessionState.intent,
        dialogAction=DialogAction(type="Delegate"),
    )

    return LexResponse(sessionState=updated_session_state, requestAttributes={}, messages=[])

get_provided_options 

get_provided_options(messages: LexMessages) -> str

Extracts the text of the buttons from a list of LexImageResponseCard messages.

This function loops through a list of messages, checks if each message is an instance of LexImageResponseCard, and if so, extracts the text of each button in the imageResponseCard attribute of the message. The function then returns a JSON-encoded list of the extracted button texts.

Parameters: messages (LexMessages): The list of messages to process.

Returns: str: A JSON-encoded list of the extracted button texts.

Source code in lex_helper/core/dialog.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def get_provided_options(messages: LexMessages) -> str:
    """
    Extracts the text of the buttons from a list of LexImageResponseCard messages.

    This function loops through a list of messages, checks if each message is an instance of LexImageResponseCard,
    and if so, extracts the text of each button in the imageResponseCard attribute of the message. The function
    then returns a JSON-encoded list of the extracted button texts.

    Parameters:
    messages (LexMessages): The list of messages to process.

    Returns:
    str: A JSON-encoded list of the extracted button texts.
    """
    options = [
        button.text
        for message in messages
        if isinstance(message, LexImageResponseCard)
        for button in message.imageResponseCard.buttons
    ]
    logger.debug("Get provided options :: %s", options)

    return json.dumps(options, cls=PydanticEncoder)

get_intent 

get_intent(lex_request: LexRequest[T]) -> Intent

Retrieves the intent from a LexRequest.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: Intent: The intent object from the session state.

Raises: ValueError: If no intent is found in the request.

Source code in lex_helper/core/dialog.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def get_intent[T: SessionAttributes](lex_request: LexRequest[T]) -> Intent:
    """
    Retrieves the intent from a LexRequest.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    Intent: The intent object from the session state.

    Raises:
    ValueError: If no intent is found in the request.
    """
    session_state = lex_request.sessionState
    if session_state:
        return session_state.intent
    else:
        raise ValueError("No intent found in request")

get_slot 

get_slot(slot_name: LexSlot | str, intent: Intent, **kwargs: Any)

Retrieves the value of a slot from an intent.

Parameters: slot_name (LexSlot | str): The name of the slot to retrieve. intent (Intent): The intent object containing the slot. kwargs (Any): Additional arguments for slot value preference.

Returns: Any: The value of the slot if available, otherwise None.

Source code in lex_helper/core/dialog.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def get_slot(slot_name: LexSlot | str, intent: Intent, **kwargs: Any):
    """
    Retrieves the value of a slot from an intent.

    Parameters:
    slot_name (LexSlot | str): The name of the slot to retrieve.
    intent (Intent): The intent object containing the slot.
    kwargs (Any): Additional arguments for slot value preference.

    Returns:
    Any: The value of the slot if available, otherwise None.
    """
    try:
        if isinstance(slot_name, str):
            slot = intent.slots.get(slot_name)
        else:
            slot = intent.slots.get(slot_name.value)
        if not slot:
            return None
        return get_slot_value(slot, **kwargs)
    except Exception:
        logger.exception("Failed to get slot")
        return None

get_composite_slot 

get_composite_slot(slot_name: str, intent: Intent, preference: str | None = None) -> dict[str, str | None] | None

Retrieves the values from sub-slots of a given slot from an intent.

Parameters:

Name Type Description Default
slot_name str

Name of the slot to be fetched.

 required
intent Intent

Intent object containing the slot.

 required
preference Optional[str], default=None

Preference for value type ('interpretedValue' or 'originalValue'). If no preference is provided and 'interpretedValue' is available, it's used. Otherwise, 'originalValue' is used.

 None

Returns:

Type Description
dict[str, str | None] | None

Dict[str, Union[str, None]] or None: Dictionary containing the subslot names and their corresponding values, or None if the slot or subslots don't exist.

Raises:

Type Description
Exception

Any exception that occurs while fetching the slot.
Source code in lex_helper/core/dialog.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def get_composite_slot(slot_name: str, intent: Intent, preference: str | None = None) -> dict[str, str | None] | None:
    """
    Retrieves the values from sub-slots of a given slot from an intent.

    Args:
        slot_name (str): Name of the slot to be fetched.
        intent (Intent): Intent object containing the slot.
        preference (Optional[str], default=None): Preference for value type ('interpretedValue' or
            'originalValue'). If no preference is provided and 'interpretedValue' is available,
            it's used. Otherwise, 'originalValue' is used.

    Returns:
        Dict[str, Union[str, None]] or None: Dictionary containing the subslot names and their
            corresponding values, or None if the slot or subslots don't exist.

    Raises:
        Exception: Any exception that occurs while fetching the slot.
    """
    subslot_dict: dict[str, Any] = {}

    # Get the slot from the intent
    slot = intent.slots.get(slot_name)
    if not slot:
        return None

    sub_slots = slot.get("subSlots", {})
    if not sub_slots:
        return None

    # Iterate through the subslots
    for key, subslot in sub_slots.items():
        subslot_value = subslot.get("value")

        # If subslot has a value
        if subslot_value:
            interpreted_value = subslot_value.get("interpretedValue")
            original_value = subslot_value.get("originalValue")
            if preference == "interpretedValue":
                subslot_dict[key] = interpreted_value
            elif preference == "originalValue":
                subslot_dict[key] = original_value
            elif interpreted_value:
                subslot_dict[key] = interpreted_value
            else:
                subslot_dict[key] = original_value
        else:
            subslot_dict[key] = None

    return subslot_dict

get_slot_value 

get_slot_value(slot: dict[str, Any], **kwargs: Any)

Retrieves the value from a slot dictionary.

Parameters: slot (dict[str, Any]): The slot dictionary containing the value. kwargs (Any): Additional arguments for slot value preference.

Returns: Any: The interpreted or original value of the slot if available, otherwise None.

Source code in lex_helper/core/dialog.py 
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
def get_slot_value(slot: dict[str, Any], **kwargs: Any):
    """
    Retrieves the value from a slot dictionary.

    Parameters:
    slot (dict[str, Any]): The slot dictionary containing the value.
    kwargs (Any): Additional arguments for slot value preference.

    Returns:
    Any: The interpreted or original value of the slot if available, otherwise None.
    """
    slot_value = slot.get("value")
    if slot_value:
        interpreted_value = slot_value.get("interpretedValue")
        original_value = slot_value.get("originalValue")
        if kwargs.get("preference") == "interpretedValue":
            return interpreted_value
        elif kwargs.get("preference") == "originalValue":
            return original_value
        elif interpreted_value:
            return interpreted_value
        else:
            return original_value
    else:
        return None

set_subslot 

set_subslot(composite_slot_name: LexSlot, subslot_name: str, subslot_value: Any | None, intent: Intent) -> Intent

Sets a subslot value within a composite slot in the given intent.

Parameters:

Name Type Description Default
composite_slot_name str

The name of the composite slot within the intent.

 required
subslot_name str

The name of the subslot to be set.

 required
subslot_value Optional[Any]

The value to be set for the subslot. If None, the subslot is set to None.

 required
intent Intent

The intent containing the slots.

 required

Returns:

Name Type Description
Intent Intent

The updated intent with the modified subslot value.
Source code in lex_helper/core/dialog.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
def set_subslot(
    composite_slot_name: LexSlot,
    subslot_name: str,
    subslot_value: Any | None,
    intent: Intent,
) -> Intent:
    """
    Sets a subslot value within a composite slot in the given intent.

    Args:
        composite_slot_name (str): The name of the composite slot within the intent.
        subslot_name (str): The name of the subslot to be set.
        subslot_value (Optional[Any]): The value to be set for the subslot. If None, the subslot is set to None.
        intent (Intent): The intent containing the slots.

    Returns:
        Intent: The updated intent with the modified subslot value.
    """

    # Ensure the composite slot and its subSlots dictionary exist
    if composite_slot_name.value not in intent.slots:
        intent.slots[composite_slot_name.value] = {"subSlots": {}}

    # Determine the value to set for the subslot
    if subslot_value is None:
        intent.slots[composite_slot_name.value]["subSlots"][subslot_name] = None  # type: ignore
    else:
        intent.slots[composite_slot_name.value]["subSlots"][subslot_name] = {  # type: ignore
            "value": {
                "interpretedValue": subslot_value,
                "originalValue": subslot_value,
                "resolvedValues": [subslot_value],
            }
        }

    # Logging for debugging purposes
    logger.debug("Setting subslot %s in composite slot %s", subslot_name, composite_slot_name)
    logger.debug("Resulting intent: %s", json.dumps(intent, cls=PydanticEncoder))

    return intent

set_slot 

set_slot(slot_name: LexSlot, slot_value: str | None, intent: Intent) -> Intent

Sets a slot value in the given intent.

Parameters: slot_name (LexSlot): The name of the slot to set. slot_value (Optional[str]): The value to set for the slot. intent (Intent): The intent containing the slots.

Returns: Intent: The updated intent with the modified slot value.

Source code in lex_helper/core/dialog.py 
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def set_slot(slot_name: LexSlot, slot_value: str | None, intent: Intent) -> Intent:
    """
    Sets a slot value in the given intent.

    Parameters:
    slot_name (LexSlot): The name of the slot to set.
    slot_value (Optional[str]): The value to set for the slot.
    intent (Intent): The intent containing the slots.

    Returns:
    Intent: The updated intent with the modified slot value.
    """
    intent.slots[slot_name.value] = {
        "value": {
            "interpretedValue": slot_value,
            "originalValue": slot_value,
            "resolvedValues": [slot_value],
        }
    }
    return intent

get_composite_slot_subslot 

get_composite_slot_subslot(composite_slot: LexSlot, sub_slot: Any, intent: Intent, **kwargs: Any) -> str | None

Retrieves the value of a subslot from a composite slot in an intent.

Parameters: composite_slot (LexSlot): The composite slot containing the subslot. sub_slot (Any): The name of the subslot to retrieve. intent (Intent): The intent object containing the slots. kwargs (Any): Additional arguments for slot value preference.

Returns: Optional[str]: The value of the subslot if available, otherwise None.

Source code in lex_helper/core/dialog.py 
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def get_composite_slot_subslot(composite_slot: LexSlot, sub_slot: Any, intent: Intent, **kwargs: Any) -> str | None:
    """
    Retrieves the value of a subslot from a composite slot in an intent.

    Parameters:
    composite_slot (LexSlot): The composite slot containing the subslot.
    sub_slot (Any): The name of the subslot to retrieve.
    intent (Intent): The intent object containing the slots.
    kwargs (Any): Additional arguments for slot value preference.

    Returns:
    Optional[str]: The value of the subslot if available, otherwise None.
    """
    try:
        slot = intent.slots[composite_slot.value]
        if not slot:
            return None
        sub_slot = slot["subSlots"][sub_slot]
        return get_slot_value(sub_slot, **kwargs)
    except Exception:
        return None

get_active_contexts 

get_active_contexts(lex_request: LexRequest[T]) -> ActiveContexts

Retrieves the active contexts from a LexRequest.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: ActiveContexts: The list of active contexts.

Source code in lex_helper/core/dialog.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
def get_active_contexts[T: SessionAttributes](lex_request: LexRequest[T]) -> ActiveContexts:
    """
    Retrieves the active contexts from a LexRequest.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    ActiveContexts: The list of active contexts.
    """
    try:
        return lex_request.sessionState.activeContexts
    except Exception:
        return []

get_invocation_label 

get_invocation_label(lex_request: LexRequest[T]) -> str | None

Retrieves the invocation label from a LexRequest.

Parameters: lex_request (LexRequest): The LexRequest object containing the invocation label.

Returns: str: The invocation label.

Source code in lex_helper/core/dialog.py 
510
511
512
513
514
515
516
517
518
519
520
521
def get_invocation_label[T: SessionAttributes](lex_request: LexRequest[T]) -> str | None:
    """
    Retrieves the invocation label from a LexRequest.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing the invocation label.

    Returns:
    str: The invocation label.
    """
    logger.debug("Invocation Label: %s", lex_request.invocationLabel)
    return lex_request.invocationLabel

safe_delete_session_attribute 

safe_delete_session_attribute(lex_request: LexRequest[T], attribute: str) -> LexRequest[T]

Safely deletes a session attribute from a LexRequest.

Parameters: lex_request (LexRequest): The LexRequest object containing session attributes. attribute (str): The name of the attribute to delete.

Returns: LexRequest: The updated LexRequest with the attribute deleted.

Source code in lex_helper/core/dialog.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
def safe_delete_session_attribute[T: SessionAttributes](lex_request: LexRequest[T], attribute: str) -> LexRequest[T]:
    """
    Safely deletes a session attribute from a LexRequest.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session attributes.
    attribute (str): The name of the attribute to delete.

    Returns:
    LexRequest: The updated LexRequest with the attribute deleted.
    """
    logger.debug("Deleting session attribute %s", attribute)
    if lex_request.sessionState.sessionAttributes and getattr(lex_request.sessionState.sessionAttributes, attribute):
        setattr(lex_request.sessionState.sessionAttributes, attribute, None)
    return lex_request

get_request_components 

get_request_components(lex_request: LexRequest[T]) -> tuple[Intent, ActiveContexts, T, str | None]

Extracts common components from the intent request.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: tuple: A tuple containing the intent, active contexts, session attributes, and invocation label.

Source code in lex_helper/core/dialog.py 
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
def get_request_components[T: SessionAttributes](
    lex_request: LexRequest[T],
) -> tuple[Intent, ActiveContexts, T, str | None]:
    """
    Extracts common components from the intent request.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    tuple: A tuple containing the intent, active contexts, session attributes, and invocation label.
    """
    intent = get_intent(lex_request)
    active_contexts = get_active_contexts(lex_request)
    session_attributes = lex_request.sessionState.sessionAttributes
    invocation_label = get_invocation_label(lex_request)
    return intent, active_contexts, session_attributes, invocation_label

any_unknown_slot_choices 

any_unknown_slot_choices(lex_request: LexRequest[T]) -> bool

Checks if the user provided an invalid response to a previous slot elicitation.

Use this function at the beginning of your intent handlers to detect when users respond with unrecognized choices (e.g., "X" when options were "A, B, C").

Usage

if dialog.any_unknown_slot_choices(lex_request): return dialog.handle_any_unknown_slot_choice(lex_request)

Continue with normal intent logic

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: bool: True if there are unknown slot choices, otherwise False.

Source code in lex_helper/core/dialog.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
def any_unknown_slot_choices[T: SessionAttributes](lex_request: LexRequest[T]) -> bool:
    """
    Checks if the user provided an invalid response to a previous slot elicitation.

    Use this function at the beginning of your intent handlers to detect when users
    respond with unrecognized choices (e.g., "X" when options were "A, B, C").

    Usage:
        if dialog.any_unknown_slot_choices(lex_request):
            return dialog.handle_any_unknown_slot_choice(lex_request)
        # Continue with normal intent logic

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    bool: True if there are unknown slot choices, otherwise False.
    """
    intent, _, session_attributes, _ = get_request_components(lex_request)

    if "ElicitSlot" != session_attributes.previous_dialog_action_type:
        lex_request.sessionState.sessionAttributes.error_count = 0
        return False

    logger.debug("ElicitSlot is the previous dialog action")
    previous_slot_to_elicit = session_attributes.previous_slot_to_elicit

    if not previous_slot_to_elicit:
        return False

    # Extract actual slot name from the stored format
    slot_name = previous_slot_to_elicit.split(".")[-1].lower() if "." in previous_slot_to_elicit else previous_slot_to_elicit

    # Check if the slot exists and has a value
    slot_data = intent.slots.get(slot_name)
    if not slot_data or not slot_data.get("value"):
        logger.debug("Unknown slot choice - no value for slot: %s", slot_name)
        return True

    # If slot has a value, it was recognized
    logger.debug("Slot recognized correctly, letting Lex continue with the conversation")
    lex_request.sessionState.sessionAttributes.error_count = 0
    return False

handle_any_unknown_slot_choice 

handle_any_unknown_slot_choice(lex_request: LexRequest[T]) -> LexResponse[T]

Automatically handles invalid slot responses by delegating back to Lex or using custom logic.

This function processes the invalid choice and either continues the conversation or triggers custom error handling via unknown_choice_handler.

Usage

if dialog.any_unknown_slot_choices(lex_request): return dialog.handle_any_unknown_slot_choice(lex_request)

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
def handle_any_unknown_slot_choice[T: SessionAttributes](lex_request: LexRequest[T]) -> LexResponse[T]:
    """
    Automatically handles invalid slot responses by delegating back to Lex or using custom logic.

    This function processes the invalid choice and either continues the conversation
    or triggers custom error handling via unknown_choice_handler.

    Usage:
        if dialog.any_unknown_slot_choices(lex_request):
            return dialog.handle_any_unknown_slot_choice(lex_request)

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    intent, _, session_attributes, _ = get_request_components(lex_request)

    logger.debug("Handle_Any_Unknown_Choice :: %s", session_attributes)
    intent = get_intent(lex_request)
    previous_slot_to_elicit = session_attributes.previous_slot_to_elicit

    logger.debug("Unparsed slot name: " + (previous_slot_to_elicit or ""))
    slot_name = previous_slot_to_elicit

    logger.debug("Identifier for bad slot: %s", slot_name)

    choice = get_slot(slot_name or "", intent, preference="interpretedValue")
    logger.debug("Bad choice is %s", choice)
    if not isinstance(choice, str):
        logger.debug("Bad slot choice")
        return unknown_choice_handler(lex_request=lex_request, choice=choice)
    return delegate(lex_request=lex_request)

unknown_choice_handler 

unknown_choice_handler(lex_request: LexRequest[T], choice: str | LexSlot | None, handle_unknown: bool | None = True, next_intent: str | None = None, next_invo_label: str | None = '') -> LexResponse[T]

Customizable handler for processing invalid user choices.

Override this function or call it directly to implement custom logic when users provide invalid responses to slot elicitations (buttons, multiple choice, etc.).

Usage

Custom handling

return dialog.unknown_choice_handler( lex_request=lex_request, choice="invalid_response", next_intent="clarification_intent" )

Or override the function entirely for global custom behavior

Parameters: lex_request (LexRequest): The LexRequest object containing session state. choice (str | LexSlot | None): The invalid choice provided by the user. handle_unknown (Optional[bool]): Whether to handle unknown choices. next_intent (Optional[str]): The next intent to transition to. next_invo_label (Optional[str]): The next invocation label.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
def unknown_choice_handler[T: SessionAttributes](
    lex_request: LexRequest[T],
    choice: str | LexSlot | None,
    handle_unknown: bool | None = True,
    next_intent: str | None = None,
    next_invo_label: str | None = "",
) -> LexResponse[T]:
    """
    Customizable handler for processing invalid user choices.

    Override this function or call it directly to implement custom logic when users
    provide invalid responses to slot elicitations (buttons, multiple choice, etc.).

    Usage:
        # Custom handling
        return dialog.unknown_choice_handler(
            lex_request=lex_request,
            choice="invalid_response",
            next_intent="clarification_intent"
        )

        # Or override the function entirely for global custom behavior

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.
    choice (str | LexSlot | None): The invalid choice provided by the user.
    handle_unknown (Optional[bool]): Whether to handle unknown choices.
    next_intent (Optional[str]): The next intent to transition to.
    next_invo_label (Optional[str]): The next invocation label.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    return delegate(lex_request=lex_request)

callback_original_intent_handler 

callback_original_intent_handler(lex_request: LexRequest[T], messages: LexMessages | None = None) -> LexResponse[T]

Handles switching the conversation flow to an initial intent. Assume a scenario in which the conversation flow starts with intent A, transitions to intent B, returns to intent A after fulfilling the steps in intent B.

For instance, from a MakeReservation intent to an Authenticate intent and back to the MakeReservation intent.

This method assumes that the session attributes callback_event and callback_handler have been populated from the calling intent A as in the example below.

lex_request.sessionState.sessionAttributes.callback_handler = lex_request.sessionState.intent.name
lex_request.sessionState.sessionAttributes.callback_event = json.dumps(lex_request, default=str)
message = (
    f"Before I can help your reservation, you will need to authenticate. "
)
return dialog.transition_to_intent(
    intent_name="Authenticate",
    lex_request=lex_request,
    messages=[LexPlainText(content=message)]
)

Invoke callback_original_intent_handler from intent B to switch back to the original intent.

Parameters:

Name Type Description Default
lex_request LexRequest[T]

description

 required

Returns:

Type Description
LexResponse[T]

LexResponse[T]: description
Source code in lex_helper/core/dialog.py 
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
def callback_original_intent_handler[T: SessionAttributes](
    lex_request: LexRequest[T], messages: LexMessages | None = None
) -> LexResponse[T]:
    """

    Handles switching the conversation flow to an initial intent.
    Assume a scenario in which the conversation flow starts with intent A,
    transitions to intent B, returns to intent A after fulfilling the steps in intent B.

    For instance, from a MakeReservation intent to an Authenticate intent
    and back to the MakeReservation intent.

    This method assumes that the session attributes callback_event and callback_handler have been
    populated from the calling intent A as in the example below.

        lex_request.sessionState.sessionAttributes.callback_handler = lex_request.sessionState.intent.name
        lex_request.sessionState.sessionAttributes.callback_event = json.dumps(lex_request, default=str)
        message = (
            f"Before I can help your reservation, you will need to authenticate. "
        )
        return dialog.transition_to_intent(
            intent_name="Authenticate",
            lex_request=lex_request,
            messages=[LexPlainText(content=message)]
        )

    Invoke callback_original_intent_handler from intent B to switch back to the original intent.


    Args:
        lex_request (LexRequest[T]): _description_

    Returns:
        LexResponse[T]: _description_
    """
    logger.debug("Calling back original handler")

    callback_event = lex_request.sessionState.sessionAttributes.callback_event
    callback_handler = lex_request.sessionState.sessionAttributes.callback_handler or ""
    if not callback_event and not callback_handler:
        logger.debug("No callback event or handler")
        lex_request.sessionState.intent.name = "greeting"
        return call_handler_for_file("greeting", lex_request)

    if callback_event:
        callback_request = json.loads(callback_event)
        logger.debug("Callback request %s", callback_request)
        del lex_request.sessionState.sessionAttributes.callback_event
        del lex_request.sessionState.sessionAttributes.callback_handler
        lex_payload: LexRequest[T] = LexRequest(**callback_request)

        logger.debug("Merging session attributes")
        merged_attrs = lex_payload.sessionState.sessionAttributes.model_dump()
        merged_attrs.update(
            {k: v for k, v in lex_request.sessionState.sessionAttributes.model_dump().items() if v is not None}
        )
        lex_payload.sessionState.sessionAttributes = type(lex_payload.sessionState.sessionAttributes)(**merged_attrs)
    else:
        lex_payload = lex_request
        lex_payload.sessionState.intent.slots = {}
        lex_payload.sessionState.intent.name = callback_handler

    response = call_handler_for_file(callback_handler, lex_payload)

    if messages:
        response.messages = list(messages) + list(response.messages)

    return response

reprompt_slot 

reprompt_slot(lex_request: LexRequest[T]) -> LexResponse[T]

Reprompts the user for a slot value by sending a message.

Parameters: lex_request (LexRequest): The LexRequest object containing session state.

Returns: LexResponse: The response object to be sent back to Lex.

Source code in lex_helper/core/dialog.py 
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
def reprompt_slot[T: SessionAttributes](lex_request: LexRequest[T]) -> LexResponse[T]:
    """
    Reprompts the user for a slot value by sending a message.

    Parameters:
    lex_request (LexRequest): The LexRequest object containing session state.

    Returns:
    LexResponse: The response object to be sent back to Lex.
    """
    logger.debug("Reprompting slot")

    session_attributes = lex_request.sessionState.sessionAttributes
    previous_slot_to_elicit = session_attributes.previous_slot_to_elicit
    if not previous_slot_to_elicit:
        return delegate(lex_request)
    logger.debug("Unparsed slot name: " + previous_slot_to_elicit)
    slot_name = previous_slot_to_elicit
    messages = []
    logger.debug("Reprompt-Messages :: %s", messages)

    return elicit_slot(slot_to_elicit=slot_name, messages=messages, lex_request=lex_request)

load_messages 

load_messages(messages: str) -> LexMessages

Loads messages from a JSON string into LexMessages objects.

Parameters: messages (str): The JSON string containing messages.

Returns: LexMessages: The list of LexMessages objects.

Source code in lex_helper/core/dialog.py 
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
def load_messages(messages: str) -> LexMessages:
    """
    Loads messages from a JSON string into LexMessages objects.

    Parameters:
    messages (str): The JSON string containing messages.

    Returns:
    LexMessages: The list of LexMessages objects.
    """
    res: LexMessages = []
    temp: list[Any] = json.loads(messages)

    for msg in temp:
        content_type: str = msg.get("contentType", "_")
        match content_type:  # type: ignore
            case "ImageResponseCard":
                res.append(LexImageResponseCard.model_validate_json(json.dumps(msg)))
            case "CustomPayload":
                res.append(LexCustomPayload.model_validate_json(json.dumps(msg)))
            case "PlainText":
                res.append(LexPlainText.model_validate_json(json.dumps(msg)))
            case _:
                res.append(msg)

    logger.debug("Previous Message :: %s", res)
    return res

parse_lex_request 

parse_lex_request(data: dict[str, Any], session_attributes: T) -> LexRequest[T]

Use this to parse a Lambda event into a LexRequest object.

Parameters: data (dict[str, Any]): The Lambda event data. sessionAttributes (Optional[T]): The session attributes to use.

Returns: LexRequest[T]: The parsed LexRequest object.

Source code in lex_helper/core/dialog.py 
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
def parse_lex_request[T: SessionAttributes](
    data: dict[str, Any],
    session_attributes: T,
) -> LexRequest[T]:
    """
    Use this to parse a Lambda event into a LexRequest object.

    Parameters:
    data (dict[str, Any]): The Lambda event data.
    sessionAttributes (Optional[T]): The session attributes to use.

    Returns:
    LexRequest[T]: The parsed LexRequest object.
    """
    # Create a copy of the data to modify
    data_copy = data.copy()

    # If there are session attributes in the event, convert them to the proper model
    if data_copy.get("sessionState", {}).get("sessionAttributes"):
        event_attrs = data_copy["sessionState"]["sessionAttributes"]
        # Create a new instance of the session attributes model with the event data
        model_attrs = type(session_attributes)(**event_attrs)
        data_copy["sessionState"]["sessionAttributes"] = model_attrs
    else:
        # If no session attributes in event, use the provided ones
        if "sessionState" not in data_copy:
            data_copy["sessionState"] = {}
        data_copy["sessionState"]["sessionAttributes"] = session_attributes

    lex_request: LexRequest[T] = LexRequest(**data_copy)
    lex_request.sessionState.activeContexts = remove_inactive_context(lex_request)  # Remove inactive contexts
    lex_request = parse_req_sess_attrs(lex_request)  # Parse Session Attributes
    return lex_request

Session Management

Session attributes provide type-safe state management across conversation turns.

Type Definitions

Core type definitions used throughout the library.

Classes

Disambiguation

Smart disambiguation functionality for handling ambiguous user input using AI.

Disambiguation Handler

Disambiguation Handler for the Smart Disambiguation feature.

This module provides the DisambiguationHandler class that orchestrates the disambiguation process, generates user-friendly clarification responses, and processes user selections to route to the appropriate intent.

Classes

DisambiguationHandler 

DisambiguationHandler(config: DisambiguationConfig | None = None)

Handles disambiguation response generation and user selection processing.

This class is responsible for creating user-friendly clarification messages when multiple intents are possible matches, and processing user responses to route them to the correct intent handler.

Parameters:

Name Type Description Default
config DisambiguationConfig | None

Configuration options for disambiguation behavior

 None
Source code in lex_helper/core/disambiguation/handler.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self, config: DisambiguationConfig | None = None):
    """
    Initialize the disambiguation handler.

    Args:
        config: Configuration options for disambiguation behavior
    """
    self.config = config or DisambiguationConfig()

    # Initialize Bedrock generator if enabled
    self.bedrock_generator = None
    if self.config.bedrock_config.enabled:
        try:
            self.bedrock_generator = BedrockDisambiguationGenerator(self.config.bedrock_config)
            logger.debug("Bedrock disambiguation generator initialized")
        except Exception as e:
            logger.warning("Failed to initialize Bedrock generator: %s", e)
            if not self.config.bedrock_config.fallback_to_static:
                raise

Functions

handle_disambiguation 
handle_disambiguation(lex_request: LexRequest[T], candidates: list[IntentCandidate]) -> LexResponse[T]

Generate disambiguation response with clarifying questions.

Creates a response that presents the user with options to clarify their intent, using buttons for easy selection.

Parameters:

Name Type Description Default
lex_request LexRequest[T]

The original Lex request

 required
candidates list[IntentCandidate]

List of intent candidates to present

 required

Returns:

Type Description
LexResponse[T]

LexResponse with disambiguation options
Source code in lex_helper/core/disambiguation/handler.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def handle_disambiguation(self, lex_request: LexRequest[T], candidates: list[IntentCandidate]) -> LexResponse[T]:
    """
    Generate disambiguation response with clarifying questions.

    Creates a response that presents the user with options to clarify
    their intent, using buttons for easy selection.

    Args:
        lex_request: The original Lex request
        candidates: List of intent candidates to present

    Returns:
        LexResponse with disambiguation options
    """
    logger.info("Generating disambiguation response with %d candidates", len(candidates))

    # Limit candidates to configured maximum
    limited_candidates = candidates[: self.config.max_candidates]

    # Store disambiguation state in session
    self._store_disambiguation_state(lex_request, limited_candidates)

    # Create clarification messages with user input context
    messages = self._create_clarification_messages(limited_candidates, lex_request.inputTranscript)

    # Use elicit_intent to get user's clarification
    return elicit_intent(messages, lex_request)
process_disambiguation_response 
process_disambiguation_response(lex_request: LexRequest[T]) -> LexResponse[T] | None

Process user's response to disambiguation and route to selected intent.

Analyzes the user's input to determine which intent they selected and updates the request to route to that intent.

Parameters:

Name Type Description Default
lex_request LexRequest[T]

The Lex request with user's disambiguation response

 required

Returns:

Type Description
LexResponse[T] | None

None if this isn't a disambiguation response, otherwise routes
LexResponse[T] | None

to the selected intent by updating the request
Source code in lex_helper/core/disambiguation/handler.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def process_disambiguation_response(self, lex_request: LexRequest[T]) -> LexResponse[T] | None:
    """
    Process user's response to disambiguation and route to selected intent.

    Analyzes the user's input to determine which intent they selected
    and updates the request to route to that intent.

    Args:
        lex_request: The Lex request with user's disambiguation response

    Returns:
        None if this isn't a disambiguation response, otherwise routes
        to the selected intent by updating the request
    """
    # Check if this is a disambiguation response
    if not self._is_disambiguation_response(lex_request):
        return None

    logger.debug("Processing disambiguation response")

    # Get stored candidates from session
    candidates = self._get_stored_candidates(lex_request)
    if not candidates:
        logger.warning("No stored disambiguation candidates found")
        return self._create_fallback_response(lex_request)

    # Determine selected intent
    selected_intent = self._determine_selected_intent(lex_request.inputTranscript, candidates)

    if not selected_intent:
        logger.warning("Could not determine selected intent from input: %s", lex_request.inputTranscript)
        return self._create_fallback_response(lex_request)

    # Update request to route to selected intent
    self._update_request_for_selected_intent(lex_request, selected_intent)

    # Clear disambiguation state
    self._clear_disambiguation_state(lex_request)

    logger.info("Routed disambiguation to intent: %s", selected_intent)

    # Return None to let the regular handler process the updated request
    return None

Functions

Disambiguation Analyzer

DisambiguationAnalyzer for intelligent intent disambiguation.

This module provides the core analysis functionality for determining when disambiguation should occur and which intent candidates to present to users.

Attributes

Classes

DisambiguationAnalyzer 

DisambiguationAnalyzer(config: DisambiguationConfig | None = None)

Analyzes user input to determine disambiguation candidates.

The analyzer uses Lex's NLU confidence scores to identify potential intent matches and determine when disambiguation should be triggered based on configurable thresholds.

Parameters:

Name Type Description Default
config DisambiguationConfig | None

Configuration options for disambiguation behavior. If None, uses default configuration.

 None
Source code in lex_helper/core/disambiguation/analyzer.py 
39
40
41
42
43
44
45
46
47
def __init__(self, config: DisambiguationConfig | None = None):
    """
    Initialize the DisambiguationAnalyzer.

    Args:
        config: Configuration options for disambiguation behavior.
               If None, uses default configuration.
    """
    self.config = config or DisambiguationConfig()

Functions

analyze_request 
analyze_request(lex_request: LexRequest[SessionAttributes]) -> DisambiguationResult

Analyze Lex request and return disambiguation candidates.

Parameters:

Name Type Description Default
lex_request LexRequest[SessionAttributes]

The Lex request containing interpretations with confidence scores

 required

Returns:

Type Description
DisambiguationResult

DisambiguationResult containing analysis results and candidates
Source code in lex_helper/core/disambiguation/analyzer.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def analyze_request(
    self,
    lex_request: LexRequest[SessionAttributes],
) -> DisambiguationResult:
    """
    Analyze Lex request and return disambiguation candidates.

    Args:
        lex_request: The Lex request containing interpretations with confidence scores

    Returns:
        DisambiguationResult containing analysis results and candidates
    """
    logger.debug("Analyzing Lex request for disambiguation: %s", lex_request.inputTranscript)

    # Extract confidence scores from Lex interpretations
    confidence_scores = self.extract_intent_scores(lex_request)

    # Determine if disambiguation should occur
    should_disambiguate = self.should_disambiguate(confidence_scores, self.config.confidence_threshold)

    # Generate candidates if disambiguation is needed
    candidates = []
    if should_disambiguate:
        candidates = self._generate_candidates(confidence_scores, lex_request)

    result = DisambiguationResult(
        should_disambiguate=should_disambiguate,
        candidates=candidates,
        confidence_scores=confidence_scores,
    )

    logger.debug(
        "Disambiguation analysis complete: should_disambiguate=%s, candidates=%d", should_disambiguate, len(candidates)
    )

    return result
extract_intent_scores 
extract_intent_scores(lex_request: LexRequest[SessionAttributes]) -> IntentScores

Extract confidence scores from Lex interpretations.

Parameters:

Name Type Description Default
lex_request LexRequest[SessionAttributes]

The Lex request containing interpretations

 required

Returns:

Type Description
IntentScores

Dictionary mapping intent names to confidence scores (0.0-1.0)
Source code in lex_helper/core/disambiguation/analyzer.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def extract_intent_scores(self, lex_request: LexRequest[SessionAttributes]) -> IntentScores:
    """
    Extract confidence scores from Lex interpretations.

    Args:
        lex_request: The Lex request containing interpretations

    Returns:
        Dictionary mapping intent names to confidence scores (0.0-1.0)
    """
    scores: dict[str, float] = {}

    for interpretation in lex_request.interpretations:
        intent_name = interpretation.intent.name
        confidence = interpretation.nluConfidence or 0.0
        scores[intent_name] = confidence

    logger.debug("Extracted intent scores from Lex: %s", scores)
    return scores
should_disambiguate 
should_disambiguate(scores: IntentScores, threshold: float) -> bool

Determine if disambiguation is needed based on confidence scores.

Parameters:

Name Type Description Default
scores IntentScores

Dictionary of intent names to confidence scores

 required
threshold float

Minimum confidence threshold to avoid disambiguation

 required

Returns:

Type Description
bool

True if disambiguation should be triggered, False otherwise
Source code in lex_helper/core/disambiguation/analyzer.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def should_disambiguate(self, scores: IntentScores, threshold: float) -> bool:
    """
    Determine if disambiguation is needed based on confidence scores.

    Args:
        scores: Dictionary of intent names to confidence scores
        threshold: Minimum confidence threshold to avoid disambiguation

    Returns:
        True if disambiguation should be triggered, False otherwise
    """
    if not scores:
        return False

    # Get the highest scoring intents
    sorted_scores = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    # Filter out very low scores
    meaningful_scores = [score for _, score in sorted_scores if score > MIN_MEANINGFUL_SCORE]
    if len(meaningful_scores) < self.config.min_candidates:
        return False

    # Get top scores for comparison
    if len(sorted_scores) < 2:
        return False

    top_score = sorted_scores[0][1]
    second_score = sorted_scores[1][1]

    # Case 1: Top score is very low (below threshold) AND we have multiple candidates
    if top_score < threshold and len(meaningful_scores) >= self.config.min_candidates:
        return True

    # Case 2: Multiple scores are close to each other (ambiguous case)
    # Only disambiguate if the difference between top scores is small
    score_difference = top_score - second_score

    # If the top two scores are within similarity_threshold of each other, consider it ambiguous
    # AND both scores are reasonably high
    similarity_threshold = self.config.similarity_threshold
    if (
        score_difference <= similarity_threshold
        and top_score >= MIN_REASONABLE_SCORE
        and second_score >= MIN_REASONABLE_SCORE
        and len(meaningful_scores) >= self.config.min_candidates
    ):
        return True

    return False

Bedrock Generator

Bedrock-powered text generation for Smart Disambiguation.

This module provides intelligent, contextual disambiguation message generation using Amazon Bedrock models to create more natural and helpful clarification responses based on user input and available intent candidates.

Classes

BedrockDisambiguationGenerator 

BedrockDisambiguationGenerator(config: BedrockDisambiguationConfig)

Generates disambiguation messages using Amazon Bedrock models.

This class creates contextual, intelligent disambiguation messages by analyzing user input and available intent candidates, then using Bedrock to generate natural language clarification text and button labels.

Parameters:

Name Type Description Default
config BedrockDisambiguationConfig

Configuration for Bedrock text generation

 required
Source code in lex_helper/core/disambiguation/bedrock_generator.py 
32
33
34
35
36
37
38
39
def __init__(self, config: BedrockDisambiguationConfig):
    """
    Initialize the Bedrock disambiguation generator.

    Args:
        config: Configuration for Bedrock text generation
    """
    self.config = config

Functions

generate_clarification_message 
generate_clarification_message(user_input: str, candidates: list[IntentCandidate], context: dict[str, Any] | None = None) -> str

Generate a contextual clarification message using Bedrock.

Parameters:

Name Type Description Default
user_input str

The original user input that was ambiguous

 required
candidates list[IntentCandidate]

List of intent candidates to choose from

 required
context dict[str, Any] | None

Optional context information (session data, etc.)

 None

Returns:

Type Description
str

Generated clarification message text
Source code in lex_helper/core/disambiguation/bedrock_generator.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def generate_clarification_message(
    self, user_input: str, candidates: list[IntentCandidate], context: dict[str, Any] | None = None
) -> str:
    """
    Generate a contextual clarification message using Bedrock.

    Args:
        user_input: The original user input that was ambiguous
        candidates: List of intent candidates to choose from
        context: Optional context information (session data, etc.)

    Returns:
        Generated clarification message text
    """
    if not self.config.enabled:
        return self._get_fallback_message(candidates)

    try:
        prompt = self._build_clarification_prompt(user_input, candidates, context)

        response = invoke_bedrock_simple_converse(
            prompt=prompt,
            model_id=self.config.model_id,
            system_prompt=self.config.system_prompt,
            max_tokens=self.config.max_tokens,
            temperature=self.config.temperature,
            region_name=self.config.region_name,
        )

        generated_text = response["text"].strip()
        logger.debug("Generated clarification message: %s", generated_text)

        return generated_text

    except BedrockInvocationError as e:
        logger.warning("Bedrock clarification generation failed: %s", e)
        if self.config.fallback_to_static:
            return self._get_fallback_message(candidates)
        raise
    except Exception as e:
        logger.error("Unexpected error in Bedrock clarification generation: %s", e)
        if self.config.fallback_to_static:
            return self._get_fallback_message(candidates)
        raise
generate_button_labels 
generate_button_labels(candidates: list[IntentCandidate], user_input: str | None = None) -> list[str]

Generate improved button labels using Bedrock.

Parameters:

Name Type Description Default
candidates list[IntentCandidate]

List of intent candidates

 required
user_input str | None

Optional user input for context

 None

Returns:

Type Description
list[str]

List of generated button labels
Source code in lex_helper/core/disambiguation/bedrock_generator.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def generate_button_labels(self, candidates: list[IntentCandidate], user_input: str | None = None) -> list[str]:
    """
    Generate improved button labels using Bedrock.

    Args:
        candidates: List of intent candidates
        user_input: Optional user input for context

    Returns:
        List of generated button labels
    """
    if not self.config.enabled:
        return [candidate.display_name for candidate in candidates]

    try:
        prompt = self._build_button_labels_prompt(candidates, user_input)

        response = invoke_bedrock_simple_converse(
            prompt=prompt,
            model_id=self.config.model_id,
            system_prompt=self.config.system_prompt,
            max_tokens=self.config.max_tokens,
            temperature=self.config.temperature,
            region_name=self.config.region_name,
        )

        # Parse the JSON response to get button labels
        generated_text = response["text"].strip()

        # Try to parse as JSON first
        try:
            if generated_text.startswith("[") and generated_text.endswith("]"):
                parsed_labels: Any = json.loads(generated_text)
                if isinstance(parsed_labels, list) and len(parsed_labels) == len(candidates):
                    # Ensure all items are strings
                    labels: list[str] = [str(item) for item in parsed_labels]  # type: ignore[misc]
                    logger.debug("Generated button labels: %s", labels)
                    return labels
        except json.JSONDecodeError:
            pass

        # If JSON parsing fails, try to extract labels from text
        extracted_labels = self._extract_labels_from_text(generated_text, len(candidates))
        if extracted_labels:
            logger.debug("Extracted button labels: %s", extracted_labels)
            return extracted_labels

        # Fallback to original display names
        logger.warning("Could not parse generated button labels, using fallback")
        return [candidate.display_name for candidate in candidates]

    except BedrockInvocationError as e:
        logger.warning("Bedrock button label generation failed: %s", e)
        if self.config.fallback_to_static:
            return [candidate.display_name for candidate in candidates]
        raise
    except Exception as e:
        logger.error("Unexpected error in Bedrock button label generation: %s", e)
        if self.config.fallback_to_static:
            return [candidate.display_name for candidate in candidates]
        raise

Functions

Disambiguation Types

Type definitions for the Smart Disambiguation feature.

This module contains all the data classes and type definitions needed for intelligent disambiguation of ambiguous user input in lex-helper.

Attributes

IntentScores module-attribute 

IntentScores = dict[str, float]

Type alias for intent name to confidence score mapping

DisambiguationMessages module-attribute 

DisambiguationMessages = dict[str, str]

Type alias for disambiguation message templates

Classes

IntentCandidate dataclass 

IntentCandidate(intent_name: str, confidence_score: float, display_name: str, description: str, required_slots: list[str] = (lambda: [])())

Represents a potential intent match for disambiguation.

Contains the intent information along with confidence scoring and user-friendly display information for presenting options to users.

Attributes

intent_name instance-attribute 
intent_name: str

Technical name of the intent (e.g., 'BookFlight')

confidence_score instance-attribute 
confidence_score: float

Confidence score between 0.0 and 1.0 for this intent match

display_name instance-attribute 
display_name: str

User-friendly name for display (e.g., 'Book a Flight')

description instance-attribute 
description: str

Brief description of what this intent does

required_slots class-attribute instance-attribute 
required_slots: list[str] = field(default_factory=lambda: [])

List of required slot names for this intent

DisambiguationResult dataclass 

DisambiguationResult(should_disambiguate: bool, candidates: list[IntentCandidate] = (lambda: [])(), confidence_scores: dict[str, float] = (lambda: {})())

Result of disambiguation analysis.

Contains the decision on whether disambiguation should occur and all the information needed to present options to the user.

Attributes

should_disambiguate instance-attribute 
should_disambiguate: bool

Whether disambiguation should be triggered based on analysis

candidates class-attribute instance-attribute 
candidates: list[IntentCandidate] = field(default_factory=lambda: [])

List of intent candidates to present to the user

confidence_scores class-attribute instance-attribute 
confidence_scores: dict[str, float] = field(default_factory=lambda: {})

Raw confidence scores for all analyzed intents

BedrockDisambiguationConfig dataclass 

BedrockDisambiguationConfig(enabled: bool = False, model_id: str = 'anthropic.claude-3-haiku-20240307-v1:0', region_name: str = 'us-east-1', max_tokens: int = 200, temperature: float = 0.3, system_prompt: str = (lambda: 'You are a helpful assistant that creates clear, concise disambiguation messages for chatbot users. When users provide ambiguous input, help them choose between available options with friendly, natural language.')(), fallback_to_static: bool = True)

Configuration for Bedrock-powered disambiguation text generation.

Allows using Amazon Bedrock models to generate contextual and intelligent disambiguation messages and button text based on the user's input and available intent candidates.

Attributes

enabled class-attribute instance-attribute 
enabled: bool = False

Whether to use Bedrock for generating disambiguation text

model_id class-attribute instance-attribute 
model_id: str = 'anthropic.claude-3-haiku-20240307-v1:0'

Bedrock model ID to use for text generation

region_name class-attribute instance-attribute 
region_name: str = 'us-east-1'

AWS region for Bedrock service

max_tokens class-attribute instance-attribute 
max_tokens: int = 200

Maximum tokens for generated response

temperature class-attribute instance-attribute 
temperature: float = 0.3

Temperature for text generation (0.0-1.0, lower = more deterministic)

system_prompt class-attribute instance-attribute 
system_prompt: str = field(default_factory=lambda: 'You are a helpful assistant that creates clear, concise disambiguation messages for chatbot users. When users provide ambiguous input, help them choose between available options with friendly, natural language.')

System prompt for the Bedrock model

fallback_to_static class-attribute instance-attribute 
fallback_to_static: bool = True

Whether to fall back to static messages if Bedrock fails

DisambiguationConfig dataclass 

DisambiguationConfig(confidence_threshold: float = 0.6, max_candidates: int = 3, fallback_to_original: bool = True, min_candidates: int = 2, similarity_threshold: float = 0.15, enable_logging: bool = True, custom_intent_groups: dict[str, list[str]] = (lambda: {})(), custom_messages: dict[str, str] = (lambda: {})(), bedrock_config: BedrockDisambiguationConfig = BedrockDisambiguationConfig())

Configuration options for the disambiguation system.

Allows developers to customize disambiguation behavior including thresholds, candidate limits, and custom intent groupings.

Attributes

confidence_threshold class-attribute instance-attribute 
confidence_threshold: float = 0.6

Minimum confidence score to avoid disambiguation (0.0-1.0)

max_candidates class-attribute instance-attribute 
max_candidates: int = 3

Maximum number of intent candidates to present to users

fallback_to_original class-attribute instance-attribute 
fallback_to_original: bool = True

Whether to fall back to original behavior if disambiguation fails

min_candidates class-attribute instance-attribute 
min_candidates: int = 2

Minimum number of candidates required to trigger disambiguation

similarity_threshold class-attribute instance-attribute 
similarity_threshold: float = 0.15

Maximum difference between top scores to trigger disambiguation (0.0-1.0)

enable_logging class-attribute instance-attribute 
enable_logging: bool = True

Whether to enable detailed logging of disambiguation events

custom_intent_groups class-attribute instance-attribute 
custom_intent_groups: dict[str, list[str]] = field(default_factory=lambda: {})

Custom groupings of related intents for better disambiguation

custom_messages class-attribute instance-attribute 
custom_messages: dict[str, str] = field(default_factory=lambda: {})

Custom clarification messages for specific disambiguation scenarios

bedrock_config class-attribute instance-attribute 
bedrock_config: BedrockDisambiguationConfig = field(default_factory=BedrockDisambiguationConfig)

Configuration for Bedrock-powered text generation

Utility Functions

Additional core utilities for common operations.

Intent Name Utilities

Bedrock Integration

Amazon Bedrock invocation utilities for Lex Helper.

This module provides functionality to invoke Amazon Bedrock models with proper error handling and response formatting.

Classes

BedrockInvocationError

Bases: Exception

Custom exception for Bedrock invocation errors.

Functions

invoke_bedrock 

invoke_bedrock(prompt: str, model_id: str, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, stop_sequences: list[str] | None = None, region_name: str = 'us-east-1', **kwargs: Any) -> dict[str, Any]

Invoke Bedrock model using InvokeModel API.

Parameters:

Name Type Description Default
prompt str

Input prompt

 required
model_id str

Bedrock model identifier

 required
max_tokens int | None

Maximum tokens to generate

 None
temperature float | None

Randomness control (0.0-1.0)

 None
top_p float | None

Nucleus sampling (0.0-1.0)

 None
stop_sequences list[str] | None

Stop generation sequences

 None
region_name str

AWS region

 'us-east-1'
**kwargs Any

Additional model parameters

 {}

Returns:

Type Description
dict[str, Any]

Dict with text, usage, raw_response
Source code in lex_helper/core/invoke_bedrock.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def invoke_bedrock(
    prompt: str,
    model_id: str,
    max_tokens: int | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    stop_sequences: list[str] | None = None,
    region_name: str = "us-east-1",
    **kwargs: Any,
) -> dict[str, Any]:
    """
    Invoke Bedrock model using InvokeModel API.

    Args:
        prompt: Input prompt
        model_id: Bedrock model identifier
        max_tokens: Maximum tokens to generate
        temperature: Randomness control (0.0-1.0)
        top_p: Nucleus sampling (0.0-1.0)
        stop_sequences: Stop generation sequences
        region_name: AWS region
        **kwargs: Additional model parameters

    Returns:
        Dict with text, usage, raw_response
    """
    try:
        bedrock_runtime = boto3.client("bedrock-runtime", region_name=region_name)  # type: ignore[misc]

        config = get_model_config(model_id)
        request_body = config.request_builder(prompt, max_tokens, temperature, top_p, stop_sequences, **kwargs)

        def invoke_func(mid: str) -> dict[str, Any]:
            return bedrock_runtime.invoke_model(  # type: ignore[misc]
                modelId=mid, body=json.dumps(request_body), contentType="application/json", accept="application/json"
            )

        response = _try_with_fallback(bedrock_runtime, model_id, invoke_func)

        response_body = json.loads(response["body"].read())
        return config.response_parser(response_body)

    except Exception as e:
        logger.error("Bedrock invocation failed: %s", e)
        raise BedrockInvocationError(f"Invocation failed: {e}") from e

invoke_bedrock_converse 

invoke_bedrock_converse(messages: list[dict[str, Any]], model_id: str, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, stop_sequences: list[str] | None = None, system_prompt: str | None = None, region_name: str = 'us-east-1', **kwargs: Any) -> dict[str, Any]

Invoke Bedrock model using Converse API with full message support.

Parameters:

Name Type Description Default
messages list[dict[str, Any]]

List of message dicts with role and content

 required
model_id str

Bedrock model identifier

 required
max_tokens int | None

Maximum tokens to generate

 None
temperature float | None

Randomness control (0.0-1.0)

 None
top_p float | None

Nucleus sampling (0.0-1.0)

 None
stop_sequences list[str] | None

Stop generation sequences

 None
system_prompt str | None

System prompt for conversation

 None
region_name str

AWS region

 'us-east-1'
**kwargs Any

Additional parameters

 {}

Returns:

Type Description
dict[str, Any]

Dict with text, usage, raw_response
Example

messages = [ ... {"role": "user", "content": [{"text": "Hello"}]}, ... {"role": "assistant", "content": [{"text": "Hi there!"}]}, ... {"role": "user", "content": [{"text": "How are you?"}]} ... ] response = invoke_bedrock_converse( ... messages=messages, ... model_id="anthropic.claude-3-haiku-20240307-v1:0", ... system_prompt="You are a helpful assistant" ... )

Source code in lex_helper/core/invoke_bedrock.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def invoke_bedrock_converse(
    messages: list[dict[str, Any]],
    model_id: str,
    max_tokens: int | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    stop_sequences: list[str] | None = None,
    system_prompt: str | None = None,
    region_name: str = "us-east-1",
    **kwargs: Any,
) -> dict[str, Any]:
    """
    Invoke Bedrock model using Converse API with full message support.

    Args:
        messages: List of message dicts with role and content
        model_id: Bedrock model identifier
        max_tokens: Maximum tokens to generate
        temperature: Randomness control (0.0-1.0)
        top_p: Nucleus sampling (0.0-1.0)
        stop_sequences: Stop generation sequences
        system_prompt: System prompt for conversation
        region_name: AWS region
        **kwargs: Additional parameters

    Returns:
        Dict with text, usage, raw_response

    Example:
        >>> messages = [
        ...     {"role": "user", "content": [{"text": "Hello"}]},
        ...     {"role": "assistant", "content": [{"text": "Hi there!"}]},
        ...     {"role": "user", "content": [{"text": "How are you?"}]}
        ... ]
        >>> response = invoke_bedrock_converse(
        ...     messages=messages,
        ...     model_id="anthropic.claude-3-haiku-20240307-v1:0",
        ...     system_prompt="You are a helpful assistant"
        ... )
    """
    try:
        bedrock_runtime = boto3.client("bedrock-runtime", region_name=region_name)  # type: ignore[misc]

        inference_config = {
            k: v
            for k, v in {
                "maxTokens": max_tokens or 1000,
                "temperature": temperature,
                "topP": top_p,
                "stopSequences": stop_sequences,
            }.items()
            if v is not None
        }

        converse_params = {"messages": messages, "inferenceConfig": inference_config, **kwargs}

        if system_prompt:
            converse_params["system"] = [{"text": system_prompt}]

        def converse_func(mid: str) -> dict[str, Any]:
            return bedrock_runtime.converse(modelId=mid, **converse_params)  # type: ignore[misc]

        response = _try_with_fallback(bedrock_runtime, model_id, converse_func)

        return {
            "text": response["output"]["message"]["content"][0]["text"],
            "usage": response.get("usage", {}),
            "raw_response": response,
        }

    except Exception as e:
        logger.error("Bedrock converse failed: %s", e)
        raise BedrockInvocationError(f"Converse failed: {e}") from e

invoke_bedrock_simple_converse 

invoke_bedrock_simple_converse(prompt: str, model_id: str, max_tokens: int | None = None, temperature: float | None = None, top_p: float | None = None, stop_sequences: list[str] | None = None, system_prompt: str | None = None, region_name: str = 'us-east-1', **kwargs: Any) -> dict[str, Any]

Simple converse wrapper that converts prompt to messages format.

Parameters:

Name Type Description Default
prompt str

Single user prompt

 required
model_id str

Bedrock model identifier

 required
system_prompt str | None

System prompt for conversation

 None
max_tokens int | None

Maximum tokens to generate

 None
temperature float | None

Sampling temperature

 None
top_p float | None

Top-p sampling parameter

 None
region_name str

AWS region name

 'us-east-1'
**kwargs Any

Additional arguments passed to invoke_bedrock_converse

 {}

Returns:

Type Description
dict[str, Any]

Dict with text, usage, raw_response
Source code in lex_helper/core/invoke_bedrock.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def invoke_bedrock_simple_converse(
    prompt: str,
    model_id: str,
    max_tokens: int | None = None,
    temperature: float | None = None,
    top_p: float | None = None,
    stop_sequences: list[str] | None = None,
    system_prompt: str | None = None,
    region_name: str = "us-east-1",
    **kwargs: Any,
) -> dict[str, Any]:
    """
    Simple converse wrapper that converts prompt to messages format.

    Args:
        prompt: Single user prompt
        model_id: Bedrock model identifier
        system_prompt: System prompt for conversation
        max_tokens: Maximum tokens to generate
        temperature: Sampling temperature
        top_p: Top-p sampling parameter
        region_name: AWS region name
        **kwargs: Additional arguments passed to invoke_bedrock_converse

    Returns:
        Dict with text, usage, raw_response
    """
    messages = [{"role": "user", "content": [{"text": prompt}]}]

    return invoke_bedrock_converse(
        messages=messages,
        model_id=model_id,
        max_tokens=max_tokens,
        temperature=temperature,
        top_p=top_p,
        stop_sequences=stop_sequences,
        system_prompt=system_prompt,
        region_name=region_name,
        **kwargs,
    )

Message Manager

Message Manager for Lex Helper Library.

This module provides centralized message management for Lambda functions using the lex-helper library. It loads messages from a YAML file in the Lambda function's directory structure.

Classes

MessageManager

Singleton class for managing messages from YAML files with locale support.

Automatically loads messages from locale-specific files: - messages_{localeId}.yaml (e.g., messages_en_US.yaml, messages_es_ES.yaml) - Falls back to messages.yaml if locale-specific file not found

Search locations: 1. Custom path from MESSAGES_YAML_PATH environment variable 2. Lambda function root directory 3. Common subdirectories: messages/, config/, resources/, data/ 4. Relative paths from current working directory

Functions

set_locale classmethod 
set_locale(locale: str) -> None

Set the current locale and load messages for it.

Source code in lex_helper/core/message_manager.py 
 96
 97
 98
 99
100
101
@classmethod
def set_locale(cls, locale: str) -> None:
    """Set the current locale and load messages for it."""
    cls._current_locale = locale
    if locale not in cls._messages:
        cls._load_messages(locale)
get_message classmethod 
get_message(key: str, default: str | None = None, locale: str | None = None) -> str

Get message by key, supporting nested keys with dot notation.

Parameters:

Name Type Description Default
key str

Message key to lookup (e.g., "agent.confirmation")

 required
default str | None

Optional default value if key not found

 None
locale str | None

Optional locale override (uses current locale if not specified)

 None

Returns:

Type Description
str

The message string
Example

MessageManager.set_locale("en_US") MessageManager.get_message("greeting") "Hello! How can I assist you today?"

MessageManager.get_message("greeting", locale="es_ES") "¡Hola! ¿Cómo puedo ayudarte hoy?"

Source code in lex_helper/core/message_manager.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def get_message(cls, key: str, default: str | None = None, locale: str | None = None) -> str:
    """
    Get message by key, supporting nested keys with dot notation.

    Args:
        key: Message key to lookup (e.g., "agent.confirmation")
        default: Optional default value if key not found
        locale: Optional locale override (uses current locale if not specified)

    Returns:
        The message string

    Example:
        >>> MessageManager.set_locale("en_US")
        >>> MessageManager.get_message("greeting")
        "Hello! How can I assist you today?"

        >>> MessageManager.get_message("greeting", locale="es_ES")
        "¡Hola! ¿Cómo puedo ayudarte hoy?"
    """
    try:
        target_locale = locale or cls._current_locale

        # Load messages for locale if not already loaded
        if target_locale not in cls._messages:
            cls._load_messages(target_locale)

        # Handle nested keys with dot notation
        keys = key.split(".")
        value = cls._messages.get(target_locale, {})

        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                value = None
                break

        if value is None:
            if default is not None:
                logger.debug(f"Message key not found: {key} for locale {target_locale}, using default")
                return default
            logger.warning(f"Message key not found: {key} for locale {target_locale}")
            return f"Message not found: {key}"

        return str(value)

    except Exception as e:
        logger.error(f"Error getting message for key {key}: {str(e)}")
        return default if default is not None else f"Error getting message: {key}"
reload_messages classmethod 
reload_messages() -> None

Reload messages from YAML file.

Source code in lex_helper/core/message_manager.py 
155
156
157
158
@classmethod
def reload_messages(cls) -> None:
    """Reload messages from YAML file."""
    cls._load_messages()
get_all_messages classmethod 
get_all_messages() -> dict[str, dict[str, str]]

Get all loaded messages.

Source code in lex_helper/core/message_manager.py 
160
161
162
163
164
165
@classmethod
def get_all_messages(cls) -> dict[str, dict[str, str]]:
    """Get all loaded messages."""
    if not cls._messages:
        cls._load_messages()
    return cls._messages.copy()

Functions

set_locale 

set_locale(locale: str) -> None

Set the current locale for message loading.

Source code in lex_helper/core/message_manager.py 
169
170
171
def set_locale(locale: str) -> None:
    """Set the current locale for message loading."""
    MessageManager.set_locale(locale)

get_message 

get_message(key: str, default: str | None = None, locale: str | None = None) -> str

Convenience function to get a message.

Parameters:

Name Type Description Default
key str

Message key to lookup

 required
default str | None

Optional default value if key not found

 None
locale str | None

Optional locale override

 None

Returns:

Type Description
str

The message string
Source code in lex_helper/core/message_manager.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
def get_message(key: str, default: str | None = None, locale: str | None = None) -> str:
    """
    Convenience function to get a message.

    Args:
        key: Message key to lookup
        default: Optional default value if key not found
        locale: Optional locale override

    Returns:
        The message string
    """
    return MessageManager.get_message(key, default, locale)

Logging Utilities

Logging utilities for the lex-helper library.

This module provides consistent logging functionality across the library, following Python logging best practices for libraries.

Functions

get_logger 

get_logger(name: str) -> Logger

Get a logger for the given module name.

Parameters:

Name Type Description Default
name str

The module name, typically name

 required

Returns:

Type Description
Logger

A logger instance configured for the module
Source code in lex_helper/core/logging_utils.py 
17
18
19
20
21
22
23
24
25
26
27
def get_logger(name: str) -> logging.Logger:
    """
    Get a logger for the given module name.

    Args:
        name: The module name, typically __name__

    Returns:
        A logger instance configured for the module
    """
    return logging.getLogger(name)

log_request_debug 

log_request_debug(logger: Logger, request: LexRequest[Any]) -> None

Log request details at DEBUG level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
request LexRequest[Any]

The Lex request to log

 required
Source code in lex_helper/core/logging_utils.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def log_request_debug(logger: logging.Logger, request: LexRequest[Any]) -> None:
    """
    Log request details at DEBUG level.

    Args:
        logger: The logger instance to use
        request: The Lex request to log
    """
    if logger.isEnabledFor(logging.DEBUG):
        logger.debug(
            "Processing Lex request - Intent: %s, Session ID: %s, User ID: %s",
            request.sessionState.intent.name if request.sessionState.intent else "None",
            request.sessionId,
            request.sessionState.sessionAttributes.get("userId", "Unknown")
            if request.sessionState.sessionAttributes
            else "Unknown",
        )

log_exception 

log_exception(logger: Logger, exc: Exception, context: str) -> None

Log exception with context at ERROR level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
exc Exception

The exception that occurred

 required
context str

Additional context about where the exception occurred

 required
Source code in lex_helper/core/logging_utils.py 
49
50
51
52
53
54
55
56
57
58
def log_exception(logger: logging.Logger, exc: Exception, context: str) -> None:
    """
    Log exception with context at ERROR level.

    Args:
        logger: The logger instance to use
        exc: The exception that occurred
        context: Additional context about where the exception occurred
    """
    logger.exception("%s: %s", context, str(exc))

log_handler_invocation 

log_handler_invocation(logger: Logger, handler_name: str, intent_name: str | None = None) -> None

Log handler invocation at DEBUG level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
handler_name str

Name of the handler being invoked

 required
intent_name str | None

Name of the intent being handled (if applicable)

 None
Source code in lex_helper/core/logging_utils.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
def log_handler_invocation(logger: logging.Logger, handler_name: str, intent_name: str | None = None) -> None:
    """
    Log handler invocation at DEBUG level.

    Args:
        logger: The logger instance to use
        handler_name: Name of the handler being invoked
        intent_name: Name of the intent being handled (if applicable)
    """
    if intent_name:
        logger.debug("Invoking handler '%s' for intent '%s'", handler_name, intent_name)
    else:
        logger.debug("Invoking handler '%s'", handler_name)

log_http_request 

log_http_request(logger: Logger, method: str, url: str, status_code: int | None = None, response_time: float | None = None) -> None

Log HTTP request details at DEBUG level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
method str

HTTP method (GET, POST, etc.)

 required
url str

Request URL

 required
status_code int | None

HTTP response status code (if available)

 None
response_time float | None

Request response time in seconds (if available)

 None
Source code in lex_helper/core/logging_utils.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def log_http_request(
    logger: logging.Logger, method: str, url: str, status_code: int | None = None, response_time: float | None = None
) -> None:
    """
    Log HTTP request details at DEBUG level.

    Args:
        logger: The logger instance to use
        method: HTTP method (GET, POST, etc.)
        url: Request URL
        status_code: HTTP response status code (if available)
        response_time: Request response time in seconds (if available)
    """
    if status_code is not None:
        if response_time is not None:
            logger.debug("HTTP %s %s - Status: %d, Time: %.3fs", method, url, status_code, response_time)
        else:
            logger.debug("HTTP %s %s - Status: %d", method, url, status_code)
    else:
        logger.debug("HTTP %s %s", method, url)

log_bedrock_invocation 

log_bedrock_invocation(logger: Logger, model_id: str, success: bool, error_message: str | None = None) -> None

Log Bedrock model invocation at appropriate level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
model_id str

The Bedrock model ID being invoked

 required
success bool

Whether the invocation was successful

 required
error_message str | None

Error message if invocation failed

 None
Source code in lex_helper/core/logging_utils.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def log_bedrock_invocation(logger: logging.Logger, model_id: str, success: bool, error_message: str | None = None) -> None:
    """
    Log Bedrock model invocation at appropriate level.

    Args:
        logger: The logger instance to use
        model_id: The Bedrock model ID being invoked
        success: Whether the invocation was successful
        error_message: Error message if invocation failed
    """
    if success:
        logger.debug("Successfully invoked Bedrock model: %s", model_id)
    else:
        logger.error("Failed to invoke Bedrock model '%s': %s", model_id, error_message or "Unknown error")

log_session_attribute_update 

log_session_attribute_update(logger: Logger, attribute_name: str, old_value: Any, new_value: Any) -> None

Log session attribute updates at DEBUG level.

Parameters:

Name Type Description Default
logger Logger

The logger instance to use

 required
attribute_name str

Name of the session attribute

 required
old_value Any

Previous value

 required
new_value Any

New value

 required
Source code in lex_helper/core/logging_utils.py 
114
115
116
117
118
119
120
121
122
123
124
def log_session_attribute_update(logger: logging.Logger, attribute_name: str, old_value: Any, new_value: Any) -> None:
    """
    Log session attribute updates at DEBUG level.

    Args:
        logger: The logger instance to use
        attribute_name: Name of the session attribute
        old_value: Previous value
        new_value: New value
    """
    logger.debug("Session attribute '%s' updated: %s -> %s", attribute_name, repr(old_value), repr(new_value))

Next: Explore the Channels API for channel-specific formatting functionality.