Warning: This document is for an old version of Rasa Core.

Source code for rasa_core.actions.action

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

import requests
import copy

import rasa_core
from rasa_core import events
from rasa_core.constants import (
    DOCS_BASE_URL,
    DEFAULT_REQUEST_TIMEOUT,
    REQUESTED_SLOT, USER_INTENT_OUT_OF_SCOPE)
from rasa_core.events import (UserUtteranceReverted, UserUttered,
                              ActionExecuted, Event)
from rasa_core.utils import EndpointConfig

if typing.TYPE_CHECKING:
    from rasa_core.trackers import DialogueStateTracker
    from rasa_core.dispatcher import Dispatcher
    from rasa_core.domain import Domain

logger = logging.getLogger(__name__)

ACTION_LISTEN_NAME = "action_listen"

ACTION_RESTART_NAME = "action_restart"

ACTION_DEFAULT_FALLBACK_NAME = "action_default_fallback"

ACTION_DEACTIVATE_FORM_NAME = "action_deactivate_form"

ACTION_REVERT_FALLBACK_EVENTS_NAME = 'action_revert_fallback_events'

ACTION_DEFAULT_ASK_AFFIRMATION_NAME = 'action_default_ask_affirmation'

ACTION_DEFAULT_ASK_REPHRASE_NAME = 'action_default_ask_rephrase'

ACTION_BACK_NAME = 'action_back'


def default_actions() -> List['Action']:
    """List default actions."""
    return [ActionListen(), ActionRestart(),
            ActionDefaultFallback(), ActionDeactivateForm(),
            ActionRevertFallbackEvents(), ActionDefaultAskAffirmation(),
            ActionDefaultAskRephrase(), ActionBack()]


def default_action_names() -> List[Text]:
    """List default action names."""
    return [a.name() for a in default_actions()]


def combine_user_with_default_actions(user_actions):
    # remove all user actions that overwrite default actions
    # this logic is a bit reversed, you'd think that we should remove
    # the action name from the default action names if the user overwrites
    # the action, but there are some locations in the code where we
    # implicitly assume that e.g. "action_listen" is always at location
    # 0 in this array. to keep it that way, we remove the duplicate
    # action names from the users list instead of the defaults
    unique_user_actions = [a
                           for a in user_actions
                           if a not in default_action_names()]
    return default_action_names() + unique_user_actions


def ensure_action_name_uniqueness(action_names: List[Text]) -> None:
    """Check and raise an exception if there are two actions with same name."""

    unique_action_names = set()  # used to collect unique action names
    for a in action_names:
        if a in unique_action_names:
            raise ValueError(
                "Action names are not unique! Found two actions with name"
                " '{}'. Either rename or remove one of them.".format(a))
        else:
            unique_action_names.add(a)


def action_from_name(name: Text, action_endpoint: Optional[EndpointConfig],
                     user_actions: List[Text]) -> 'Action':
    """Return an action instance for the name."""

    defaults = {a.name(): a for a in default_actions()}

    if name in defaults and name not in user_actions:
        return defaults.get(name)
    elif name.startswith("utter_"):
        return UtterAction(name)
    else:
        return RemoteAction(name, action_endpoint)


def actions_from_names(action_names: List[Text],
                       action_endpoint: Optional[EndpointConfig],
                       user_actions: List[Text]) -> List['Action']:
    """Converts the names of actions into class instances."""

    return [action_from_name(name, action_endpoint, user_actions)
            for name in action_names]


class Action(object):
    """Next action to be taken in response to a dialogue state."""

    def name(self) -> Text:
        """Unique identifier of this simple action."""

        raise NotImplementedError

[docs] def run(self, dispatcher: 'Dispatcher', tracker: 'DialogueStateTracker', domain: 'Domain') -> List['Event']: """ Execute the side effects of this action. Args: dispatcher (Dispatcher): the dispatcher which is used to send messages back to the user. Use ``dispatcher.utter_message()`` or any other :class:`rasa_core.dispatcher.Dispatcher` method. tracker (DialogueStateTracker): the state tracker for the current user. You can access slot values using ``tracker.get_slot(slot_name)`` and the most recent user message is ``tracker.latest_message.text``. domain (Domain): the bot's domain Returns: List[Event]: A list of :class:`rasa_core.events.Event` instances """ raise NotImplementedError
def __str__(self) -> Text: return "Action('{}')".format(self.name()) class UtterAction(Action): """An action which only effect is to utter a template when it is run. Both, name and utter template, need to be specified using the `name` method.""" def __init__(self, name): self._name = name def run(self, dispatcher, tracker, domain): """Simple run implementation uttering a (hopefully defined) template.""" dispatcher.utter_template(self.name(), tracker) return [] def name(self) -> Text: return self._name def __str__(self) -> Text: return "UtterAction('{}')".format(self.name()) class ActionBack(Action): """Revert the tracker state by two user utterances.""" def name(self) -> Text: return ACTION_BACK_NAME def run(self, dispatcher, tracker, domain): # only utter the template if it is available dispatcher.utter_template("utter_back", tracker, silent_fail=True) return [UserUtteranceReverted(), UserUtteranceReverted()] class ActionListen(Action): """The first action in any turn - bot waits for a user message. The bot should stop taking further actions and wait for the user to say something.""" def name(self) -> Text: return ACTION_LISTEN_NAME def run(self, dispatcher, tracker, domain): return [] class ActionRestart(Action): """Resets the tracker to its initial state. Utters the restart template if available.""" def name(self) -> Text: return ACTION_RESTART_NAME def run(self, dispatcher, tracker, domain): from rasa_core.events import Restarted # only utter the template if it is available dispatcher.utter_template("utter_restart", tracker, silent_fail=True) return [Restarted()]
[docs]class ActionDefaultFallback(Action): """Executes the fallback action and goes back to the previous state of the dialogue""" def name(self) -> Text: return ACTION_DEFAULT_FALLBACK_NAME def run(self, dispatcher, tracker, domain): from rasa_core.events import UserUtteranceReverted dispatcher.utter_template("utter_default", tracker, silent_fail=True) return [UserUtteranceReverted()]
class ActionDeactivateForm(Action): """Deactivates a form""" def name(self) -> Text: return ACTION_DEACTIVATE_FORM_NAME def run(self, dispatcher, tracker, domain): from rasa_core.events import Form, SlotSet return [Form(None), SlotSet(REQUESTED_SLOT, None)] class RemoteAction(Action): def __init__(self, name: Text, action_endpoint: Optional[EndpointConfig]) -> None: self._name = name self.action_endpoint = action_endpoint def _action_call_format(self, tracker: 'DialogueStateTracker', domain: 'Domain') -> Dict[Text, Any]: """Create the request json send to the action server.""" from rasa_core.trackers import EventVerbosity tracker_state = tracker.current_state(EventVerbosity.ALL) return { "next_action": self._name, "sender_id": tracker.sender_id, "tracker": tracker_state, "domain": domain.as_dict(), "version": rasa_core.__version__ } @staticmethod def action_response_format_spec(): """Expected response schema for an Action endpoint. Used for validation of the response returned from the Action endpoint.""" return { "type": "object", "properties": { "events": { "type": "array", "items": { "type": "object", "properties": { "event": {"type": "string"} } } }, "responses": { "type": "array", "items": { "type": "object", } } }, } def _validate_action_result(self, result): from jsonschema import validate from jsonschema import ValidationError try: validate(result, self.action_response_format_spec()) return True except ValidationError as e: e.message += ( ". Failed to validate Action server response from API, " "make sure your response from the Action endpoint is valid. " "For more information about the format visit " "{}/customactions/".format(DOCS_BASE_URL)) raise e @staticmethod def _utter_responses(responses: List[Dict[Text, Any]], dispatcher: 'Dispatcher', tracker: 'DialogueStateTracker' ) -> None: """Use the responses generated by the action endpoint and utter them. Uses the normal dispatcher to utter the responses from the action endpoint.""" for response in responses: if "template" in response: kwargs = response.copy() del kwargs["template"] draft = dispatcher.nlg.generate( response["template"], tracker, dispatcher.output_channel.name(), **kwargs) if not draft: continue del response["template"] else: draft = {} if "buttons" in response: if "buttons" not in draft: draft["buttons"] = [] draft["buttons"].extend(response["buttons"]) del response["buttons"] draft.update(response) dispatcher.utter_response(draft) def run(self, dispatcher, tracker, domain): json = self._action_call_format(tracker, domain) if not self.action_endpoint: raise Exception("The model predicted the custom action '{}' " "but you didn't configure an endpoint to " "run this custom action. Please take a look at " "the docs and set an endpoint configuration. " "{}/customactions/" "".format(self.name(), DOCS_BASE_URL)) try: logger.debug("Calling action endpoint to run action '{}'." "".format(self.name())) response = self.action_endpoint.request( json=json, method="post", timeout=DEFAULT_REQUEST_TIMEOUT) if response.status_code == 400: response_data = response.json() exception = ActionExecutionRejection( response_data["action_name"], response_data.get("error") ) logger.debug(exception.message) raise exception response.raise_for_status() response_data = response.json() self._validate_action_result(response_data) except requests.exceptions.ConnectionError as e: logger.error("Failed to run custom action '{}'. Couldn't connect " "to the server at '{}'. Is the server running? " "Error: {}".format(self.name(), self.action_endpoint.url, e)) raise Exception("Failed to execute custom action.") except requests.exceptions.HTTPError as e: logger.error("Failed to run custom action '{}'. Action server " "responded with a non 200 status code of {}. " "Make sure your action server properly runs actions " "and returns a 200 once the action is executed. " "Error: {}".format(self.name(), e.response.status_code, e)) raise Exception("Failed to execute custom action.") events_json = response_data.get("events", []) responses = response_data.get("responses", []) self._utter_responses(responses, dispatcher, tracker) evts = events.deserialise_events(events_json) return evts def name(self) -> Text: return self._name class ActionExecutionRejection(Exception): """Raising this exception will allow other policies to predict a different action""" def __init__(self, action_name, message=None): self.action_name = action_name self.message = (message or "Custom action '{}' rejected to run" "".format(action_name)) def __str__(self): return self.message class ActionRevertFallbackEvents(Action): """Reverts events which were done during the `TwoStageFallbackPolicy`. This reverts user messages and bot utterances done during a fallback of the `TwoStageFallbackPolicy`. By doing so it is not necessary to write custom stories for the different paths, but only of the happy path. """ def name(self) -> Text: return ACTION_REVERT_FALLBACK_EVENTS_NAME def run(self, dispatcher: 'Dispatcher', tracker: 'DialogueStateTracker', domain: 'Domain') -> List[Event]: from rasa_core.policies.two_stage_fallback import has_user_rephrased revert_events = [] # User rephrased if has_user_rephrased(tracker): revert_events = _revert_successful_rephrasing(tracker) # User affirmed elif has_user_affirmed(tracker): revert_events = _revert_affirmation_events(tracker) return revert_events def has_user_affirmed(tracker: 'DialogueStateTracker') -> bool: return tracker.last_executed_action_has( ACTION_DEFAULT_ASK_AFFIRMATION_NAME) def _revert_affirmation_events(tracker: 'DialogueStateTracker') -> List[Event]: revert_events = _revert_single_affirmation_events() last_user_event = tracker.get_last_event_for(UserUttered) last_user_event = copy.deepcopy(last_user_event) last_user_event.parse_data['intent']['confidence'] = 1.0 # User affirms the rephrased intent rephrased_intent = tracker.last_executed_action_has( name=ACTION_DEFAULT_ASK_REPHRASE_NAME, skip=1) if rephrased_intent: revert_events += _revert_rephrasing_events() return revert_events + [last_user_event] def _revert_single_affirmation_events() -> List[Event]: return [UserUtteranceReverted(), # revert affirmation and request # revert original intent (has to be re-added later) UserUtteranceReverted(), # add action listen intent ActionExecuted(action_name=ACTION_LISTEN_NAME)] def _revert_successful_rephrasing(tracker) -> List[Event]: last_user_event = tracker.get_last_event_for(UserUttered) last_user_event = copy.deepcopy(last_user_event) return _revert_rephrasing_events() + [last_user_event] def _revert_rephrasing_events() -> List[Event]: return [UserUtteranceReverted(), # remove rephrasing # remove feedback and rephrase request UserUtteranceReverted(), # remove affirmation request and false intent UserUtteranceReverted(), # replace action with action listen ActionExecuted(action_name=ACTION_LISTEN_NAME)] class ActionDefaultAskAffirmation(Action): """Default implementation which asks the user to affirm his intent. It is suggested to overwrite this default action with a custom action to have more meaningful prompts for the affirmations. E.g. have a description of the intent instead of its identifier name. """ def name(self) -> Text: return ACTION_DEFAULT_ASK_AFFIRMATION_NAME def run(self, dispatcher: 'Dispatcher', tracker: 'DialogueStateTracker', domain: 'Domain') -> List[Event]: intent_to_affirm = tracker.latest_message.intent.get('name') affirmation_message = "Did you mean '{}'?".format(intent_to_affirm) dispatcher.utter_button_message(text=affirmation_message, buttons=[{'title': 'Yes', 'payload': '/{}'.format( intent_to_affirm)}, {'title': 'No', 'payload': '/{}'.format( USER_INTENT_OUT_OF_SCOPE)} ]) return [] class ActionDefaultAskRephrase(Action): """Default implementation which asks the user to rephrase his intent.""" def name(self) -> Text: return ACTION_DEFAULT_ASK_REPHRASE_NAME def run(self, dispatcher: 'Dispatcher', tracker: 'DialogueStateTracker', domain: 'Domain') -> List[Event]: dispatcher.utter_template("utter_ask_rephrase", tracker, silent_fail=True) return []