import os import logging from abc import ABC import requests from langchain_community.llms import LlamaCpp from llm.utils.config import config from llm.utils.lc_interface import LCInterface logger = logging.getLogger(__name__) logger.setLevel(logging.ERROR) # because if something went wrong in execution application can't be work anymore file_handler = logging.FileHandler( "logs/chelsea_llm_llamacpp.log") # for all modules template for logs file is "logs/chelsea_{module_name}_{dir_name}.log" logger.setLevel(logging.INFO) # informed formatted = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") file_handler.setFormatter(formatted) logger.addHandler(file_handler) work_dir = os.getcwd() models_dir = os.path.join(work_dir, "llm/models") class LC_TinyLlama(LCInterface, ABC): def __init__(self): self.model_config = config["LC_TinyLlama-1.1B-Chat-v1.0-GGUF"] try: get_file = requests.get(self.model_config["model_url"]) if get_file.status_code == 200: path_to_model = os.path.join(models_dir, self.model_config["model_name"]) with open(path_to_model, "wb") as f: f.write(get_file.content) logger.info("Model file successfully recorded") f.close() except FileExistsError: print(f"Model file {path_to_model} already exists. Skipping download.") logger.info(f"Model file {path_to_model} already exists. Skipping download.") except OSError as e: print(f"Error while writing a file to directory : {e}") logger.error(msg="Error while write a file to directory", exc_info=e) self.llm = LlamaCpp( model_path=os.path.join(models_dir, self.model_config["model_name"]), temperature=self.model_config["temperature"], max_tokens=self.model_config["max_tokens"], top_p=self.model_config["top_p"], top_k=self.model_config["top_k"], # callback_manager=callback_manager, verbose=True, # Verbose is required to pass to the callback manager ) def execution(self): try: return self.llm except Exception as e: print(f"Execution filed in LC_TinyLlama execution function: {e}") logger.critical(msg="Execution filed in LC_TinyLlama execution function", exc_info=e) return None def clear_llm(self, unused_model_dict, current_lc): # If unused_model_dict is not empty if len(unused_model_dict) > 1 or unused_model_dict is not None: # go through key and value for key, value in unused_model_dict.items(): # check if path is existing and key is not current using model if os.path.exists(value) and key != current_lc: # delete files from models directory except of current_lc os.remove(value) logger.info(f"Successfully deleted file {value}") print(f"Successfully deleted file {value}") else: logger.info(f"Unfortunately dictionary empty or None") print(f"Unfortunately dictionary {unused_model_dict} empty or None") def get_unused(self, current_lc): if len(os.listdir(models_dir)) > 1: file_names = [os.path.basename(md) for md in os.listdir(models_dir)] for item in file_names: if item != current_lc: unused_model_file = os.path.join(models_dir, item) return {item: unused_model_file} else: return None def model_name(self): return self.model_config["model_name"] def __str__(self): return f"{self.__class__.__name__}_{self.model_name()}" def __repr__(self): llm_info = f"llm={self.llm}" if hasattr(self, 'llm') else 'llm=not initialized' return f"{self.__class__.__name__}({llm_info})" class LC_Phi3(LCInterface, ABC): def __init__(self): self.model_config = config["LC_Phi-3-mini-4k-instruct-gguf"] try: get_file = requests.get(self.model_config["model_url"]) if get_file.status_code == 200: path_to_model = os.path.join(models_dir, self.model_config["model_name"]) with open(path_to_model, "wb") as f: f.write(get_file.content) logger.info("Model file successfully recorded") print("Model file successfully recorded") f.close() except FileExistsError: print(f"Model file {path_to_model} already exists. Skipping download.") logger.info(f"Model file {path_to_model} already exists. Skipping download.") except OSError as e: print(f"Error while writing a file to directory : {e}") logger.error(msg="Error while write a file to directory", exc_info=e) self.llm = LlamaCpp( model_path=os.path.join(models_dir, self.model_config["model_name"]), temperature=self.model_config["temperature"], max_tokens=self.model_config["max_tokens"], top_p=self.model_config["top_p"], top_k=self.model_config["top_k"], # callback_manager=callback_manager, verbose=True, # Verbose is required to pass to the callback manager ) def execution(self): try: return self.llm except Exception as e: print(f"Execution filed in LC_Phi3 execution function: {e}") logger.critical(msg="Execution filed in LC_Phi3 execution function:", exc_info=e) return None def clear_llm(self, unused_model_dict, current_lc): # If unused_model_dict is not empty if len(unused_model_dict) > 1 or unused_model_dict is not None: # go through key and value for key, value in unused_model_dict.items(): # check if path is existing and key is not current using model if os.path.exists(value) and key != current_lc: # delete files from models directory except of current_lc os.remove(value) logger.info(f"Successfully deleted file {value}") print(f"Successfully deleted file {value}") else: logger.info(f"Unfortunately dictionary empty or None") print(f"Unfortunately dictionary {unused_model_dict} empty or None") def get_unused(self, current_lc): if len(os.listdir(models_dir)) > 1: file_names = [os.path.basename(md) for md in os.listdir(models_dir)] for item in file_names: if item != current_lc: unused_model_file = os.path.join(models_dir, item) return {item: unused_model_file} else: return None def model_name(self): return self.model_config["model_name"] def __str__(self): return f"{self.__class__.__name__}_{self.model_name()}" def __repr__(self): llm_info = f"llm={self.llm}" if hasattr(self, 'llm') else 'llm=not initialized' return f"{self.__class__.__name__}({llm_info})"