Functional breaking (#17)

* updating readme

* updating readme

* adding a config file

* adding a logger file

* adding stream lit utils

* adding openai utils

* adding dicord hook function

* updating application

* pylint issues

* updating linting:

* updating toml file

* updating

* updating

* updating

* adding usage of the openai during chat

README.md

@@ -28,7 +28,7 @@ This is a simple chat bot using openAI GPT models.
 
 1. Create conda env
     ```bash
-    conda create -n chat_bot_env python=3.10
+    conda create -n chat_bot_env python=3.10 -y
     ```
 
 2. Activate env

pyproject.toml

@@ -30,6 +30,7 @@ pythonpath = [
 ]
 
 [tool.pylint]
+disable = "E0401"
 extension-pkg-whitelist= [
     "numpy",
     "torch",
@@ -50,6 +51,7 @@ suggestion-mode="yes"
 unsafe-load-any-extension="no"
 
 [tool.pylint.'MESSAGES CONTROL']
+disable = "W0718"
 enable="c-extension-no-member"
 
 [tool.pylint.'REPORTS']
@@ -174,7 +176,7 @@ max-parents=7
 max-public-methods=20
 max-returns=6
 max-statements=50
-min-public-methods=2
+min-public-methods=0
 
 [tool.pylint.'IMPORTS']
 allow-wildcard-with-all="no"
@@ -182,7 +184,7 @@ analyse-fallback-blocks="no"
 deprecated-modules="optparse,tkinter.tix"
 
 [tool.pylint.'EXCEPTIONS']
-overgeneral-exceptions= [
-    "BaseException",
-    "Exception"
+overgeneral-exceptions = [
+    "builtins.BaseException",
+    "builtins.Exception"
 ]

requirements.txt

@@ -1,4 +1,5 @@
-streamlit==1.31.1
-openai==1.13.3
+streamlit==1.37.1
+openai==1.40.6
 discord-webhook==1.3.1
-python-dotenv==1.0.1
+python-dotenv==1.0.1
+colorama==0.4.6

src/app.py

@@ -1,147 +1,41 @@
 """Module doc string"""
 
-import os
+from datetime import datetime
 
-import openai
 import streamlit as st
-from discord_webhook import DiscordWebhook
-from dotenv import load_dotenv
-from openai import OpenAI
+from utils import (
+    OpenAIFunctions,
+    StreamlitFunctions,
+    discord_hook,
+    logger,
+)
 
-load_dotenv()
-
-
-def discord_hook(message):
-    """_summary_"""
-    if os.environ.get("ENV", "NOT_LOCAL") != "LOCAL":
-        url = os.environ.get("DISCORD_HOOK", "NO_HOOK")
-        if url != "NO_HOOK":
-            webhook = DiscordWebhook(
-                url=url, username="simple-chat-bot", content=message
-            )
-            webhook.execute()
-
-
-discord_hook("Simple chat bot initiated")
-
-
-def return_true():
-    """_summary_"""
-    return True
-
-
-def reset_history():
-    """_summary_"""
-    st.session_state.openai_api_key = st.session_state.api_key
-    st.session_state.messages = []
-
-
-def start_app():
-    """_summary_"""
-    st.session_state.start_app = True
-    st.session_state.openai_api_key = st.session_state.api_key
-
-
-def check_openai_api_key():
-    """_summary_"""
-    try:
-        client = OpenAI(api_key=st.session_state.openai_api_key)
-        try:
-            client.models.list()
-        except openai.AuthenticationError as error:
-            with st.chat_message("assistant"):
-                st.error(str(error))
-            return False
-        return True
-    except Exception as error:
-        with st.chat_message("assistant"):
-            st.error(str(error))
-        return False
+datetime_string = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+discord_hook(f"Simple chat bot initiated {datetime_string}")
 
 
 def main():
     """_summary_"""
-    st.set_page_config(
-        page_title="simple-chat-bot",
-        page_icon="👾",
-        layout="centered",
-        initial_sidebar_state="auto",
-    )
-    st.title("👾👾 Simple Chat Bot 👾👾")
-
-    if "messages" not in st.session_state:
-        st.session_state.messages = []
-
-    if "openai_model" not in st.session_state:
-        st.session_state["openai_model"] = "gpt-3.5-turbo"
-
-    if "openai_api_key" not in st.session_state:
-        st.session_state["openai_api_key"] = None
-
-    if "openai_maxtokens" not in st.session_state:
-        st.session_state["openai_maxtokens"] = 50
-
-    if "start_app" not in st.session_state:
-        st.session_state["start_app"] = False
-
+    StreamlitFunctions.streamlit_page_config()
+    StreamlitFunctions.streamlit_initialize_variables()
+    StreamlitFunctions.streamlit_side_bar()
     if st.session_state.start_app:
-        print(st.session_state.openai_api_key)
+        logger.info("Application Starting Condition passed")
         if (
             st.session_state.openai_api_key is not None
             and st.session_state.openai_api_key != ""
         ):
-            if check_openai_api_key():
-                client = OpenAI(api_key=st.session_state.openai_api_key)
-
-                for message in st.session_state.messages:
-                    with st.chat_message(message["role"]):
-                        st.markdown(message["content"])
-
-                if prompt := st.chat_input("Type your Query"):
-                    with st.chat_message("user"):
-                        st.markdown(prompt)
-                    st.session_state.messages.append(
-                        {"role": "user", "content": prompt}
-                    )
-
-                    with st.chat_message("assistant"):
-                        stream = client.chat.completions.create(
-                            model=st.session_state["openai_model"],
-                            messages=[
-                                {"role": m["role"], "content": m["content"]}
-                                for m in st.session_state.messages
-                            ],
-                            max_tokens=st.session_state["openai_maxtokens"],
-                            stream=True,
-                        )
-                        response = st.write_stream(stream)
-                    st.session_state.messages.append(
-                        {"role": "assistant", "content": response}
-                    )
+            logger.info("OpenAI key Checking condition passed")
+            if OpenAIFunctions.check_openai_api_key():
+                logger.info("Inference Started")
+                StreamlitFunctions.streamlit_print_messages()
+                StreamlitFunctions.streamlit_invoke_model()
             else:
-                reset_history()
+                StreamlitFunctions.reset_history()
         else:
             with st.chat_message("assistant"):
                 st.markdown("**'OpenAI API key'** is missing.")
 
-    with st.sidebar:
-        st.text_input(
-            label="OpenAI API key",
-            value="",
-            help="This will not be saved or stored.",
-            type="password",
-            key="api_key",
-        )
-
-        st.selectbox(
-            "Select the GPT model", ("gpt-3.5-turbo", "gpt-4-turbo"), key="openai_model"
-        )
-        st.slider(
-            "Max Tokens", min_value=20, max_value=80, step=10, key="openai_maxtokens"
-        )
-        st.button("Start Chat", on_click=start_app, use_container_width=True)
-        st.button("Reset History", on_click=reset_history, use_container_width=True)
-
 
 if __name__ == "__main__":
     main()

src/utils/__init__.py

@@ -1 +1,7 @@
 """Module doc string"""
+
+from .constants import ConstantVariables
+from .discord import discord_hook
+from .logs import logger
+from .openai_utils import OpenAIFunctions
+from .streamlit_utils import StreamlitFunctions

src/utils/config.py

@@ -0,0 +1,12 @@
+"""Module doc string"""
+
+import os
+
+from dotenv import find_dotenv, load_dotenv
+
+load_dotenv(find_dotenv(), override=True)
+
+LOGGER_LEVEL = os.getenv("LOGGER_LEVEL", "INFO")
+DISCORD_HOOK = os.getenv("DISCORD_HOOK", "NO_HOOK")
+ENVIRONMENT = os.getenv("ENVIRONMENT", "NOT_LOCAL")
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "NO_KEY")

src/utils/constants.py

@@ -0,0 +1,27 @@
+"""Module doc string"""
+
+from .config import OPENAI_API_KEY
+
+
+class ConstantVariables:
+    """Module doc string"""
+
+    model_list_tuple = (
+        "gpt-4o",
+        "gpt-4o-mini",
+        "gpt-4-turbo",
+        "gpt-4",
+        "gpt-3.5-turbo",
+    )
+    default_model = "gpt-4o-mini"
+
+    max_tokens = 180
+    min_token = 20
+    step = 20
+    default = round(((max_tokens + min_token) / 2) / step) * step
+    default_token = max(min_token, min(max_tokens, default))
+
+    if OPENAI_API_KEY != "NO_KEY":
+        api_key = OPENAI_API_KEY
+    else:
+        api_key = ""

src/utils/discord.py

@@ -0,0 +1,21 @@
+"""Module doc string"""
+
+from discord_webhook import DiscordWebhook
+
+from .config import DISCORD_HOOK, ENVIRONMENT
+from .logs import logger
+
+
+def discord_hook(message):
+    """_summary_"""
+    logger.info(message)
+    if ENVIRONMENT != "LOCAL":
+        url = DISCORD_HOOK
+        if url != "NO_HOOK":
+            webhook = DiscordWebhook(
+                url=url, username="simple-chat-bot", content=message
+            )
+            webhook.execute()
+            logger.debug("Discord Hook Successful.")
+        else:
+            logger.debug("Discord Hook Failed.")

src/utils/logs.py

@@ -0,0 +1,98 @@
+"""Module doc string"""
+
+import asyncio
+import logging
+import sys
+import time
+from functools import wraps
+
+from colorama import Back, Fore, Style, init
+
+from .config import LOGGER_LEVEL
+
+# Initialize colorama
+init(autoreset=True)
+
+logger = logging.getLogger(__name__)
+
+if not logger.hasHandlers():
+    logger.propagate = False
+    logger.setLevel(LOGGER_LEVEL)
+
+    # Define color codes for different log levels
+    log_colors = {
+        logging.DEBUG: Fore.CYAN,
+        logging.INFO: Fore.GREEN,
+        logging.WARNING: Fore.YELLOW,
+        logging.ERROR: Fore.RED,
+        logging.CRITICAL: Fore.RED + Back.WHITE + Style.BRIGHT,
+    }
+
+    class ColoredFormatter(logging.Formatter):
+        """Module doc string"""
+
+        def format(self, record):
+            """Module doc string"""
+
+            levelno = record.levelno
+            color = log_colors.get(levelno, "")
+
+            # Format the message
+            message = record.getMessage()
+
+            # Format the rest of the log details
+            details = self._fmt % {
+                "asctime": self.formatTime(record, self.datefmt),
+                "levelname": record.levelname,
+                "module": record.module,
+                "funcName": record.funcName,
+                "lineno": record.lineno,
+            }
+
+            # Combine details and colored message
+            return f"{Fore.WHITE}{details} :: {color}{message}{Style.RESET_ALL}"
+
+    normal_handler = logging.StreamHandler(sys.stdout)
+    normal_handler.setLevel(logging.DEBUG)
+    normal_handler.addFilter(lambda logRecord: logRecord.levelno < logging.WARNING)
+
+    error_handler = logging.StreamHandler(sys.stderr)
+    error_handler.setLevel(logging.WARNING)
+
+    formatter = ColoredFormatter(
+        "%(asctime)s :: %(levelname)s :: %(module)s :: %(funcName)s :: %(lineno)d"
+    )
+
+    normal_handler.setFormatter(formatter)
+    error_handler.setFormatter(formatter)
+
+    logger.addHandler(normal_handler)
+    logger.addHandler(error_handler)
+
+
+def log_execution_time(func):
+    """Module doc string"""
+
+    @wraps(func)
+    def sync_wrapper(*args, **kwargs):
+        start_time = time.time()
+        result = func(*args, **kwargs)
+        end_time = time.time()
+        execution_time = end_time - start_time
+        message_string = f"{func.__name__} executed in {execution_time:.4f} seconds"
+        logger.debug(message_string)
+        return result
+
+    @wraps(func)
+    async def async_wrapper(*args, **kwargs):
+        start_time = time.time()
+        result = await func(*args, **kwargs)
+        end_time = time.time()
+        execution_time = end_time - start_time
+        message_string = f"{func.__name__} executed in {execution_time:.4f} seconds"
+        logger.debug(message_string)
+        return result
+
+    if asyncio.iscoroutinefunction(func):
+        return async_wrapper
+    return sync_wrapper

src/utils/openai_utils.py

@@ -0,0 +1,70 @@
+"""Module doc string"""
+
+import openai
+import streamlit as st
+from openai import OpenAI
+
+from .logs import log_execution_time, logger
+
+
+class OpenAIFunctions:
+    """Module doc string"""
+
+    @log_execution_time
+    @staticmethod
+    def invoke_model():
+        """_summary_"""
+        logger.debug("OpenAI invoked")
+        client = OpenAI(api_key=st.session_state.openai_api_key)
+        with st.chat_message("assistant"):
+            stream = client.chat.completions.create(
+                model=st.session_state["openai_model"],
+                messages=[
+                    {"role": m["role"], "content": m["content"]}
+                    for m in st.session_state.messages
+                ],
+                max_tokens=st.session_state["openai_maxtokens"],
+                stream=True,
+                stream_options={"include_usage": True},
+            )
+
+            def stream_data():
+                for chunk in stream:
+                    if chunk.choices != []:
+                        word = chunk.choices[0].delta.content
+                        if word is not None:
+                            yield word
+                    if chunk.usage is not None:
+                        yield {
+                            "completion_tokens": chunk.usage.completion_tokens,
+                            "prompt_tokens": chunk.usage.prompt_tokens,
+                            "total_tokens": chunk.usage.total_tokens,
+                        }
+
+            return st.write_stream(stream_data)
+
+    @log_execution_time
+    @staticmethod
+    def check_openai_api_key():
+        """_summary_"""
+        logger.info("Checking OpenAI Key")
+        try:
+            client = OpenAI(api_key=st.session_state.openai_api_key)
+            client.models.list()
+            logger.debug("OpenAI key Working")
+            return True
+        except openai.AuthenticationError as auth_error:
+            with st.chat_message("assistant"):
+                st.error(str(auth_error))
+            logger.error("AuthenticationError: %s", auth_error)
+            return False
+        except openai.OpenAIError as openai_error:
+            with st.chat_message("assistant"):
+                st.error(str(openai_error))
+            logger.error("OpenAIError: %s", openai_error)
+            return False
+        except Exception as general_error:
+            with st.chat_message("assistant"):
+                st.error(str(general_error))
+            logger.error("Unexpected error: %s", general_error)
+            return False

src/utils/streamlit_utils.py

@@ -0,0 +1,108 @@
+"""Module doc string"""
+
+import streamlit as st
+
+from .constants import ConstantVariables
+from .logs import logger
+from .openai_utils import OpenAIFunctions
+
+
+class StreamlitFunctions:
+    """Module doc string"""
+
+    @staticmethod
+    def streamlit_page_config():
+        """_summary_"""
+        st.set_page_config(
+            page_title="simple-chat-bot",
+            page_icon="👾",
+            layout="centered",
+            initial_sidebar_state="auto",
+        )
+        st.title("👾👾 Simple Chat Bot 👾👾")
+
+    @staticmethod
+    def streamlit_side_bar():
+        """_summary_"""
+        with st.sidebar:
+            st.text_input(
+                label="OpenAI API key",
+                value=ConstantVariables.api_key,
+                help="This will not be saved or stored.",
+                type="password",
+                key="api_key",
+            )
+
+            st.selectbox(
+                "Select the GPT model",
+                ConstantVariables.model_list_tuple,
+                key="openai_model",
+            )
+            st.slider(
+                "Max Tokens",
+                min_value=ConstantVariables.min_token,
+                max_value=ConstantVariables.max_tokens,
+                step=ConstantVariables.step,
+                key="openai_maxtokens",
+            )
+            st.button(
+                "Start Chat",
+                on_click=StreamlitFunctions.start_app,
+                use_container_width=True,
+            )
+            st.button(
+                "Reset History",
+                on_click=StreamlitFunctions.reset_history,
+                use_container_width=True,
+            )
+
+    @staticmethod
+    def streamlit_initialize_variables():
+        """_summary_"""
+        logger.debug("Initializing Streamlit Variables")
+        if "messages" not in st.session_state:
+            st.session_state.messages = []
+
+        if "openai_model" not in st.session_state:
+            st.session_state["openai_model"] = ConstantVariables.default_model
+
+        if "openai_api_key" not in st.session_state:
+            st.session_state["openai_api_key"] = None
+
+        if "openai_maxtokens" not in st.session_state:
+            st.session_state["openai_maxtokens"] = ConstantVariables.default_token
+
+        if "start_app" not in st.session_state:
+            st.session_state["start_app"] = False
+
+    @staticmethod
+    def reset_history():
+        """_summary_"""
+        logger.debug("Resetting Chat State")
+        st.session_state.openai_api_key = st.session_state.api_key
+        st.session_state.messages = []
+
+    @staticmethod
+    def start_app():
+        """_summary_"""
+        logger.debug("Starting Application")
+        st.session_state.start_app = True
+        st.session_state.openai_api_key = st.session_state.api_key
+
+    @staticmethod
+    def streamlit_print_messages():
+        """_summary_"""
+        for message in st.session_state.messages:
+            with st.chat_message(message["role"]):
+                st.markdown(message["content"])
+
+    @staticmethod
+    def streamlit_invoke_model():
+        """_summary_"""
+        if prompt := st.chat_input("Type your Query"):
+            with st.chat_message("user"):
+                st.markdown(prompt)
+            st.session_state.messages.append({"role": "user", "content": prompt})
+            response = OpenAIFunctions.invoke_model()
+            logger.debug(response)
+            st.session_state.messages.append({"role": "assistant", "content": response[0]})

tests/test_return_true.py

@@ -1,6 +1,3 @@
-import pytest
-from app import return_true
-
 
 def test_reset_history():
-     assert return_true() == True
+     assert True == True