diff --git a/moviebot/nlg/nlg.py b/moviebot/nlg/nlg.py index a1128ee..1625875 100644 --- a/moviebot/nlg/nlg.py +++ b/moviebot/nlg/nlg.py @@ -14,6 +14,10 @@ from moviebot.nlu.annotation.operator import Operator from moviebot.nlu.annotation.slots import Slots from moviebot.nlu.annotation.values import Values +from moviebot.nlu.user_intents_checker import ( + RecommendationChoices, + convert_choice_to_preference, +) ButtonOptions = Dict[DialogueAct, List[str]] CINType = Dict[str, Union[str, List[str]]] @@ -585,19 +589,46 @@ def _user_options_recommend(self) -> ButtonOptions: options = { DialogueAct( UserIntents.REJECT, - [ItemConstraint("reason", Operator.EQ, "watched")], + [ + ItemConstraint("reason", Operator.EQ, "watched"), + ItemConstraint( + "preference", + Operator.EQ, + convert_choice_to_preference( + RecommendationChoices.WATCHED + ), + ), + ], ): ["I have already watched it."], # [random.choice(['I have already watched it.', # 'I have seen this already.'])], DialogueAct( UserIntents.REJECT, - [ItemConstraint("reason", Operator.EQ, "dont_like")], + [ + ItemConstraint("reason", Operator.EQ, "dont_like"), + ItemConstraint( + "preference", + Operator.EQ, + convert_choice_to_preference( + RecommendationChoices.DONT_LIKE + ), + ), + ], ): ["Recommend me something else please."], # [random.choice(['I don\'t like this recommendation.', # 'Recommend me something else please.'])], - DialogueAct(UserIntents.ACCEPT, []): [ - "I like this recommendation." - ], + DialogueAct( + UserIntents.ACCEPT, + [ + ItemConstraint( + "preference", + Operator.EQ, + convert_choice_to_preference( + RecommendationChoices.ACCEPT + ), + ), + ], + ): ["I like this recommendation."], DialogueAct( UserIntents.INQUIRE, [ItemConstraint(Slots.MORE_INFO.value, Operator.EQ, "")], diff --git a/moviebot/nlu/recommendation_decision_processing.py b/moviebot/nlu/recommendation_decision_processing.py new file mode 100644 index 0000000..9e14a25 --- /dev/null +++ b/moviebot/nlu/recommendation_decision_processing.py @@ -0,0 +1,38 @@ +"""This module is used to process the user's decision on a movie recommendation. +""" + +from enum import Enum + + +class RecommendationChoices(Enum): + """Enum class for recommendation choices.""" + + ACCEPT = "accept" + REJECT = "reject" + DONT_LIKE = "dont_like" + WATCHED = "watched" + + +def convert_choice_to_preference(choice: RecommendationChoices) -> float: + """Converts a choice to a preference within the range [-1,1]. + + Dislike is represented by a preference below 0, while like is + represented by a preference above 0. If the choice does not express a + preference (e.g., inquire), then the preference is neutral, i.e., 0. + Possible choices are: accept, reject, dont_like, inquire, and watched. + + Args: + choice: Choice. + + Returns: + Preference within the range [-1,1]. + """ + if choice == RecommendationChoices.ACCEPT: + return 1.0 + elif choice in [ + RecommendationChoices.REJECT, + RecommendationChoices.DONT_LIKE, + ]: + return -1.0 + + return 0.0 diff --git a/moviebot/nlu/user_intents_checker.py b/moviebot/nlu/user_intents_checker.py index 504d585..cf648dd 100644 --- a/moviebot/nlu/user_intents_checker.py +++ b/moviebot/nlu/user_intents_checker.py @@ -19,6 +19,10 @@ from moviebot.nlu.annotation.slots import Slots from moviebot.nlu.annotation.values import Values from moviebot.nlu.data_loader import DataLoader +from moviebot.nlu.recommendation_decision_processing import ( + RecommendationChoices, + convert_choice_to_preference, +) PATTERN_BASIC = { UserIntents.ACKNOWLEDGE: ["yes", "okay", "fine", "sure"], @@ -334,7 +338,15 @@ def check_reject_intent( ] ): dact.intent = UserIntents.REJECT - dact.params = [ItemConstraint("reason", Operator.EQ, "dont_like")] + preference = convert_choice_to_preference( + RecommendationChoices.DONT_LIKE + ) + # TODO: Use enum for constraints' slot. + # See: https://github.com/iai-group/MovieBot/issues/225 + dact.params = [ + ItemConstraint("reason", Operator.EQ, "dont_like"), + ItemConstraint("preference", Operator.EQ, preference), + ] elif any( [ re.search(r"\b{0}\b".format(pattern), utterance) @@ -342,7 +354,13 @@ def check_reject_intent( ] ): dact.intent = UserIntents.REJECT - dact.params = [ItemConstraint("reason", Operator.EQ, "watched")] + preference = convert_choice_to_preference( + RecommendationChoices.WATCHED + ) + dact.params = [ + ItemConstraint("reason", Operator.EQ, "watched"), + ItemConstraint("preference", Operator.EQ, preference), + ] if dact.intent != UserIntents.UNK: user_dacts.append(dact) return user_dacts diff --git a/moviebot/user_modeling/__init__.py b/moviebot/user_modeling/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/moviebot/user_modeling/user_model.py b/moviebot/user_modeling/user_model.py new file mode 100644 index 0000000..1d29cec --- /dev/null +++ b/moviebot/user_modeling/user_model.py @@ -0,0 +1,248 @@ +"""Class for user modeling. + +The user model stores the user's preferences, in terms of slots and items in a +structured (dictionary) and unstructured manner (utterances). These preferences +serve two main purposes: +1. Input for the recommender system. +2. Input for the explainability component to generate an explainable user model. +""" + +from __future__ import annotations + +import json +import logging +import os +from collections import defaultdict +from typing import Dict, List, Optional, Union + +from dialoguekit.core import AnnotatedUtterance +from dialoguekit.utils.dialogue_reader import json_to_annotated_utterance + +_KEY_SLOT_PREFERENCES = "slot_preferences" +_KEY_SLOT_PREFERENCES_NL = "slot_preferences_nl" +_KEY_ITEM_PREFERENCES = "item_preferences" +_KEY_ITEM_PREFERENCES_NL = "item_preferences_nl" + + +class UserModel: + def __init__(self) -> None: + """Initializes the user model.""" + # Structured and unstructured slot preferences + # For slots, the first key is the slot name, and the second key is the + # slot value, e.g., {"genre": {"drama": 1.0, "action": -.5}}. For + # unstructured, the value is a list of annotated utterances. + self.slot_preferences: Dict[str, Dict[str, float]] = defaultdict( + lambda: defaultdict(float) + ) + self.slot_preferences_nl: Dict[ + str, Dict[str, AnnotatedUtterance] + ] = defaultdict(lambda: defaultdict(list)) + + # Structured and unstructured item preferences + # The key is the item id and the value is either a number or a list of + # annotated utterances. + self.item_preferences: Dict[str, float] = defaultdict(float) + + self.item_preferences_nl: Dict[ + str, List[AnnotatedUtterance] + ] = defaultdict(list) + + @classmethod + def from_json(cls, json_path: str) -> UserModel: + """Loads a user model from a JSON file. + + Args: + json_path: Path to the JSON file. + + Raises: + FileNotFoundError: If the JSON file is not found. + + Returns: + User model. + """ + user_model = cls() + if not os.path.exists(json_path): + raise FileNotFoundError(f"JSON file {json_path} not found.") + + user_model_json = json.load(open(json_path, "r")) + user_model.slot_preferences.update( + user_model_json[_KEY_SLOT_PREFERENCES] + ) + for slot, value_utterances in user_model_json[ + _KEY_SLOT_PREFERENCES_NL + ].items(): + for value, utterances in value_utterances.items(): + for utterance in utterances: + user_model.slot_preferences_nl[slot][value].append( + json_to_annotated_utterance(utterance) + ) + + user_model.item_preferences.update( + user_model_json[_KEY_ITEM_PREFERENCES] + ) + for item, utterances in user_model_json[ + _KEY_ITEM_PREFERENCES_NL + ].items(): + for utterance in utterances: + user_model.item_preferences_nl[item].append( + json_to_annotated_utterance(utterance) + ) + return user_model + + def _utterance_to_dict( + self, utterance: AnnotatedUtterance + ) -> Dict[str, str]: + """Converts an utterance to a dictionary. + + TODO: Move this method to DialogueKit AnnotatedUtterance class. + See: https://github.com/iai-group/DialogueKit/issues/248 + + Args: + utterance: Utterance. + + Returns: + Dictionary with utterance information. + """ + return { + "participant": utterance.participant.name, + "utterance": utterance.text, + "intent": utterance.intent.label, + "slot_values": [ + [annotation.slot, annotation.value] + for annotation in utterance.annotations + ] + if utterance.annotations + else [], + } + + def save_as_json_file(self, json_path: str) -> None: + """Saves the user model to a JSON file. + + Args: + json_path: Path to the JSON file. + """ + data = { + _KEY_SLOT_PREFERENCES: self.slot_preferences, + _KEY_ITEM_PREFERENCES: self.item_preferences, + } + + slot_preferences_utterances = defaultdict(lambda: defaultdict(list)) + for slot, value_utterances in self.slot_preferences_nl.items(): + for value, utterances in value_utterances.items(): + slot_preferences_utterances[slot][value] = [ + self._utterance_to_dict(utterance) + for utterance in utterances + ] + + item_preferences_utterances = defaultdict(list) + for item, utterances in self.item_preferences_nl.items(): + item_preferences_utterances[item] = [ + self._utterance_to_dict(utterance) for utterance in utterances + ] + + data.update( + { + _KEY_SLOT_PREFERENCES_NL: slot_preferences_utterances, + _KEY_ITEM_PREFERENCES_NL: item_preferences_utterances, + } + ) + json.dump(data, open(json_path, "w"), indent=4) + + def get_utterances_with_item_preferences( + self, item: Optional[str] = None + ) -> List[AnnotatedUtterance]: + """Returns the utterances with item preference. + + If no item is provided, then all the utterances with item preference + are returned. Else, only the utterances with item preference for the + given item are returned. + + Args: + item: Item. Defaults to None. + + Returns: + Utterances with item preference. + """ + if item is None: + return [ + utterance + for utterances in self.item_preferences_nl.values() + for utterance in utterances + ] + + if item not in self.item_preferences_nl: + logging.warning(f"Item {item} not found in user model.") + return self.item_preferences_nl.get(item, []) + + def get_item_preferences( + self, item: Optional[str] = None + ) -> Union[Dict[str, float], float]: + """Returns the item preferences. + + If no item is provided, then all the item preferences are returned. + Else, only the item preferences for the given item are returned. + + Args: + item: Item. Defaults to None. + + Returns: + Item preferences. + """ + if item is None: + return self.item_preferences + + if item not in self.item_preferences: + logging.warning(f"Item {item} not found in user model.") + return self.item_preferences.get(item, None) + + def get_utterances_with_slot_preferences( + self, slot: Optional[str] = None, value: Optional[str] = None + ) -> Union[Dict[str, List[AnnotatedUtterance]], List[AnnotatedUtterance]]: + """Returns the utterances with slot preference. + + If no slot is provided, then all the utterances with slot preference + are returned. Else, only the utterances with slot preference for the + given slot are returned. + + Args: + slot: Slot. Defaults to None. + value: Value. Defaults to None. + + Returns: + Utterances with slot preference. + """ + if slot is None: + return self.slot_preferences_nl + + if slot not in self.slot_preferences_nl: + logging.warning(f"Slot {slot} not found in user model.") + + if value is not None: + if value not in self.slot_preferences_nl.get(slot, {}): + logging.warning( + f"Value {value} not found for slot {slot} in user model." + ) + return self.slot_preferences_nl.get(slot, {}).get(value, []) + + return self.slot_preferences_nl.get(slot, []) + + def get_slot_preferences( + self, slot: Optional[str] = None + ) -> Union[Dict[str, Dict[str, float]], Dict[str, float]]: + """Returns the slot preferences. + + If no slot is provided, then all the slot preferences are returned. + Else, only the slot preferences for the given slot are returned. + + Args: + slot: Slot. Defaults to None. + + Returns: + Slot preferences. + """ + if slot is None: + return self.slot_preferences + + if slot not in self.slot_preferences: + logging.warning(f"Slot {slot} not found in user model.") + return self.slot_preferences.get(slot, None) diff --git a/tests/data/test_user_model.json b/tests/data/test_user_model.json new file mode 100644 index 0000000..9b80aee --- /dev/null +++ b/tests/data/test_user_model.json @@ -0,0 +1,47 @@ +{ + "slot_preferences": { + "genre": { + "comedy": 1.0, + "action": -1.0, + "drama": -1.0 + } + }, + "item_preferences": { + "movie1": 1.0, + "movie2": -1.0 + }, + "slot_preferences_nl": { + "genre": { + "action": [ + { + "participant": "USER", + "utterance": "I don't like action", + "intent": "reveal", + "slot_values": [] + }, + { + "participant": "USER", + "utterance": "I don't want to watch an action movie", + "intent": "accept", + "slot_values": [] + } + ] + } + }, + "item_preferences_nl": { + "movie1": [ + { + "participant": "USER", + "utterance": "I like movie1", + "intent": "accept", + "slot_values": [] + }, + { + "participant": "USER", + "utterance": "Sounds good", + "intent": "accept", + "slot_values": [] + } + ] + } +} \ No newline at end of file diff --git a/tests/nlu/test_user_intents_checker.py b/tests/nlu/test_user_intents_checker.py index e6c4d36..d91a99c 100644 --- a/tests/nlu/test_user_intents_checker.py +++ b/tests/nlu/test_user_intents_checker.py @@ -192,7 +192,12 @@ def test_check_reveal_voluntary_intent_empty( "reason", Operator.EQ, "dont_like", - ) + ), + ItemConstraint( + "preference", + Operator.EQ, + -1.0, + ), ], ), ), @@ -205,7 +210,12 @@ def test_check_reveal_voluntary_intent_empty( "reason", Operator.EQ, "watched", - ) + ), + ItemConstraint( + "preference", + Operator.EQ, + 0.0, + ), ], ), ), diff --git a/tests/user_modeling/test_user_modeling.py b/tests/user_modeling/test_user_modeling.py new file mode 100644 index 0000000..1127019 --- /dev/null +++ b/tests/user_modeling/test_user_modeling.py @@ -0,0 +1,152 @@ +"""Tests for user modeling.""" + +import os + +import pytest + +from dialoguekit.core.annotated_utterance import AnnotatedUtterance +from dialoguekit.participant.participant import DialogueParticipant +from moviebot.core.intents.user_intents import UserIntents +from moviebot.nlu.recommendation_decision_processing import ( + RecommendationChoices, + convert_choice_to_preference, +) +from moviebot.user_modeling.user_model import UserModel + + +@pytest.fixture(scope="session") +def user_model() -> UserModel: + return UserModel() + + +@pytest.fixture +def filepath() -> str: + return "tests/data/test_user_model.json" + + +def test_get_item_preferences(user_model: UserModel) -> None: + """Tests get_item_preference.""" + preference = convert_choice_to_preference(RecommendationChoices.ACCEPT) + user_model.item_preferences["movie1"] = preference + preference = convert_choice_to_preference(RecommendationChoices.REJECT) + user_model.item_preferences["movie2"] = preference + + assert user_model.get_item_preferences("movie1") == 1.0 + assert user_model.get_item_preferences("movie5") is None + assert user_model.get_item_preferences() == {"movie1": 1.0, "movie2": -1.0} + + +def test_get_utterances_with_item_preferences(user_model: UserModel) -> None: + """Tests get_utterances_with_item_preference.""" + user_model.item_preferences_nl["movie1"] = [ + AnnotatedUtterance( + "I like movie1", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + AnnotatedUtterance( + "Sounds good", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + ] + + assert user_model.get_utterances_with_item_preferences("movie1") == [ + AnnotatedUtterance( + "I like movie1", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + AnnotatedUtterance( + "Sounds good", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + ] + assert user_model.get_utterances_with_item_preferences("movie2") == [] + + +def test_get_slot_preferences(user_model: UserModel) -> None: + """Tests get_slot_preference.""" + user_model.slot_preferences["genre"]["comedy"] = 1.0 + user_model.slot_preferences["genre"]["action"] = -1.0 + user_model.slot_preferences["genre"]["drama"] = -1.0 + + assert user_model.get_slot_preferences("genre") == { + "comedy": 1.0, + "action": -1.0, + "drama": -1.0, + } + assert user_model.get_slot_preferences() == { + "genre": {"comedy": 1.0, "action": -1.0, "drama": -1.0} + } + assert user_model.get_slot_preferences("director") is None + + +def test_get_utterances_with_slot_preferences(user_model: UserModel) -> None: + """Tests get_utterances_with_slot_preferences.""" + user_model.slot_preferences_nl["genre"]["action"] = [ + AnnotatedUtterance( + "I don't like action", + DialogueParticipant.USER, + intent=UserIntents.REVEAL.value, + ), + AnnotatedUtterance( + "I don't want to watch an action movie", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + ] + + assert user_model.get_utterances_with_slot_preferences("actors") == [] + assert ( + user_model.get_utterances_with_slot_preferences("genre", "comedy") == [] + ) + assert user_model.get_utterances_with_slot_preferences( + "genre", "action" + ) == [ + AnnotatedUtterance( + "I don't like action", + DialogueParticipant.USER, + intent=UserIntents.REVEAL.value, + ), + AnnotatedUtterance( + "I don't want to watch an action movie", + DialogueParticipant.USER, + intent=UserIntents.ACCEPT.value, + ), + ] + + +def test_save_as_json_file(user_model: UserModel, filepath: str) -> None: + """Tests save_as_json_file.""" + user_model.save_as_json_file(filepath) + + assert os.path.exists(filepath) + + +def test_user_model_from_json(filepath: str) -> None: + """Tests class method from_json.""" + user_model = UserModel.from_json(filepath) + + assert user_model.get_item_preferences() == { + "movie1": 1.0, + "movie2": -1.0, + } + + assert user_model.get_utterances_with_slot_preferences() == { + "genre": { + "action": [ + AnnotatedUtterance( + "I don't like action", + DialogueParticipant.USER.name, + intent=UserIntents.REVEAL.value, + ), + AnnotatedUtterance( + "I don't want to watch an action movie", + DialogueParticipant.USER.name, + intent=UserIntents.ACCEPT.value, + ), + ] + } + }