dream

1
#!/usr/bin/env python
2

3
import json
4

5
from respond_funcs import get_respond_funcs
6

7
INTENT_RESPONSES_PATH = "./data/intent_response_phrases.json"
8

9

10
class Responder:
11
    def __init__(self, logger):
12
        """
13

14
        The responder for each of the detected intents.
15
        If the are more than one detected intent, we select the one with the maximum confidence.
16
        The confidence of the response is equal to the confidence of detection.
17

18
        """
19
        self.logger = logger
20
        self.intent_responses = self.load_responses(INTENT_RESPONSES_PATH)
21
        self.response_funcs = get_respond_funcs()
22
        self.seen_intents = set()  # Whether the intent have already been detected
23

24
    def respond(self, dialog: dict):
25
        response = ""
26
        confidence = 0.0
27

28
        utt = dialog["utterances"][-1]
29
        for intent_name, intent_data in utt["annotations"].get("intent_catcher", {}).items():
30
            print("intent name: " + intent_name, flush=True)
31
            if intent_data["detected"] and intent_data["confidence"] > confidence:
32
                if intent_name in self.response_funcs:
33
                    dialog["seen"] = dialog["called_intents"][intent_name]
34
                    response = self.response_funcs[intent_name](dialog, self.intent_responses[intent_name])
35
                    # Special formatter which used in AWS Lambda to identify what was the intent
36
                    while "#+#" in response:
37
                        response = response[: response.rfind(" #+#")]
38
                    self.logger.info(f"Response: {response}; intent_name: {intent_name}")
39
                    try:
40
                        response += " #+#{}".format(intent_name)
41
                    except TypeError:
42
                        self.logger.error(f"TypeError intent_name: {intent_name} response: {response};")
43
                        response = "Hmmm... #+#{}".format(intent_name)
44
                    confidence = intent_data["confidence"]
45
                    self.seen_intents.add(intent_name)
46
                    # todo: we need to know what intent was called
47
                    # current workaround is to use only one intent if several were detected
48
                    # and to append special token with intent_name
49
                else:
50
                    # skip
51
                    # self.logger.error(f'responder for intent_name: {intent_name} not found')
52
                    continue
53
        if response == "":
54
            self.logger.error(f"response is empty for intents: {utt['annotations'].get('intent_catcher', {}).items()}")
55
        return response, confidence
56

57
    def load_responses(self, intent_responses_filename: str):
58
        with open(intent_responses_filename, "r") as fp:
59
            return json.load(fp)
60