Joakim Persson
119 lines
6.1 KiB
Python
119 lines
6.1 KiB
Python
# Start all services
|
|
import subprocess
|
|
import os
|
|
import yaml
|
|
import json
|
|
import socket
|
|
import urllib.parse
|
|
from backend import run_flask
|
|
import logging
|
|
import requests
|
|
import utils
|
|
from utils import GlobalState
|
|
from enums import LogLevel
|
|
|
|
global_state = GlobalState() # Configure root logger. The level will be adjusted later based on config file
|
|
logger = global_state.get_logger(__name__) # Logger for this module, inherit properties of the root logger
|
|
|
|
def configure():
|
|
"""
|
|
Reads YAML configruation file into dictionary, parse it and fill all referenceed
|
|
environment variables with their values.
|
|
"""
|
|
####################################
|
|
# Read YAML config
|
|
####################################
|
|
# Load configuration file that defines parameters for services
|
|
with open('./smartassist/config/smartassist.yaml') as f:
|
|
config = yaml.safe_load(f)
|
|
|
|
def resolve_env_var(value):
|
|
if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
|
|
env_var_name = value[2:-1] # Extract name between ${}
|
|
return os.getenv(env_var_name, None)
|
|
return value
|
|
|
|
def update_value(value):
|
|
if isinstance(value, dict): # Dictionaries need recursive check
|
|
return update_dict_with_env_vars(value)
|
|
elif isinstance(value, list): # Lists must be traversed element by element
|
|
return [update_value(item) for item in value]
|
|
elif isinstance(value, str): # If value is a string it might be an environmnet variable
|
|
return resolve_env_var(value)
|
|
else: # Anything else, just keep the old value
|
|
return value
|
|
|
|
def update_dict_with_env_vars(d): # Check all keys in d
|
|
for key in d: # Iterate over all keys in the dictionary. The keys seen are all at the top-level of d
|
|
# logger.info(f"key investigated now: {key}")
|
|
d[key] = update_value(d[key])
|
|
return d
|
|
|
|
updated_config = update_dict_with_env_vars(config)
|
|
|
|
####################################
|
|
# Extract global logging level
|
|
####################################
|
|
if isinstance(updated_config.get('logging'), dict): # Look for 'logging' key in config file
|
|
logging_config = updated_config['logging']
|
|
if isinstance(logging_config.get('level'), str): # Set to value of the yaml file if specified
|
|
# global_state.set_log_level(logging_config['level'])
|
|
global_state.set_log_level(LogLevel(logging_config['level']))
|
|
logger.info("configure(): This logger now has effective log level %s", logger.getEffectiveLevel())
|
|
|
|
####################################
|
|
# Extract models (server url, api_key, model, et cetera)
|
|
####################################
|
|
if isinstance(updated_config.get('backend'),dict): # Extract backend info from dictionary
|
|
global_state.set_backend(backend=updated_config.get('backend'))
|
|
logger.debug("backend = \n{}".format(json.dumps(global_state.get_backend(), indent=4)))
|
|
logger.debug(f"Backend API endpoint is set to: {global_state.get_backend_api_ep()}")
|
|
|
|
preferred_ep = updated_config.get('preferred_ep', None) # Get the preferred endpoint if specified, otherwise None
|
|
|
|
if isinstance(updated_config.get('endpoints'), list): # Extract info on endpoint, model, url, provider et cetera from list
|
|
global_state.set_endpoints(endpoints=updated_config.get('endpoints')) # Extract and set list of endpoints
|
|
# logger.debug("endpoints = \n{}".format(json.dumps(global_state.get_endpoints(), indent=4)))
|
|
global_state.fetch_models()
|
|
endpoints = global_state.get_endpoints()
|
|
for endpoint in endpoints: # Set default LLM for each endpoint
|
|
available_llms = global_state.get_list_of_available_llms(endpoint=endpoint)
|
|
llm = next(iter(available_llms),None) # First available LLM or None. Default for AUTODETECT and requests for non-existing LLMs
|
|
logger.debug(f"url {endpoint['url']} = {available_llms}")
|
|
if endpoint["model"] in available_llms: # Check if specific LLM requested, AUTODETECT evaluates to False
|
|
llm = endpoint["model"]
|
|
endpoint["default_llm"] = llm
|
|
|
|
if preferred_ep: # If preferred_ep is specified, set it as the default endpoint
|
|
list_of_eps = global_state.get_endpoints_with_key_values("title", preferred_ep) # Should only be one element in the list...
|
|
default_endpoint = next(iter(list_of_eps),None) # Same as default_endpoint = list_of_eps[0] if list_of_eps else None
|
|
else:
|
|
default_endpoint = next(iter(endpoints),None) # Set default_endpoint to first endpoint from list of all endpoints
|
|
default_llm = default_endpoint["default_llm"] # Get default LLM for default_endpoint
|
|
default_ulr = default_endpoint["url"] # Get ulr of default_endpoint
|
|
global_state.set_host_url(default_ulr) # Set initial host to the first item in endpoints (or None)
|
|
global_state.set_llm(default_llm) # Set which llm to use
|
|
logger.debug(f"Desired default endpoint: {default_ulr},\tDesired default LLM: {default_llm}")
|
|
logger.debug(f"Returned default endpoint: {global_state.get_host_url()},\tReturned default LLM: {global_state.get_llm()}")
|
|
|
|
return updated_config
|
|
|
|
|
|
def start_backend(config):
|
|
parsed_url = urllib.parse.urlparse(config['backend']['url'])
|
|
# hostname = parsed_url.netloc.split(':')[0] # Split by ':' and take the first part, i.e., 'localhost', IP, or domain name
|
|
port = parsed_url.port # This is the server port
|
|
logger.debug('Backend parsed url set to {}'.format(parsed_url))
|
|
logger.debug('Backend port set to {}'.format(port))
|
|
|
|
try:
|
|
run_flask(fport = port)
|
|
except Exception as e:
|
|
logger.error("Failed to start backend: %s", str(e)) # Corresponds to print(f"Failed to start backend: {e}")
|
|
|
|
if __name__ == '__main__':
|
|
conf = configure() # Read config from file and set up config dict
|
|
# logger.debug('conf dictionary set to \n{}'.format(json.dumps(conf, indent=4)))
|
|
# start_frontend(config=conf) # Not needed as we are using Flask for backend now
|
|
start_backend(config=conf)
|