chatbotlib/demo_chatbot.py

#!/usr/bin/env python

"""The chatbot demo app is the "Demo the chatbot page" in the main page"""

# the code for the page
def run_app():
    """
    Application code for the Chatbot demo
    """
    ############################
    # :: IMPORTS AND CONSTANTS #
    ############################

    import nltk
    from nltk.stem import WordNetLemmatizer
    lemmatizer = WordNetLemmatizer()

    import pickle
    import numpy as np
    import pandas as pd

    import time
    import json
    import random
    import datetime

    import streamlit as st
    from streamlit_chat import message

    from keras.models import load_model

    # Load commands
    commands = json.loads(open('commands.json').read())

    ######################################
    # :: HELPER FUNCTIONS FOR PROCESSING #
    ######################################

    def clean_up_sentence(sentence):
        """
        Tokenize and lemmatize the input sentence.
    
        Args:
            sentence (str): The input sentence to be preprocessed.
        
        Returns:
            list: The list of preprocessed words.
        """
        sentence_words = nltk.word_tokenize(sentence)
        sentence_words = [lemmatizer.lemmatize(word.lower()) for word in sentence_words]
        return sentence_words

    def bag_of_words(sentence):
        """Converts a sentence into a bag of words representation.
    
        Args:
            sentence (str): The input sentence to convert.
        
        Returns:
            numpy.ndarray: The bag of words representation as a NumPy array.
        
        """
        sentence_words = clean_up_sentence(sentence)
        bag = np.zeros(len(words), dtype=np.float32)
        indices = np.where(np.isin(words, sentence_words))
        bag[indices] = 1
        return bag

    def predict_class(sentence):
        """
        Predicts the intent based on the input sentence.

        Args:
            sentence (str): The input sentence.

        Returns:
            list: A list of dictionaries containing the predicted intents and their probabilities, sorted by probability.

        """
        p = bag_of_words(sentence)
        res = chatbot_model.predict(np.array([p]))[0]
        ERROR_THRESHOLD = 0.25

        threshold_indices = np.where(res > ERROR_THRESHOLD)[0]
        results = [{"intent": classes[i], "probability": str(res[i])} for i in threshold_indices]
        results.sort(key=lambda x: x["probability"], reverse=True)
        return results

    def get_random_response(commands_json, tag):
        """
        Retrieves a random response for the given tag from the commands JSON.

        Args:
            commands_json (dict): The JSON object containing the commands.
            tag (str): The tag associated with the intent.

        Returns:
            str: A random response for the given tag.

        """
        list_of_commands = commands_json["intents"]
        for i in list_of_commands:
            if i["tag"] == tag:
                result = random.choice(i["responses"])
                return result
        return "I'm sorry, I don't understand."


    def chatbot_response(text):
        """
        Generate a response from the chatbot based on the user input.
    
        Args:
            text (str): The user input message.
        
        Returns:
            tuple: A tuple containing the generated response from the chatbot and the execution time.
        """
        start_time = time.time()
        ints = predict_class(text)
        elapsed_time = time.time() - start_time

        if ints:
            tag = ints[0]['intent']
            res = get_random_response(commands, tag)
            return res, elapsed_time
        else:
            return "I'm sorry, I don't understand.", elapsed_time

    def get_text():
        """
        Displays a text input box and returns the user input.

        Returns:
            str: The user input.
        """
        input_text = st.text_input("You: ", "Hello, how are you?", key="input")
        return input_text 

    def get_chat_history_df():
        """
        Retrieve the chat history from the session state and create a DataFrame.

        Returns:
            pd.DataFrame: The DataFrame containing the chat history with columns 'User Input' and 'Bot Response'.
        """
        # Get the chat history from the session state
        chat_history = zip(st.session_state['past'], st.session_state['generated_responses'])
        # Convert chat history to a list
        chat_history_list = list(chat_history)
        # Create a dataframe from the chat history list
        chat_history_df = pd.DataFrame(chat_history_list, columns=['User Input', 'Bot Response'])

        return chat_history_df

    def export_chat_history(chat_history_df):
        """
        Export the chat history DataFrame to a CSV file and provide a download link.

        Args:
            chat_history_df (pd.DataFrame): The DataFrame containing the chat history.

        Returns:
            bool: True if the export is successful, False otherwise.
        """
        try:
            # Get the current datetime
            current_datetime = datetime.datetime.now().strftime("%m/%d/%Y_%I_%M_%S_%p")

            # Define the CSV file path
            chat_history_filename = f'chat_history_{current_datetime}.csv'

            # Write the dataframe to a CSV file
            chat_history_csv = chat_history_df.to_csv(index=False).encode("utf-8")
        
            # Provide a download link for the CSV file
            st.download_button(
                label="Download Chat History",
                data=chat_history_csv,
                file_name=chat_history_filename,
                mime="text/csv",
                help="Download the chat history session to a CSV file."
            )

            return True
        except Exception as e:
            st.error(f"Export failed: {str(e)}")
            return False

    def clear_session_state():
        """
        Clear the session state variables related to chat history.

        This function clears the session state variables 'generated_responses' and 'past',
        which store the generated responses and user inputs in the chat history.

        """
        # Clear the session state variables
        st.session_state['generated_responses'] = []
        st.session_state['past'] = []

    ###############
    # :: MAIN APP #
    ###############

    st.markdown("<h1 style='text-align: left;'>NLP Chatbot Demo 💬</h1>", unsafe_allow_html=True)
    st.subheader(
    """
    NLP Chatbot is a conversational chat bot. Begin by entering in a prompt below. 
    """
    )

    try:
        # Load preprocessed data and the model
        words = pickle.load(open('words.pkl', 'rb'))
        classes = pickle.load(open('classes.pkl', 'rb'))
        chatbot_model = load_model('chatbot_model.h5')
        
    except Exception as e:
            st.error("An error occurred from missing a generated file...")
            st.error(str(e))
            st.info("""
                    Double check if you followed all the steps in **Train the Chatbot Model** prior to running the demo.
                    
                    - Upload the commands.json file for processing
                    - Load the pickle data
                    - Create training data
                    - Build the model
                    
                    """)
    
    # Explanation of code using st.markdown with bullet points
    st.markdown("""
    - This chatbot app offers the following options:
    - Option 1: Clear Chat Log
    - Option 2: Preview the Chat History
    - Option 3: Export Chat History

    - To manage these options, three columns are created using `st.columns(3)`.
        - Column 1 contains a checkbox labeled 'Clear Chat Log'. Selecting it will clear the chat log. Make sure to leave the text input box empty as well.
        - Column 2 contains a checkbox labeled 'Preview the Chat History'. Selecting it will display the chat history.
        - Column 3 contains a checkbox labeled 'Export Chat History'. Selecting it will export the chat history as a CSV file.

    - Please interact with the checkboxes to perform the desired actions.
    """)
    st.write("---")
    # Create a container for the columns
    container = st.container()

    # Add the columns inside the container
    with container:
        col1, col2, col3 = st.columns(3)

    # Add a button to clear session state
    if col1.checkbox("Clear Chat Log"):
        clear_session_state()

    if col2.checkbox("Preview the Chat History"):
        chat_history_df = get_chat_history_df()
        st.write(chat_history_df)

    # Add a button to export chat history in the third column
    if col3.checkbox("Export Chat History"):
        chat_history_df = get_chat_history_df()
        export_chat_history(chat_history_df)

    # Initialize session states
    if 'generated_responses' not in st.session_state:
        st.session_state['generated_responses'] = []

    if 'execution_times' not in st.session_state:
        st.session_state['execution_times'] = []

    if 'past' not in st.session_state:
        st.session_state['past'] = []

    # Get user input
    user_input = get_text()

    if user_input:
        # Generate response
        response, exec_time = chatbot_response(user_input)

        # Update session states
        st.session_state.past.append(user_input)
        st.session_state.generated_responses.append(response)
        st.session_state.execution_times.append(exec_time)

        # Display the execution time of the response as a metric
        st.metric("Execution Time", f"{exec_time:.2f} seconds")

    if st.session_state['generated_responses']:
        # Display generated responses and user inputs
        for i in range(len(st.session_state['generated_responses']) - 1, -1, -1):
            # Unique key for each generated response widget
            message(f"Bot: {st.session_state['generated_responses'][i]}",
                    is_user=False, 
                    avatar_style='bottts-neutral',
                    seed=10, 
                    key=f"response_{i}"
            )  
            # Unique key for each user input widget
            message(f"You: {st.session_state['past'][i]}", 
                    is_user=True, 
                    avatar_style="open-peeps", 
                    seed=1, 
                    key=f"user_{i}"
            )  
            
# End of app
if __name__ == "__main__":

    run_app()