dream

1
import json
2
import logging
3
import requests
4
from copy import deepcopy
5
from os import getenv
6

7

8
logger = logging.getLogger(__name__)
9

10
with open("common/prompts/goals_for_prompts.json", "r") as f:
11
    META_GOALS_PROMPT = json.load(f)["prompt"]
12

13
with open("common/prompts/goals_for_prompts_ru.json", "r") as f:
14
    META_GOALS_PROMPT_RU = json.load(f)["prompt"]
15

16

17
def send_request_to_prompted_generative_service(dialog_context, prompt, url, config, timeout, sending_variables):
18
    response = requests.post(
19
        url,
20
        json={
21
            "dialog_contexts": [dialog_context],
22
            "prompts": [prompt],
23
            "configs": [config],
24
            **sending_variables,
25
        },
26
        timeout=timeout,
27
    )
28
    hypotheses = response.json()[0]
29
    return hypotheses
30

31

32
def get_goals_from_prompt(prompt, url, generative_timeout, sending_variables):
33
    new_url = "/".join(url.split("/")[:-1])
34
    try:
35
        goals_description = requests.post(
36
            f"{new_url}/generate_goals",
37
            json={
38
                "prompts": [prompt],
39
                **sending_variables,
40
            },
41
            timeout=generative_timeout,
42
        ).json()[0]
43
    except Exception as e:
44
        logger.info(f"Exception in `/generate_goals` endpoint:\n{e}")
45
        goals_description = prompt
46
    return goals_description
47

48

49
def if_none_var_values(sending_variables):
50
    if len(sending_variables) > 0 and all(
51
        [var_value[0] is None or var_value[0] == "" for var_value in sending_variables.values()]
52
    ):
53
        return True
54
    return False
55

56

57
def compose_sending_variables(lm_service_kwargs, envvars_to_send, **kwargs):
58
    if len(envvars_to_send):
59
        # get variables which names are in `envvars_to_send` (splitted by comma if many)
60
        # from the last human utterance's attributes
61
        sending_variables = {f"{var.lower()}s": [kwargs.get(var.lower(), None)] for var in envvars_to_send}
62
        if if_none_var_values(sending_variables):
63
            # get variables which names are in `envvars_to_send` (splitted by comma if many)
64
            # from env variables
65
            sending_variables = {f"{var.lower()}s": [getenv(var, None)] for var in envvars_to_send}
66
            if if_none_var_values(sending_variables):
67
                logger.info(f"Did not get {envvars_to_send}'s values. Sending without them.")
68
            else:
69
                logger.info(f"Got {envvars_to_send}'s values from environment.")
70
        else:
71
            logger.info(f"Got {envvars_to_send}'s values from attributes.")
72
    else:
73
        sending_variables = {}
74

75
    # adding kwargs to request from the last human utterance's attributes
76
    for _key, _value in lm_service_kwargs.items():
77
        logger.info(f"Got/Re-writing {_key}s values from kwargs.")
78
        sending_variables[f"{_key}s"] = [deepcopy(_value)]
79
    return sending_variables
80