BA-Chatbot_Ali_Thesis/chatbot/actions/actions.py

"""
This file contains a series of custom actions for a Rasa-based chatbot, designed to interact with a question-answering component. 
The actions cater to various scenarios including expert searches, module recommendations (WPM Recommendations), 
and providing answers related to the Studienprüfungsordnung (examination regulations) and crawled web data from Hochschule Mannheim.
The actions are structured to check slot values for different parameters like reader model, retrieval model, and reranking, allowing for dynamic configuration during runtime. 
These actions enable the chatbot to adapt its response strategy based on user input and context.
Additionally, the chatbot includes a feedback mechanism and a function to provide top-k document references, enhancing user interaction and information retrieval effectiveness.
Some actions are equipped with fallback mechanisms to search in alternative indexes if the primary search doesn't yield an answer, 
ensuring the bot can still provide relevant information in various situations.
"""
import datetime
from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
from rasa_sdk.events import FollowupAction, SlotSet, ReminderScheduled
import requests
import json
from .helper.credit_mappings import CREDITS
import os

BACKEND_HOST = os.environ.get("FLASK_HOST", "localhost")

FEEDBACK_BUTTONS = [
    {"title": "👍", "payload": "/liked_answer"},
    {"title": "👎", "payload": "/disliked_answer"},
]
COULD_NOT_FIND_ANSWER = "Ich konnte keine Antwort finden"
NO_ANSWER = "Keine Antwort"
NO_INFO = "keine Information"


def get_intent_before_last(tracker: Tracker) -> str:
    user_events = [event for event in tracker.events if event.get("event") == "user"]
    if len(user_events) < 2:
        return None
    intent_before_last = (
        user_events[-2].get("parse_data", {}).get("intent", {}).get("name")
    )
    return intent_before_last


def get_last_executed_action(tracker: Tracker, domain) -> str:
    lastAction = next(
        e["name"]
        for e in reversed(tracker.events)
        if e["event"] == "action"
        and "name" in e
        and (e["name"] in domain["actions"] or e["name"] in domain["forms"])
        and e["name"] != "action_set_reminder"
    )
    return lastAction


def extract_answer_from_response(response, reader_model):
    if reader_model == "GPT":
        answer = response.json()["answer"]["choices"][0]["message"]["content"]
    else:
        answer = response.json()["answer"]["answers"][0]["answer"]
    return answer


class ActionGreet(Action):
    """Sends a greeting message to the user and follows up with 'action_listen'."""
    def name(self) -> Text:
        return "action_greet"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(response="utter_greet")
        return [FollowupAction("action_listen")]


class ActionGetCreditsForModule(Action):
    """Fetches and responds with credit information for a specific module based on user query."""
    def name(self) -> Text:
        return "action_get_credits"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        entities = tracker.latest_message["entities"]
        print(entities, flush=True)
        module = next(
            (
                entity["value"]
                for entity in entities
                if "entity" in entity and "module" == entity["entity"]
            ),
            None,
        )
        print(module, flush=True)
        print("entity", flush=True)
        if module:
            payload = json.dumps({"module": module + "\n"})
            headers = {
                "Content-Type": "application/json",
            }
            resp = requests.post(
                f"http://{BACKEND_HOST}:8000/get_module_credits",
                headers=headers,
                data=payload,
            )
            result = resp.json()
            found_credits = None
            if result:
                print(result, flush=True)
                found_module = result[0]
                found_title = found_module["meta"]["title"].strip()
                if "credits" in found_module["meta"]:
                    found_credits = found_module["meta"]["credits"]
                if not found_credits and module in CREDITS:
                    found_credits = CREDITS[module] + " credits"
                if found_credits:
                    dispatcher.utter_message(
                        f"Das Module: **{found_title}** gibt {found_credits} "
                    )
                    return []

            dispatcher.utter_message(
                f"Konnte zu dem Module: {module} keine Informationen finden..."
            )

        return []


class ActionRecommendModule(Action):
    """Provides module recommendations based on user's interests, career goals, and previous courses."""
    def name(self) -> Text:
        return "action_recommend_module"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        events = []
        semester = tracker.get_slot("semester")
        interests = tracker.get_slot("interests")
        previous_courses = tracker.get_slot("previous_courses")
        future_carrer = tracker.get_slot("future_carrer")
        retrieval_method_or_model = (
            tracker.get_slot("retrieval_method_or_model") or "ada"
        )
        rerank = (
            tracker.get_slot("rerank")
            if tracker.get_slot("rerank") is not None
            else False
        )
        reader_model = tracker.get_slot("reader_model") or "GPT"
        payload = json.dumps(
            {
                "interests": interests,
                "future_carrer": future_carrer,
                "previous_courses": previous_courses,
                "index": "ib",
                "retrieval_method_or_model": retrieval_method_or_model,
                "recommendation_method": "generate_llm_answer" if reader_model == "GPT" else "generate_farm_reader_answer",
                "rerank_retrieved_results": rerank,
            }
        )
        headers = {
            "Content-Type": "application/json",
        }
        resp = requests.post(
            f"http://{BACKEND_HOST}:8080/recommend_wpms",
            headers=headers,
            data=payload,
        )
        if reader_model =="Bert":
            courses = resp.json()
            print(courses,'courses')
            if courses:
                dispatcher.utter_message(
                    "Basierend auf deinen Interessen kann ich dir folgende Wahlpflichtmodule empfehlen:"
                )
                title = ""
                profs = ""
                credits = ""
                semester = None
                uttered_modules = 0
                for course in courses[:3]:
                    meta = course.get("meta", {})
                    is_wmp = meta.get("is_wpm")
                    if not is_wmp:
                        continue
                    title = meta.get("name_de")
                    description = meta.get("inhalte_de")
                    credits = meta.get("credits")
                    profs = meta.get("dozenten")
                    semester = meta.get("semester")
                    if uttered_modules > 3:
                        break 
                    if (
                        semester
                        and semester in ["6", "7", "6/7"]
                        and title.strip()
                        not in ["Bachelorarbeit (BA)", "Wissenschaftliches Arbeiten (WIA)"]
                    ):
                        dispatcher.utter_message(
                            text=f"**Titel** : {title}\n **ECTS**: {credits}\n **Beschreibung**:\n{description}\n\n **Dozenten**:{profs}"
                        )
                        uttered_modules += 1
                if uttered_modules == 0:
                    dispatcher.utter_message(
                        "Basierend auf deinen Anfragen konnte ich leider keine Wahlpflichtmodule finden"
                    )
                else:
                    dispatcher.utter_message(text="Haben dir die Empfehlungen gefallen?",  buttons= FEEDBACK_BUTTONS )

        else:
            answer = resp.json()["answer"]["choices"][0]["message"]["content"]
            events.append(SlotSet("wpm_recommendation_answer", answer))
            dispatcher.utter_message(text=answer, buttons=FEEDBACK_BUTTONS)
        return events


class ActionAnswerStupo(Action):
    """Finds and provides answers related to the Studienprüfungsordnung based on the user's query."""
    def name(self) -> Text:
        return "action_answer_stupo"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        url = f"http://{BACKEND_HOST}:8080/get_answer"
        retrieval_method_or_model = (
            tracker.get_slot("retrieval_method_or_model") or "ada"
        )
        reader_model = tracker.get_slot("reader_model") or "GPT"
        rerank = (
            tracker.get_slot("rerank")
            if tracker.get_slot("rerank") is not None
            else False
        )

        events = []
        buttons = []
        latest_message = tracker.latest_message
        index = "stupo"
        user_text = None
        if latest_message:
            user_text = latest_message["text"]
            payload = {
                "query": user_text,
                "index": index,
                "retrieval_method_or_model": retrieval_method_or_model,
                "reader_model": reader_model,
                "rerank_documents": rerank,
            }

            headers = {
                "Content-Type": "application/json",
                "Authorization": "Basic Og==",
            }
            response = requests.request(
                "POST", url, headers=headers, data=json.dumps(payload)
            )
            buttons.extend(FEEDBACK_BUTTONS)
            buttons.append(
                {"title": "Liste die Referenzen auf.", "payload": "/ask_for_references"}
            )
            answer = extract_answer_from_response(
                reader_model=reader_model, response=response
            )
            if answer is not None and (
                COULD_NOT_FIND_ANSWER in answer
                or NO_ANSWER in answer
                or NO_INFO in answer
            ):
                dispatcher.utter_message(
                    "Ich konnte keine Antwort in der Studienprüfungsordnung finden..."
                )
                dispatcher.utter_message(
                    "Ich suche nun nach Informationen auf den Hochschulseiten..."
                )
                index = "crawled_hsma"
                payload["index"] = index
                response = requests.request(
                    "POST", url, headers=headers, data=json.dumps(payload)
                )
                answer = extract_answer_from_response(
                    reader_model=reader_model, response=response
                )

            dispatcher.utter_message(
                text=answer,
                buttons=buttons,
            )
            events.append(SlotSet("query_stupo", user_text))
            events.append(SlotSet("answer_stupo", answer))
            events.append(
                SlotSet("retrieval_method_or_model", retrieval_method_or_model)
            )
            events.append(SlotSet("reader_model", reader_model))
            events.append(SlotSet("last_searched_index", index))
        references = response.json().get("documents")
        if references:
            events.append(SlotSet("references", references))
        return events


class ActionAskAboutCrawledHSMAData(Action):
    """Addresses general inquiries by searching through crawled data from the Hochschule Mannheim website."""

    def name(self) -> Text:
        return "action_ask_about_crawled_hsma_data"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        url = f"http://{BACKEND_HOST}:8080/get_answer"
        latest_message = tracker.latest_message
        retrieval_method_or_model = (
            tracker.get_slot("retrieval_method_or_model") or "ada"
        )
        reader_model = tracker.get_slot("reader_model") or "GPT"
        rerank = (
            tracker.get_slot("rerank")
            if tracker.get_slot("rerank") is not None
            else False
        )
        user_text = None
        index = "crawled_hsma"
        events = []
        buttons = []
        if latest_message:
            user_text = latest_message["text"]
            payload = {
                "query": user_text,
                "index": index,
                "retrieval_method_or_model": retrieval_method_or_model,
                "reader_model": reader_model,
                "rerank_documents": rerank,
            }

            headers = {
                "Authorization": "Basic Og==",
                "Content-Type": "application/json",
            }
            response = requests.request(
                "POST", url, headers=headers, data=json.dumps(payload)
            )
            references = response.json().get("documents")
            if references:
                events.append(SlotSet("references", references))
            buttons.extend(FEEDBACK_BUTTONS)
            buttons.append(
                {"title": "Liste die Referenzen auf", "payload": "/ask_for_references"}
            )
            answer = extract_answer_from_response(
                reader_model=reader_model, response=response
            )
            if answer is not None and (
                COULD_NOT_FIND_ANSWER in answer
                or NO_ANSWER in answer
                or NO_INFO in answer
            ):
                dispatcher.utter_message(
                    "Ich konnte keine Antwort in auf den Hochschulseiten finden..."
                )
                dispatcher.utter_message(
                    "Ich suche nun nach Informationen in der Studienprüfungsordnung..."
                )
                index = "stupo"
                payload["index"] = index
                response = requests.request(
                    "POST", url, headers=headers, data=json.dumps(payload)
                )
                answer = extract_answer_from_response(
                    reader_model=reader_model, response=response
                )

            dispatcher.utter_message(
                text=answer,
                buttons=buttons,
            )
            events.append(SlotSet("query_crawled_data", user_text))
            events.append(SlotSet("answer_crawled_data", answer))
            events.append(
                SlotSet("retrieval_method_or_model", retrieval_method_or_model)
            )
            events.append(SlotSet("reader_model", reader_model))
        return events


class ActionProvideReferences(Action):
    """Lists references related to the user's last query from either StuPO or crawled data."""

    def name(self) -> Text:
        return "action_provide_references"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        references = tracker.get_slot("references")
        intent_before_last = get_intent_before_last(tracker=tracker)
        header_text = "Hier sind die Referenzen: \n"
        if intent_before_last == "stupo_question":
            header_text = "Hier sind die Referenzen aus der [StuPO](https://www.hs-mannheim.de/fileadmin/user_upload/hauptseite/pdf/SCS/Satzungen/SPO/Bachelor/230824_SPO_Bachelor.pdf): \n"

        # Max length for each reference content
        MAX_LEN = 150  # Adjust based on your requirements

        markdown_references = []
        for ref in references:
            content = ref.get("content", "")
            meta = ref.get("meta")
            truncated_content = (
                (content[:MAX_LEN] + "...") if len(content) > MAX_LEN else content
            )

            # Check if a URL is provided
            url = meta.get("url", None)
            if url:
                markdown_references.append(f"- {truncated_content} [Link]({url})")
            else:
                markdown_references.append(f"- {truncated_content}")

        # Join the references and dispatch them in markdown format
        markdown_message = header_text + "\n".join(markdown_references)
        dispatcher.utter_message(markdown_message)

        return []


class ActionExampleStupoQuestions(Action):
    """Offers example questions related to Studienprüfungsordnung for user guidance."""
    def name(self) -> Text:
        return "action_provide_stupo_example_questions"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        events = []
        buttons = [
            {"title": question, "payload": question}
            for question in [
                "Wie kann man Elternzeit beantragen?",
                "Darf ich die Frist der Bachelorthesis verlängern?",
                "Welche Vorteile haben Studierende mit Kindern?",
                "Wieviele Präsenztage muss ich im Praktischen Studiensemester ableisten?",
                "Wie lange werden meine Daten und Prüfungsleistungen von der Hochschule gepeichert?",
            ]
        ]
        text = "Ich kann versuchen, dir bei inhaltichen Fragen über die StuPo zu helfen.\n  Einige Beispielfragen:"
        dispatcher.utter_message(
            text=text,
            buttons=buttons,
        )
        return events


class ActionExampleGeneralQuestions(Action):
    """Presents sample general questions about Hochschule Mannheim to assist the user."""
    def name(self) -> Text:
        return "action_provide_general_example_questions"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        buttons = [
            {"title": question, "payload": question}
            for question in [
                "Wo finde ich einen 3D-Drucker?",
                "Wo befindet sich der lasercutter?",
                "Wie setze ich mein Passwort für das Hochschulportal zurück?",
                "An wen kann ich mich wenden, wenn ich mein Passwort und meine zentrale Kennung vergessen habe?",
                "Wo befindet sich der Mars-Raum?",
            ]
        ]
        text = "Ich kann versuchen, dir bei allgemeinen Fragen über die Hochschule zu helfen.\n  Einige Beispielfragen:"
        dispatcher.utter_message(
            text=text,
            buttons=buttons,
        )
        return []


class ActionExpertSearch(Action):
    """Conducts an expert search based on the user's query and provides relevant results."""
    def name(self) -> Text:
        return "action_expert_search"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        url = f"http://{BACKEND_HOST}:8080/search_experts"
        buttons = [*FEEDBACK_BUTTONS]
        events = []
        query = tracker.get_slot("expert_search_query")
        events.append(SlotSet("expert_search_query", query))
        events.append(SlotSet("retrieval_method_or_model", "mpnet"))
        events.append(SlotSet("reader_model", "GPT"))
        events.append(FollowupAction("action_listen"))

        retrieval_method_or_model = (
            tracker.get_slot("retrieval_method_or_model") or "ada"
        )
        reader_model = tracker.get_slot("reader_model") or "GPT"

        print(reader_model,'reader_model')
        rerank = (
            tracker.get_slot("rerank")
            if tracker.get_slot("rerank") is not None
            else False
        )
        payload = json.dumps(
            {
                "query": query,
                "index": "stupo",
                "retriever_model": retrieval_method_or_model,
                "generate_answer": True,
                "rerank_retrieved_results": rerank,
                "search_method": "classic_retriever_reader" if reader_model =="GPT" else "retriever_farm_reader"
            }
        )
        headers = {"Authorization": "Basic Og==", "Content-Type": "application/json"}
        response = requests.request("POST", url, headers=headers, data=payload)
        if response is not None:
            response_json: Dict = response.json()
            if reader_model =="Bert":
                if response_json:
                    dispatcher.utter_message(
                        "Basierend auf deinen Interessen kann ich dir folgende Experten sortiert nach Relevanz empfehlen:"
                    )
                    expert = ""
                    title_work = None
                    uttered_experts = []
                    for doc in response_json:
                        meta = doc.get("meta", {})
                        expert = meta.get("author")
                        title = meta.get("title")
                        description = meta.get("abstract")
                        if len(uttered_experts) > 3:
                            break
                        if (
                            expert and title and expert not in uttered_experts
                        ):
                            dispatcher.utter_message(
                                text= f"**Experte**: {expert}\n **Relevantes Paper** :{title}\n"
                            )
                            uttered_experts.append(expert)
                    if len(uttered_experts) == 0:
                        dispatcher.utter_message(
                            "Basierend auf deinen Anfragen konnte ich leider keine Experten finden"
                        )
                    else:
                        dispatcher.utter_message(text="Hat dir die Expertensuche gefallen?",  buttons= FEEDBACK_BUTTONS )
                return events
            answer= extract_answer_from_response(response=response, reader_model=reader_model)
            documents = response_json.get("documents")
            if answer:
                events.append(
                    SlotSet("expert_search_answer", answer)
                )
                answer = answer
                dispatcher.utter_message(
                    text=answer,
                    buttons=buttons,
                )
        return events


class ActionHandleFeedback(Action):
    """Collects user feedback on the bot's responses and forwards it for processing."""

    def name(self) -> Text:
        return "action_handle_feedback"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        last_intent = tracker.get_intent_of_latest_message()
        last_action = get_last_executed_action(tracker=tracker, domain=domain)
        query_stupo = tracker.get_slot("query_stupo")
        answer_stupo = tracker.get_slot("answer_stupo")
        query_crawled_data = tracker.get_slot("query_crawled_data")
        answer_crawled_data = tracker.get_slot("answer_crawled_data")
        expert_search_query = tracker.get_slot("expert_search_query")
        expert_search_answer = tracker.get_slot("expert_search_answer")
        retrieval_method_or_model = tracker.get_slot("retrieval_method_or_model")
        future_carrer = tracker.get_slot("future_carrer")
        interests = tracker.get_slot("interests")
        previous_courses = tracker.get_slot("previous_courses")
        reader_model = tracker.get_slot("reader_model")
        wpm_recommendation_answer = tracker.get_slot("wpm_recommendation_answer")
        last_searched_index = tracker.get_slot("last_searched_index")
        payload = {}
        if last_action == "action_ask_about_crawled_hsma_data":
            payload = json.dumps(
                {
                    "type": "crawled_data",
                    "user_queston": query_crawled_data,
                    "provided_answer": answer_crawled_data,
                    "retrieval_method_or_model": retrieval_method_or_model,
                    "reader_model": reader_model,
                    "feedback": last_intent,
                    "last_searched_index": last_searched_index,
                }
            )
        if last_action == "action_answer_stupo":
            payload = json.dumps(
                {
                    "type": "stupo",
                    "user_queston": query_stupo,
                    "provided_answer": answer_stupo,
                    "retrieval_method_or_model": retrieval_method_or_model,
                    "reader_model": reader_model,
                    "feedback": last_intent,
                    "last_searched_index": last_searched_index,
                }
            )
        if last_action == "action_expert_search":
            payload = json.dumps(
                {
                    "type": "expert_search",
                    "user_queston": expert_search_query,
                    "provided_answer": expert_search_answer,
                    "retrieval_method_or_model": retrieval_method_or_model,
                    "reader_model": "GPT",
                    "feedback": last_intent,
                }
            )
        if last_action == "action_recommend_module":
            payload = json.dumps(
                {
                    "type": "wpm_recommendation",
                    "user_queston": f"future_carrer:{future_carrer}\n interests: {interests}\n previous_courses:{previous_courses}",
                    "provided_answer": wpm_recommendation_answer,
                    "reader_model": "GPT",
                    "retrieval_method_or_model": retrieval_method_or_model,
                    "feedback": last_intent,
                }
            )
        headers = {
            "Content-Type": "application/json",
        }
        resp = requests.post(
            f"http://{BACKEND_HOST}:8080/feedback",
            headers=headers,
            data=payload,
        )
        dispatcher.utter_message(text="Vielen Dank für das Feedback!")
        dispatcher.utter_message(response="utter_how_can_i_help")
        return [FollowupAction("action_listen")]


class ActionSetReminder(Action):
    """Sets a reminder for the user based on their request with a default one-minute delay."""

    def name(self) -> Text:
        return "action_set_reminder"

    async def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        date = datetime.datetime.now() + datetime.timedelta(minutes=1)
        entities = tracker.latest_message.get("entities")

        reminder = ReminderScheduled(
            "EXTERNAL_reminder",
            trigger_date_time=date,
            entities=entities,
            name="my_reminder",
            kill_on_user_message=True,
        )

        return [reminder]


class ActionResetSlots(Action):
    """Resets conversation slots to clear stored values from previous interactions."""
    def name(self) -> Text:
        return "action_reset_slots"

    def run(
        self,
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any],
    ) -> List[Dict[Text, Any]]:
        events = []
        events.append(SlotSet("expert_search_query", None))
        events.append(SlotSet("interests", None))
        events.append(SlotSet("future_carrer", None))
        events.append(SlotSet("previous_courses", None))
        events.append(SlotSet("references", None))
        return events