Utförligare beskrivningar (docstrings) och lagt till "type hinting". Nya metod get_endpoints_with_key(). Döpt om getLogger till get_logger

smartassist/config/smartassist.yaml

@@ -6,7 +6,7 @@ backend:
 
 endpoints: 
   - model: "AUTODETECT"
-    title: "Ollama"
+    title: "Ollama-local"
     url: "http://localhost:11434"
     provider: "ollama"
   # - model: "AUTODETECT"  
@@ -18,7 +18,7 @@ endpoints:
         Authorization: "${OLLAMA_API_KEY}" # on MacOS: echo "Authorization: Basic $(echo -n 'user:password' | gbase64 -w 0)"
     provider: "ollama"
 
-# Ollama Server Configuration
+# Ollama Server Configuration – this section is to be removed when endpoints are being parsed correctly
 ollama:
   title: "Ollama-local"
   # url: "http://localhost:11434"

smartassist/src/backend.py

@@ -12,7 +12,7 @@ from utils import GlobalState
 
 # Create a logger for this module
 global_state = GlobalState() # Import the singleton that holds global states (e.g., logger)
-logger = global_state.getLogger(__name__)  # Logger for this module, inherit properties of the root logger
+logger = global_state.get_logger(__name__)  # Logger for this module, inherit properties of the root logger
 
 
 # Find out the path to current directory according to the Python interpreter (venv)
@@ -45,13 +45,15 @@ def index():
 
     session['chat_history'] = []   # The session object (actually, a dictonary) holds the chat session
     logger.debug("Entering route '/'")
-    # api_endpoint = os.environ['BE_API_ENDPOINT']  # Retrieve the environment variable 
     api_endpoint = global_state.get_backend_api_ep()  # Retrieve the environment variable 
-    logger.debug("Backend API endpoint: %s", api_endpoint)
+    host_url = global_state.get_host_url()
     use_model = global_state.get_llm()
+    logger.debug("Backend API endpoint:\t%s", api_endpoint)
+    logger.debug("Host of LLMs:\t\t%s", host_url)
+    logger.debug("LLM to use:\t\t\t%s", use_model)
     with open('smartassist/src/html/client.html', 'r') as f:
         client_html = f.read()
-    logger.debug("Client HTML (first few characters): %s", client_html[:50]) # Print to see if it's loading
+    # logger.debug("Client HTML (first few characters): %s", client_html[:50]) # Print to see if it's loading
     # logger.debug("Client HTML (all characters): %s", client_html) # Print to see if it's loading
 
     return render_template('index.html', api_endpoint=api_endpoint, use_model = use_model, client_content=client_html)
@@ -93,7 +95,6 @@ CORS(app, resources={
     
 @app.route('/api/tags', methods=['GET'])
 def tag(url = "http://localhost:11434/api/tags", headers = None):
-# def tag(url = "http://localhost:11434/api/tags", headers = {"Content-Type": "application/json"}):
     """Get a list of models for the server located at url."""
     try:
         logger.debug(f"url: {url} headers: {headers}")
@@ -106,8 +107,8 @@ def tag(url = "http://localhost:11434/api/tags", headers = None):
 
 
 @app.route('/api/chat', methods=['POST'])
-def chat(model = "phi3:mini"):
+def chat():
-# def chat(url_server = "http://localhost:11434/api/generate", model = "phi3:mini"):
+# def chat(model = "phi3:mini"):
     """
     This function handles the chat. The frontend client (web browser) calls the 
     backend server through this endpoint (/api/chat) that manage queries
@@ -117,8 +118,9 @@ def chat(model = "phi3:mini"):
     # Get the message from the JSON in the request body
     data = request.get_json()
     message = data.get('query')
-    url_server = data.get('url_server', "https://ollama-test.wara-ops.org/api/generate")  # Use provided URL or default
+    url_server = data.get('url_server', global_state.get_host_url())  # Use provided URL or current if not provided
-    model = data.get('model', model)  # Use provided model or default if not provided
+    # url_server = data.get('url_server', "https://ollama-test.wara-ops.org/api/generate")  # Use provided URL or default
+    model = data.get('model', global_state.get_llm())  # Use provided model or current if not provided
     
     # Get chat history from session storage (e.g., a dictionary)
     chat_history = session.get('chat_history', [])
@@ -135,13 +137,30 @@ def chat(model = "phi3:mini"):
         'prompt': '\n'.join([f"{item['role']}: {item['message']}" for item in chat_history]),
         "stream": False           
     }
+    url = url_server
+
+    # TODO: This section should only run when changing to new endpoint...
+    # begin refactor ###################################
+    headers = { # Set default header
+        "Content-Type": "application/json",
+    }
+    found_endpoint = False
+    endpoints = global_state.get_endpoints()
+    for endpoint in endpoints:
+        if endpoint["url"] == url: # Look for endpoint with this url
+            found_endpoint = True
+            if endpoint["provider"] == "ollama": # Currently only supporting ollama servers
+                if "requestOptions" in endpoint:  # Check if authentication is needed
+                    headers.update({
+                        "Authorization": endpoint["requestOptions"]["headers"]["Authorization"]
+                    })
+    if found_endpoint == False:
+        # Raise some error or whatever...
+        logger.debug(f"Host {url} not found")
+    # end refactor ###################################
 
     try:
-        url = url_server
+        url = url + "/api/generate"
-        headers = {
-                    "Content-Type": "application/json",
-                    "Authorization": "Basic ZWNzanBlcjoxM2JjMTU4ZDhmNmY5YTU4YTkzZDNmY2I="
-        }
         logger.debug(f"url: {url} headers: {headers}")
         response = requests.post(url,
                                  headers=headers, 

smartassist/src/enums.py

@@ -0,0 +1,10 @@
+# Some useful constants to be used in the code
+
+from enum import Enum
+
+class LogLevel(Enum):
+    DEBUG = 'DEBUG'
+    INFO = 'INFO'
+    WARNING = 'WARNING'
+    ERROR = 'ERROR'
+    CRITICAL = 'CRITICAL'

smartassist/src/startservices.py

@@ -5,14 +5,14 @@ import yaml
 import json
 import socket
 import urllib.parse
-from backend import run_flask, tag
+from backend import run_flask
 import logging
 import requests
 import utils
-from utils import GlobalState, fetch_models_from_endpoints
+from utils import GlobalState
 
 global_state = GlobalState() # Configure root logger. The level will be adjusted later based on config file
-logger = global_state.getLogger(__name__)  # Logger for this module, inherit properties of the root logger
+logger = global_state.get_logger(__name__)  # Logger for this module, inherit properties of the root logger
 
 def configure():
     """
@@ -44,7 +44,7 @@ def configure():
 
     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}")
+            # logger.info(f"key investigated now: {key}")
             d[key] = update_value(d[key])
         return d
     
@@ -70,19 +70,23 @@ def configure():
     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)))
-        fetch_models_from_endpoints(global_state.get_endpoints(), global_state)  # Call the new function
+        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
+            if endpoint["model"] in available_llms: # Check if specific LLM requested, AUTODETECT evaluates to False
                 llm = endpoint["model"] 
             endpoint["default_llm"] = llm
  
- 
+        default_endpoint = next(iter(endpoints),None) # Set default_endpoint to first endpoint from list of endpoints
-        global_state.set_host_url(next(iter(endpoints),None)) # Set initial host to the first item in endpoints (or None)
+        default_llm = default_endpoint["default_llm"] # Get default LLM for default_endpoint
-        global_state.set_llm(llm) # Set which server and llm to use
+        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            
 

smartassist/src/utils.py

@@ -4,45 +4,8 @@
 import logging
 import json
 import requests
-#from backend import GlobalState  # Assuming GlobalState is defined there
+from typing import Optional
-
+from enums import LogLevel
-def fetch_models_from_endpoints(endpoints, global_state):
-    """
-    Fetch models from endpoints and update the endpoint dictionaries.
-
-    Args:
-        endpoints (list): List of endpoint dictionaries.
-        global_state: The global state object.
-
-    Returns:
-        None
-    """
-    logger = logging.getLogger(__name__)
-
-    for endpoint in endpoints:
-        if endpoint["provider"] == "ollama":
-            headers = {
-                "Content-Type": "application/json",
-            }
-            if "requestOptions" in endpoint:  # Check if authentication is needed
-                headers.update({
-                    "Authorization": endpoint["requestOptions"]["headers"]["Authorization"]
-                })
-
-            try:
-                models_response = requests.get(endpoint["url"] + "/api/tags", headers=headers)
-                models_response.raise_for_status()  # Raise an exception for HTTP errors
-                models = models_response.json()
-            except requests.exceptions.RequestException as e:
-                logger.error("Error fetching models from backend: %s", str(e))
-                continue
-
-            if isinstance(models, dict) and 'error' in models:
-                logger.error('Error fetching models from backend: %s', models['error'])
-            else:
-                endpoint["models"] = models.get("models", [])  # Get the list of models directly
-                # logger.debug("models = \n{}".format(json.dumps(models, indent=4)))
-
 
 class GlobalState:
     """
@@ -54,7 +17,12 @@ class GlobalState:
     """
     _instance = None  # Private class attribute to hold the single instance of the class
 
-    def __new__(cls):
+    def __new__(cls) -> 'GlobalState':
+        """
+        Create a new instance of the GlobalState class.
+        
+        This is a singleton implementation, so only one instance will be created.
+        """
         if cls._instance is None:
             cls._instance = super(GlobalState, cls).__new__(cls)
             cls._instance.log_level = 'INFO'  # Default logging level
@@ -75,8 +43,27 @@ class GlobalState:
 
         return cls._instance
 
-    def configure_logging(self, level=None):
+    # def configure_logging(self, level: Optional[LogLevel] = None) -> None:
-        """Set up logging for the project."""
+    #     """
+    #     Configure the logging system for this project.
+        
+    #     Args:
+    #         level (LogLevel): The log level to use. If None, uses the default log level set in `self.log_level`.
+        
+    #     Notes:
+    #         This method sets up logging for the project and logs a message at the debug level indicating the effective log level.
+    #    """
+    def configure_logging(self, level: Optional[str] = None) -> None:
+
+        """
+        Configure the logging system for this project.
+
+        Args:
+            level (str): The log level to use. Can be one of the standard Python log levels (e.g., 'DEBUG', 'INFO', 'WARNING', etc.). If None, uses the default log level set in `self.log_level`.
+
+        Notes:
+            This method sets up logging for the project and logs a message at the debug level indicating the effective log level.
+        """
         if level is None:
             level = self.log_level
         # numeric_level = getattr(logging, level.upper())  # Convert string to numeric level
@@ -84,68 +71,220 @@ class GlobalState:
         self.logger.setLevel(numeric_level)
         self.logger.debug(f"utils.py -- configure_logging(): effective log level is {level} which is {self.logger.getEffectiveLevel()}")
 
-    def set_log_level(self, level = 'INFO'):
+    # def set_log_level(self, level: LogLevel) -> None:
-        """Set the logging level."""
+        #  """
+        # Set the log level for this project.
+
+        # Args:
+        #     level (LogLevel): The new log level to use. Can be one of the evels defined in enum.py (e.g., DEBUG, INFO, WARNING, CRITICAL etc.).
+
+        # Notes:
+        #     This method updates the `self.log_level` attribute and calls `configure_logging()` to apply the change.
+        # """    
+    def set_log_level(self, level: str = 'INFO') -> None:
+        """
+        Set the log level for this project.
+
+        Args:
+            level (str): The new log level to use. Can be one of the standard Python log levels (e.g., 'DEBUG', 'INFO', 'WARNING', etc.).
+
+        Notes:
+            This method updates the `self.log_level` attribute and calls `configure_logging()` to apply the change.
+        """
         self.log_level = level
         self.configure_logging()
 
-    def get_log_level(self):
+    def get_log_level(self) -> str:
-        """Getter for log_level attribute."""
+        """
+        Get the current log level.
+
+        Returns:
+            str: The current log level (e.g., 'DEBUG', 'INFO', 'WARNING', etc.).
+        """
         return self.log_level
     
-    def get_effective_log_level(self):
+    def get_effective_log_level(self) -> int:
-        """Getter for effective log level of loggerattribute."""
+        """
+        Get the effective log level of the logger.
+
+        Returns:
+            int: The numeric value of the effective log level.
+        """
         return self.logger.getEffectiveLevel()
     
-    def getLogger(self, module_name = None):
+    def get_logger(self, module_name: Optional[str] = None) -> logging.Logger:
-        """Return a logger based on the module name."""
+
+        """
+        Get a logger instance based on the module name.
+
+        Args:
+            module_name (str): The name of the module to get a logger for. If None, uses the current module name (`__name__`).
+
+        Returns:
+            Logger: A logger instance configured for the specified module.
+        """
         if module_name is None:
             module_name = __name__
         logger = logging.getLogger(module_name)
         return logger
     
-    def set_host_url(self, url="http://localhost:11434"):
+    def set_host_url(self, url: str = "http://localhost:11434") -> None:
-        """Set the host url to which LLM requests are sent"""
+        """
-        self.host = url
+        Set the URL of the host to which LLM requests are sent.
 
-    def get_host_url(self):
+        Args:
-        """Get the url for the currently used host for LLMs"""
+            url (str): The new URL to use. Defaults to 'http://localhost:11434' if not specified.
+        """
+        self.host_url = url
+
+    def get_host_url(self) -> str:
+        """
+        Get the current URL of the host used for LLMs.
+
+        Returns:
+            str: The current URL of the host.
+        """
         return self.host_url
     
-    def set_llm(self, model_name="phi3:mini"):
+    def set_llm(self, model_name: str = "phi3:mini") -> None:
-        """Set LLM for queries"""
+        """
+        Set the LLM to use for queries.
+
+        Args:
+            model_name (str): The name of the LLM to use. Defaults to 'phi3:mini' if not specified.
+        """
         self.llm = model_name
 
-    def get_llm(self):
+    def get_llm(self) -> str:
-        """Getter for which LLM is used for queries"""
+        """
+        Get the current LLM used for queries.
+
+        Returns:
+            str: The name of the current LLM.
+        """
         return self.llm
     
-    def set_backend(self, backend=None):
+    def set_backend(self, backend: Optional[dict] = None) -> None:
-        """Set backend server that web clients connect to"""
+
+        """
+        Set the backend server that web clients connect to.
+
+        Args:
+            backend (dict): A dictionary containing information about the backend server. If None, resets the backend server to its default value.
+        """
         self.backend = backend
 
-    def get_backend(self):
+    def get_backend(self) -> dict:
-        """Getter for backend server that web clients connect to"""
+        """
+        Get the current backend server used by web clients.
+
+        Returns:
+            dict: A dictionary containing information about the current backend server.
+        """
         return self.backend
     
-    def get_backend_api_ep(self):
+    def get_backend_api_ep(self) -> str:
-        """Getter for backend API endpoint"""
+        """
+        Get the API endpoint of the backend server.
+
+        Returns:
+            str: The URL of the API endpoint.
+        """
         return self.backend["url"]+self.backend["api"]
 
-    def set_endpoints(self, endpoints=None):
+    def set_endpoints(self, endpoints: Optional[list[dict]] = None) -> None:
-        """Set the list of endpoints."""
+        """
+        Set the list of endpoints used by this object.
+
+        Args:
+            endpoints (list): A list of endpoint dictionaries. Each dictionary should contain information about an endpoint.
+                If None, resets the endpoints to their default value.
+
+        Raises:
+            ValueError: If endpoints is not a list.
+
+        Notes:
+            Endpoints can be reset to their default value by passing None as the argument.
+        """
         if endpoints is not None:
             if not isinstance(endpoints, list):
                 raise ValueError("Endpoints must be a list, even if there is just one model")
         self.endpoints = endpoints
    
-    def get_endpoints(self):
+    def get_endpoints(self) -> list[dict]:
-        """Return the list of endpoints"""
+        """
+        Get the complete list of endpoints.
+        
+        Returns:
+            List of endpoints
+        """
         return self.endpoints
     
-    def get_list_of_available_llms(self, endpoint=None):
+    def get_endpoints_with_key(self, key: str) -> list[dict]:
-        """Return a sorted list of LLMs available at endpoint"""
+        """
+        Returns a list of endpoint dictionaries that contain the specified key.
+
+        Args:
+            key (str): The key to search for in the endpoint dictionaries.
+
+        Returns:
+            List[Dict]: A list of endpoint dictionaries containing the specified key.
+        """
+        return [ep for ep in self.endpoints if key in ep]
+
+    
+    def fetch_models(self) -> None:
+        """
+        Fetch models from endpoints and update the endpoint dictionaries.
+        Returns:
+            None
+        """
+        logger = logging.getLogger(__name__)
+
+        for endpoint in self.endpoints:
+            try:
+                if endpoint["provider"] == "ollama":
+                    headers = {
+                        "Content-Type": "application/json",
+                    }
+                    if "requestOptions" in endpoint:  # Check if authentication is needed
+                        headers.update({
+                            "Authorization": endpoint["requestOptions"]["headers"]["Authorization"]
+                        })
+
+                    models_response = requests.get(endpoint["url"] + "/api/tags", headers=headers)
+                    models_response.raise_for_status()  # Raise an exception for HTTP errors
+
+                    try:
+                        models = models_response.json()
+                    except json.JSONDecodeError as e:
+                        logger.error(f"Failed to parse JSON response: {e}")
+                        continue
+
+                    if isinstance(models, dict) and 'error' in models: # Unclear if requests to any API actually add this in the response
+                        logger.error('Error fetching models from backend: %s', models['error'])
+                    else:
+                        endpoint["models"] = models.get("models", [])  # Get the list of models directly
+            except requests.exceptions.RequestException as e:
+                logger.error(f"Request error: {e}")
+            except Exception as e:
+                logger.error(f"Unexpected error: {e}")
+
+        return # No value returned
+
+    
+    def get_list_of_available_llms(self, endpoint: Optional[dict] = None) -> Optional[list[str]]:
+        """
+        Returns a sorted list of Large Language Models (LLMs) available at the specified endpoint.
+
+        Args:
+            endpoint (dict): Optional endpoint dictionary to retrieve LLMs from. If not provided, will use internal endpoint configuration.
+
+        Returns:
+            list: A sorted list of LLM names (strings). Returns None if no LLMs are found or endpoint is invalid.
+        """        
         llm_list = None
         if isinstance(endpoint["models"], list):
             llm_list = sorted([list_item['name'] for list_item in endpoint["models"]], key=str.lower)
         return llm_list
+