v0.0.1_DigItBMTH

app.py

@@ -1,38 +1,38 @@
-# version - ArticMonkey:19.03.24:1743
+# version - ArcticMonkey:19.03.24:1743
 
 import psutil
-import sys
-
-import streamlit as st 
+# components from other authors
+from streamlit_mic_recorder import mic_recorder
 
-# my modules 
+# core modules
 from audio_processing.A2T import A2T
 from audio_processing.T2A import T2A
-from llm.llm import LLM_chain
+from command.utils import build_chain
+from llm.llm_factory import LLM_Factory
 
-# libraries from other authors 
-from streamlit_mic_recorder import mic_recorder
+llm_model = LLM_Factory()
+
+trigger = {"lc": "small"}
+
+
+def prepare_cor(input_text: str):
+    return build_chain.build_command_chain().handle_command(input_text)
 
-llmchain = LLM_chain()
 
 def main():
-    mic = mic_recorder(start_prompt="Record",stop_prompt="Stop", just_once=True)
+    mic = mic_recorder(start_prompt="Record", stop_prompt="Stop", just_once=True)
 
     if mic is not None:
         a2t = A2T(mic["bytes"])
         text = a2t.predict()
-        response = llmchain(entity=text, id=1)
-    
+        # prepare_cor(input_text=text)
+        llm = llm_model.create_llm(prompt_entity=text, prompt_id=1, trigger=trigger)
+        response = llm.execution() if llm is not None else "Oops occurred some error. Please try again"
+
         t2a = T2A(response)
         t2a.autoplay()
 
-        print(sys.getsizeof(response))
-        
-        del response
-        del a2t
-        
-    print('RAM memory % used after:', psutil.virtual_memory()[2])  
 
 if __name__ == "__main__":
-    print('RAM memory % used:', psutil.virtual_memory()[2]) # ~ 94 GB full memory
-    main()
+    print('RAM memory % used:', psutil.virtual_memory()[2])  # ~ 94 GB full memory
+    main()

audio_processing/A2T.py

@@ -2,6 +2,8 @@ import numpy as np
 import librosa
 import io
 
+from typing import Optional
+
 from .config import pipe
 
 TASK = "transcribe"
@@ -11,30 +13,31 @@ class A2T:
     def __init__(self, mic):
         self.mic = mic
 
-    def __generate_text(self, inputs, task: str = None):
+    def __generate_text(self, inputs, task: Optional[str] = None) -> str:
         if inputs is  None:
             raise Exception("Inputs is None")
 
         transcribed_text = pipe(inputs, batch_size=BATCH_SIZE, generate_kwargs={"task": task}, return_timestamps=True)["text"]
         return transcribed_text
 
-    def __preprocces(self, raw: bytes):
+    @staticmethod
+    def __preprocess(raw: bytes) -> np.ndarray:
         print(f"Raw type : {type(raw)}")
         chunk = io.BytesIO(raw)
         audio, sample_rate = librosa.load(chunk, sr=16000)
         print(f"Sample rate : {sample_rate}")
         return audio
  
-    def predict(self):
+    def predict(self) -> str:
         try:
             if self.mic is not None:
                 raw = self.mic
-                audio = self.__preprocces(raw=raw)
+                audio = self.__preprocess(raw=raw)
                 print(f"audio type : {type(audio)} \n shape : {audio.shape} \n audio max value : {np.max(audio)}")
             else:
                 raise Exception("please provide audio")
 
-            if isinstance(audio , np.ndarray):
+            if isinstance(audio, np.ndarray):
                 return self.__generate_text(inputs=audio, task=TASK)
             else:
                 raise Exception("Audio is not np array")

audio_processing/T2A.py

@@ -1,22 +1,23 @@
 import logging
 
-import torch
+from io import BytesIO
+from typing import Optional
+
 import librosa
 import soundfile as sf
+from streamlit_TTS import auto_play, text_to_audio
 
-from io import BytesIO
 from .config import pipe_tts
 
-from streamlit_TTS import auto_play, text_to_audio
+SAMPLING_RATE = 16_000
 
-SAMPLING_RATE = 16000
 
 class T2A:
-    def __init__(self, input_text: str = None):
+    def __init__(self, input_text: Optional[str] = None):
         self.text = input_text
         self.output_model = pipe_tts(input_text)
 
-    def __get_duration(self, raw: bytes):
+    def __get_duration(self, raw: bytes) -> float:
         chunk = BytesIO(raw)
         audio, sample_rate = librosa.load(chunk, sr=SAMPLING_RATE)
         duration = librosa.get_duration(y=audio, sr=sample_rate)
@@ -27,14 +28,14 @@ class T2A:
             if isinstance(self.text, str):
                 audio = text_to_audio(self.text, language=lang)
                 auto_play(audio)
-            else:
+            else: # more checking
                 text = f"Text you provide is {type(self.text)} accepted only string type"
-                audio = text_to_audio(text, language=language)
+                audio = text_to_audio(text, language=lang)
                 auto_play(audio)
         else:
             raise Exception("Text is None")
 
-    def get_audio(self):
+    def get_audio(self) -> tuple[bytes, int, float]:
         try:
             synth = self.output_model["audio"][0]
 
@@ -42,7 +43,7 @@ class T2A:
 
             with BytesIO() as buffer:
                 sf.write(buffer, synth, SAMPLING_RATE, format='wav')
-                output = buffer.getvalue() # bytes
+                output = buffer.getvalue()  # bytes
 
             print(f"type : {type(output)}")
 
@@ -52,4 +53,4 @@ class T2A:
 
             return output, SAMPLING_RATE, duration
         except Exception as e:
-            logging.error(e)
+            logging.error(e)

audio_processing/config.py

@@ -1,9 +1,10 @@
+# ArcticMonkey:19.03.24:1700 example of version name in plaintext  will be converted into hex using this site ->
+# https://magictool.ai/tool/text-to-hex-converter/ Here ArcticMonkey is name of version and rest of all is data and time
+
 import torch
 
 from transformers import pipeline
 
-# ArticMonkey:19.03.24:1700 example of version name in plaintext  will be convert into hex using this site -> https://magictool.ai/tool/text-to-hex-converter/
-# Here ArticMonkey is name of version and rest of all is data and time
 
 device = 0 if torch.cuda.is_available() else "cpu"
 

command/basic/documentation.py

@@ -1,15 +1,15 @@
-import os 
+import os
 import logging
 
-import streamlit as st 
-import core.form_documentation as fd 
+import streamlit as st
+from ..utils.form_documentation import generate_doc
 
 from typing import Optional
 from ..command_handler import CommandHandler
 
 
 class DocumentationHandler(CommandHandler):
-   def __init__(self, commands, successor: Optional["CommandHandler"] = None):
+    def __init__(self, commands, successor: Optional["CommandHandler"] = None):
         super().__init__(successor)
         self.commands = commands
 
@@ -20,13 +20,12 @@ class DocumentationHandler(CommandHandler):
             super().handle_command(command)
 
     def execute_command(self):
-        current_dir = os.path.dirname(os.path.realpath(__file__)) 
-        path_to_file = fd.generate_doc(path=current_dir)
+        current_dir = os.path.dirname(os.path.realpath(__file__))
+        path_to_file = generate_doc(path=current_dir)
 
         if path_to_file is not None:
-             with st.sidebar:
-                 st.download_button('Download doc', path_to_file)
-                 st.success("Done", icon="✅")
+            with st.sidebar:
+                st.download_button('Download doc', path_to_file)
+                st.success("Done", icon="✅")
         else:
             logging.error("path_to_file is None")
-        

command/command_handler.py

@@ -1,6 +1,7 @@
 from typing import Optional
 from abc import ABC, abstractmethod
 
+
 class CommandHandler(ABC):
     def __init__(self, successor: Optional["CommandHandler"] = None):
         self.successor = successor
@@ -11,4 +12,4 @@ class CommandHandler(ABC):
 
     @abstractmethod
     def execute_command(self):
-        """Method of processing command execution logic"""
+        """Method of processing command execution logic"""

command/utils/build_chain.py

@@ -0,0 +1,19 @@
+from ..basic.documentation import DocumentationHandler
+from ..utils.read_yaml import load_commands_from_yaml
+
+
+def build_command_chain():
+    commands = load_commands_from_yaml(file_path="../commands.yaml")
+    handlers = None
+    for cmd_data in commands:
+        command_dict = cmd_data.get('command_dict', {})
+        commands_list = [cmd.lower() for cmds in command_dict.values() for cmd in cmds]
+        description = cmd_data.get('description', '')
+
+        if description.lower() == "download text file with all commands":
+            handler = DocumentationHandler(commands_list, handlers)
+        else:
+            handler = None
+
+        handlers = handler
+    return handlers

command/{core → utils}/form_documentation.py

@@ -19,23 +19,22 @@ The list of commands and their use.
 Command documentation: the user downloads a text file that describes the commands in detail. 
 The information on how to use the very first and main command will be announced during the first use. 
 The following list of voice commands in English is available for calling the command: "Download commands", "download commands", "load commands", "boot commands", "install commands", "commands", "load", "boot", "download", "install". 
-In Ukrainian: "Завантажити команди", "Скачати команди", "Загрузити команди", "Переглянути команди", "Завантажити", "Скачати", "Загрузити", "Переглянути", "команди".
+In Ukrainian: "Завантаж команди", "Скачай команди", "Загрузи команди", "Завантаж", "Скачай", "Загрузи", "команди".
 """
 
-def generate_doc(path: Optional[str] = None) -> Optional[str]:  
-    
+
+def generate_doc(path: Optional[str] = None) -> Optional[str]:
     if path is not None:
-        file = os.path.join(path, NAME) 
+        file = os.path.join(path, NAME)
     else:
-        current_dir = os.path.dirname(os.path.realpath(__file__)) 
+        current_dir = os.path.dirname(os.path.realpath(__file__))
         file = os.path.join(current_dir, NAME)
 
     logging.info(file)
-    
+
     try:
         with open(file, 'w') as f:
             f.write(TEMPLATE)
         return file
     except IOError as e:
         logging.error(e)
-    

command/utils/read_yaml.py

@@ -0,0 +1,12 @@
+import yaml
+import logging
+from typing import Optional, Any
+
+
+def load_commands_from_yaml(file_path: Optional[str] = None) -> Any:
+    try:
+        with open(file_path, 'r') as file:
+            commands_data = yaml.safe_load(file)
+            return commands_data.get('commands', [])
+    except IOError as e:
+        logging.error(e)

llm/config.py

@@ -1 +1,34 @@
-config = {"model": "mistralai/Mixtral-8x7B-Instruct-v0.1", "temperature": 0.1, "max_new_tokens": 1024, "top_k": 5, "load_in_8bit": True}
+config = {
+    "HF_Mistrail": {
+        "model": "mistralai/Mixtral-8x7B-Instruct-v0.1",
+        "temperature": 0.1,
+        "max_new_tokens": 1024,
+        "top_k": 5,
+        "load_in_8bit": True
+    },
+    "HF_TinyLlama": {
+        "model": "TinyLlama/TinyLlama-1.1B-Chat-v1.0",
+        "temperature": 0.1,
+        "max_new_tokens": 1024,
+        "top_k": 5,
+        "top_p":0.95,
+        "load_in_8bit": True,
+        "do_sample": True
+    },
+    "LC_TinyLlama-1.1B-Chat-v1.0-GGUF": {
+        "model_url": "https://huggingface.co/TheBloke/TinyLlama-1.1B-Chat-v1.0-GGUF/resolve/main/tinyllama-1.1b-chat-v1.0.Q8_0.gguf",
+        "model_name": "tinyllama-1.1b-chat-v1.0.Q8_0.gguf.bin",
+        "temperature": 0.4,
+        "max_tokens": 868,
+        "top_p": 0.8,
+        "top_k": 5,
+    },
+    "LC_Phi-3-mini-4k-instruct-gguf": {
+        "model_url": "https://huggingface.co/microsoft/Phi-3-mini-4k-instruct-gguf/resolve/main/Phi-3-mini-4k-instruct-q4.gguf",
+        "model_name": "Phi-3-mini-4k-instruct-gguf.bin",
+        "temperature": 0.4,
+        "max_tokens": 868,
+        "top_p": 0.8,
+        "top_k": 5,
+    }
+}

llm/huggingfacehub/hf_model.py

@@ -0,0 +1,113 @@
+import logging
+import os
+import yaml
+
+from abc import ABC
+
+from llm.llm_interface import LLMInterface
+from llm.config import config
+
+from langchain.prompts import PromptTemplate
+from langchain.chains import LLMChain
+from langchain.llms import HuggingFaceEndpoint
+
+logger = logging.getLogger(__name__)
+
+logger.setLevel(logging.CRITICAL)  # because if something went wrong in execution application can't be work anymore
+
+file_handler = logging.FileHandler(
+    "logs/chelsea_llm_huggingfacehub.log")  # for all modules here 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)
+
+logger.info("Getting information from hf_model module")
+
+
+class HF_Mistaril(LLMInterface, ABC):
+    def __init__(self, prompt_entity: str, prompt_id: int = 0):
+        self.prompt_entity = prompt_entity
+        self.prompt_id = prompt_id
+
+        self.model_config = config["HF_Mistrail"]
+
+        self.llm = HuggingFaceEndpoint(
+            repo_id=self.model_config["model"],
+            model_kwargs={"temperature": self.model_config["temperature"],
+                          "max_new_tokens": self.model_config["max_new_tokens"],
+                          "top_k": self.model_config["top_k"], "load_in_8bit": self.model_config["load_in_8bit"]})
+
+    @staticmethod
+    def __read_yaml():
+        try:
+            yaml_file = os.path.join("../", 'prompts.yaml')
+            with open(yaml_file, 'r') as file:
+                data = yaml.safe_load(file)
+            return data
+        except Exception as e:
+            logger.error(msg="Execution filed", exc_info=e)
+
+    def execution(self):
+        try:
+            data = self.__read_yaml()
+            prompts = data["prompts"][
+                self.prompt_id]  #get second prompt from yaml, need change id parameter to get other prompt
+            template = prompts["prompt_template"]
+            prompt = PromptTemplate(template=template, input_variables=["entity"])
+            llm_chain = LLMChain(prompt=prompt, llm=self.llm, verbose=True)
+            output = llm_chain.invoke(self.prompt_entity)
+            return output["text"]
+        except Exception as e:
+            logger.critical(msg="Execution filed", exc_info=e)
+
+    def __str__(self):
+        return f"prompt_entity={self.prompt_entity}, prompt_id={self.prompt_id}"
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(prompt_entity: {type(self.prompt_entity)} = {self.prompt_entity}, prompt_id: {type(self.prompt_id)} = {self.prompt_id})"
+
+
+class HF_TinyLlama(LLMInterface, ABC):
+    def __init__(self, prompt_entity: str, prompt_id: int = 0):
+        self.prompt_entity = prompt_entity
+        self.prompt_id = prompt_id
+
+        self.model_config = config["HF_TinyLlama"]
+
+        self.llm = HuggingFaceEndpoint(
+            repo_id=self.model_config["model"],
+            model_kwargs={"temperature": self.model_config["temperature"],
+                          "max_new_tokens": self.model_config["max_new_tokens"],
+                          "top_k": self.model_config["top_k"], "load_in_8bit": self.model_config["load_in_8bit"]})
+
+    @staticmethod
+    def __read_yaml():
+        try:
+            yaml_file = os.path.join("../", 'prompts.yaml')
+            with open(yaml_file, 'r') as file:
+                data = yaml.safe_load(file)
+            return data
+        except Exception as e:
+            logger.error(msg="Execution filed", exc_info=e)
+
+    def execution(self):
+        try:
+            data = self.__read_yaml()
+            prompts = data["prompts"][
+                self.prompt_id]  #get second prompt from yaml, need change id parameter to get other prompt
+            template = prompts["prompt_template"]
+            prompt = PromptTemplate(template=template, input_variables=["entity"])
+            llm_chain = LLMChain(prompt=prompt, llm=self.llm, verbose=True)
+            output = llm_chain.invoke(self.prompt_entity)
+            return output["text"]
+        except Exception as e:
+            logger.critical(msg="Execution filed", exc_info=e)
+
+    def __str__(self):
+        return f"prompt_entity={self.prompt_entity}, prompt_id={self.prompt_id}"
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(prompt_entity: {type(self.prompt_entity)} = {self.prompt_entity}, prompt_id: {type(self.prompt_id)} = {self.prompt_id})"

llm/llamacpp/lc_model.py

@@ -0,0 +1,196 @@
+import logging
+import os
+from abc import ABC
+
+import requests
+import yaml
+from langchain.prompts import PromptTemplate
+from langchain_community.llms import LlamaCpp
+
+from llm.config import config
+from llm.llm_interface import LLMInterface
+
+logger = logging.getLogger(__name__)
+
+logger.setLevel(logging.CRITICAL)  # 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)
+
+
+class LC_TinyLlama(LLMInterface, ABC):
+    def __init__(self, prompt_entity: str, prompt_id: int = 0):
+        self.prompt_entity = prompt_entity
+        self.prompt_id = prompt_id
+
+        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", 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 OSError as e:
+            logger.error(msg="Error while write a file to directory", exc_info=e)
+
+    @staticmethod
+    def __read_yaml():
+        try:
+            yaml_file = os.path.join("../", 'prompts.yaml')
+            with open(yaml_file, 'r') as file:
+                data = yaml.safe_load(file)
+            return data
+        except Exception as e:
+            logger.error(msg="Execution filed", exc_info=e)
+
+    def execution(self):
+        try:
+            data = self.__read_yaml()
+            prompts = data["prompts"][
+                self.prompt_id]  # get second prompt from yaml, need change id parameter to get other prompt
+            template = prompts["prompt_template"]
+            prompt = PromptTemplate(template=template, input_variables=["entity"])
+
+            llm = LlamaCpp(
+                model_path=os.path.join("../models", 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
+            )
+
+            logger.info(f"Check llm : {llm}")
+
+            llm_chain = prompt | llm
+            output = llm_chain.invoke({"question": self.prompt_entity})
+            return output
+        except Exception as e:
+            logger.critical(msg="Execution filed", exc_info=e)
+
+    def clear_llm(self, unused_model_dict, current_lc):
+        # If unused_model_dict is not empty
+        if len(unused_model_dict) > 1 and unused_model_dict is not None:
+            # go through key and value
+            for key, value in zip(unused_model_dict.keys(), unused_model_dict.values()):
+                # 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}")
+        else:
+            logger.info(f"Unfortunately dictionary empty or None")
+
+    def get_unused(self, current_lc):
+        models_dir = "../models"
+
+        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 __str__(self):
+        return f"prompt_entity={self.prompt_entity}, prompt_id={self.prompt_id}"
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(prompt_entity: {type(self.prompt_entity)} = {self.prompt_entity}, prompt_id: {type(self.prompt_id)} = {self.prompt_id})"
+
+
+class LC_Phi3(LLMInterface, ABC):
+    def __init__(self, prompt_entity: str, prompt_id: int = 0):
+        self.prompt_entity = prompt_entity
+        self.prompt_id = prompt_id
+
+        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", 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 OSError as e:
+            logger.error(msg="Error while write a file to directory", exc_info=e)
+
+    @staticmethod
+    def __read_yaml():
+        try:
+            yaml_file = os.path.join("../", 'prompts.yaml')
+            with open(yaml_file, 'r') as file:
+                data = yaml.safe_load(file)
+            return data
+        except Exception as e:
+            logger.error(msg="Execution filed", exc_info=e)
+
+    def execution(self):
+        try:
+            data = self.__read_yaml()
+            prompts = data["prompts"][
+                self.prompt_id]  # get second prompt from yaml, need change id parameter to get other prompt
+            template = prompts["prompt_template"]
+            prompt = PromptTemplate(template=template, input_variables=["entity"])
+
+            llm = LlamaCpp(
+                model_path=os.path.join("../models", 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
+            )
+
+            logger.info(f"Check llm : {llm}")
+
+            llm_chain = prompt | llm
+            output = llm_chain.invoke({"question": self.prompt_entity})
+            return output
+        except Exception as e:
+            logger.critical(msg="Execution filed", exc_info=e)
+
+    def clear_llm(self, unused_model_dict, current_lc):
+        # If unused_model_dict is not empty
+        if len(unused_model_dict) > 1:
+            # go through key and value
+            for key, value in zip(unused_model_dict.keys(), unused_model_dict.values()):
+                # 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}")
+        else:
+            logger.info(f"Unfortunately dictionary empty")
+
+    def get_unused(self, current_lc):
+        models_dir = "../models"
+
+        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 __str__(self):
+        return f"prompt_entity={self.prompt_entity}, prompt_id={self.prompt_id}"
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(prompt_entity: {type(self.prompt_entity)} = {self.prompt_entity}, prompt_id: {type(self.prompt_id)} = {self.prompt_id})"

llm/llm.py

@@ -1,31 +1,29 @@
-# {'prompts': [
-#    {'id': 1, 'prompt_template': 'Question: {question}\nAnswer: \n', 'description': 'simple question without a prompt', 'rate': 1}, 
-#    {'id': 2, 'prompt_template': "Question: {question}\nAnswer: Write a concise answer on the question with 
-#               one example if it's possible. CONCISE ANSWER.\n", 'description': 'simple concise prompt', 'rate': 3}]}
-
-
 import os
 
 import yaml
 import logging
 
 from langchain.prompts import PromptTemplate
-from langchain.chains import LLMChain 
-from langchain.llms import HuggingFaceHub
-from .config import config 
+from langchain.chains import LLMChain
+from langchain.llms import HuggingFaceEndpoint
+from .config import config
+
+os.environ["HUGGINGFACEHUB_API_TOKEN"] = "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"
 
 class LLM_chain:
     def __init__(self):
-        self.llm = HuggingFaceHub(
-            repo_id=config["model"], 
-            model_kwargs={"temperature": config["temperature"], "max_new_tokens": config["max_new_tokens"], "top_k": config["top_k"], "load_in_8bit": config["load_in_8bit"]})
+        self.llm = HuggingFaceEndpoint(
+            repo_id=config["model"],
+            model_kwargs={"temperature": config["temperature"], "max_new_tokens": config["max_new_tokens"],
+                          "top_k": config["top_k"], "load_in_8bit": config["load_in_8bit"]})
 
-    def __read_yaml(self):
+    @staticmethod
+    def __read_yaml():
         try:
-            # get current dir 
+            # get current dir
             current_dir = os.path.dirname(os.path.realpath(__file__))
             yaml_file = os.path.join(current_dir, 'prompts.yaml')
-            
+
             with open(yaml_file, 'r') as file:
                 data = yaml.safe_load(file)
             return data
@@ -35,11 +33,11 @@ class LLM_chain:
     def __call__(self, entity: str, id: int = 0):
         try:
             data = self.__read_yaml()
-            prompts = data["prompts"][id] # get second prompt from yaml, need change id parameter to get other prompt
+            prompts = data["prompts"][id]  # get second prompt from yaml, need change id parameter to get other prompt
             template = prompts["prompt_template"]
             prompt = PromptTemplate(template=template, input_variables=["entity"])
             llm_chain = LLMChain(prompt=prompt, llm=self.llm, verbose=True)
             output = llm_chain.invoke(entity)
             return output["text"]
         except Exception as e:
-            logging.error(e)
+            logging.error(e)

llm/llm_factory.py

@@ -0,0 +1,24 @@
+from typing import Dict, List
+
+from .huggingfacehub.hf_model import HF_Mistaril, HF_TinyLlama
+from .llamacpp.lc_model import LC_TinyLlama, LC_Phi3
+
+
+class LLM_Factory:
+
+    # trigger = {"model_type": "execution_type"} -> {"hf": "small"}
+    @staticmethod
+    def create_llm(prompt_entity: str, prompt_id: int, trigger: Dict[str, str]):
+
+        if trigger.keys() == "hf" and trigger.values() == "effective":
+            model = HF_Mistaril(prompt_entity=prompt_entity, prompt_id=prompt_id)
+        elif trigger.keys() == "hf" and trigger.values() == "small":
+            model = HF_TinyLlama(prompt_entity=prompt_entity, prompt_id=prompt_id)
+        elif trigger.keys() == "lc" and trigger.values() == "effective":
+            model = LC_Phi3(prompt_entity=prompt_entity, prompt_id=prompt_id)
+        elif trigger.keys() == "lc" and trigger.values() == "small":
+            model = LC_TinyLlama(prompt_entity=prompt_entity, prompt_id=prompt_id)
+        else:
+            model = None
+
+        return model

llm/llm_interface.py

@@ -0,0 +1,20 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Optional
+
+
+class LLMInterface(ABC):
+    @abstractmethod
+    def execution(self):
+        """Method execution LLM model based on HuggingFace or Langchain"""
+        pass
+
+    # {"model_name": "./models/model_name"}
+    @abstractmethod
+    def clear_llm(self, unused_model_dict: Optional[Dict[str, str]], current_lc: str) -> None:
+        """Method clear unused LLM"""
+        pass
+
+    @abstractmethod
+    def get_unused(self, current_lc: str) -> Optional[Dict[str, str]]:
+        """Method getting LLM that unused and forming corresponding dict"""
+        pass

packages.txt

@@ -1 +0,0 @@
-ffmpeg

requirements.txt

@@ -1,30 +1,35 @@
-psutil
+# other
+psutil~=5.9.8
+PyYAML~=6.0.1
 
 # related to stremalit
-streamlit-mic-recorder==0.0.4
+streamlit-mic-recorder==0.0.8
 streamlit-TTS
-streamlit
+streamlit~=1.33.0
 
 # related to streamlit-TTS
 python-dotenv
 gTTS
 pydub
 openai
+ffprobe
+ffmpeg
 
 # related to huggingface
-torch
-transformers
+torch~=2.2.2
+transformers~=4.40.1
 accelerate
 sentence-transformers
 bitsandbytes
+einops
 
 # related to langchain
-langchain 
+langchain~=0.1.17
 pypdf
 chromadb
 
 # related to audio
 librosa==0.10.1
-soundfile
-
-einops
+soundfile~=0.12.1
+numpy~=1.26.4
+streamlit_TTS~=0.0.7