Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
) -> EndpointConfig:
import questionary
default_event_broker_endpoint = EndpointConfig(
type="sql", dialect="sqlite", db=DEFAULT_EVENTS_DB
)
if not event_broker_endpoint:
return default_event_broker_endpoint
elif not _is_correct_event_broker(event_broker_endpoint):
cli_utils.print_error(
"Rasa X currently only supports a SQLite event broker with path '{}' "
"when running locally. You can deploy Rasa X with Docker "
"(https://rasa.com/docs/rasa-x/deploy/) if you want to use "
"other event broker configurations.".format(DEFAULT_EVENTS_DB)
)
continue_with_default_event_broker = questionary.confirm(
"Do you want to continue with the default SQLite event broker?"
).ask()
if not continue_with_default_event_broker:
exit(0)
return default_event_broker_endpoint
else:
return event_broker_endpoint
endpoints.model = EndpointConfig(
"{}/projects/default/models/tags/production".format(rasa_x_url),
token=rasa_x_token,
wait_time_between_pulls=2,
)
overwrite_existing_event_broker = False
if endpoints.event_broker and not _is_correct_event_broker(endpoints.event_broker):
cli_utils.print_error(
"Rasa X currently only supports a SQLite event broker with path '{}' "
"when running locally. You can deploy Rasa X with Docker "
"(https://rasa.com/docs/rasa-x/deploy/) if you want to use "
"other event broker configurations.".format(DEFAULT_EVENTS_DB)
)
overwrite_existing_event_broker = questionary.confirm(
"Do you want to continue with the default SQLite event broker?"
).ask()
if not overwrite_existing_event_broker:
exit(0)
if not endpoints.tracker_store or overwrite_existing_event_broker:
endpoints.event_broker = EndpointConfig(type="sql", db=DEFAULT_EVENTS_DB)
def confirm_version(version: Version) -> bool:
"""Allow the user to confirm the version number."""
if str(version) in git_existing_tags():
confirmed = questionary.confirm(
f"Tag with version '{version}' already exists, overwrite?", default=False
).ask()
else:
confirmed = questionary.confirm(
f"Is the next version '{version}' correct "
f"(current version is '{get_current_version()}')?",
default=True,
).ask()
if confirmed:
return True
else:
print("Aborting.")
sys.exit(1)
def _ask_overwrite(path: Text) -> None:
import questionary
overwrite = questionary.confirm(
"Directory '{}' is not empty. Continue?".format(os.path.abspath(path))
).ask()
if not overwrite:
print_cancel()
def print_train_or_instructions(args: argparse.Namespace, path: Text) -> None:
import questionary
print_success("Finished creating project structure.")
should_train = (
questionary.confirm("Do you want to train an initial model? 💪🏽")
.skip_if(args.no_prompt, default=True)
.ask()
)
if should_train:
print_success("Training an initial model...")
config = os.path.join(path, DEFAULT_CONFIG_PATH)
training_files = os.path.join(path, DEFAULT_DATA_PATH)
domain = os.path.join(path, DEFAULT_DOMAIN_PATH)
output = os.path.join(path, create_output_path())
args.model = rasa.train(domain, config, training_files, output)
print_run_or_instructions(args, path)
else:
def _validate_action(action_name: Text,
policy: Text,
confidence: float,
predictions: List[Dict[Text, Any]],
endpoint: EndpointConfig,
sender_id: Text,
finetune: bool = False
) -> bool:
"""Query the user to validate if an action prediction is correct.
Returns `True` if the prediction is correct, `False` otherwise."""
question = questionary.confirm(
"The bot wants to run '{}', correct?".format(action_name))
is_correct = _ask_or_abort(question, sender_id, endpoint)
if not is_correct:
action_name, is_new_action = _request_action_from_user(
predictions, sender_id, endpoint)
else:
is_new_action = False
tracker = retrieve_tracker(endpoint, sender_id,
EventVerbosity.AFTER_RESTART)
if _form_is_rejected(action_name, tracker):
# notify the tracker that form was rejected
send_event(endpoint, sender_id,
requested_slot = tracker.get("slots", {}).get(REQUESTED_SLOT)
validation_questions = questionary.confirm(
"Should '{}' validate user input to fill "
"the slot '{}'?".format(action_name, requested_slot))
validate_input = _ask_or_abort(validation_questions, sender_id, endpoint)
if not validate_input:
# notify form action to skip validation
send_event(endpoint, sender_id,
{"event": "form_validation", "validate": False})
elif not tracker.get('active_form', {}).get('validate'):
# handle contradiction with learned behaviour
warning_question = questionary.confirm(
"ERROR: FormPolicy predicted no form validation "
"based on previous training stories. "
"Make sure to remove contradictory stories "
"from training data. "
"Otherwise predicting no form validation "
"will not work as expected.")
_ask_or_abort(warning_question, sender_id, endpoint)
# notify form action to validate an input
send_event(endpoint, sender_id,
{"event": "form_validation", "validate": True})
action = questionary.select(
'What do you want to do next?',
choices=[
"Find papers by keywords",
"Find papers by ID",
questionary.Separator(),
f"Continue review ({n_included} included, "
f"{n_excluded} excluded)",
"Export",
questionary.Separator(),
"Stop"
]
).ask()
if action is None or action.startswith("Stop"):
stop = questionary.confirm(
"Are you sure you want to stop?",
default=False
).ask()
if stop:
raise KeyboardInterrupt
elif action.endswith("by keywords"):
self._papers_from_finder(state)
elif action.endswith("by ID"):
self._papers_from_id(state)
elif action.startswith("Export"):
self._export()
elif action.startswith("Continue review"):
try:
self._do_review(state, *args, **kwargs)
except KeyboardInterrupt:
pass
"type": "text",
"name": "task_dir",
"message": "Task directory path:",
"default": str(dirs["data"])
}
])
config["databases"] = q.prompt([
{
"type": "text",
"name": "default",
"message": "Default database path:"
}
])
i = 1
while q.confirm("Do you want to add another database?").ask():
q2 = q.prompt([
{
"type": "text",
"name": "label",
"message": "Enter the label for the database:",
"default": f"database_{i}"
},
{
"type": "text",
"name": "path",
"message": "The path of the database file:",
"default": str(
Path(config.get("databases").get("default")).parent)
}])
config["databases"][q2.get("label")] = q2.get("path")
i += 1
def print_train_or_instructions(args: argparse.Namespace, path: Text) -> None:
import questionary
print_success("Finished creating project structure.")
should_train = questionary.confirm(
"Do you want to train an initial model? 💪🏽"
).skip_if(args.no_prompt, default=True)
if should_train:
print_success("Training an initial model...")
config = os.path.join(path, DEFAULT_CONFIG_PATH)
training_files = os.path.join(path, DEFAULT_DATA_PATH)
domain = os.path.join(path, DEFAULT_DOMAIN_PATH)
output = os.path.join(path, create_output_path())
args.model = rasa.train(domain, config, training_files, output)
print_run_or_instructions(args, path)
else:
print_success(