dream

1
import logging
2
from copy import deepcopy
3

4
import common.dff.integration.context as int_ctx
5
import scenario.response_funcs as response_funcs
6
from common.robot import command_intents
7
from common.utils import get_intents
8
from df_engine.core import Actor, Context
9

10

11
logger = logging.getLogger(__name__)
12

13

14
def command_selector_response(ctx: Context, actor: Actor, *args, **kwargs) -> str:
15
    annotated_utterance = int_ctx.get_last_human_utterance(ctx, actor)
16
    intention, confidence = get_detected_intents(annotated_utterance)
17
    logger.info(f"Detected intents: {intention}")
18

19
    response, conf, human_attr, bot_attr, attr = "", 0.0, {}, {}, {}
20
    if intention is not None and confidence > 0 and intention in response_funcs.get_respond_funcs():
21
        logger.debug(f"Intent is defined as {intention}")
22
        dialog = int_ctx.get_dialog(ctx, actor)
23
        dialog["seen"] = dialog["called_intents"][intention]
24
        funcs = response_funcs.get_respond_funcs()[intention]
25
        response = funcs(ctx, actor)
26
        if not isinstance(response, str):
27
            conf = deepcopy(response[1])
28
            human_attr = deepcopy(response[2])
29
            bot_attr = deepcopy(response[3])
30
            attr = deepcopy(response[4])
31
            response = deepcopy(response[0])
32
        # Special formatter which used in AWS Lambda to identify what was the intent
33
        while "#+#" in response:
34
            response = response[: response.rfind(" #+#")]
35
        logger.info(f"Response: {response}; intent_name: {intention}")
36
        try:
37
            response += " #+#{}".format(intention)
38
        except TypeError:
39
            logger.error(f"TypeError intent_name: {intention} response: {response};")
40
            response = "Hmmm... #+#{}".format(intention)
41
        # todo: we need to know what intent was called
42
        # current workaround is to use only one intent if several were detected
43
        # and to append special token with intent_name
44
    else:
45
        logger.debug("Intent is not defined")
46

47
    if response == "":
48
        intents = get_intents(annotated_utterance, probs=True, which="intent_catcher")
49
        logger.error(f"response is empty for intents: {intents}")
50
    elif conf == 0.0:
51
        return response
52
    return [[response, conf, human_attr, bot_attr, attr]]
53

54

55
def default_response(ctx: Context, actor: Actor, *args, **kwargs) -> str:
56
    annotated_utterance = int_ctx.get_last_human_utterance(ctx, actor)
57

58
    intents = get_intents(annotated_utterance, probs=True, which="intent_catcher")
59
    logger.error(f"response is empty for intents: {intents}")
60
    return ""
61

62

63
def set_confidence_from_input(ctx: Context, actor: Actor, *args, **kwargs) -> Context:
64
    intent, confidence = get_detected_intents(int_ctx.get_last_human_utterance(ctx, actor))
65
    if intent in command_intents:
66
        int_ctx.set_confidence(ctx, actor, 1.0)
67
    else:
68
        int_ctx.set_confidence(ctx, actor, confidence)
69
    return ctx
70

71

72
def get_detected_intents(annotated_utterance):
73
    intents = get_intents(annotated_utterance, probs=True, which="intent_catcher")
74
    intent, confidence = None, 0.0
75
    for intent_name, intent_conf in intents.items():
76
        if intent_conf > 0 and intent_name in response_funcs.get_respond_funcs():
77
            confidence_current = intent_conf
78
            if confidence_current > confidence:
79
                intent, confidence = intent_name, float(confidence_current)
80

81
    return intent, confidence
82