From 796eb67d10e9eea624db58ff54f1e66380901ef9 Mon Sep 17 00:00:00 2001 From: Ivica Kostric Date: Tue, 22 Aug 2023 17:26:53 +0200 Subject: [PATCH] Refactor NLU.generate_dact (#194) --- moviebot/agent/agent.py | 7 +- moviebot/nlu/nlu.py | 269 +++++++++++++++++++++++++--------------- tests/nlu/test_nlu.py | 171 +++++++++++++++++++++++++ 3 files changed, 341 insertions(+), 106 deletions(-) create mode 100644 tests/nlu/test_nlu.py diff --git a/moviebot/agent/agent.py b/moviebot/agent/agent.py index a01cc8f..a550c87 100644 --- a/moviebot/agent/agent.py +++ b/moviebot/agent/agent.py @@ -214,11 +214,8 @@ def receive_utterance( previous turn. """ self.dialogue_manager.get_state().user_utterance = user_utterance - - user_dacts = self.nlu.generate_dact( - user_utterance, - user_options, - self.dialogue_manager.get_state(), + user_dacts = self.nlu.generate_dacts( + user_utterance, user_options, self.dialogue_manager.get_state() ) self.dialogue_manager.receive_input(user_dacts) diff --git a/moviebot/nlu/nlu.py b/moviebot/nlu/nlu.py index b1dfed4..782c7f0 100644 --- a/moviebot/nlu/nlu.py +++ b/moviebot/nlu/nlu.py @@ -10,12 +10,10 @@ from moviebot.core.intents.agent_intents import AgentIntents from moviebot.core.intents.user_intents import UserIntents from moviebot.core.utterance.utterance import UserUtterance -from moviebot.database.db_movies import DataBase from moviebot.dialogue_manager.dialogue_act import DialogueAct from moviebot.dialogue_manager.dialogue_state import DialogueState from moviebot.nlu.annotation.values import Values from moviebot.nlu.user_intents_checker import UserIntentsChecker -from moviebot.ontology.ontology import Ontology DialogueOptions = Dict[DialogueAct, Union[str, List[str]]] @@ -35,18 +33,159 @@ def __init__(self, config): Args: config: Paths to ontology, database and tag words for slots in NLU. """ - self.ontology: Ontology = config["ontology"] - self.database: DataBase = config["database"] self.intents_checker = UserIntentsChecker(config) - def generate_dact( # noqa: C901 + def _process_first_turn( + self, user_utterance: UserUtterance + ) -> List[DialogueAct]: + """Generates dialogue acts for the first turn of the conversation. + + The system chceks if the user provided any voluntary preferences or + if the user is just saying hi. + + Args: + user_utterance: User utterance. + + Returns: + A list of dialogue acts. + """ + return ( + self.intents_checker.check_reveal_voluntary_intent(user_utterance) + or self.intents_checker.check_basic_intent( + user_utterance, UserIntents.HI + ) + or [DialogueAct(UserIntents.UNK, [])] + ) + + def _process_last_agent_dacts( + self, user_utterance: UserUtterance, last_agent_dacts: List[DialogueAct] + ) -> List[DialogueAct]: + """Processes response to agent dialogue acts from previous turn. + + + Args: + user_utterance: User utterance. + last_agent_dacts: Last agent dialogue acts. + + Returns: + A list of dialogue acts. Returns an empty list if user haven't + provided any voluntary preferences or preferences after elicitation. + """ + for last_agent_dact in last_agent_dacts: + if last_agent_dact.intent == AgentIntents.WELCOME: + user_dacts = self._follow_up_welcome(user_utterance) + if user_dacts: + return user_dacts + elif last_agent_dact.intent == AgentIntents.ELICIT: + user_dacts = self._follow_up_elicit( + user_utterance, last_agent_dact + ) + if user_dacts: + return user_dacts + return [] + + def _process_recommendation_feedback( + self, user_utterance: UserUtterance + ) -> List[DialogueAct]: + """Processes recommendation feedback from the user. The function checks + if the user is rejecting the recommendation, inquiring about the + recommendation, or providing voluntary preferences. + + Args: + user_utterance: User utterance. + + Returns: + A list of dialogue acts. + """ + feedback_intents = [ + self.intents_checker.check_reject_intent, + self.intents_checker.check_inquire_intent, + self.intents_checker.check_reveal_voluntary_intent, + self._convert_deny_to_inquire, + ] + + for check_intent in feedback_intents: + user_dacts = check_intent(user_utterance) + if user_dacts: + return user_dacts + return [] + + def _follow_up_welcome( + self, user_utterance: UserUtterance + ) -> List[DialogueAct]: + """Follow up on welcome intent. + + Args: + user_utterance: User utterance. + + Returns: + A list of dialogue acts. + """ + return self.intents_checker.check_reveal_voluntary_intent( + user_utterance + ) or self.intents_checker.check_basic_intent( + user_utterance, UserIntents.ACKNOWLEDGE + ) + + def _follow_up_elicit( + self, user_utterance: UserUtterance, last_agent_dact: DialogueAct + ) -> List[DialogueAct]: + """Follow up on elicit intent. + + Args: + user_utterance: User utterance. + last_agent_dact: Last agent dialogue act. + + Returns: + A list of dialogue acts. + """ + user_dacts = self.intents_checker.check_reveal_intent( + user_utterance, last_agent_dact + ) + elicitation_is_irrelevant = any( + [ + param.value in Values.__dict__.values() + for dact in user_dacts + for param in dact.params + ] + ) + if not user_dacts or elicitation_is_irrelevant: + user_dacts.extend( + self.intents_checker.check_reveal_voluntary_intent( + user_utterance + ) + ) + return user_dacts + + def _convert_deny_to_inquire( + self, user_utterance: UserUtterance + ) -> List[DialogueAct]: + """Converts deny intent to inquire intent. + + Args: + user_utterance: User utterance. + + Returns: + A list of dialogue acts. + """ + # TODO: It is unclear the purpose of this function. It should be + # removed or refactored. + # https://github.com/iai-group/MovieBot/issues/199 + deny_dact = self.intents_checker.check_basic_intent( + user_utterance, UserIntents.DENY + ) + if deny_dact: + deny_dact[0].intent = UserIntents.INQUIRE + return deny_dact + + def generate_dacts( self, user_utterance: UserUtterance, options: DialogueOptions, - dialogue_state: DialogueState = None, - ): - """Processes the utterance according to dialogue state and context and - generate a user dialogue act for Agent to understand. + dialogue_state: DialogueState, + ) -> List[DialogueAct]: + """Processes the utterance according to dialogue state and + generates a user dialogue act for Agent to understand. Args: user_utterance: UserUtterance class containing user input. @@ -57,111 +196,39 @@ def generate_dact( # noqa: C901 Returns: A list of dialogue acts. """ - # this is the top priority. The agent must check if user selected - # any option + # This is the top priority. The agent must check if user selected + # any option. selected_option = self.get_selected_option( user_utterance, options, dialogue_state.item_in_focus ) if selected_option: return selected_option - # Define a list of dialogue acts for this specific utterance - user_dacts = [] + # Check if user is ending the conversation. + bye_dacts = self.intents_checker.check_basic_intent( + user_utterance, UserIntents.BYE + ) + if bye_dacts: + return bye_dacts - # process the utterance for necessary - # utterance = self.intents_checker._lemmatize_value(raw_utterance) - self.dialogue_state = dialogue_state + # Check if it's the start of a conversation. + if not dialogue_state.last_agent_dacts: + return self._process_first_turn(user_utterance) - # check if user is ending the conversation - user_dacts.extend( - self.intents_checker.check_basic_intent( - user_utterance, UserIntents.BYE - ) + # Start eliciting or follow up on elicitation. + user_dacts = self._process_last_agent_dacts( + user_utterance, dialogue_state.last_agent_dacts ) - if len(user_dacts) > 0: + if user_dacts: return user_dacts - # check if it's the start of a conversation - if not self.dialogue_state.last_agent_dacts: - user_dacts.extend( - self.intents_checker.check_reveal_voluntary_intent( - user_utterance - ) - ) - if len(user_dacts) == 0: - user_dacts.extend( - self.intents_checker.check_basic_intent( - user_utterance, UserIntents.HI - ) - ) - if len(user_dacts) > 0: - return user_dacts - else: - return None - - for last_agent_dact in self.dialogue_state.last_agent_dacts: - if last_agent_dact.intent == AgentIntents.WELCOME: - user_dacts.extend( - self.intents_checker.check_reveal_voluntary_intent( - user_utterance - ) - ) - if len(user_dacts) == 0: - user_dacts.extend( - self.intents_checker.check_basic_intent( - user_utterance, UserIntents.ACKNOWLEDGE - ) - ) - if len(user_dacts) > 0: - return user_dacts - elif last_agent_dact.intent == AgentIntents.ELICIT: - user_dacts.extend( - self.intents_checker.check_reveal_intent( - user_utterance, last_agent_dact - ) - ) - if len(user_dacts) == 0 or any( - [ - param.value in Values.__dict__.values() - for dact in user_dacts - for param in dact.params - ] - ): - user_dacts.extend( - self.intents_checker.check_reveal_voluntary_intent( - user_utterance - ) - ) - if len(user_dacts) > 0: - return user_dacts - + # Handle feedback after recommendation. if dialogue_state.agent_made_offer: - user_dacts.extend( - self.intents_checker.check_reject_intent(user_utterance) - ) - if len(user_dacts) == 0: - user_dacts.extend( - self.intents_checker.check_inquire_intent(user_utterance) - ) - if len(user_dacts) == 0: - user_dacts.extend( - self.intents_checker.check_reveal_voluntary_intent( - user_utterance - ) - ) - if len(user_dacts) == 0: - deny_dact = self.intents_checker.check_basic_intent( - user_utterance, UserIntents.DENY - ) - if len(deny_dact) > 0: - deny_dact[0].intent = UserIntents.INQUIRE - user_dacts.extend(deny_dact) - if len(user_dacts) > 0: - return user_dacts + user_dacts = self._process_recommendation_feedback(user_utterance) + if user_dacts: + return user_dacts - if len(user_dacts) == 0: - user_dacts.append(DialogueAct(UserIntents.UNK, [])) - return user_dacts + return [DialogueAct(UserIntents.UNK, [])] def get_selected_option( self, diff --git a/tests/nlu/test_nlu.py b/tests/nlu/test_nlu.py new file mode 100644 index 0000000..76efd38 --- /dev/null +++ b/tests/nlu/test_nlu.py @@ -0,0 +1,171 @@ +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from moviebot.core.intents.agent_intents import AgentIntents +from moviebot.core.intents.user_intents import UserIntents +from moviebot.core.utterance.utterance import UserUtterance +from moviebot.dialogue_manager.dialogue_act import DialogueAct +from moviebot.nlu.annotation.item_constraint import ItemConstraint +from moviebot.nlu.annotation.operator import Operator +from moviebot.nlu.annotation.values import Values +from moviebot.nlu.nlu import NLU +from tests.mocks.mock_data_loader import MockDataLoader + + +class MockIntentChecker(MagicMock): + def check_basic_intent(self, user_utterance, intent): + if intent == UserIntents.BYE and user_utterance.text == "bye": + return [DialogueAct(UserIntents.BYE, [])] + if intent == UserIntents.DENY and user_utterance.text == "no": + return [DialogueAct(UserIntents.DENY, [])] + return [] + + def check_reveal_voluntary_intent(self, user_utterance): + if user_utterance.text == "voluntary reveal text": + return [DialogueAct(UserIntents.REVEAL, [])] + return [] + + def check_reveal_intent(self, user_utterance, last_agent_dact): + if last_agent_dact.intent == AgentIntents.ELICIT: + if user_utterance.text == "reveal intent text": + return [DialogueAct(UserIntents.REVEAL, [])] + elif user_utterance.text == "dont care text": + constraint = ItemConstraint("", Operator.EQ, Values.DONT_CARE) + return [ + DialogueAct( + UserIntents.REVEAL, + [constraint], + ) + ] + return [] + + def check_reject_intent(self, user_utterance): + if user_utterance.text == "reject text": + return [DialogueAct(UserIntents.REJECT, [])] + return [] + + def check_inquire_intent(self, user_utterance): + if user_utterance.text == "inquire text": + return [DialogueAct(UserIntents.INQUIRE, [])] + return [] + + +@pytest.fixture +@patch("moviebot.nlu.user_intents_checker.DataLoader", new=MockDataLoader) +def nlu(): + config = { + "ontology": "", + "database": "", + "slot_values_path": "", + "tag_words_slots_path": "", + } + nlu = NLU(config) + nlu.intents_checker = MockIntentChecker() + return nlu + + +@pytest.fixture +def dialogue_state(): + dialogue_state = Mock() + dialogue_state.item_in_focus = None + dialogue_state.last_agent_dacts = [] + return dialogue_state + + +@pytest.mark.parametrize( + "last_dacts", [[], [DialogueAct(AgentIntents.ACKNOWLEDGE, [])]] +) +def test_no_intent(nlu, dialogue_state, last_dacts): + utterance = UserUtterance("random text that doesn't match any intent") + dialogue_state.last_agent_dacts = last_dacts + options = {} + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == UserIntents.UNK + + +def test_basic_bye_intent(nlu, dialogue_state): + utterance = UserUtterance("bye") + options = {} + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == UserIntents.BYE + + +@pytest.mark.parametrize( + "user_intent", + [UserIntents.ACKNOWLEDGE, UserIntents.CONTINUE_RECOMMENDATION], +) +def test_selected_option(nlu, dialogue_state, user_intent): + utterance = UserUtterance("selected option") + options = {DialogueAct(user_intent): "selected option"} + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == user_intent + + +@pytest.mark.parametrize( + "last_dacts, utterance, expected_intent", + [ + ( + [DialogueAct(AgentIntents.WELCOME, [])], + "reject text", + UserIntents.REJECT, + ), + ([DialogueAct(AgentIntents.RECOMMEND, [])], "no", UserIntents.INQUIRE), + ], +) +def test_deny_intent( + nlu, dialogue_state, last_dacts, utterance, expected_intent +): + utterance = UserUtterance(utterance) + options = {} + + dialogue_state.last_agent_dacts = last_dacts + dialogue_state.agent_made_offer = True + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == expected_intent + + +@pytest.mark.parametrize( + "utterance, expected_intent", + [ + ("voluntary reveal text", UserIntents.REVEAL), + ("not reveal text", UserIntents.UNK), + ], +) +def test_reveal_voluntary_intent( + nlu, dialogue_state, utterance, expected_intent +): + utterance = UserUtterance(utterance) + options = {} + + dialogue_state.last_agent_dacts = [DialogueAct(AgentIntents.WELCOME, [])] + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == expected_intent + + +@pytest.mark.parametrize("utterance", ["reveal intent text", "dont care text"]) +def test_elicit_intent(nlu, dialogue_state, utterance): + utterance = UserUtterance(utterance) + options = {} + + dialogue_state.last_agent_dacts = [DialogueAct(AgentIntents.ELICIT, [])] + + dacts = nlu.generate_dacts(utterance, options, dialogue_state) + + assert len(dacts) == 1 + assert dacts[0].intent == UserIntents.REVEAL