Skip to content

Commit

Permalink
Refactor NLU.generate_dact (#194)
Browse files Browse the repository at this point in the history
  • Loading branch information
@IKostric
IKostric authored Aug 22, 2023
1 parent 5170a5e commit 796eb67
Show file tree
Hide file tree
Showing 3 changed files with 341 additions and 106 deletions.
7 changes: 2 additions & 5 deletions moviebot/agent/agent.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
269 changes: 168 additions & 101 deletions moviebot/nlu/nlu.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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]]]

Expand All @@ -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.
Expand All @@ -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,
Expand Down
Loading

0 comments on commit 796eb67

Please sign in to comment.