closing_actions.py

from typing import Text, Dict, List, Any

from rasa_sdk import Tracker, FormValidationAction
from rasa_sdk.executor import CollectingDispatcher
from rasa_sdk.forms import Action
from rasa_sdk.events import (
    SlotSet,
    SessionStarted,
    UserUtteranceReverted,
    ActionReverted,
)
from rasa_sdk.types import DomainDict

from actions.actions import form_action_is_starting
from common.metrics import flow_started_count, flow_finished_count, Flow


class ValidateFormClosing(FormValidationAction):
    def name(self) -> Text:
        return "validate_form_closing"

    @staticmethod
    def break_form_if_not_extracted_requested_slot(
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: DomainDict,
        slot: Text,
    ):
        if (
            tracker.slots.get("requested_slot") == slot
            and tracker.slots.get(slot) is None
        ):
            return {"core_break_form": True}

        return {slot: tracker.slots.get(slot)}

    @staticmethod
    def extract_closing_got_help(
        dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict
    ) -> Dict[Text, Any]:
        return ValidateFormClosing.break_form_if_not_extracted_requested_slot(
            dispatcher, tracker, domain, "closing_got_help"
        )

    @staticmethod
    def extract_closing_leave_feedback(
        dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict
    ) -> Dict[Text, Any]:
        return ValidateFormClosing.break_form_if_not_extracted_requested_slot(
            dispatcher, tracker, domain, "closing_leave_feedback"
        )

    async def should_ask_closing_feedback(self, dispatcher, tracker, domain):
        requested_slot = tracker.get_slot("requested_slot")

        if (
            requested_slot == "closing_leave_feedback"
            and tracker.get_slot("closing_leave_feedback") is True
        ):
            return True

        next_requested_slot = await self.next_requested_slot(
            dispatcher, tracker, domain
        )
        if (
            next_requested_slot is not None
            and next_requested_slot.get("value") == "closing_feedback"
        ):
            return True

        return False

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        # feedback_type is set by the feedback flow
        if await form_action_is_starting(self, dispatcher, tracker, domain):
            if tracker.get_slot("closing_feedback_type") is None:
                flow_started_count(Flow.CLOSING)
                return [
                    SlotSet(key, None)
                    for key in [
                        "closing_got_help",
                        "closing_leave_feedback",
                        "closing_feedback",
                        "closing_skip_got_help",
                        "closing_feedback_type",
                    ]
                ]

        requested_slot = tracker.get_slot("requested_slot")

        if requested_slot == "closing_got_help":
            closing_got_help = tracker.get_slot("closing_got_help")
            if closing_got_help is True:
                dispatcher.utter_message(response="utter_closing_got_help_yes")
            elif closing_got_help is False:
                dispatcher.utter_message(response="utter_closing_got_help_no")

        if await self.should_ask_closing_feedback(dispatcher, tracker, domain):
            closing_feedback_type = tracker.get_slot("closing_feedback_type")
            if closing_feedback_type == "general":
                dispatcher.utter_message(response="utter_closing_feedback_general")
            elif closing_feedback_type == "bad_experience":
                dispatcher.utter_message(
                    response="utter_closing_feedback_bad_experience"
                )
            elif closing_feedback_type == "this_conversation":
                dispatcher.utter_message(
                    response="utter_closing_feedback_this_conversation"
                )
            else:
                dispatcher.utter_message(response="utter_closing_feedback_general")

        form_result = await super().run(dispatcher, tracker, domain)

        core_break_form = tracker.get_slot("core_break_form")

        if core_break_form is True:
            form_start = tracker.get_last_event_for("active_loop")
            form_start_index = tracker.events.index(form_start)
            user_utterances_count = 1
            for event in tracker.events[form_start_index:]:
                if event["event"] == "user":
                    user_utterances_count += 1

            return [UserUtteranceReverted()] * user_utterances_count + [
                SlotSet("requested_slot", None),
                ActionReverted(),
            ]

        return form_result

    async def required_slots(
        self,
        domain_slots: List[Text],
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: DomainDict,
    ) -> List[Text]:
        updated_slots = domain_slots.copy()

        if tracker.get_slot("closing_skip_got_help") is True:
            updated_slots.remove("closing_got_help")

        if tracker.get_slot("closing_feedback_type") is not None:
            updated_slots.remove("closing_leave_feedback")

        if tracker.get_slot("closing_leave_feedback") is False:
            updated_slots.remove("closing_feedback")

        return updated_slots


class ExecuteFormClosing(Action):
    def name(self) -> Text:
        return "execute_form_closing"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        if tracker.get_slot("closing_feedback_type") is None:
            flow_finished_count(Flow.CLOSING)

        if tracker.get_slot("closing_leave_feedback") is True:
            closing_feedback_type = (
                tracker.get_slot("closing_feedback_type") or "general"
            )
            closing_skip_got_help = tracker.get_slot("closing_skip_got_help") or False
            closing_feedback = tracker.get_slot("closing_feedback")
            closing_got_help = tracker.get_slot("closing_got_help")

            feedback_type_label = (
                "-".join(closing_feedback_type.split("_")) + "-feedback"
            )
            feedback_type = " ".join(closing_feedback_type.split("_")).capitalize()

            user_got_help_section = "* Because of the feedback flow, the user was not asked if the conversation was helpful."
            if closing_skip_got_help is False:
                user_got_help_section = (
                    f"Did the user get help from the assistant? {closing_got_help}"
                )

            dispatcher.utter_message(
                json_message={
                    "type": "command",
                    "command": "feedback",
                    "params": {
                        "summary": f"Virtual assistant feedback",
                        "description": f"""
                        Feedback type: {feedback_type}
                        Feedback: {closing_feedback}
                        Sender: {tracker.sender_id}
                        
                        {user_got_help_section}
                        """,
                        "labels": ["virtual-assistant", feedback_type_label],
                    },
                },
            )

            dispatcher.utter_message(response="utter_feedback_transparency")
            dispatcher.utter_message(response="utter_closing_finally")
        return [
            SlotSet(key, None)
            for key in [
                "closing_got_help",
                "closing_leave_feedback",
                "closing_feedback",
                "closing_skip_got_help",
                "closing_feedback_type",
            ]
        ]


class ValidateFormClosingAnythingElse(FormValidationAction):
    def name(self) -> Text:
        return "validate_form_closing_anything_else"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        if await form_action_is_starting(self, dispatcher, tracker, domain):
            flow_started_count(Flow.CLOSING_ANYTHING_ELSE)

        return await super().run(dispatcher, tracker, domain)


class ExecuteFormClosingAnythingElse(Action):
    def name(self) -> Text:
        return "execute_form_closing_anything_else"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        flow_finished_count(Flow.CLOSING_ANYTHING_ELSE)
        closing_anything_else = tracker.get_slot("closing_anything_else")
        if closing_anything_else is True:
            dispatcher.utter_message(response="utter_core_how_can_i_help")
        elif closing_anything_else is False:
            dispatcher.utter_message(response="utter_closing_bye")
        else:
            return []

        return [SlotSet(key, None) for key in ["closing_anything_else"]] + [
            SessionStarted()
        ]


class ActionSkipGotHelp(Action):
    def name(self) -> Text:
        return "action_skip_got_help"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        return [SlotSet("closing_skip_got_help", True)]


class ActionClosingFeedbackTypeBadExperience(Action):
    def name(self) -> Text:
        return "action_closing_feedback_type_bad_experience"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        return [
            SlotSet("closing_feedback_type", "bad_experience"),
            SlotSet("closing_leave_feedback", True),
        ]


class ActionClosingFeedbackTypeGeneral(Action):
    def name(self) -> Text:
        return "action_closing_feedback_type_general"

    async def run(
        self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
    ) -> List[Dict[Text, Any]]:
        return [
            SlotSet("closing_feedback_type", "general"),
            SlotSet("closing_leave_feedback", True),
        ]