721 lines
28 KiB
Python
721 lines
28 KiB
Python
|
"""
|
||
|
|
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
|