init commit

.env.template

@@ -0,0 +1,11 @@
+# ["hf", "openai"]
+ASSISTANT_TYPE=openai
+EMBEDDING_TYPE=hf
+
+# if openai
+OPENAI_API_KEY=sk-xxxxx
+OPENAI_KNOWLEDGEBASE=kb_openai
+
+# if hf
+HUGGINGFACEHUB_API_TOKEN=hf_xxxxx
+HF_KNOWLEDGEBASE=kb_hf

.gitignore

@@ -0,0 +1,163 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/
+
+
+# App-specific

.streamlit/config.toml

@@ -0,0 +1,6 @@
+[theme]
+primaryColor="#2c7b2c"
+backgroundColor="#171e1a"
+secondaryBackgroundColor="#111811"
+textColor="#cfd8dc"
+font="sans serif"

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.10.9
+
+WORKDIR /app
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8501
+
+CMD ["streamlit", "run", "app.py", "--server.port", "8501"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Ishara Dissanayake
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

app.py

@@ -0,0 +1,294 @@
+import logging
+import os
+import re
+
+import streamlit as st
+from streamlit.logger import get_logger
+
+from knowledgebase import Knowledgebase
+from utils.constants import (
+    AssistantType,
+    OPENAI_KNOWLEDGEBASE_KEY,
+    HUGGINGFACEHUB_API_TOKEN_KEY,
+    HF_KNOWLEDGEBASE_KEY,
+    SOURCES_TAG,
+    ANSWER_TAG,
+    NONE_TAG,
+    EMPTY_TAG,
+    MESSAGE_HISTORY_TAG,
+    TEXT_TAG,
+    USER_TAG,
+    ASSISTANT_TAG,
+    FROM_TAG,
+    IN_PROGRESS_TAG,
+    QUERY_INPUT_TAG,
+    VALID_TOKEN_TAG,
+    StNotificationType,
+    API_KEY_TAG,
+    ASSISTANT_TYPE_TAG,
+    ASSISTANT_AVATAR,
+    USER_AVATAR,
+    EmbeddingType,
+    APIKeyType,
+)
+from utils.llm import validate_api_token
+
+# initialize a logger
+logger = get_logger(__name__)
+
+
+def retrieve_answer(query: str):
+    try:
+        assistant_type = st.session_state.selected_assistant_type
+        embedding_type = EmbeddingType.HUGGINGFACE
+        assistant_api_key = st.session_state.verified_api_key
+        embedding_api_key = st.session_state.embedding_api_key
+        knowledgebase_name = st.session_state.knowledgebase_name
+
+        knowledgebase = Knowledgebase(
+            assistant_type=assistant_type,
+            embedding_type=embedding_type,
+            assistant_api_key=assistant_api_key,
+            embedding_api_key=embedding_api_key,
+            knowledgebase_name=knowledgebase_name,
+        )
+        answer, metadata = knowledgebase.query_knowledgebase(query=query)
+        if not metadata:
+            metadata = "$0.00"
+
+        final_answer = re.sub(
+            r"\bSOURCES:[\n\s]*$", "", str(answer[ANSWER_TAG]).strip()
+        ).strip()
+        logger.info(f"final answer: {final_answer}")
+
+        if answer.get(SOURCES_TAG, None) not in [None, NONE_TAG, EMPTY_TAG]:
+            return f"{final_answer}\n\nSources:\n{answer[SOURCES_TAG]}\n\nCost (USD):\n`{metadata}`"
+        else:
+            return f"{final_answer}\n\nCost:\n`{metadata}`"
+    except Exception as e:
+        logger.exception(f"Invalid API key. {e}")
+        return (
+            f"Could not retrieve the answer. This could be due to "
+            f"various reasons such as Invalid API Tokens or hitting "
+            f"the Rate limit enforced by LLM vendors."
+        )
+
+
+def show_chat_ui():
+    if (
+        st.session_state.selected_assistant_type == AssistantType.HUGGINGFACE
+        and not st.session_state.get(MESSAGE_HISTORY_TAG, None)
+    ):
+        show_notification_banner_ui(
+            notification_type=StNotificationType.WARNING,
+            notification="🤗🤏🏽 HuggingFace assistant is not always guaranteed "
+            "to return a valid response and often exceeds the "
+            "maximum token limit. Use the OpenAI assistant for "
+            "more reliable responses.",
+        )
+
+    if not st.session_state.get(MESSAGE_HISTORY_TAG, None):
+        st.subheader("Let's start chatting, shall we?")
+
+    if st.session_state.get(IN_PROGRESS_TAG, False):
+        query = st.chat_input(
+            "Ask me about ShoutOUT AI stuff", key=QUERY_INPUT_TAG, disabled=True
+        )
+    else:
+        query = st.chat_input("Ask me about ShoutOUT AI stuff", key=QUERY_INPUT_TAG)
+
+    if query:
+        st.session_state.in_progress = True
+        current_messages = st.session_state.get(MESSAGE_HISTORY_TAG, [])
+        current_messages.append({TEXT_TAG: query, FROM_TAG: USER_TAG})
+        st.session_state.message_history = current_messages
+        answer = retrieve_answer(query=query)
+        current_messages.append({TEXT_TAG: answer, FROM_TAG: ASSISTANT_TAG})
+        st.session_state.message_history = current_messages
+        st.session_state.in_progress = False
+
+    if st.session_state.get(MESSAGE_HISTORY_TAG, None):
+        messages = st.session_state.message_history
+        for message in messages:
+            if message.get(FROM_TAG) == USER_TAG:
+                with st.chat_message(USER_TAG, avatar=USER_AVATAR):
+                    st.write(message.get(TEXT_TAG))
+
+            if message.get(FROM_TAG) == ASSISTANT_TAG:
+                with st.chat_message(ASSISTANT_TAG, avatar=ASSISTANT_AVATAR):
+                    st.write(message.get(TEXT_TAG))
+
+
+def show_hf_chat_ui():
+    st.sidebar.info(
+        "🤗 You are using the Hugging Face Hub models for the QA task and "
+        "performance might not be as good as proprietary LLMs."
+    )
+
+    verify_token()
+    validated_token = st.session_state.get(VALID_TOKEN_TAG, None)
+    if validated_token is None:
+        st.stop()
+    if not validated_token:
+        st.sidebar.error("❌ Failed to get connected to the HuggingFace Hub")
+        show_notification_banner_ui(
+            notification_type=StNotificationType.INFO,
+            notification="Failed to get connected to the HuggingFace Hub",
+        )
+        st.stop()
+
+    st.sidebar.success(f"✅ Connected to the HF Hub")
+    show_chat_ui()
+
+
+def show_openai_chat_ui():
+    st.sidebar.info(
+        "🚀 To get started, enter your OpenAI API key. Once that's done, "
+        "you can ask start asking questions. Oh! one more thing, we take "
+        "security seriously and we are NOT storing the API keys in any manner, "
+        "so you're safe. Just revoke it after usage to make sure nothing "
+        "unexpected happens."
+    )
+    if st.sidebar.text_input(
+        "Enter the OpenAI API Key",
+        key=API_KEY_TAG,
+        label_visibility="hidden",
+        placeholder="OpenAI API Key",
+        type="password",
+    ):
+        verify_token()
+
+    validated_token = st.session_state.get(VALID_TOKEN_TAG, None)
+    if validated_token is None:
+        st.sidebar.info(f"🗝️ Provide the API Key")
+        st.stop()
+    if not validated_token:
+        st.sidebar.error("❌ API Key you provided is invalid")
+        show_notification_banner_ui(
+            notification_type=StNotificationType.INFO,
+            notification="Please provide a valid OpenAI API Key",
+        )
+        st.stop()
+
+    st.sidebar.success(f"✅ Token Validated!")
+    show_chat_ui()
+
+
+def show_notification_banner_ui(
+    notification_type: StNotificationType, notification: str
+):
+    if notification_type == StNotificationType.INFO:
+        st.info(notification)
+    elif notification_type == StNotificationType.WARNING:
+        st.warning(notification)
+    elif notification_type == StNotificationType.ERROR:
+        st.error(notification)
+
+
+def verify_token():
+    from dotenv import load_dotenv
+
+    load_dotenv()
+
+    embedding_api_key = os.getenv(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+    st_assistant_type = st.session_state.selected_assistant_type
+    if st_assistant_type == AssistantType.OPENAI:
+        assistant_api_key = st.session_state.get(API_KEY_TAG, None)
+        assistant_api_key_type = APIKeyType.OPENAI
+        knowledgebase_name = os.environ.get(OPENAI_KNOWLEDGEBASE_KEY, None)
+    else:
+        assistant_api_key = os.getenv(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+        assistant_api_key_type = APIKeyType.HUGGINGFACE
+        knowledgebase_name = os.environ.get(HF_KNOWLEDGEBASE_KEY, None)
+
+    logger.info(
+        f"The API key for the current st session: {assistant_api_key}\n"
+        f"The Knowledgebase for the current st session: {knowledgebase_name}"
+    )
+
+    assistant_valid, assistant_err = validate_api_token(
+        api_key_type=assistant_api_key_type,
+        api_key=assistant_api_key,
+    )
+    embedding_valid, embedding_err = validate_api_token(
+        api_key_type=APIKeyType.HUGGINGFACE,
+        api_key=embedding_api_key,
+    )
+
+    if assistant_valid and embedding_valid:
+        st.session_state.valid_token = True
+        st.session_state.verified_api_key = assistant_api_key
+        st.session_state.embedding_api_key = embedding_api_key
+        st.session_state.knowledgebase_name = knowledgebase_name
+    elif not assistant_valid and not embedding_valid:
+        st.session_state.valid_token = False
+        st.session_state.token_err = f"{assistant_err}\n{embedding_err}"
+    elif not assistant_valid:
+        st.session_state.valid_token = False
+        st.session_state.token_err = assistant_err
+    elif not embedding_valid:
+        st.session_state.valid_token = False
+        st.session_state.token_err = embedding_err
+    else:
+        st.session_state.valid_token = False
+        st.session_state.token_err = (
+            "An unknown error occurred while validating the API keys"
+        )
+
+
+def app():
+    # sidebar
+    st.sidebar.image(
+        "https://thisisishara.com/res/images/favicon/android-chrome-192x192.png",
+        width=80,
+    )
+    if st.sidebar.selectbox(
+        "Assistant Type",
+        ["OpenAI", "Hugging Face"],
+        key=ASSISTANT_TYPE_TAG,
+        placeholder="Select Assistant Type",
+    ):
+        if str(st.session_state.assistant_type).lower() == AssistantType.OPENAI.value:
+            st.session_state.selected_assistant_type = AssistantType.OPENAI
+        else:
+            st.session_state.selected_assistant_type = AssistantType.HUGGINGFACE
+        st.session_state.valid_token = None
+        st.session_state.verified_api_key = None
+        st.session_state.knowledgebase_name = None
+
+    st.write(st.session_state.selected_assistant_type)
+
+    # main section
+    st.header("LLM Website QA Demo")
+    st.caption("⚡ Powered by :blue[LangChain], :green[OpenAI] & :green[Hugging Face]")
+
+    assistant_type = st.session_state.selected_assistant_type
+    if assistant_type == AssistantType.OPENAI:
+        show_openai_chat_ui()
+    elif assistant_type == AssistantType.HUGGINGFACE:
+        show_hf_chat_ui()
+    else:
+        show_notification_banner_ui(
+            notification_type=StNotificationType.INFO,
+            notification="Please select an assistant type to get started!",
+        )
+
+
+if __name__ == "__main__":
+    st.set_page_config(
+        page_title="Website QA powered by LangChain & LLMs",
+        page_icon="https://thisisishara.com/res/images/favicon/android-chrome-192x192.png",
+        layout="wide",
+        initial_sidebar_state="expanded",
+    )
+    hide_streamlit_style = """
+                <style>
+                # #MainMenu {visibility: hidden;}
+                # footer {visibility: hidden;}
+                [data-testid="stDecoration"] {background: linear-gradient(to right, #9EE51A, #208BBC) !important;}
+                </style>
+                """
+    st.markdown(hide_streamlit_style, unsafe_allow_html=True)
+
+    # run the app
+    app()

backup/docker/kb_openai_ishara.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:540dfd6d7bda272777ee04edb931074a548217f1abbd76e772d6a36dea44c5bc
+size 40432

backup/windows/kb_openai_ishara.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:16937df0644608b4c1a5d3c6dcfa3cbd12b8afd3dac9d35d5654ac5d727ffefb
+size 40432

build_knowledgebase.py

@@ -0,0 +1,94 @@
+import logging
+import os
+import sys
+
+from dotenv import load_dotenv
+
+from knowledgebase import create_knowledgebase
+from utils.constants import (
+    ASSISTANT_TYPE_KEY,
+    AssistantType,
+    OPENAI_API_TOKEN_KEY,
+    HUGGINGFACEHUB_API_TOKEN_KEY,
+    OPENAI_KNOWLEDGEBASE_KEY,
+    HF_KNOWLEDGEBASE_KEY,
+    ENV_FILE,
+    EMBEDDING_TYPE_KEY,
+    EmbeddingType,
+    APIKeyType,
+)
+from utils.llm import validate_api_token
+
+logger = logging.getLogger(__name__)
+
+# load the .env
+load_dotenv(dotenv_path=os.path.join(os.getcwd(), ENV_FILE))
+
+
+if __name__ == "__main__":
+    # initialize the knowledgebase
+    logger.info("⚡ Initializing the URLs...")
+
+    # determine assistant type
+    assistant_type = os.getenv(ASSISTANT_TYPE_KEY, AssistantType.HUGGINGFACE.value)
+    embedding_type = os.getenv(EMBEDDING_TYPE_KEY, EmbeddingType.HUGGINGFACE.value)
+
+    if assistant_type == AssistantType.OPENAI.value:
+        assistant_type = AssistantType.OPENAI
+        knowledgebase_name = os.environ.get(OPENAI_KNOWLEDGEBASE_KEY, None)
+
+        if embedding_type == EmbeddingType.OPENAI.value:
+            embedding_type = EmbeddingType.OPENAI
+            embedding_api_key = os.getenv(OPENAI_API_TOKEN_KEY, None)
+            embedding_api_key_type = APIKeyType.OPENAI
+        else:
+            embedding_type = EmbeddingType.HUGGINGFACE
+            embedding_api_key = os.getenv(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+            embedding_api_key_type = APIKeyType.HUGGINGFACE
+
+    else:
+        assistant_type = AssistantType.HUGGINGFACE
+        knowledgebase_name = os.environ.get(HF_KNOWLEDGEBASE_KEY, None)
+        embedding_type = EmbeddingType.HUGGINGFACE
+        embedding_api_key = os.getenv(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+        embedding_api_key_type = APIKeyType.HUGGINGFACE
+
+    if embedding_type == EmbeddingType.OPENAI:
+        urls = [
+            "https://thisisishara.com/",
+            "https://github.com/thisisishara",
+            "https://github.com/thisisishara?tab=repositories",
+            "https://www.hackerrank.com/thisisishara?hr_r=1",
+            "https://www.npmjs.com/~thisisishara",
+            "https://pypi.org/user/thisisishara/",
+            "https://www.linkedin.com/in/isharadissanayake/",
+        ]
+
+    else:
+        urls = [
+            "https://thisisishara.com/",
+            "https://github.com/thisisishara",
+            "https://github.com/thisisishara?tab=repositories",
+            "https://www.hackerrank.com/thisisishara?hr_r=1",
+            "https://www.npmjs.com/~thisisishara",
+            "https://pypi.org/user/thisisishara/",
+            "https://www.linkedin.com/in/isharadissanayake/",
+        ]
+
+    logger.info("🗝️ Validating the embedding API token...")
+    embedding_valid, embedding_err = validate_api_token(
+        api_key_type=embedding_api_key_type, api_key=embedding_api_key
+    )
+    if not embedding_valid:
+        logger.error(embedding_err)
+        sys.exit(1)
+
+    create_knowledgebase(
+        urls=urls,
+        assistant_type=assistant_type,
+        embedding_type=embedding_type,
+        embedding_api_key=embedding_api_key,
+        knowledgebase_name=knowledgebase_name,
+    )
+
+    logger.info("✅ Knowledgebase created")

chat.py

@@ -0,0 +1,92 @@
+import argparse
+import logging
+import os
+import sys
+
+from dotenv import load_dotenv
+
+from knowledgebase import Knowledgebase
+from utils.constants import (
+    ENV_FILE,
+    ASSISTANT_TYPE_KEY,
+    AssistantType,
+    OPENAI_API_TOKEN_KEY,
+    OPENAI_KNOWLEDGEBASE_KEY,
+    HUGGINGFACEHUB_API_TOKEN_KEY,
+    HF_KNOWLEDGEBASE_KEY,
+    QUERY_TAG,
+    ANSWER_TAG,
+    SOURCES_TAG,
+    EMBEDDING_TYPE_KEY,
+    APIKeyType,
+    EmbeddingType,
+)
+from utils.llm import validate_api_token
+
+# load the .env
+load_dotenv(dotenv_path=os.path.join(os.getcwd(), ENV_FILE))
+
+logger = logging.getLogger(__name__)
+
+
+if __name__ == "__main__":
+    assistant_type = os.getenv(ASSISTANT_TYPE_KEY, AssistantType.HUGGINGFACE.value)
+    embedding_type = os.getenv(EMBEDDING_TYPE_KEY, EmbeddingType.HUGGINGFACE.value)
+
+    if assistant_type == AssistantType.OPENAI.value:
+        assistant_type = AssistantType.OPENAI
+        assistant_api_key = os.environ.get(OPENAI_API_TOKEN_KEY, None)
+        assistant_api_key_type = APIKeyType.OPENAI
+        knowledgebase_name = os.environ.get(OPENAI_KNOWLEDGEBASE_KEY, None)
+
+        if embedding_type == EmbeddingType.OPENAI.value:
+            embedding_type = EmbeddingType.OPENAI
+            embedding_api_key = assistant_api_key
+            embedding_api_key_type = APIKeyType.OPENAI
+        else:
+            embedding_type = EmbeddingType.HUGGINGFACE
+            embedding_api_key = os.getenv(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+            embedding_api_key_type = APIKeyType.HUGGINGFACE
+    else:
+        assistant_type = AssistantType.HUGGINGFACE
+        assistant_api_key = os.environ.get(HUGGINGFACEHUB_API_TOKEN_KEY, None)
+        assistant_api_key_type = APIKeyType.HUGGINGFACE
+        knowledgebase_name = os.environ.get(HF_KNOWLEDGEBASE_KEY, None)
+        embedding_type = EmbeddingType.HUGGINGFACE
+        embedding_api_key = assistant_api_key
+        embedding_api_key_type = APIKeyType.HUGGINGFACE
+
+    logger.info("🗝️ Validating the API tokens...")
+    assistant_valid, assistant_err = validate_api_token(
+        api_key_type=assistant_api_key_type, api_key=assistant_api_key
+    )
+    if not assistant_valid:
+        logger.error(assistant_err)
+        sys.exit(1)
+
+    embedding_valid, embedding_err = validate_api_token(
+        api_key_type=embedding_api_key_type, api_key=embedding_api_key
+    )
+    if not embedding_valid:
+        logger.error(embedding_err)
+        sys.exit(1)
+
+    parser = argparse.ArgumentParser(description="LLM Website QA - CLI")
+    parser.add_argument(
+        QUERY_TAG, type=str, help="Question to be asked from the assistant"
+    )
+    args = parser.parse_args()
+    query = args.query
+
+    knowledgebase = Knowledgebase(
+        assistant_type=assistant_type,
+        embedding_type=embedding_type,
+        assistant_api_key=assistant_api_key,
+        embedding_api_key=embedding_api_key,
+        knowledgebase_name=knowledgebase_name,
+    )
+    result, metadata = knowledgebase.query_knowledgebase(query=query)
+
+    print(f"\nAnswer: \n{str(result.get(ANSWER_TAG, '').strip())}")
+    print(f"\nSources: \n{str(result.get(SOURCES_TAG, '').strip())}")
+    print(f"\nCost: \n{metadata}")

knowledgebase.py

@@ -0,0 +1,203 @@
+import requests
+from bs4 import BeautifulSoup
+from langchain.callbacks import get_openai_callback
+from langchain.chains import RetrievalQAWithSourcesChain
+from langchain.embeddings import OpenAIEmbeddings, HuggingFaceHubEmbeddings
+from langchain.llms import OpenAIChat, HuggingFaceHub
+from langchain.text_splitter import CharacterTextSplitter
+from langchain.vectorstores import FAISS
+from streamlit.logger import get_logger
+
+from utils.constants import (
+    KNOWLEDGEBASE_DIR,
+    AssistantType,
+    BS_HTML_PARSER,
+    TEXT_TAG,
+    SOURCE_TAG,
+    ANSWER_TAG,
+    QUESTION_TAG,
+    HF_TEXT_GENERATION_REPO_ID,
+    EmbeddingType,
+    TOTAL_TOKENS_TAG,
+    PROMPT_TOKENS_TAG,
+    COMPLETION_TOKENS_TAG,
+    TOTAL_COST_TAG,
+    OPENAI_CHAT_COMPLETIONS_MODEL,
+)
+
+logger = get_logger(__name__)
+
+
+def extract_text_from(url_: str):
+    html = requests.get(url_).text
+    soup = BeautifulSoup(html, features=BS_HTML_PARSER)
+    text = soup.get_text()
+
+    lines = (line.strip() for line in text.splitlines())
+    return "\n".join(line for line in lines if line)
+
+
+def create_knowledgebase(
+    urls: list,
+    assistant_type: AssistantType,
+    embedding_type: EmbeddingType,
+    embedding_api_key: str,
+    knowledgebase_name: str,
+):
+    pages: list[dict] = []
+    for url in urls:
+        pages.append({TEXT_TAG: extract_text_from(url_=url), SOURCE_TAG: url})
+
+    chunk_size = 500
+    chunk_overlap = 30
+    if assistant_type == AssistantType.OPENAI:
+        # # override the default chunk configs
+        # chunk_size = 1500
+        # chunk_overlap = 200
+        if embedding_type == EmbeddingType.HUGGINGFACE:
+            embeddings = HuggingFaceHubEmbeddings(
+                huggingfacehub_api_token=embedding_api_key
+            )
+            logger.info(f"Using `hf` embeddings")
+        else:
+            embeddings = OpenAIEmbeddings(openai_api_key=embedding_api_key)
+            logger.info(f"Using `openai` embeddings")
+    else:
+        embeddings = HuggingFaceHubEmbeddings(
+            huggingfacehub_api_token=embedding_api_key
+        )
+        logger.info(
+            f"Since the assistant type is set to `hf`, `hf` embeddings are used by default."
+        )
+
+    text_splitter = CharacterTextSplitter(
+        chunk_size=chunk_size, chunk_overlap=chunk_overlap, separator="\n"
+    )
+
+    docs, metadata = [], []
+    for page in pages:
+        splits = text_splitter.split_text(page[TEXT_TAG])
+        docs.extend(splits)
+        metadata.extend([{SOURCE_TAG: page[SOURCE_TAG]}] * len(splits))
+        print(f"Split {page[SOURCE_TAG]} into {len(splits)} chunks")
+
+    vectorstore = FAISS.from_texts(texts=docs, embedding=embeddings, metadatas=metadata)
+    vectorstore.save_local(folder_path=KNOWLEDGEBASE_DIR, index_name=knowledgebase_name)
+
+
+def load_vectorstore(
+    embedding_type: EmbeddingType,
+    embedding_api_key: str,
+    knowledgebase_name: str,
+):
+    if embedding_type == EmbeddingType.OPENAI:
+        embeddings = OpenAIEmbeddings(openai_api_key=embedding_api_key)
+    else:
+        embeddings = HuggingFaceHubEmbeddings(
+            huggingfacehub_api_token=embedding_api_key
+        )
+        logger.info(
+            f"Since the assistant type is set to `hf`, `hf` embeddings are used by default."
+        )
+
+    store = FAISS.load_local(
+        folder_path=KNOWLEDGEBASE_DIR,
+        embeddings=embeddings,
+        index_name=knowledgebase_name,
+    )
+    return store
+
+
+def construct_query_response(result: dict) -> dict:
+    return {ANSWER_TAG: result}
+
+
+class Knowledgebase:
+    def __init__(
+        self,
+        assistant_type: AssistantType,
+        embedding_type: EmbeddingType,
+        assistant_api_key: str,
+        embedding_api_key: str,
+        knowledgebase_name: str,
+    ):
+        self.assistant_type = assistant_type
+        self.embedding_type = embedding_type
+        self.assistant_api_key = assistant_api_key
+        self.embedding_api_key = embedding_api_key
+        self.knowledgebase = load_vectorstore(
+            embedding_type=embedding_type,
+            embedding_api_key=embedding_api_key,
+            knowledgebase_name=knowledgebase_name,
+        )
+
+    def query_knowledgebase(self, query: str) -> tuple[dict, dict]:
+        try:
+            logger.info(
+                f"The assistant API key for the current session: ***{self.assistant_api_key[-4:]}"
+            )
+            logger.info(
+                f"The embedding API key for the current session: ***{self.embedding_api_key[-4:]}"
+            )
+
+            query = query.strip()
+            if not query:
+                return {
+                    ANSWER_TAG: "Oh snap! did you hit send accidentally, because I can't see any questions 🤔",
+                }, {}
+
+            if self.assistant_type == AssistantType.OPENAI:
+                llm = OpenAIChat(
+                    model_name=OPENAI_CHAT_COMPLETIONS_MODEL,
+                    temperature=0,
+                    verbose=True,
+                    openai_api_key=self.assistant_api_key,
+                )
+                # # this is deprecated
+                # chain = VectorDBQAWithSourcesChain.from_llm(
+                #     llm=llm,
+                #     vectorstore=self.knowledgebase,
+                #     max_tokens_limit=2048,
+                #     k=2,
+                #     reduce_k_below_max_tokens=True,
+                # )
+                chain = RetrievalQAWithSourcesChain.from_chain_type(
+                    llm=llm,
+                    chain_type="stuff",
+                    retriever=self.knowledgebase.as_retriever(),
+                    reduce_k_below_max_tokens=True,
+                    chain_type_kwargs={"verbose": True},
+                )
+            else:
+                llm = HuggingFaceHub(
+                    repo_id=HF_TEXT_GENERATION_REPO_ID,
+                    model_kwargs={"temperature": 0.5, "max_length": 64},
+                    huggingfacehub_api_token=self.assistant_api_key,
+                    verbose=True,
+                )
+                chain = RetrievalQAWithSourcesChain.from_chain_type(
+                    llm=llm,
+                    chain_type="refine",
+                    retriever=self.knowledgebase.as_retriever(),
+                    max_tokens_limit=1024,
+                    reduce_k_below_max_tokens=True,
+                    chain_type_kwargs={"verbose": True},
+                )
+
+            with get_openai_callback() as cb:
+                result = chain({QUESTION_TAG: query})
+                print(f"Total Tokens: {cb.total_tokens}")
+                print(f"Prompt Tokens: {cb.prompt_tokens}")
+                print(f"Completion Tokens: {cb.completion_tokens}")
+                print(f"Total Cost (USD): ${cb.total_cost}")
+
+                metadata = {
+                    TOTAL_TOKENS_TAG: cb.total_tokens,
+                    PROMPT_TOKENS_TAG: cb.prompt_tokens,
+                    COMPLETION_TOKENS_TAG: cb.completion_tokens,
+                    TOTAL_COST_TAG: cb.total_cost,
+                }
+            return result, metadata
+        except Exception as e:
+            logger.error(f"{e.__class__.__name__}: {e}")
+            return {ANSWER_TAG: f"{e.__class__.__name__}: {e}"}, {}

knowledgebases/kb_openai_ishara.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:540dfd6d7bda272777ee04edb931074a548217f1abbd76e772d6a36dea44c5bc
+size 40432

requirements.txt

@@ -0,0 +1,10 @@
+openai~=0.27.8
+streamlit~=1.25.0
+streamlit-chat~=0.1.1
+langchain~=0.0.238
+bs4==0.0.1
+tiktoken==0.3.0
+faiss-cpu==1.7.4
+requests~=2.31.0
+python-dotenv==1.0.0
+huggingface-hub==0.16.4

utils/constants.py

@@ -0,0 +1,65 @@
+from enum import Enum
+
+KNOWLEDGEBASE_DIR = "knowledgebases"
+BS_HTML_PARSER = "html.parser"
+OPENAI_COMPLETIONS_MODEL = "gpt-3.5-turbo"
+OPENAI_CHAT_COMPLETIONS_MODEL = "gpt-3.5-turbo"
+OPENAI_TEST_MODEL = "text-ada-001"
+ENV_FILE = ".env"
+HF_TEXT_GENERATION_REPO_ID = "google/flan-t5-xxl"
+# HF_TEXT_GENERATION_REPO_ID = "OpenAssistant/falcon-40b-sft-mix-1226"
+# HF_TEXT_GENERATION_REPO_ID = "OpenAssistant/oasst-sft-4-pythia-12b-epoch-3.5"
+TEST_PROMPT = "test"
+
+ASSISTANT_TYPE_KEY = "ASSISTANT_TYPE"
+EMBEDDING_TYPE_KEY = "EMBEDDING_TYPE"
+OPENAI_API_TOKEN_KEY = "OPENAI_API_KEY"
+HUGGINGFACEHUB_API_TOKEN_KEY = "HUGGINGFACEHUB_API_TOKEN"
+OPENAI_KNOWLEDGEBASE_KEY = "OPENAI_KNOWLEDGEBASE"
+HF_KNOWLEDGEBASE_KEY = "HF_KNOWLEDGEBASE"
+
+TEXT_TAG = "text"
+SOURCE_TAG = "source"
+SOURCES_TAG = "sources"
+ANSWER_TAG = "answer"
+QUESTION_TAG = "question"
+QUERY_TAG = "query"
+NONE_TAG = "None"
+EMPTY_TAG = ""
+MESSAGE_HISTORY_TAG = "message_history"
+USER_TAG = "user"
+ASSISTANT_TAG = "assistant"
+FROM_TAG = "from"
+IN_PROGRESS_TAG = "in_progress"
+QUERY_INPUT_TAG = "query_input"
+VALID_TOKEN_TAG = "valid_token"
+API_KEY_TAG = "api_key"
+ASSISTANT_TYPE_TAG = "assistant_type"
+TOTAL_TOKENS_TAG = "total_tokens"
+PROMPT_TOKENS_TAG = "prompt_tokens"
+COMPLETION_TOKENS_TAG = "completion_tokens"
+TOTAL_COST_TAG = "total_cost"
+
+USER_AVATAR = "https://i.imgur.com/Rf63hWt.png"
+ASSISTANT_AVATAR = "https://i.imgur.com/NQwsRn2.png"
+
+
+class AssistantType(Enum):
+    HUGGINGFACE = "hf"
+    OPENAI = "openai"
+
+
+class APIKeyType(Enum):
+    HUGGINGFACE = "hf"
+    OPENAI = "openai"
+
+
+class EmbeddingType(Enum):
+    HUGGINGFACE = "hf"
+    OPENAI = "openai"
+
+
+class StNotificationType(Enum):
+    INFO = "info"
+    WARNING = "warning"
+    ERROR = "err"

utils/llm.py

@@ -0,0 +1,42 @@
+import openai
+import streamlit as st
+from huggingface_hub import InferenceClient
+from streamlit.logger import get_logger
+
+from utils.constants import APIKeyType, TEST_PROMPT, OPENAI_TEST_MODEL
+
+logger = get_logger(__name__)
+
+
+@st.cache_data(show_spinner=False)
+def validate_api_token(api_key_type: APIKeyType, api_key: str) -> tuple[bool, str]:
+    if not api_key_type:
+        return (
+            False,
+            "API key type is not mentioned",
+        )
+
+    if not api_key:
+        return (
+            False,
+            "Invalid API key detected",
+        )
+
+    try:
+        if api_key_type == APIKeyType.OPENAI:
+            openai.Completion.create(
+                model=OPENAI_TEST_MODEL,
+                prompt=TEST_PROMPT,
+                api_key=api_key,
+                max_tokens=1,
+            )
+            logger.info("OpenAI token validated")
+        else:
+            client = InferenceClient(token=api_key)
+            client.text_generation(prompt=TEST_PROMPT, max_new_tokens=1)
+            logger.info("HuggingFace token validated")
+
+    except Exception as e:
+        logger.error(f"{e.__class__.__name__}: {e}")
+        return False, f"{e.__class__.__name__}: {e}"
+    return True, ""