Merge branch 'master' into add_website_memory

2026-02-01 10:24:56 -05:00 · 2023-04-14 18:34:53 +02:00
parent a67818648e f6f4f1fbc0
commit 483abb1da1
17 changed files with 501 additions and 182 deletions
--- a/scripts/ai_config.py
+++ b/scripts/ai_config.py
@@ -70,8 +70,8 @@ class AIConfig:
        """

        config = {"ai_name": self.ai_name, "ai_role": self.ai_role, "ai_goals": self.ai_goals}
-        with open(config_file, "w") as file:
-            yaml.dump(config, file)
+        with open(config_file, "w",  encoding='utf-8') as file:
+            yaml.dump(config, file, allow_unicode=True)

    def construct_full_prompt(self) -> str:
        """
--- a/scripts/call_ai_function.py
+++ b/scripts/call_ai_function.py
@@ -13,7 +13,7 @@ def call_ai_function(function, args, description, model=None):
        model = cfg.smart_llm_model
    # For each arg, if any are None, convert to "None":
    args = [str(arg) if arg is not None else "None" for arg in args]
-    # parse args to comma seperated string
+    # parse args to comma separated string
    args = ", ".join(args)
    messages = [
        {
--- a/scripts/config.py
+++ b/scripts/config.py
@@ -38,7 +38,9 @@ class Config(metaclass=Singleton):
        self.continuous_mode = False
        self.continuous_limit = 0
        self.speak_mode = False
+        self.skip_reprompt = False

+        self.ai_settings_file = os.getenv("AI_SETTINGS_FILE", "ai_settings.yaml")
        self.fast_llm_model = os.getenv("FAST_LLM_MODEL", "gpt-3.5-turbo")
        self.smart_llm_model = os.getenv("SMART_LLM_MODEL", "gpt-4")
        self.fast_token_limit = int(os.getenv("FAST_TOKEN_LIMIT", 4000))
@@ -64,6 +66,9 @@ class Config(metaclass=Singleton):
        self.use_mac_os_tts = False
        self.use_mac_os_tts = os.getenv("USE_MAC_OS_TTS")

+        self.use_brian_tts = False
+        self.use_brian_tts = os.getenv("USE_BRIAN_TTS")
+
        self.google_api_key = os.getenv("GOOGLE_API_KEY")
        self.custom_search_engine_id = os.getenv("CUSTOM_SEARCH_ENGINE_ID")

--- a/scripts/data_ingestion.py
+++ b/scripts/data_ingestion.py
@@ -0,0 +1,70 @@
+import argparse
+import logging
+from config import Config
+from memory import get_memory
+from file_operations import ingest_file, search_files
+
+cfg = Config()
+
+
+def configure_logging():
+    logging.basicConfig(filename='log-ingestion.txt',
+                    filemode='a',
+                    format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
+                    datefmt='%H:%M:%S',
+                    level=logging.DEBUG)
+    return logging.getLogger('AutoGPT-Ingestion')
+
+
+def ingest_directory(directory, memory, args):
+    """
+    Ingest all files in a directory by calling the ingest_file function for each file.
+
+    :param directory: The directory containing the files to ingest
+    :param memory: An object with an add() method to store the chunks in memory
+    """
+    try:
+        files = search_files(directory)
+        for file in files:
+            ingest_file(file, memory, args.max_length, args.overlap)
+    except Exception as e:
+        print(f"Error while ingesting directory '{directory}': {str(e)}")
+
+
+def main():
+    logger = configure_logging()
+
+    parser = argparse.ArgumentParser(description="Ingest a file or a directory with multiple files into memory. Make sure to set your .env before running this script.")
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--file", type=str, help="The file to ingest.")
+    group.add_argument("--dir", type=str, help="The directory containing the files to ingest.")
+    parser.add_argument("--init", action='store_true', help="Init the memory and wipe its content (default: False)", default=False)
+    parser.add_argument("--overlap", type=int, help="The overlap size between chunks when ingesting files (default: 200)", default=200)
+    parser.add_argument("--max_length", type=int, help="The max_length of each chunk when ingesting files (default: 4000)", default=4000)
+
+    args = parser.parse_args()
+
+    # Initialize memory
+    memory = get_memory(cfg, init=args.init)
+    print('Using memory of type: ' + memory.__class__.__name__)
+
+    if args.file:
+        try:
+            ingest_file(args.file, memory, args.max_length, args.overlap)
+            print(f"File '{args.file}' ingested successfully.")
+        except Exception as e:
+            logger.error(f"Error while ingesting file '{args.file}': {str(e)}")
+            print(f"Error while ingesting file '{args.file}': {str(e)}")
+    elif args.dir:
+        try:
+            ingest_directory(args.dir, memory, args)
+            print(f"Directory '{args.dir}' ingested successfully.")
+        except Exception as e:
+            logger.error(f"Error while ingesting directory '{args.dir}': {str(e)}")
+            print(f"Error while ingesting directory '{args.dir}': {str(e)}")
+    else:
+        print("Please provide either a file path (--file) or a directory name (--dir) inside the auto_gpt_workspace directory as input.")
+
+
+if __name__ == "__main__":
+    main()
--- a/scripts/file_operations.py
+++ b/scripts/file_operations.py
@@ -20,6 +20,29 @@ def safe_join(base, *paths):
    return norm_new_path


+def split_file(content, max_length=4000, overlap=0):
+    """
+    Split text into chunks of a specified maximum length with a specified overlap
+    between chunks.
+
+    :param text: The input text to be split into chunks
+    :param max_length: The maximum length of each chunk, default is 4000 (about 1k token)
+    :param overlap: The number of overlapping characters between chunks, default is no overlap
+    :return: A generator yielding chunks of text
+    """
+    start = 0
+    content_length = len(content)
+
+    while start < content_length:
+        end = start + max_length
+        if end + overlap < content_length:
+            chunk = content[start:end+overlap]
+        else:
+            chunk = content[start:content_length]
+        yield chunk
+        start += max_length - overlap
+
+
 def read_file(filename):
    """Read a file and return the contents"""
    try:
@@ -31,6 +54,37 @@ def read_file(filename):
        return "Error: " + str(e)


+def ingest_file(filename, memory, max_length=4000, overlap=200):
+    """
+    Ingest a file by reading its content, splitting it into chunks with a specified
+    maximum length and overlap, and adding the chunks to the memory storage.
+
+    :param filename: The name of the file to ingest
+    :param memory: An object with an add() method to store the chunks in memory
+    :param max_length: The maximum length of each chunk, default is 4000
+    :param overlap: The number of overlapping characters between chunks, default is 200
+    """
+    try:
+        print(f"Working with file {filename}")
+        content = read_file(filename)
+        content_length = len(content)
+        print(f"File length: {content_length} characters")
+
+        chunks = list(split_file(content, max_length=max_length, overlap=overlap))
+
+        num_chunks = len(chunks)
+        for i, chunk in enumerate(chunks):
+            print(f"Ingesting chunk {i + 1} / {num_chunks} into memory")
+            memory_to_add = f"Filename: {filename}\n" \
+                            f"Content part#{i + 1}/{num_chunks}: {chunk}"
+
+            memory.add(memory_to_add)
+
+        print(f"Done ingesting {num_chunks} chunks from {filename}.")
+    except Exception as e:
+        print(f"Error while ingesting file '{filename}': {str(e)}")
+
+
 def write_to_file(filename, text):
    """Write text to a file"""
    try:
--- a/scripts/logger.py
+++ b/scripts/logger.py
@@ -24,7 +24,8 @@ For console handler: simulates typing
 class Logger(metaclass=Singleton):
    def __init__(self):
        # create log directory if it doesn't exist
-        log_dir = os.path.join('..', 'logs')
+        this_files_dir_path = os.path.dirname(__file__)
+        log_dir = os.path.join(this_files_dir_path, '../logs')
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)

--- a/scripts/main.py
+++ b/scripts/main.py
@@ -129,64 +129,14 @@ def print_assistant_thoughts(assistant_reply):
        logger.error("Error: \n", call_stack)


-def load_variables(config_file="config.yaml"):
-    """Load variables from yaml file if it exists, otherwise prompt the user for input"""
-    try:
-        with open(config_file) as file:
-            config = yaml.load(file, Loader=yaml.FullLoader)
-        ai_name = config.get("ai_name")
-        ai_role = config.get("ai_role")
-        ai_goals = config.get("ai_goals")
-    except FileNotFoundError:
-        ai_name = ""
-        ai_role = ""
-        ai_goals = []
-
-    # Prompt the user for input if config file is missing or empty values
-    if not ai_name:
-        ai_name = utils.clean_input("Name your AI: ")
-        if ai_name == "":
-            ai_name = "Entrepreneur-GPT"
-
-    if not ai_role:
-        ai_role = utils.clean_input(f"{ai_name} is: ")
-        if ai_role == "":
-            ai_role = "an AI designed to autonomously develop and run businesses with the sole goal of increasing your net worth."
-
-    if not ai_goals:
-        print("Enter up to 5 goals for your AI: ")
-        print("For example: \nIncrease net worth, Grow Twitter Account, Develop and manage multiple businesses autonomously'")
-        print("Enter nothing to load defaults, enter nothing when finished.")
-        ai_goals = []
-        for i in range(5):
-            ai_goal = utils.clean_input(f"Goal {i+1}: ")
-            if ai_goal == "":
-                break
-            ai_goals.append(ai_goal)
-        if len(ai_goals) == 0:
-            ai_goals = ["Increase net worth", "Grow Twitter Account", "Develop and manage multiple businesses autonomously"]
-
-    # Save variables to yaml file
-    config = {"ai_name": ai_name, "ai_role": ai_role, "ai_goals": ai_goals}
-    with open(config_file, "w") as file:
-        documents = yaml.dump(config, file)
-
-    prompt = get_prompt()
-    prompt_start = """Your decisions must always be made independently without seeking user assistance. Play to your strengths as an LLM and pursue simple strategies with no legal complications."""
-
-    # Construct full prompt
-    full_prompt = f"You are {ai_name}, {ai_role}\n{prompt_start}\n\nGOALS:\n\n"
-    for i, goal in enumerate(ai_goals):
-        full_prompt += f"{i+1}. {goal}\n"
-
-    full_prompt += f"\n\n{prompt}"
-    return full_prompt
-
-
 def construct_prompt():
    """Construct the prompt for the AI to respond to"""
-    config = AIConfig.load()
-    if config.ai_name:
+    config = AIConfig.load(cfg.ai_settings_file)
+    if cfg.skip_reprompt and config.ai_name:
+        logger.typewriter_log("Name :", Fore.GREEN, config.ai_name)
+        logger.typewriter_log("Role :", Fore.GREEN, config.ai_role)
+        logger.typewriter_log("Goals:", Fore.GREEN, config.ai_goals)
+    elif config.ai_name:
        logger.typewriter_log(
            f"Welcome back! ",
            Fore.GREEN,
@@ -274,13 +224,15 @@ def parse_arguments():
    cfg.set_speak_mode(False)

    parser = argparse.ArgumentParser(description='Process arguments.')
-    parser.add_argument('--continuous', action='store_true', help='Enable Continuous Mode')
+    parser.add_argument('--continuous', '-c', action='store_true', help='Enable Continuous Mode')
    parser.add_argument('--continuous-limit', '-l', type=int, dest="continuous_limit", help='Defines the number of times to run in continuous mode')
    parser.add_argument('--speak', action='store_true', help='Enable Speak Mode')
    parser.add_argument('--debug', action='store_true', help='Enable Debug Mode')
    parser.add_argument('--gpt3only', action='store_true', help='Enable GPT3.5 Only Mode')
    parser.add_argument('--gpt4only', action='store_true', help='Enable GPT4 Only Mode')
    parser.add_argument('--use-memory', '-m', dest="memory_type", help='Defines which Memory backend to use')
+    parser.add_argument('--skip-reprompt', '-y', dest='skip_reprompt', action='store_true', help='Skips the re-prompting messages at the beginning of the script')
+    parser.add_argument('--ai-settings', '-C', dest='ai_settings_file', help="Specifies which ai_settings.yaml file to use, will also automatically skip the re-prompt.")
    args = parser.parse_args()

    if args.debug:
@@ -318,10 +270,6 @@ def parse_arguments():
        logger.typewriter_log("GPT4 Only Mode: ", Fore.GREEN, "ENABLED")
        cfg.set_fast_llm_model(cfg.smart_llm_model)

-    if args.debug:
-        logger.typewriter_log("Debug Mode: ", Fore.GREEN, "ENABLED")
-        cfg.set_debug_mode(True)
-
    if args.memory_type:
        supported_memory = get_supported_memory_backends()
        chosen = args.memory_type
@@ -331,6 +279,24 @@ def parse_arguments():
        else:
            cfg.memory_backend = chosen

+    if args.skip_reprompt:
+        logger.typewriter_log("Skip Re-prompt: ", Fore.GREEN, "ENABLED")
+        cfg.skip_reprompt = True
+
+    if args.ai_settings_file:
+        file = args.ai_settings_file
+
+        # Validate file
+        (validated, message) = utils.validate_yaml_file(file)
+        if not validated:
+            logger.typewriter_log("FAILED FILE VALIDATION", Fore.RED, message)
+            logger.double_check()
+            exit(1)
+
+        logger.typewriter_log("Using AI Settings File:", Fore.GREEN, file)
+        cfg.ai_settings_file = file
+        cfg.skip_reprompt = True
+

 def main():
    global ai_name, memory
@@ -351,110 +317,148 @@ def main():
    # this is particularly important for indexing and referencing pinecone memory
    memory = get_memory(cfg, init=True)
    print('Using memory of type: ' + memory.__class__.__name__)
-    # Interaction Loop
-    loop_count = 0
-    while True:
-         # Discontinue if continuous limit is reached
-        loop_count += 1
-        if cfg.continuous_mode and cfg.continuous_limit > 0 and loop_count > cfg.continuous_limit:
-            logger.typewriter_log("Continuous Limit Reached: ", Fore.YELLOW, f"{cfg.continuous_limit}")
-            break
+    agent = Agent(
+        ai_name=ai_name,
+        memory=memory,
+        full_message_history=full_message_history,
+        next_action_count=next_action_count,
+        prompt=prompt,
+        user_input=user_input
+    )
+    agent.start_interaction_loop()

-        # Send message to AI, get response
-        with Spinner("Thinking... "):
-            assistant_reply = chat.chat_with_ai(
-                prompt,
-                user_input,
-                full_message_history,
-                memory,
-                cfg.fast_token_limit)  # TODO: This hardcodes the model to use GPT3.5. Make this an argument

-        # Print Assistant thoughts
-        print_assistant_thoughts(assistant_reply)
+class Agent:
+    """Agent class for interacting with Auto-GPT.

-        # Get command name and arguments
-        try:
-            command_name, arguments = cmd.get_command(
-                attempt_to_fix_json_by_finding_outermost_brackets(assistant_reply))
-            if cfg.speak_mode:
-                speak.say_text(f"I want to execute {command_name}")
-        except Exception as e:
-            logger.error("Error: \n", str(e))
+    Attributes:
+        ai_name: The name of the agent.
+        memory: The memory object to use.
+        full_message_history: The full message history.
+        next_action_count: The number of actions to execute.
+        prompt: The prompt to use.
+        user_input: The user input.

-        if not cfg.continuous_mode and next_action_count == 0:
-            ### GET USER AUTHORIZATION TO EXECUTE COMMAND ###
-            # Get key press: Prompt the user to press enter to continue or escape
-            # to exit
-            user_input = ""
-            logger.typewriter_log(
-                "NEXT ACTION: ",
-                Fore.CYAN,
-                f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL}  ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
-            print(
-                f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {ai_name}...",
-                flush=True)
-            while True:
-                console_input = utils.clean_input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
-                if console_input.lower().rstrip() == "y":
-                    user_input = "GENERATE NEXT COMMAND JSON"
-                    break
-                elif console_input.lower().startswith("y -"):
-                    try:
-                        next_action_count = abs(int(console_input.split(" ")[1]))
-                        user_input = "GENERATE NEXT COMMAND JSON"
-                    except ValueError:
-                        print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
-                        continue
-                    break
-                elif console_input.lower() == "n":
-                    user_input = "EXIT"
-                    break
-                else:
-                    user_input = console_input
-                    command_name = "human_feedback"
-                    break
+    """
+    def __init__(self,
+                 ai_name,
+                 memory,
+                 full_message_history,
+                 next_action_count,
+                 prompt,
+                 user_input):
+        self.ai_name = ai_name
+        self.memory = memory
+        self.full_message_history = full_message_history
+        self.next_action_count = next_action_count
+        self.prompt = prompt
+        self.user_input = user_input

-            if user_input == "GENERATE NEXT COMMAND JSON":
-                logger.typewriter_log(
-                    "-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
-                    Fore.MAGENTA,
-                    "")
-            elif user_input == "EXIT":
-                print("Exiting...", flush=True)
+    def start_interaction_loop(self):
+        # Interaction Loop
+        loop_count = 0
+        while True:
+             # Discontinue if continuous limit is reached
+            loop_count += 1
+            if cfg.continuous_mode and cfg.continuous_limit > 0 and loop_count > cfg.continuous_limit:
+                logger.typewriter_log("Continuous Limit Reached: ", Fore.YELLOW, f"{cfg.continuous_limit}")
                break
-        else:
-            # Print command
-            logger.typewriter_log(
-                "NEXT ACTION: ",
-                Fore.CYAN,
-                f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL}  ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")

-        # Execute command
-        if command_name is not None and command_name.lower().startswith("error"):
-            result = f"Command {command_name} threw the following error: " + arguments
-        elif command_name == "human_feedback":
-            result = f"Human feedback: {user_input}"
-        else:
-            result = f"Command {command_name} returned: {cmd.execute_command(command_name, arguments)}"
-            if next_action_count > 0:
-                next_action_count -= 1
+            # Send message to AI, get response
+            with Spinner("Thinking... "):
+                assistant_reply = chat.chat_with_ai(
+                    self.prompt,
+                    self.user_input,
+                    self.full_message_history,
+                    self.memory,
+                    cfg.fast_token_limit)  # TODO: This hardcodes the model to use GPT3.5. Make this an argument

-        memory_to_add = f"Assistant Reply: {assistant_reply} " \
-                        f"\nResult: {result} " \
-                        f"\nHuman Feedback: {user_input} "
+            # Print Assistant thoughts
+            print_assistant_thoughts(assistant_reply)

-        memory.add(memory_to_add)
+            # Get command name and arguments
+            try:
+                command_name, arguments = cmd.get_command(
+                    attempt_to_fix_json_by_finding_outermost_brackets(assistant_reply))
+                if cfg.speak_mode:
+                    speak.say_text(f"I want to execute {command_name}")
+            except Exception as e:
+                logger.error("Error: \n", str(e))

-        # Check if there's a result from the command append it to the message
-        # history
-        if result is not None:
-            full_message_history.append(chat.create_chat_message("system", result))
-            logger.typewriter_log("SYSTEM: ", Fore.YELLOW, result)
-        else:
-            full_message_history.append(
-                chat.create_chat_message(
-                    "system", "Unable to execute command"))
-            logger.typewriter_log("SYSTEM: ", Fore.YELLOW, "Unable to execute command")
+            if not cfg.continuous_mode and self.next_action_count == 0:
+                ### GET USER AUTHORIZATION TO EXECUTE COMMAND ###
+                # Get key press: Prompt the user to press enter to continue or escape
+                # to exit
+                self.user_input = ""
+                logger.typewriter_log(
+                    "NEXT ACTION: ",
+                    Fore.CYAN,
+                    f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL}  ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
+                print(
+                    f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {self.ai_name}...",
+                    flush=True)
+                while True:
+                    console_input = utils.clean_input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
+                    if console_input.lower().rstrip() == "y":
+                        self.user_input = "GENERATE NEXT COMMAND JSON"
+                        break
+                    elif console_input.lower().startswith("y -"):
+                        try:
+                            self.next_action_count = abs(int(console_input.split(" ")[1]))
+                            self.user_input = "GENERATE NEXT COMMAND JSON"
+                        except ValueError:
+                            print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
+                            continue
+                        break
+                    elif console_input.lower() == "n":
+                        self.user_input = "EXIT"
+                        break
+                    else:
+                        self.user_input = console_input
+                        command_name = "human_feedback"
+                        break
+
+                if self.user_input == "GENERATE NEXT COMMAND JSON":
+                    logger.typewriter_log(
+                        "-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
+                        Fore.MAGENTA,
+                        "")
+                elif self.user_input == "EXIT":
+                    print("Exiting...", flush=True)
+                    break
+            else:
+                # Print command
+                logger.typewriter_log(
+                    "NEXT ACTION: ",
+                    Fore.CYAN,
+                    f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL}  ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
+
+            # Execute command
+            if command_name is not None and command_name.lower().startswith("error"):
+                result = f"Command {command_name} threw the following error: " + arguments
+            elif command_name == "human_feedback":
+                result = f"Human feedback: {self.user_input}"
+            else:
+                result = f"Command {command_name} returned: {cmd.execute_command(command_name, arguments)}"
+                if self.next_action_count > 0:
+                    self.next_action_count -= 1
+
+            memory_to_add = f"Assistant Reply: {assistant_reply} " \
+                            f"\nResult: {result} " \
+                            f"\nHuman Feedback: {self.user_input} "
+
+            self.memory.add(memory_to_add)
+
+            # Check if there's a result from the command append it to the message
+            # history
+            if result is not None:
+                self.full_message_history.append(chat.create_chat_message("system", result))
+                logger.typewriter_log("SYSTEM: ", Fore.YELLOW, result)
+            else:
+                self.full_message_history.append(
+                    chat.create_chat_message(
+                        "system", "Unable to execute command"))
+                logger.typewriter_log("SYSTEM: ", Fore.YELLOW, "Unable to execute command")


 if __name__ == "__main__":
--- a/scripts/memory/init.py
+++ b/scripts/memory/init.py
@@ -3,7 +3,7 @@ from memory.no_memory import NoMemory

 # List of supported memory backends
 # Add a backend to this list if the import attempt is successful
-supported_memory = ['local']
+supported_memory = ['local', 'no_memory']

 try:
    from memory.redismem import RedisMemory
--- a/scripts/speak.py
+++ b/scripts/speak.py
@@ -53,6 +53,24 @@ def eleven_labs_speech(text, voice_index=0):
        return False


+def brian_speech(text):
+    """Speak text using Brian with the streamelements API"""
+    tts_url = f"https://api.streamelements.com/kappa/v2/speech?voice=Brian&text={text}"
+    response = requests.get(tts_url)
+
+    if response.status_code == 200:
+        with mutex_lock:
+            with open("speech.mp3", "wb") as f:
+                f.write(response.content)
+            playsound("speech.mp3")
+            os.remove("speech.mp3")
+        return True
+    else:
+        print("Request failed with status code:", response.status_code)
+        print("Response content:", response.content)
+        return False
+
+
 def gtts_speech(text):
    tts = gtts.gTTS(text)
    with mutex_lock:
@@ -76,7 +94,11 @@ def say_text(text, voice_index=0):
    def speak():
        if not cfg.elevenlabs_api_key:
            if cfg.use_mac_os_tts == 'True':
-                macos_tts_speech(text, voice_index)
+                macos_tts_speech(text)
+            elif cfg.use_brian_tts == 'True':
+                success = brian_speech(text)
+                if not success:
+                    gtts_speech(text)
            else:
                gtts_speech(text)
        else:
--- a/scripts/spinner.py
+++ b/scripts/spinner.py
@@ -17,10 +17,10 @@ class Spinner:
    def spin(self):
        """Spin the spinner"""
        while self.running:
-            sys.stdout.write(next(self.spinner) + " " + self.message + "\r")
+            sys.stdout.write(f"{next(self.spinner)} {self.message}\r")
            sys.stdout.flush()
            time.sleep(self.delay)
-            sys.stdout.write('\r' + ' ' * (len(self.message) + 2) + '\r')
+            sys.stdout.write(f"\r{' ' * (len(self.message) + 2)}\r")

    def __enter__(self):
        """Start the spinner"""
@@ -32,5 +32,5 @@ class Spinner:
        """Stop the spinner"""
        self.running = False
        self.spinner_thread.join()
-        sys.stdout.write('\r' + ' ' * (len(self.message) + 2) + '\r')
+        sys.stdout.write(f"\r{' ' * (len(self.message) + 2)}\r")
        sys.stdout.flush()
--- a/scripts/utils.py
+++ b/scripts/utils.py
@@ -1,3 +1,7 @@
+import yaml
+from colorama import Fore
+
+
 def clean_input(prompt: str=''):
    try:
        return input(prompt)
@@ -6,3 +10,14 @@ def clean_input(prompt: str=''):
        print("Quitting...")
        exit(0)

+
+def validate_yaml_file(file: str):
+    try:
+        with open(file) as file:
+            yaml.load(file, Loader=yaml.FullLoader)
+    except FileNotFoundError:
+        return (False, f"The file {Fore.CYAN}`{file}`{Fore.RESET} wasn't found")
+    except yaml.YAMLError as e:
+        return (False, f"There was an issue while trying to read with your AI Settings file: {e}")
+
+    return (True, f"Successfully validated {Fore.CYAN}`{file}`{Fore.RESET}!")