import logging import textwrap from typing import Dict, Text, List, Any, Union from unittest.mock import Mock import pytest from _pytest.logging import LogCaptureFixture from _pytest.monkeypatch import MonkeyPatch from aioresponses import aioresponses from rasa.core.agent import Agent from rasa.core.policies.policy import PolicyPrediction from rasa.core.actions import action from rasa.core.actions.action import ActionExecutionRejection, ActionExtractSlots from rasa.shared.constants import ( LATEST_TRAINING_DATA_FORMAT_VERSION, REQUIRED_SLOTS_KEY, IGNORED_INTENTS, ) from rasa.shared.core.constants import ACTION_LISTEN_NAME, REQUESTED_SLOT from rasa.core.actions.forms import FormAction from rasa.core.channels import CollectingOutputChannel from rasa.shared.core.domain import Domain from rasa.shared.core.events import ( ActiveLoop, SlotSet, UserUttered, ActionExecuted, BotUttered, Restarted, Event, ActionExecutionRejected, DefinePrevUserUtteredFeaturization, ) from rasa.core.nlg import TemplatedNaturalLanguageGenerator from rasa.shared.core.trackers import DialogueStateTracker from rasa.utils.endpoints import EndpointConfig ACTION_SERVER_URL = "http://my-action-server:5055/webhook" async def test_activate(): tracker = DialogueStateTracker.from_events(sender_id="bla", evts=[]) form_name = "my form" action = FormAction(form_name, None) slot_name = "num_people" domain = textwrap.dedent( f""" slots: {slot_name}: type: float mappings: - type: from_entity entity: number forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} responses: utter_ask_num_people: - text: "How many people?" """ ) domain = Domain.from_yaml(domain) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert isinstance(events[-1], BotUttered) assert events[:-1] == [ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, slot_name)] async def test_activate_with_custom_slot_mapping(): tracker = DialogueStateTracker.from_events(sender_id="bla", evts=[]) form_name = "my_form" action_server = EndpointConfig(ACTION_SERVER_URL) action = FormAction(form_name, action_server) domain_required_slot_name = "num_people" slot_set_by_remote_custom_extraction_method = "some_slot" slot_value_set_by_remote_custom_extraction_method = "anything" domain = textwrap.dedent( f""" slots: {domain_required_slot_name}: type: float mappings: - type: from_entity entity: number {slot_set_by_remote_custom_extraction_method}: type: any mappings: - type: custom forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {domain_required_slot_name} responses: utter_ask_num_people: - text: "How many people?" actions: - validate_{form_name} """ ) domain = Domain.from_yaml(domain) form_validation_events = [ { "event": "slot", "timestamp": None, "name": slot_set_by_remote_custom_extraction_method, "value": slot_value_set_by_remote_custom_extraction_method, } ] with aioresponses() as mocked: mocked.post( ACTION_SERVER_URL, payload={"events": form_validation_events}, ) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events[:-1] == [ ActiveLoop(form_name), SlotSet( slot_set_by_remote_custom_extraction_method, slot_value_set_by_remote_custom_extraction_method, ), SlotSet(REQUESTED_SLOT, domain_required_slot_name), ] assert isinstance(events[-1], BotUttered) async def test_activate_with_mapping_conditions_slot(): tracker = DialogueStateTracker.from_events(sender_id="bla", evts=[]) form_name = "my form" action = FormAction(form_name, None) slot_name = "num_people" domain = textwrap.dedent( f""" slots: {slot_name}: type: float mappings: - type: from_entity entity: number conditions: - active_loop: {form_name} forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} responses: utter_ask_num_people: - text: "How many people?" """ ) domain = Domain.from_yaml(domain) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events[:-1] == [ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, slot_name)] assert isinstance(events[-1], BotUttered) async def test_activate_with_prefilled_slot(): slot_name = "num_people" slot_value = 5 tracker = DialogueStateTracker.from_events( sender_id="bla", evts=[SlotSet(slot_name, slot_value)] ) form_name = "my_form" action = FormAction(form_name, None) next_slot_to_request = "next slot to request" domain = f""" forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} - {next_slot_to_request} slots: {slot_name}: type: any mappings: - type: from_entity entity: {slot_name} {next_slot_to_request}: type: text mappings: - type: from_text """ domain = Domain.from_yaml(domain) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, next_slot_to_request), ] async def test_activate_with_prefilled_slot_with_mapping_conditions(): slot_name = "num_people" slot_value = 5 tracker = DialogueStateTracker.from_events( sender_id="bla", evts=[SlotSet(slot_name, slot_value)] ) form_name = "my form" action = FormAction(form_name, None) next_slot_to_request = "next slot to request" domain = f""" forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} - {next_slot_to_request} slots: {slot_name}: type: any mappings: - type: from_entity entity: {slot_name} conditions: - active_loop: {form_name} {next_slot_to_request}: type: text mappings: - type: from_text conditions: - active_loop: {form_name} """ domain = Domain.from_yaml(domain) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, next_slot_to_request), ] async def test_switch_forms_with_same_slot(default_agent: Agent): """Tests switching of forms, where the first slot is the same in both forms. Tests the fix for issue 7710""" # Define two forms in the domain, with same first slot slot_a = "my_slot_a" form_1 = "my_form_1" utter_ask_form_1 = f"Please provide the value for {slot_a} of form 1" form_2 = "my_form_2" utter_ask_form_2 = f"Please provide the value for {slot_a} of form 2" domain = f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" nlu: - intent: order_status examples: | - check status of my order - when are my shoes coming in - intent: return examples: | - start a return - I don't want my shoes anymore slots: {slot_a}: type: float mappings: - type: from_entity entity: number forms: {form_1}: {REQUIRED_SLOTS_KEY}: - {slot_a} {form_2}: {REQUIRED_SLOTS_KEY}: - {slot_a} responses: utter_ask_{form_1}_{slot_a}: - text: {utter_ask_form_1} utter_ask_{form_2}_{slot_a}: - text: {utter_ask_form_2} """ domain = Domain.from_yaml(domain) # Driving it like rasa/core/processor processor = default_agent.processor processor.domain = domain # activate the first form tracker = DialogueStateTracker.from_events( "some-sender", evts=[ ActionExecuted(ACTION_LISTEN_NAME), UserUttered("order status", {"name": "form_1", "confidence": 1.0}), DefinePrevUserUtteredFeaturization(False), ], ) # rasa/core/processor.predict_next_action prediction = PolicyPrediction([], "some_policy") action_1 = FormAction(form_1, None) await processor._run_action( action_1, tracker, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), prediction, ) events_expected = [ ActionExecuted(ACTION_LISTEN_NAME), UserUttered("order status", {"name": "form_1", "confidence": 1.0}), DefinePrevUserUtteredFeaturization(False), ActionExecuted(form_1), ActiveLoop(form_1), SlotSet(REQUESTED_SLOT, slot_a), BotUttered( text=utter_ask_form_1, metadata={"utter_action": f"utter_ask_{form_1}_{slot_a}"}, ), ] assert tracker.applied_events() == events_expected next_events = [ ActionExecuted(ACTION_LISTEN_NAME), UserUttered("return my shoes", {"name": "form_2", "confidence": 1.0}), DefinePrevUserUtteredFeaturization(False), ] tracker.update_with_events(next_events, domain) events_expected.extend(next_events) # form_1 is still active, and bot will first validate if the user utterance # provides valid data for the requested slot, which is rejected await processor._run_action( action_1, tracker, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), prediction, ) events_expected.extend([ActionExecutionRejected(action_name=form_1)]) assert tracker.applied_events() == events_expected # Next, bot predicts form_2 action_2 = FormAction(form_2, None) await processor._run_action( action_2, tracker, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), prediction, ) events_expected.extend( [ ActionExecuted(form_2), ActiveLoop(form_2), SlotSet(REQUESTED_SLOT, slot_a), BotUttered( text=utter_ask_form_2, metadata={"utter_action": f"utter_ask_{form_2}_{slot_a}"}, ), ] ) assert tracker.applied_events() == events_expected async def test_activate_and_immediate_deactivate(): slot_name = "num_people" slot_value = 5 tracker = DialogueStateTracker.from_events( sender_id="bla", evts=[ ActionExecuted(ACTION_LISTEN_NAME), UserUttered( "haha", {"name": "greet"}, entities=[{"entity": slot_name, "value": slot_value}], ), SlotSet(slot_name, slot_value), ], ) form_name = "my form" action = FormAction(form_name, None) domain = f""" forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} slots: {slot_name}: type: any mappings: - type: from_entity entity: {slot_name} """ domain = Domain.from_yaml(domain) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ] async def test_set_slot_and_deactivate(): form_name = "my form" slot_name = "num_people" slot_value = "dasdasdfasdf" events = [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, slot_name), ActionExecuted(ACTION_LISTEN_NAME), UserUttered(slot_value), SlotSet(slot_name, slot_value), ] tracker = DialogueStateTracker.from_events(sender_id="bla", evts=events) domain = f""" forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} slots: {slot_name}: type: text influence_conversation: false mappings: - type: from_text """ domain = Domain.from_yaml(domain) form_action = FormAction(form_name, None) events = await form_action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [SlotSet(REQUESTED_SLOT, None), ActiveLoop(None)] async def test_action_rejection(): form_name = "my form" slot_to_fill = "some slot" tracker = DialogueStateTracker.from_events( sender_id="bla", evts=[ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, slot_to_fill), ActionExecuted(ACTION_LISTEN_NAME), UserUttered("haha", {"name": "greet"}), ], ) form_name = "my form" action = FormAction(form_name, None) domain = f""" forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_to_fill} slots: {slot_to_fill}: type: any mappings: - type: from_entity entity: some_entity """ domain = Domain.from_yaml(domain) with pytest.raises(ActionExecutionRejection): await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) @pytest.mark.parametrize( "validate_return_events, expected_events", [ # Validate function returns SlotSet events for every slot to fill ( [ {"event": "slot", "name": "num_people", "value": "so_clean"}, {"event": "slot", "name": "num_tables", "value": 5}, ], [ SlotSet("num_people", "so_clean"), SlotSet("num_tables", 5), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ], ), # Validate function returns extra Slot Event ( [ {"event": "slot", "name": "num_people", "value": "so_clean"}, {"event": "slot", "name": "some_other_slot", "value": 2}, ], [ SlotSet("num_people", "so_clean"), SlotSet("some_other_slot", 2), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ], ), # Validate function only validates one of the candidates ( [{"event": "slot", "name": "num_people", "value": "so_clean"}], [ SlotSet("num_people", "so_clean"), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ], ), # Validate function says slot is invalid ( [{"event": "slot", "name": "num_people", "value": None}], [SlotSet("num_people", None), SlotSet(REQUESTED_SLOT, "num_people")], ), # Validate function decides to request a slot which is not part of the default # slot mapping ( [{"event": "slot", "name": "requested_slot", "value": "is_outside"}], [SlotSet(REQUESTED_SLOT, "is_outside")], ), # Validate function decides that no more slots should be requested ( [ {"event": "slot", "name": "num_people", "value": None}, {"event": "slot", "name": REQUESTED_SLOT, "value": None}, ], [ SlotSet("num_people", None), SlotSet(REQUESTED_SLOT, None), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ], ), # Validate function deactivates loop ( [ {"event": "slot", "name": "num_people", "value": None}, {"event": "active_loop", "name": None}, ], [ SlotSet("num_people", None), ActiveLoop(None), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ], ), # User rejected manually ( [{"event": "action_execution_rejected", "name": "my form"}], [ActionExecutionRejected("my form")], ), ], ) async def test_validate_slots( validate_return_events: List[Dict], expected_events: List[Event] ): form_name = "my form" slot_name = "num_people" slot_value = "hi" events = [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, slot_name), ActionExecuted(ACTION_LISTEN_NAME), UserUttered(slot_value, entities=[{"entity": "num_tables", "value": 5}]), ] tracker = DialogueStateTracker.from_events(sender_id="bla", evts=events) domain = f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" entities: - num_tables - some_entity - some_other_entity - some_slot slots: {slot_name}: type: any mappings: - type: from_text num_tables: type: any mappings: - type: from_entity entity: num_tables forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} - num_tables actions: - validate_{form_name} """ domain = Domain.from_yaml(textwrap.dedent(domain)) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert slot_events == [SlotSet(slot_name, slot_value), SlotSet("num_tables", 5)] tracker.update_with_events(slot_events, domain) with aioresponses() as mocked: mocked.post(ACTION_SERVER_URL, payload={"events": validate_return_events}) action_server = EndpointConfig(ACTION_SERVER_URL) action = FormAction(form_name, action_server) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == expected_events async def test_request_correct_slots_after_unhappy_path_with_custom_required_slots(): form_name = "some_form" slot_name_1 = "slot_1" slot_name_2 = "slot_2" domain = f""" slots: {slot_name_1}: type: any mappings: - type: from_intent intent: some_intent value: some_value {slot_name_2}: type: any mappings: - type: from_intent intent: some_intent value: some_value forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name_1} - {slot_name_2} actions: - validate_{form_name} """ domain = Domain.from_yaml(domain) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, "slot_2"), ActionExecuted(ACTION_LISTEN_NAME), UserUttered("hello", intent={"name": "greet", "confidence": 1.0}), ActionExecutionRejected(form_name), ActionExecuted("utter_greet"), ], ) # Custom form validation action changes the order of the requested slots validate_return_events = [ {"event": "slot", "name": REQUESTED_SLOT, "value": slot_name_2} ] # The form should ask the same slot again when coming back after unhappy path expected_events = [SlotSet(REQUESTED_SLOT, slot_name_2)] with aioresponses() as mocked: mocked.post(ACTION_SERVER_URL, payload={"events": validate_return_events}) action_server = EndpointConfig(ACTION_SERVER_URL) action = FormAction(form_name, action_server) events = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == expected_events @pytest.mark.parametrize( "custom_events", [ # Custom action returned no events [], # Custom action returned events but no `SlotSet` events [BotUttered("some text").as_dict()], # Custom action returned only `SlotSet` event for `required_slot` [SlotSet(REQUESTED_SLOT, "some value").as_dict()], ], ) async def test_no_slots_extracted_with_custom_slot_mappings(custom_events: List[Event]): form_name = "my form" events = [ ActiveLoop(form_name), SlotSet(REQUESTED_SLOT, "num_tables"), ActionExecuted(ACTION_LISTEN_NAME), UserUttered("off topic"), ] tracker = DialogueStateTracker.from_events(sender_id="bla", evts=events) domain = f""" slots: num_tables: type: any mappings: - type: from_entity entity: num_tables forms: {form_name}: {REQUIRED_SLOTS_KEY}: - num_tables actions: - validate_{form_name} """ domain = Domain.from_yaml(domain) with aioresponses() as mocked: mocked.post(ACTION_SERVER_URL, payload={"events": custom_events}) action_server = EndpointConfig(ACTION_SERVER_URL) action = FormAction(form_name, action_server) with pytest.raises(ActionExecutionRejection): await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) async def test_validate_slots_on_activation_with_other_action_after_user_utterance(): form_name = "my form" slot_name = "num_people" slot_value = "hi" events = [ ActionExecuted(ACTION_LISTEN_NAME), UserUttered(slot_value, entities=[{"entity": "num_tables", "value": 5}]), ActionExecuted("action_in_between"), ] tracker = DialogueStateTracker.from_events(sender_id="bla", evts=events) domain = f""" slots: {slot_name}: type: any mappings: - type: from_text forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} actions: - validate_{form_name} """ domain = Domain.from_yaml(domain) expected_slot_value = "✅" with aioresponses() as mocked: mocked.post( ACTION_SERVER_URL, payload={ "events": [ {"event": "slot", "name": slot_name, "value": expected_slot_value} ] }, ) action_server = EndpointConfig(ACTION_SERVER_URL) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) form_action = FormAction(form_name, action_server) mocked.post( ACTION_SERVER_URL, payload={"events": []}, ) events = await form_action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [ ActiveLoop(form_name), SlotSet(slot_name, expected_slot_value), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ] @pytest.mark.parametrize( "utterance_name", ["utter_ask_my_form_num_people", "utter_ask_num_people"] ) def test_name_of_utterance(utterance_name: Text): form_name = "my_form" slot_name = "num_people" domain = f""" slots: {slot_name}: type: any mappings: - type: from_text forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} responses: {utterance_name}: - text: "How many people?" """ domain = Domain.from_yaml(domain) action = FormAction(form_name, None) assert action._name_of_utterance(domain, slot_name) == utterance_name def test_temporary_tracker(): extra_slot = "some_slot" sender_id = "test" domain = Domain.from_yaml( f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" slots: {extra_slot}: type: any mappings: - type: from_text """ ) previous_events = [ActionExecuted(ACTION_LISTEN_NAME)] old_tracker = DialogueStateTracker.from_events( sender_id, previous_events, slots=domain.slots ) new_events = [Restarted()] form_action = FormAction("some name", None) temp_tracker = form_action._temporary_tracker(old_tracker, new_events, domain) assert extra_slot in temp_tracker.slots.keys() assert list(temp_tracker.events) == [ *previous_events, SlotSet(REQUESTED_SLOT), ActionExecuted(form_action.name()), *new_events, ] async def test_extract_requested_slot_default(): """Test default extraction of a slot value from entity with the same name.""" form_name = "some_form" form = FormAction(form_name, None) domain = Domain.from_dict( { "slots": { "some_slot": { "type": "text", "mappings": [ { "type": "from_entity", "entity": "some_slot", "value": "some_value", } ], } }, "forms": {form_name: {REQUIRED_SLOTS_KEY: ["some_slot"]}}, } ) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop("some form"), SlotSet(REQUESTED_SLOT, "some_slot"), UserUttered( "bla", entities=[{"entity": "some_slot", "value": "some_value"}] ), SlotSet("some_slot", "some_value"), ActionExecuted(ACTION_LISTEN_NAME), ], ) slot_values = await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) assert slot_values == [] @pytest.mark.parametrize( "some_other_slot_mapping, some_slot_mapping, entities, " "intent, expected_slot_events", [ ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "role": "some_role", } ], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [ { "entity": "some_entity", "value": "some_value", "role": "some_other_role", } ], "some_intent", [], ), ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "role": "some_role", } ], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [{"entity": "some_entity", "value": "some_value", "role": "some_role"}], "some_intent", [SlotSet("some_other_slot", "some_value")], ), ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "group": "some_group", } ], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [ { "entity": "some_entity", "value": "some_value", "group": "some_other_group", } ], "some_intent", [], ), ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "group": "some_group", } ], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [{"entity": "some_entity", "value": "some_value", "group": "some_group"}], "some_intent", [SlotSet("some_other_slot", "some_value")], ), ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "group": "some_group", "role": "some_role", } ], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [ { "entity": "some_entity", "value": "some_value", "role": "some_role", "group": "some_group", } ], "some_intent", [SlotSet("some_other_slot", "some_value")], ), ( [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [ { "type": "from_entity", "intent": "some_intent", "entity": "some_other_entity", } ], [{"entity": "some_entity", "value": "some_value"}], "some_intent", # other slot should be extracted because slot mapping is unique [SlotSet("some_other_slot", "some_value")], ), ( [ { "type": "from_entity", "intent": "some_intent", "entity": "some_entity", "role": "some_role", } ], [ { "type": "from_entity", "intent": "some_intent", "entity": "some_other_entity", } ], [{"entity": "some_entity", "value": "some_value", "role": "some_role"}], "some_intent", # other slot should be extracted because slot mapping is unique [SlotSet("some_other_slot", "some_value")], ), ( [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [ { "type": "from_entity", "intent": "some_intent", "entity": "some_other_entity", } ], [{"entity": "some_entity", "value": "some_value", "role": "some_role"}], "some_intent", # other slot should not be extracted # because even though slot mapping is unique it doesn't contain the role [], ), ( [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [{"type": "from_entity", "intent": "some_intent", "entity": "some_entity"}], [{"entity": "some_entity", "value": "some_value"}], "some_intent", # other slot should not be extracted because slot mapping is not unique [SlotSet("some_slot", "some_value")], ), ], ) async def test_extract_other_slots_with_entity( some_other_slot_mapping: List[Dict[Text, Any]], some_slot_mapping: List[Dict[Text, Any]], entities: List[Dict[Text, Any]], intent: Text, expected_slot_events: List[SlotSet], ): """Test extraction of other not requested slots values from entities.""" form_name = "some_form" form = FormAction(form_name, None) domain = Domain.from_dict( { "slots": { "some_other_slot": {"type": "any", "mappings": some_other_slot_mapping}, "some_slot": {"type": "any", "mappings": some_slot_mapping}, }, "forms": { form_name: {REQUIRED_SLOTS_KEY: ["some_other_slot", "some_slot"]} }, } ) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "some_slot"), UserUttered( "bla", intent={"name": intent, "confidence": 1.0}, entities=entities ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) if slot_events: slot_values = await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) assert slot_values == expected_slot_events else: with pytest.raises(ActionExecutionRejection): await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) @pytest.mark.parametrize( "domain_dict, expected_action", [ ( { "actions": ["action_ask_my_form_sun", "action_ask_sun"], "responses": {"utter_ask_my_form_sun": [{"text": "ask"}]}, }, "action_ask_my_form_sun", ), ( { "actions": ["action_ask_sun"], "responses": {"utter_ask_my_form_sun": [{"text": "ask"}]}, }, "utter_ask_my_form_sun", ), ( { "actions": ["action_ask_sun"], "responses": {"utter_ask_sun": [{"text": "hi"}]}, }, "action_ask_sun", ), ( { "actions": ["action_ask_my_form_sun"], "responses": {"utter_ask_my_form_sun": [{"text": "hi"}]}, }, "action_ask_my_form_sun", ), ], ) async def test_ask_for_slot( domain_dict: Dict, expected_action: Text, monkeypatch: MonkeyPatch, default_nlg: TemplatedNaturalLanguageGenerator, ): slot_name = "sun" action_from_name = Mock(return_value=action.ActionListen()) endpoint_config = Mock() monkeypatch.setattr( action, action.action_for_name_or_text.__name__, action_from_name ) form = FormAction("my_form", endpoint_config) domain = Domain.from_dict(domain_dict) await form._ask_for_slot( domain, default_nlg, CollectingOutputChannel(), slot_name, DialogueStateTracker.from_events("dasd", []), ) action_from_name.assert_called_once_with(expected_action, domain, endpoint_config) async def test_ask_for_slot_if_not_utter_ask( monkeypatch: MonkeyPatch, default_nlg: TemplatedNaturalLanguageGenerator ): action_from_name = Mock(return_value=action.ActionListen()) endpoint_config = Mock() monkeypatch.setattr( action, action.action_for_name_or_text.__name__, action_from_name ) form = FormAction("my_form", endpoint_config) events = await form._ask_for_slot( Domain.empty(), default_nlg, CollectingOutputChannel(), "some slot", DialogueStateTracker.from_events("dasd", []), ) assert not events action_from_name.assert_not_called() @pytest.mark.parametrize( "ignored_intents, slot_not_intent", [ # for entity_type -> from_entity ( # `ignored_intents` as a string and slot's not_intent as an empty list. "greet", [], ), ( # `ignored_intents` as an empty list and slot's not_intent has a value. [], ["greet"], ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value different than the ones in `ignored_intents`. ["chitchat", "greet"], ["inform"], ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value that is included also in `ignored_intents`. ["chitchat", "greet"], ["chitchat"], ), ], ) async def test_ignored_intents_with_slot_type_from_entity( ignored_intents: Union[Text, List[Text]], slot_not_intent: Union[Text, List[Text]] ): form_name = "some_form" entity_name = "some_slot" form = FormAction(form_name, None) domain = Domain.from_dict( { "slots": { entity_name: { "type": "any", "mappings": [ { "type": "from_entity", "entity": entity_name, "not_intent": slot_not_intent, } ], } }, "forms": { form_name: { IGNORED_INTENTS: ignored_intents, REQUIRED_SLOTS_KEY: [entity_name], } }, } ) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "some_slot"), UserUttered( "hello", intent={"name": "greet", "confidence": 1.0}, entities=[{"entity": entity_name, "value": "some_value"}], ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) assert slot_events == [] with pytest.raises(ActionExecutionRejection): await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) @pytest.mark.parametrize( "ignored_intents, slot_not_intent", [ # same examples for entity_type -> from_text ( # `ignored_intents` as a string and slot's not_intent as an empty list. "greet", [], ), ( # `ignored_intents` as an empty list and slot's not_intent has a value. [], ["greet"], ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value different than the ones in `ignored_intents`. ["chitchat", "greet"], ["inform"], ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value that is included also in `ignored_intents`. ["chitchat", "greet"], ["chitchat"], ), ], ) async def test_ignored_intents_with_slot_type_from_text( ignored_intents: Union[Text, List[Text]], slot_not_intent: Union[Text, List[Text]] ): form_name = "some_form" entity_name = "some_slot" form = FormAction(form_name, None) domain = Domain.from_dict( { "slots": { entity_name: { "type": "any", "mappings": [ { "type": "from_text", "intent": "some_intent", "not_intent": slot_not_intent, } ], } }, "forms": { form_name: { IGNORED_INTENTS: ignored_intents, REQUIRED_SLOTS_KEY: [entity_name], } }, } ) tracker = DialogueStateTracker.from_events( "default", [ SlotSet(REQUESTED_SLOT, "some_slot"), UserUttered( "hello", intent={"name": "greet", "confidence": 1.0}, entities=[{"entity": entity_name, "value": "some_value"}], ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) assert slot_events == [] form_slot_events = await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) assert form_slot_events == [] @pytest.mark.parametrize( "ignored_intents, slot_not_intent, entity_type", [ # same examples for entity_type -> from_intent ( # `ignored_intents` as a string and slot's not_intent as an empty list. "greet", [], "from_intent", ), ( # `ignored_intents` as an empty list and slot's not_intent has a value. [], ["greet"], "from_intent", ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value different than the ones in `ignored_intents`. ["chitchat", "greet"], ["inform"], "from_intent", ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value that is included also in `ignored_intents`. ["chitchat", "greet"], ["chitchat"], "from_intent", ), # same examples for entity_type -> from_trigger_intent ( # `ignored_intents` as a string and slot's not_intent as an empty list. "greet", [], "from_trigger_intent", ), ( # `ignored_intents` as an empty list and slot's not_intent has a value. [], ["greet"], "from_trigger_intent", ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value different than the ones in `ignored_intents`. ["chitchat", "greet"], ["inform"], "from_trigger_intent", ), ( # `ignored_intents` as a list of 2 values and slot's not_intent has one # value that is included also in `ignored_intents`. ["chitchat", "greet"], ["chitchat"], "from_trigger_intent", ), ], ) async def test_ignored_intents_with_other_type_of_slots( ignored_intents: Union[Text, List[Text]], slot_not_intent: Union[Text, List[Text]], entity_type: Text, ): form_name = "some_form" entity_name = "some_slot" form = FormAction(form_name, None) domain = Domain.from_dict( { "slots": { entity_name: { "type": "any", "mappings": [ { "type": entity_type, "value": "affirm", "intent": "true", "not_intent": slot_not_intent, } ], } }, "forms": { form_name: { IGNORED_INTENTS: ignored_intents, REQUIRED_SLOTS_KEY: [entity_name], } }, } ) tracker = DialogueStateTracker.from_events( "default", [ SlotSet(REQUESTED_SLOT, "some_slot"), UserUttered( "hello", intent={"name": "greet", "confidence": 1.0}, entities=[{"entity": entity_name, "value": "some_value"}], ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) assert slot_events == [] form_slot_events = await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) assert form_slot_events == [] async def test_extract_other_slots_with_matched_mapping_conditions(): form_name = "some_form" form = FormAction(form_name, None) domain = Domain.from_yaml( textwrap.dedent( f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" intent: - greet - inform entities: - email - name slots: name: type: text influence_conversation: false mappings: - type: from_entity entity: name conditions: - active_loop: some_form requested_slot: email email: type: text influence_conversation: false mappings: - type: from_entity entity: email forms: some_form: required_slots: - email - name """ ) ) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "email"), UserUttered( "My name is Emily.", intent={"name": "inform", "confidence": 1.0}, entities=[{"entity": "name", "value": "Emily"}], ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert slot_events == [SlotSet("name", "Emily")] tracker.update_with_events(slot_events, domain) form_slot_events = await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) assert form_slot_events == [] async def test_extract_other_slots_raises_no_matched_conditions(): form_name = "some_form" form = FormAction(form_name, None) domain = Domain.from_yaml( textwrap.dedent( f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" intent: - greet - inform entities: - email - name slots: name: type: text influence_conversation: false mappings: - type: from_entity entity: name conditions: - active_loop: some_form requested_slot: name email: type: text influence_conversation: false mappings: - type: from_entity entity: email forms: some_form: required_slots: - email - name """ ) ) tracker = DialogueStateTracker.from_events( "default", [ ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "email"), UserUttered( "My name is Emily.", intent={"name": "inform", "confidence": 1.0}, entities=[{"entity": "name", "value": "Emily"}], ), ActionExecuted(ACTION_LISTEN_NAME), ], ) action_extract_slots = ActionExtractSlots(action_endpoint=None) slot_events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) tracker.update_with_events(slot_events, domain) assert slot_events == [] with pytest.raises(ActionExecutionRejection): await form.validate( tracker, domain, CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), ) async def test_action_extract_slots_custom_mapping_with_condition(): domain_yaml = textwrap.dedent( f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" slots: custom_slot: type: text influence_conversation: false mappings: - type: custom conditions: - active_loop: some_form forms: some_form: required_slots: - custom_slot actions: - validate_some_form """ ) domain = Domain.from_yaml(domain_yaml) events = [ActiveLoop("some_form"), UserUttered("Hi")] tracker = DialogueStateTracker.from_events( sender_id="test_id", evts=events, slots=domain.slots ) with aioresponses() as mocked: mocked.post( ACTION_SERVER_URL, payload={ "events": [{"event": "slot", "name": "custom_slot", "value": "test"}] }, ) action_server = EndpointConfig(ACTION_SERVER_URL) action_extract_slots = ActionExtractSlots(action_server) events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [] form = FormAction("some_form", action_server) form_events = await form.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert form_events == [ SlotSet("custom_slot", "test"), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ] async def test_form_slots_empty_with_restart(): domain = Domain.from_yaml( textwrap.dedent( f""" version: "{LATEST_TRAINING_DATA_FORMAT_VERSION}" intent: - greet - inform entities: - name slots: name: type: text influence_conversation: false mappings: - type: from_entity entity: name conditions: - active_loop: some_form forms: some_form: required_slots: - name """ ) ) tracker = DialogueStateTracker.from_events( "default", [ UserUttered("hi", intent={"name": "greet", "confidence": 1.0}), ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "name"), UserUttered( "My name is Emily.", intent={"name": "inform", "confidence": 1.0}, entities=[{"entity": "name", "value": "Emily"}], ), SlotSet("name", "emily"), Restarted(), ActiveLoop("some_form"), SlotSet(REQUESTED_SLOT, "name"), UserUttered("hi", intent={"name": "greet", "confidence": 1.0}), ActionExecuted(ACTION_LISTEN_NAME), ], ) action = FormAction("some_form", None) # FormAction execution is rejected because a slot was requested but none # were extracted (events before restart are not considered). with pytest.raises(ActionExecutionRejection): await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) async def test_extract_slots_with_mapping_conditions_during_form_activation(): slot_name = "city" entity_value = "London" entity_name = "location" form_name = "test_form" domain = Domain.from_yaml( f""" entities: - {entity_name} slots: {slot_name}: type: text mappings: - type: from_entity entity: {entity_name} conditions: - active_loop: {form_name} forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} """ ) events = [ ActionExecuted("action_listen"), UserUttered( "I live in London", entities=[{"entity": entity_name, "value": entity_value}], ), ] tracker = DialogueStateTracker.from_events( sender_id="test", evts=events, domain=domain, slots=domain.slots ) assert tracker.active_loop_name is None action_extract_slots = ActionExtractSlots(None) events = await action_extract_slots.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert events == [] expected_events = [ ActiveLoop(form_name), SlotSet(slot_name, entity_value), SlotSet(REQUESTED_SLOT, None), ActiveLoop(None), ] form_action = FormAction(form_name, None) form_events = await form_action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert form_events == expected_events async def test_form_validation_happens_once(caplog: LogCaptureFixture): """ Tests if form validation happens once instead of twice. Solves the bug presented in https://rasahq.atlassian.net/browse/ENG-117 """ tracker = DialogueStateTracker.from_events(sender_id="bla", evts=[]) form_name = "my_form" action_server = EndpointConfig(ACTION_SERVER_URL) action = FormAction(form_name, action_server) slot_name = "num_people" domain = textwrap.dedent( f""" slots: {slot_name}: type: float mappings: - type: from_entity entity: number forms: {form_name}: {REQUIRED_SLOTS_KEY}: - {slot_name} responses: utter_ask_num_people: - text: "How many people?" actions: - validate_{form_name} """ ) domain = Domain.from_yaml(domain) form_validation_events = [ { "event": "slot", "timestamp": None, "name": "num_people", "value": 5, } ] with aioresponses() as mocked, caplog.at_level(logging.DEBUG): mocked.post( ACTION_SERVER_URL, payload={"events": form_validation_events}, ) _ = await action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert ( sum( [ 1 for message in caplog.messages if f"Calling action endpoint to run action 'validate_{form_name}'." in message ] ) == 1 ) async def test_form_validation_happens_at_form_activation(caplog: LogCaptureFixture): """Test if form validation happens at form activation. The particular case is when a non-required slot for the form is also filled at form activation. Fixes the bug in https://rasahq.atlassian.net/browse/ATO-1104. """ form = "help_form" action_server = EndpointConfig(ACTION_SERVER_URL) form_action = FormAction(form, action_server) entity_a = "device" entity_b = "account_type" entity_b_value = "bronze" context_slot = "device_type" required_slot = "send_sms" global_slot = "membership" domain = textwrap.dedent( f""" intents: - start_form entities: - {entity_a} - {entity_b} slots: {context_slot}: type: categorical influence_conversation: false initial_value: "mobile" values: - mobile - desktop - other mappings: - type: from_entity entity: {entity_a} {required_slot}: type: float influence_conversation: false mappings: - type: custom {global_slot}: type: text influence_conversation: false mappings: - type: from_entity entity: {entity_b} forms: {form}: {REQUIRED_SLOTS_KEY}: - {required_slot} responses: utter_ask_{required_slot}: - text: "Would you like to receive an SMS?" actions: - validate_{form} """ ) domain = Domain.from_yaml(domain) tracker = DialogueStateTracker.from_events( sender_id="test", evts=[ UserUttered( "Can you help me with my bronze account ", intent={"name": "start_form", "confidence": 1.0}, entities=[{"entity": entity_b, "value": entity_b_value}], ), ], ) # Mock the custom validation action to update form required_slots to empty list form_validation_events = [ { "event": "slot", "timestamp": None, "name": "requested_slot", "value": None, } ] with aioresponses() as mocked, caplog.at_level(logging.DEBUG): mocked.post( ACTION_SERVER_URL, payload={"events": form_validation_events}, ) events = await form_action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert ( sum( [ 1 for message in caplog.messages if f"Calling action endpoint to run action 'validate_{form}'." in message ] ) == 1 ) assert events == [ ActiveLoop(form), SlotSet(REQUESTED_SLOT, None), SlotSet(global_slot, entity_b_value), ActiveLoop(None), ] async def test_form_validation_happens_at_form_activation_with_empty_required_slots( caplog: LogCaptureFixture, ): """Test if required_slots get updated dynamically at form activation. The particular case is when a form has an empty required_slots list. """ form = "booking_form" action_server = EndpointConfig(ACTION_SERVER_URL) form_action = FormAction(form, action_server) dynamic_slot = "customer_name" bot_utterance = "What is your name?" domain = textwrap.dedent( f""" intents: - start_form slots: {dynamic_slot}: type: text influence_conversation: false mappings: - type: custom forms: {form}: {REQUIRED_SLOTS_KEY}: [] responses: utter_ask_{dynamic_slot}: - text: "{bot_utterance}" actions: - validate_{form} """ ) domain = Domain.from_yaml(domain) tracker = DialogueStateTracker.from_events( sender_id="test", evts=[ UserUttered( "I'd like to make a booking. ", intent={"name": "start_form", "confidence": 1.0}, ), ], ) # Mock the custom validation action to update form required_slots form_validation_events = [ { "event": "slot", "timestamp": None, "name": "requested_slot", "value": dynamic_slot, } ] with aioresponses() as mocked, caplog.at_level(logging.DEBUG): mocked.post( ACTION_SERVER_URL, payload={"events": form_validation_events}, ) events = await form_action.run( CollectingOutputChannel(), TemplatedNaturalLanguageGenerator(domain.responses), tracker, domain, ) assert ( sum( [ 1 for message in caplog.messages if f"Calling action endpoint to run action 'validate_{form}'." in message ] ) == 1 ) assert len(events) == 3 assert events[0] == ActiveLoop(form) assert events[1] == SlotSet(REQUESTED_SLOT, dynamic_slot) assert isinstance(events[2], BotUttered) is True assert events[2].text == bot_utterance