Source code for rasa_core_sdk.forms

# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import logging
import typing
from typing import Dict, Text, Any, List, Union, Optional, Tuple

from rasa_core_sdk import Action, ActionExecutionRejection
from rasa_core_sdk.events import SlotSet, Form

logger = logging.getLogger(__name__)

if typing.TYPE_CHECKING:
    from rasa_core_sdk import Tracker
    from rasa_core_sdk.executor import CollectingDispatcher

# this slot is used to store information needed
# to do the form handling
REQUESTED_SLOT = "requested_slot"


[docs]class FormAction(Action): def name(self): # type: () -> Text """Unique identifier of the form""" raise NotImplementedError("A form must implement a name") @staticmethod def required_slots(tracker): # type: (Tracker) -> List[Text] """A list of required slots that the form has to fill. Use `tracker` to request different list of slots depending on the state of the dialogue """ raise NotImplementedError( "A form must implement required slots that it has to fill" ) def from_entity( self, entity, # type: Text intent=None, # type: Optional[Union[Text, List[Text]]] not_intent=None, # type: Optional[Union[Text, List[Text]]] ): # type: (...) -> Dict[Text: Any] """A dictionary for slot mapping to extract slot value. From: - an extracted entity - conditioned on - intent if it is not None - not_intent if it is not None, meaning user intent should not be this intent """ intent, not_intent = self._list_intents(intent, not_intent) return { "type": "from_entity", "entity": entity, "intent": intent, "not_intent": not_intent, } def from_trigger_intent( self, value, # type: Any intent=None, # type: Optional[Union[Text, List[Text]]] not_intent=None, # type: Optional[Union[Text, List[Text]]] ): # type: (...) -> Dict[Text: Any] """A dictionary for slot mapping to extract slot value. From: - trigger_intent: value pair - conditioned on - intent if it is not None - not_intent if it is not None, meaning user intent should not be this intent Only used on form activation. """ intent, not_intent = self._list_intents(intent, not_intent) return { "type": "from_trigger_intent", "value": value, "intent": intent, "not_intent": not_intent, } def from_intent( self, value, # type: Any intent=None, # type: Optional[Union[Text, List[Text]]] not_intent=None, # type: Optional[Union[Text, List[Text]]] ): # type: (...) -> Dict[Text: Any] """A dictionary for slot mapping to extract slot value. From: - intent: value pair - conditioned on - intent if it is not None - not_intent if it is not None, meaning user intent should not be this intent """ intent, not_intent = self._list_intents(intent, not_intent) return { "type": "from_intent", "value": value, "intent": intent, "not_intent": not_intent, } def from_text( self, intent=None, # type: Optional[Union[Text, List[Text]]] not_intent=None, # type: Optional[Union[Text, List[Text]]] ): # type: (...) -> Dict[Text: Any] """A dictionary for slot mapping to extract slot value. From: - a whole message - conditioned on - intent if it is not None - not_intent if it is not None, meaning user intent should not be this intent """ intent, not_intent = self._list_intents(intent, not_intent) return {"type": "from_text", "intent": intent, "not_intent": not_intent} # noinspection PyMethodMayBeStatic def slot_mappings(self): # type: () -> Dict[Text: Union[Dict, List[Dict]]] """A dictionary to map required slots. Options: - an extracted entity - intent: value pairs - trigger_intent: value pairs - a whole message or a list of them, where the first match will be picked Empty dict is converted to a mapping of the slot to the extracted entity with the same name """ return {} def get_mappings_for_slot(self, slot_to_fill): # type: (Text) -> List[Dict[Text: Any]] """Get mappings for requested slot. If None, map requested slot to an entity with the same name """ requested_slot_mappings = self._to_list( self.slot_mappings().get(slot_to_fill, self.from_entity(slot_to_fill)) ) # check provided slot mappings for requested_slot_mapping in requested_slot_mappings: if ( not isinstance(requested_slot_mapping, dict) or requested_slot_mapping.get("type") is None ): raise TypeError("Provided incompatible slot mapping") return requested_slot_mappings @staticmethod def intent_is_desired(requested_slot_mapping, tracker): # type: (Dict[Text: Any], Tracker) -> bool """Check whether user intent matches intent conditions""" mapping_intents = requested_slot_mapping.get("intent", []) mapping_not_intents = requested_slot_mapping.get("not_intent", []) intent = tracker.latest_message.get("intent", {}).get("name") intent_not_blacklisted = ( not mapping_intents and intent not in mapping_not_intents ) return intent_not_blacklisted or intent in mapping_intents @staticmethod def get_entity_value(name, tracker): # type: (Text, Tracker) -> Any """Extract entities for given name""" # list is used to cover the case of list slot type value = list(tracker.get_latest_entity_values(name)) if len(value) == 0: value = None elif len(value) == 1: value = value[0] return value # noinspection PyUnusedLocal def extract_other_slots( self, dispatcher, # type: CollectingDispatcher tracker, # type: Tracker domain, # type: Dict[Text, Any] ): # type: (...) -> Dict[Text: Any] """Extract the values of the other slots if they are set by corresponding entities from the user input else return None """ slot_to_fill = tracker.get_slot(REQUESTED_SLOT) slot_values = {} for slot in self.required_slots(tracker): # look for other slots if slot != slot_to_fill: # list is used to cover the case of list slot type other_slot_mappings = self.get_mappings_for_slot(slot) for other_slot_mapping in other_slot_mappings: intent = tracker.latest_message.get("intent", {}).get("name") # check whether the slot should be filled # by entity with the same name should_fill_entity_slot = ( other_slot_mapping["type"] == "from_entity" and other_slot_mapping.get("entity") == slot and self.intent_is_desired(other_slot_mapping, tracker) ) # check whether the slot should be # filled from trigger intent mapping should_fill_trigger_slot = ( tracker.active_form.get("name") != self.name() and other_slot_mapping["type"] == "from_trigger_intent" and self.intent_is_desired(other_slot_mapping, tracker) ) if should_fill_entity_slot: value = self.get_entity_value(slot, tracker) elif should_fill_trigger_slot: value = other_slot_mapping.get("value") else: value = None if value is not None: logger.debug( "Extracted '{}' " "for extra slot '{}'" "".format(value, slot) ) slot_values[slot] = value # this slot is done, check next break return slot_values # noinspection PyUnusedLocal def extract_requested_slot( self, dispatcher, # type: CollectingDispatcher tracker, # type: Tracker domain, # type: Dict[Text, Any] ): # type: (...) -> Dict[Text: Any] """Extract the value of requested slot from a user input else return None """ slot_to_fill = tracker.get_slot(REQUESTED_SLOT) logger.debug("Trying to extract requested slot '{}' ...".format(slot_to_fill)) # get mapping for requested slot requested_slot_mappings = self.get_mappings_for_slot(slot_to_fill) for requested_slot_mapping in requested_slot_mappings: logger.debug("Got mapping '{}'".format(requested_slot_mapping)) if self.intent_is_desired(requested_slot_mapping, tracker): mapping_type = requested_slot_mapping["type"] if mapping_type == "from_entity": value = self.get_entity_value( requested_slot_mapping.get("entity"), tracker ) elif mapping_type == "from_intent": value = requested_slot_mapping.get("value") elif mapping_type == "from_trigger_intent": # from_trigger_intent is only used on form activation continue elif mapping_type == "from_text": value = tracker.latest_message.get("text") else: raise ValueError("Provided slot mapping type is not supported") if value is not None: logger.debug( "Successfully extracted '{}' " "for requested slot '{}'" "".format(value, slot_to_fill) ) return {slot_to_fill: value} logger.debug("Failed to extract requested slot '{}'".format(slot_to_fill)) return {} def validate_slots(self, slot_dict, dispatcher, tracker, domain): # type: (Dict[Text, Any], CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Validate slots using helper validation functions. Call validate_{slot} function for each slot, value pair to be validated. If this function is not implemented, set the slot to the value. """ for slot, value in list(slot_dict.items()): validate_func = getattr( self, "validate_{}".format(slot), lambda *x: {slot: value} ) validation_output = validate_func(value, dispatcher, tracker, domain) if not isinstance(validation_output, dict): logger.warning( "Returning values in helper validation methods is deprecated. " + "Your `validate_{}()` method should return ".format(slot) + "a dict of {'slot_name': value} instead." ) validation_output = {slot: validation_output} slot_dict.update(validation_output) # validation succeed, set slots to extracted values return [SlotSet(slot, value) for slot, value in slot_dict.items()] def validate(self, dispatcher, tracker, domain): # type: (CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Extract and validate value of requested slot. If nothing was extracted reject execution of the form action. Subclass this method to add custom validation and rejection logic """ # extract other slots that were not requested # but set by corresponding entity or trigger intent mapping slot_values = self.extract_other_slots(dispatcher, tracker, domain) # extract requested slot slot_to_fill = tracker.get_slot(REQUESTED_SLOT) if slot_to_fill: slot_values.update(self.extract_requested_slot(dispatcher, tracker, domain)) if not slot_values: # reject to execute the form action # if some slot was requested but nothing was extracted # it will allow other policies to predict another action raise ActionExecutionRejection( self.name(), "Failed to extract slot {0} " "with action {1}" "".format(slot_to_fill, self.name()), ) logger.debug("Validating extracted slots: {}".format(slot_values)) return self.validate_slots(slot_values, dispatcher, tracker, domain) # noinspection PyUnusedLocal def request_next_slot( self, dispatcher, # type: CollectingDispatcher tracker, # type: Tracker domain, # type: Dict[Text, Any] ): # type: (...) -> Optional[List[Dict]] """Request the next slot and utter template if needed, else return None""" for slot in self.required_slots(tracker): if self._should_request_slot(tracker, slot): logger.debug("Request next slot '{}'".format(slot)) dispatcher.utter_template( "utter_ask_{}".format(slot), tracker, silent_fail=False, **tracker.slots ) return [SlotSet(REQUESTED_SLOT, slot)] # no more required slots to fill return None def deactivate(self): # type: () -> List[Dict] """Return `Form` event with `None` as name to deactivate the form and reset the requested slot""" logger.debug("Deactivating the form '{}'".format(self.name())) return [Form(None), SlotSet(REQUESTED_SLOT, None)] def submit(self, dispatcher, tracker, domain): # type: (CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Define what the form has to do after all required slots are filled""" raise NotImplementedError("A form must implement a submit method") # helpers @staticmethod def _to_list(x): # type: (Optional[Any]) -> List[Any] """Convert object to a list if it is not a list, None converted to empty list """ if x is None: x = [] elif not isinstance(x, list): x = [x] return x def _list_intents( self, intent=None, # type: Optional[Union[Text, List[Text]]] not_intent=None, # type: Optional[Union[Text, List[Text]]] ): # type: (...) -> Tuple[List[Text], List[Text]] """Check provided intent and not_intent""" if intent and not_intent: raise ValueError( "Providing both intent '{}' and not_intent '{}' " "is not supported".format(intent, not_intent) ) return self._to_list(intent), self._to_list(not_intent) def _log_form_slots(self, tracker): """Logs the values of all required slots before submitting the form.""" req_slots = self.required_slots(tracker) slot_values = "\n".join( ["\t{}: {}".format(slot, tracker.get_slot(slot)) for slot in req_slots] ) logger.debug( "No slots left to request, all required slots are filled:\n{}".format( slot_values ) ) def _activate_if_required(self, dispatcher, tracker, domain): # type: (CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Activate form if the form is called for the first time. If activating, validate any required slots that were filled before form activation and return `Form` event with the name of the form, as well as any `SlotSet` events from validation of pre-filled slots. """ if tracker.active_form.get("name") is not None: logger.debug("The form '{}' is active".format(tracker.active_form)) else: logger.debug("There is no active form") if tracker.active_form.get("name") == self.name(): return [] else: logger.debug("Activated the form '{}'".format(self.name())) events = [Form(self.name())] # collect values of required slots filled before activation prefilled_slots = {} for slot_name in self.required_slots(tracker): if not self._should_request_slot(tracker, slot_name): prefilled_slots[slot_name] = tracker.get_slot(slot_name) if prefilled_slots: logger.debug( "Validating pre-filled required slots: {}".format(prefilled_slots) ) events.extend( self.validate_slots(prefilled_slots, dispatcher, tracker, domain) ) else: logger.debug("No pre-filled required slots to validate.") return events def _validate_if_required(self, dispatcher, tracker, domain): # type: (CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Return a list of events from `self.validate(...)` if validation is required: - the form is active - the form is called after `action_listen` - form validation was not cancelled """ if tracker.latest_action_name == "action_listen" and tracker.active_form.get( "validate", True ): logger.debug("Validating user input '{}'".format(tracker.latest_message)) return self.validate(dispatcher, tracker, domain) else: logger.debug("Skipping validation") return [] @staticmethod def _should_request_slot(tracker, slot_name): # type: (Tracker, Text) -> bool """Check whether form action should request given slot""" return tracker.get_slot(slot_name) is None def run(self, dispatcher, tracker, domain): # type: (CollectingDispatcher, Tracker, Dict[Text, Any]) -> List[Dict] """Execute the side effects of this form. Steps: - activate if needed - validate user input if needed - set validated slots - utter_ask_{slot} template with the next required slot - submit the form if all required slots are set - deactivate the form """ # activate the form events = self._activate_if_required(dispatcher, tracker, domain) # validate user input events.extend(self._validate_if_required(dispatcher, tracker, domain)) # check that the form wasn't deactivated in validation if Form(None) not in events: # create temp tracker with populated slots from `validate` method temp_tracker = tracker.copy() for e in events: if e["event"] == "slot": temp_tracker.slots[e["name"]] = e["value"] next_slot_events = self.request_next_slot(dispatcher, temp_tracker, domain) if next_slot_events is not None: # request next slot events.extend(next_slot_events) else: # there is nothing more to request, so we can submit self._log_form_slots(tracker) logger.debug("Submitting the form '{}'".format(self.name())) events.extend(self.submit(dispatcher, temp_tracker, domain)) # deactivate the form after submission events.extend(self.deactivate()) return events def __str__(self): return "FormAction('{}')".format(self.name())