dream

1
import logging
2
from importlib import import_module
3
from typing import Dict
4

5
from .core.db import DataBase
6
from .core.state_manager import StateManager, Dialog
7
from .core.workflow_manager import WorkflowManager
8
from .state_formatters.output_formatters import http_api_output_formatter, http_debug_output_formatter
9

10

11
# Default parameters
12
BASE_PARAMETERS = {
13
    "debug": True,
14
    "state_manager_class": StateManager,
15
    "workflow_manager_class": WorkflowManager,
16
    "db_class": DataBase,
17
    "pipeline_config": "pipeline_conf.json",
18
    "db_config": "db_conf.json",
19
    "overwrite_last_chance": None,
20
    "overwrite_timeout": None,
21
    "formatters_module": None,
22
    "connectors_module": None,
23
    "response_logger": True,
24
    "time_limit": 0,
25
    "output_formatter": http_api_output_formatter,
26
    "debug_output_formatter": http_debug_output_formatter,
27
    "port": 4242,
28
    "cors": False,
29
    "telegram_token": "",
30
    "telegram_proxy": "",
31
}
32

33

34
# Replasing constants with ones from user settings
35
def setup_parameter(name, user_settings):
36
    res = None
37
    if user_settings:
38
        res = getattr(user_settings, name, None)
39
    if res is None:
40
        res = BASE_PARAMETERS[name]
41
    return res
42

43

44
user_settings = None
45
try:
46
    user_settings = import_module("dp_agent_settings")
47
except ModuleNotFoundError:
48
    logging.info("settings.py was not found. Default settings are used")
49

50
# Set up common parameters
51
DEBUG = setup_parameter("debug", user_settings)
52

53

54
class ExtendedStateManager(StateManager):
55
    async def update_attributes(self, dialog, payload, label: str, **kwargs):
56
        if isinstance(payload.get("human_attributes"), dict):
57
            await self.update_human(dialog.human, payload)
58
        if isinstance(payload.get("bot_attributes"), dict):
59
            await self.update_bot(dialog.bot, payload)
60

61
    async def add_annotation_and_reset_human_attributes_for_first_turn(
62
        self, dialog: Dialog, payload: Dict, label: str, **kwargs
63
    ):
64
        dialog.utterances[-1].annotations[label] = payload
65
        if len(dialog.utterances) == 1:
66
            dialog.human.attributes = {"disliked_skills": dialog.human.attributes.get("disliked_skills", [])}
67

68

69
# Basic agent configuration parameters (some are currently unavailable)
70
STATE_MANAGER_CLASS = ExtendedStateManager
71
WORKFLOW_MANAGER_CLASS = WorkflowManager
72
DB_CLASS = DataBase
73

74
PIPELINE_CONFIG = setup_parameter("pipeline_config", user_settings)
75
DB_CONFIG = setup_parameter("db_config", user_settings)
76

77
OVERWRITE_LAST_CHANCE = setup_parameter("overwrite_last_chance", user_settings)
78
OVERWRITE_TIMEOUT = setup_parameter("overwrite_timeout", user_settings)
79

80
RESPONSE_LOGGER = setup_parameter("response_logger", user_settings)
81

82
# HTTP app configuraion parameters
83
TIME_LIMIT = setup_parameter("time_limit", user_settings)  # Without engaging the timeout by default
84
CORS = setup_parameter("cors", user_settings)
85

86
OUTPUT_FORMATTER = setup_parameter("output_formatter", user_settings)
87
DEBUG_OUTPUT_FORMATTER = setup_parameter("debug_output_formatter", user_settings)
88

89
# HTTP api run parameters
90
PORT = setup_parameter("port", user_settings)
91

92
# Telegram client configuration parameters
93
TELEGRAM_TOKEN = setup_parameter("telegram_token", user_settings)
94
TELEGRAM_PROXY = setup_parameter("telegram_proxy", user_settings)
95