Initial Draft

app.py

@@ -1,36 +1,25 @@
+import streamlit as st
 import os
+import requests
 import re
-import streamlit as st
-import google.generativeai as genai
-from dotenv import load_dotenv
 
-from langchain_community.document_loaders import TextLoader
 from langchain_community.document_loaders import PyPDFLoader
 from langchain.docstore.document import Document
-from langchain import PromptTemplate
-from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain.text_splitter import CharacterTextSplitter
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain.vectorstores.faiss import FAISS
 
-# Loading Google Gemini API Key from Environment Variables
+from langchain.prompts.prompt import PromptTemplate
+from langchain_community.llms import LlamaCpp
+from langchain.chains import RetrievalQA
+
+from dotenv import load_dotenv
+import google.generativeai as genai
+
+# Loading Google Gemini
 load_dotenv()
 genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
 
-# Display user Error, Warning or Success Message
-def fn_display_user_messages(lv_text, lv_type, mv_processing_message):
-    """Display user Info, Error, Warning or Success Message"""
-    
-    if lv_type == "Success":
-        with mv_processing_message.container(): 
-            st.success(lv_text)
-    elif lv_type == "Error":
-        with mv_processing_message.container(): 
-            st.error(lv_text)
-    elif lv_type == "Warning":
-        with mv_processing_message.container(): 
-            st.warning(lv_text)
-    else:
-        with mv_processing_message.container(): 
-            st.info(lv_text)
-
 # Upload pdf file into 'pdf-data' folder if it does not exist
 def fn_upload_pdf(mv_pdf_input_file, mv_processing_message):
     """Upload pdf file into 'pdf-data' folder if it does not exist"""
@@ -43,8 +32,8 @@ def fn_upload_pdf(mv_pdf_input_file, mv_processing_message):
     lv_temp_file_path = os.path.join("pdf-data",lv_file_name)
     
     if os.path.exists(lv_temp_file_path):
-        print("Step1: File already available")
-        fn_display_user_messages("Step1: File already available","Warning", mv_processing_message)
+        print("File already available")
+        fn_display_user_messages("File already available","Warning", mv_processing_message)
     else:
         with open(lv_temp_file_path,"wb") as lv_file:
             lv_file.write(mv_pdf_input_file.getbuffer())
@@ -52,77 +41,137 @@ def fn_upload_pdf(mv_pdf_input_file, mv_processing_message):
         print("Step1: PDF uploaded successfully at -> " + lv_temp_file_path)
         fn_display_user_messages("Step1: PDF uploaded successfully at -> " + lv_temp_file_path, "Info", mv_processing_message)
 
-# Extract uploaded pdf data
-def fn_extract_pdf_data(mv_pdf_input_file, mv_processing_message):
-    """Extract uploaded pdf data"""
-
-    lv_temp_pdf_file_path = os.path.join("pdf-data",mv_pdf_input_file.name)
+# Create Vector DB of uploaded PDF
+def fn_create_vector_db(mv_pdf_input_file, mv_processing_message):
+    """Create Vector DB of uploaded PDF"""
 
-    # -- Loading PDF Data
-    lv_pdf_loader = PyPDFLoader(lv_temp_pdf_file_path)
-    lv_pdf_content = lv_pdf_loader.load()
+    lv_file_name = mv_pdf_input_file.name[:-4] + ".vectorstore"
 
-    # -- Define patterns with flexibility
-    pattern1 = r"(\w+)-\n(\w+)"  # Match hyphenated words separated by a line break
-    pattern2 = r"(?<!\n\s)\n(?!\s\n)"  # Match line breaks not surrounded by whitespace
-    pattern3 = r"\n\s*\n"  # Match multiple line breaks with optional whitespace
-
-    lv_pdf_formatted_content = []
+    if not os.path.exists(os.path.join("vectordb","fiaas")):
+        os.makedirs(os.path.join("vectordb","fiaas"))
     
-    for lv_page in lv_pdf_content:
-        # -- Apply substitutions with flexibility
-        lv_pdf_page_content = re.sub(pattern1, r"\1\2", lv_page.page_content)
-        lv_pdf_page_content = re.sub(pattern2, " ", lv_pdf_page_content.strip())
-        lv_pdf_page_content = re.sub(pattern3, " ", lv_pdf_page_content)
-        lv_pdf_page_content = re.sub("\n", " ", lv_pdf_page_content)
-
-        lv_pdf_formatted_content.append(
-                                            Document( page_content= lv_pdf_page_content,
-                                                    metadata= lv_page.metadata
-                                                )
-                                       )
+    lv_temp_file_path = os.path.join(os.path.join("vectordb","fiaas"),lv_file_name)
+    lv_embeddings = HuggingFaceEmbeddings(
+                                            model_name="sentence-transformers/all-mpnet-base-v2",
+                                            model_kwargs={'device': 'cpu'}
+                                        )
     
-        # print("Page Details of "+str(lv_page.metadata)+" is - "+lv_pdf_page_content)
+    if os.path.exists(lv_temp_file_path):
+        print("VectorDB already available for uploaded file")
+        fn_display_user_messages("VectorDB already available for uploaded file","Warning", mv_processing_message)
 
-    print("Step2: PDF content extracted")
-    fn_display_user_messages("Step2: PDF content extracted", "Info", mv_processing_message)
+        lv_vector_store = FAISS.load_local(lv_temp_file_path, lv_embeddings,allow_dangerous_deserialization=True)
+        return lv_vector_store
+    else:
+        lv_temp_pdf_file_path = os.path.join("pdf-data",mv_pdf_input_file.name)
+        
+        # -- Loading PDF Data
+        lv_pdf_loader = PyPDFLoader(lv_temp_pdf_file_path)
+        lv_pdf_content = lv_pdf_loader.load()
+
+        # -- Define patterns with flexibility
+        pattern1 = r"(\w+)-\n(\w+)"  # Match hyphenated words separated by a line break
+        pattern2 = r"(?<!\n\s)\n(?!\s\n)"  # Match line breaks not surrounded by whitespace
+        pattern3 = r"\n\s*\n"  # Match multiple line breaks with optional whitespace
+
+        lv_pdf_formatted_content = []
+        for lv_page in lv_pdf_content:
+            # -- Apply substitutions with flexibility
+            lv_pdf_page_content = re.sub(pattern1, r"\1\2", lv_page.page_content)
+            lv_pdf_page_content = re.sub(pattern2, " ", lv_pdf_page_content.strip())
+            lv_pdf_page_content = re.sub(pattern3, " ", lv_pdf_page_content)
+            lv_pdf_page_content = re.sub("\n", " ", lv_pdf_page_content)
+
+            lv_pdf_formatted_content.append(Document( page_content= lv_pdf_page_content,
+                                                      metadata= lv_page.metadata)
+                                           )
+            
+            # print("Page Details of "+str(lv_page.metadata)+" is - "+lv_pdf_page_content)
+
+        print("Step2: PDF content extracted")
+        fn_display_user_messages("Step2: PDF content extracted", "Info", mv_processing_message)
+
+        # -- Chunking PDF Data
+        lv_text_splitter = CharacterTextSplitter(
+                                                    separator="\n",
+                                                    chunk_size=300,
+                                                    chunk_overlap=30,
+                                                    length_function=len
+                                                )
+        lv_pdf_chunk_documents = lv_text_splitter.split_documents(lv_pdf_formatted_content)
+        print("Step3: PDF content chucked and document object created")
+        fn_display_user_messages("Step3: PDF content chucked and document object created", "Info", mv_processing_message)
 
-    return lv_pdf_formatted_content
+        # -- Creating FIASS Vector Store
+        lv_vector_store = FAISS.from_documents(lv_pdf_chunk_documents, lv_embeddings)
+        print("Step4: Vector store created")
+        fn_display_user_messages("Step4: Vector store created", "Info", mv_processing_message)
+        lv_vector_store.save_local(lv_temp_file_path)
 
-# Load PDF data as Text File
-def fn_process_pf_data(mv_pdf_input_file, mv_processing_message):
-    """Load PDF data as Text File"""
+        return lv_vector_store
 
-    # -- Create txt folder inside vectordb folder if it does not exist
-    if not os.path.exists(os.path.join("vectordb","txt")):
-        os.makedirs(os.path.join("vectordb","txt"))
+# Display user Error, Warning or Success Message
+def fn_display_user_messages(lv_text, lv_type, mv_processing_message):
+    """Display user Info, Error, Warning or Success Message"""
+    
+    if lv_type == "Success":
+        with mv_processing_message.container(): 
+            st.success(lv_text)
+    elif lv_type == "Error":
+        with mv_processing_message.container(): 
+            st.error(lv_text)
+    elif lv_type == "Warning":
+        with mv_processing_message.container(): 
+            st.warning(lv_text)
+    else:
+        with mv_processing_message.container(): 
+            st.info(lv_text)
 
-    lv_file_name = mv_pdf_input_file.name[:-4] + ".txt"
-    lv_temp_file_path = os.path.join(os.path.join("vectordb","txt"),lv_file_name)
+# Download TheBloke Models
+def fn_download_llm_models(mv_selected_model, mv_processing_message):
+    """Download TheBloke Models"""
 
-    if os.path.isfile(lv_temp_file_path):
-        print("Step2: Processed file details exists")
-        fn_display_user_messages("Step2: Processed file details exists", "Warning", mv_processing_message)
-    else:
-        lv_pdf_formatted_content = fn_extract_pdf_data(mv_pdf_input_file, mv_processing_message)
-        lv_text_data = ""
-        
-        for lv_page in lv_pdf_formatted_content:
-            # print(lv_page.page_content)
-            lv_text_data = lv_text_data + lv_page.page_content
-        
-        # print(lv_text_data)
-        f = open(lv_temp_file_path, "w")
-        f.write(lv_text_data)
-        f.close()
+    lv_download_url = ""
 
-# Return QA Response
-def fn_generate_QnA_response(mv_user_question, mv_pdf_input_file, mv_processing_message):
-    """Returns QA Response"""
+    print("Downloading TheBloke of "+mv_selected_model)
+    fn_display_user_messages("Downloading TheBloke of "+mv_selected_model, "Info", mv_processing_message)
 
-    print("Step4: Generating LLM response")
-    fn_display_user_messages("Step4: Generating LLM response","Info", mv_processing_message)
+    if mv_selected_model == 'microsoft/phi-2':
+        lv_download_url = "https://huggingface.co/TheBloke/phi-2-GGUF/resolve/main/phi-2.Q2_K.gguf"
+    elif mv_selected_model == 'google/gemma-2b':
+        lv_download_url = "https://huggingface.co/MaziyarPanahi/gemma-2b-it-GGUF/resolve/main/gemma-2b-it.Q2_K.gguf"
+    elif mv_selected_model == 'mistralai/Mistral-7B-Instruct-v0.2':
+        lv_download_url = "https://huggingface.co/TheBloke/Mistral-7B-Instruct-v0.2-GGUF/resolve/main/mistral-7b-instruct-v0.2.Q2_K.gguf"
+
+    if not os.path.exists("model"):
+        os.makedirs("model")
+    
+    lv_filename = os.path.basename(lv_download_url)
+    lv_temp_file_path = os.path.join("model",lv_filename)
 
+    if os.path.exists(lv_temp_file_path):
+        print("Model already available")
+        fn_display_user_messages("Model already available","Warning", mv_processing_message)
+    else:
+        lv_response = requests.get(lv_download_url, stream=True)
+        if lv_response.status_code == 200:
+            with open(lv_temp_file_path, 'wb') as f:
+                for chunk in lv_response.iter_content(chunk_size=1024):
+                    if chunk:
+                        f.write(chunk)
+            
+            print("Download completed")
+            fn_display_user_messages("Model download completed","Info", mv_processing_message)
+        else:
+            print(f"Model download completed {response.status_code}")
+            fn_display_user_messages(f"Model download completed {response.status_code}","Error", mv_processing_message)
+
+# Function return QA Response using Vector Store
+def fn_generate_QnA_response(mv_selected_model, mv_user_question, lv_vector_store, mv_processing_message):
+    """Returns QA Response using Vector Store"""
+
+    lv_model_path = ""
+    lv_model_type = ""
     lv_template   = """Instruction:
                     You are an AI assistant for answering questions about the provided context.
                     You are given the following extracted parts of a long document and a question. Provide a detailed answer.
@@ -132,96 +181,194 @@ def fn_generate_QnA_response(mv_user_question, mv_pdf_input_file, mv_processing_
                     =======
                     Question: {question}
                     Output:\n"""
-    
     lv_qa_prompt = PromptTemplate(
-                                    template=lv_template,
-                                    input_variables=["question", "context"]
-                                 )    
-    # lv_model = ChatGoogleGenerativeAI(model="gemini-pro",
-    #              temperature=0.7, top_p=0.85)
-
-    lv_model = genai.GenerativeModel('gemini-pro')
-
-    lv_file_name = mv_pdf_input_file.name[:-4] + ".txt"
-    lv_temp_file_path = os.path.join(os.path.join("vectordb","txt"),lv_file_name)
-    lv_text_loader = TextLoader(lv_temp_file_path)
-    lv_pdf_formatted_content = lv_text_loader.load()
-    lv_text_data = ""    
-    for lv_page in lv_pdf_formatted_content:
-        lv_text_data = lv_text_data + lv_page.page_content
+                                template=lv_template,
+                                input_variables=["question", "context"]
+                              )
+
+    if mv_selected_model == 'microsoft/phi-2':
+        lv_model_path = "model/phi-2.Q2_K.gguf"
+        lv_model_type = "pi"
+    elif mv_selected_model == 'google/gemma-2b':
+        lv_model_path = "model/gemma-2b-it.Q2_K.gguf"
+        lv_model_type = "gemma"
+    elif mv_selected_model == 'mistralai/Mistral-7B-Instruct-v0.2':
+        lv_model_path = "model/mistral-7b-instruct-v0.2.Q2_K.gguf"
+        lv_model_type = "mistral"
+        
+    print("Step4: Generating LLM response")
+    fn_display_user_messages("Step4: Generating LLM response","Info", mv_processing_message)
+
+    lv_model = LlamaCpp(
+                            model_path=lv_model_path,
+                            temperature=0.00,
+                            max_tokens=2048,
+                            top_p=1,
+                            n_ctx=2048,
+                            verbose=False
+                       )
+    lv_vector_search_result = lv_vector_store.similarity_search(mv_user_question, k=2)
+    # print("Vector Search Result - ")
+    # print(lv_vector_search_result)
+
+    # -- Creating formatted document result
+    lv_document_context = ""
+    lv_count = 0
+    for lv_result in lv_vector_search_result:
+        print("Concatenating Result of page - " + str(lv_count) + " with content of document page no - "+str(lv_result.metadata["page"]))
+        lv_document_context += lv_result.page_content 
+        lv_count += 1
+
+    # print("Formatted Document Search Result - ")
+    # print(lv_document_context)
 
     lv_qa_formatted_prompt = lv_qa_prompt.format(  
-                                                    question=mv_user_question,
-                                                    context=lv_text_data
-                                                )
-    
-    # lv_llm_response = lv_model.invoke(lv_qa_formatted_prompt).content
-    lv_llm_response = lv_model.generate_content(lv_qa_formatted_prompt).text
+                                            question=mv_user_question,
+                                            context=lv_document_context
+                                         )
+    print("Formatted Prompt - " + lv_qa_formatted_prompt)
+
+    lv_llm_response = lv_model(lv_qa_formatted_prompt)
+    # print("LLM Response" +lv_llm_response)
 
     print("Step5: LLM response generated")
     fn_display_user_messages("Step5: LLM response generated","Info", mv_processing_message)
 
     return lv_llm_response
 
+# Function return API based QA Response using Vector Store
+def fn_generate_API_QnA_response(mv_selected_model, mv_user_question, lv_vector_store, mv_processing_message):
+    """Returns QA Response using Vector Store"""
+
+    lv_template   = """Instruction:
+                    You are an AI assistant for answering questions about the provided context.
+                    You are given the following extracted parts of a long document and a question. Provide a detailed answer.
+                    If you don't know the answer, just say "Hmm, I'm not sure." Don't try to make up an answer.
+                    =======
+                    {context}
+                    =======
+                    Question: {question}
+                    Output:\n"""
+    lv_qa_prompt = PromptTemplate(
+                                template=lv_template,
+                                input_variables=["question", "context"]
+                              )
+
+    lv_vector_search_result = lv_vector_store.similarity_search(mv_user_question, k=2)
+    # print("Vector Search Result - ")
+    # print(lv_vector_search_result)
+
+    # -- Creating formatted document result
+    lv_document_context = ""
+    lv_count = 0
+    for lv_result in lv_vector_search_result:
+        # print("Concatenating Result of page - " + str(lv_count) + " with content of document page no - "+str(lv_result.metadata["page"]))
+        lv_document_context += lv_result.page_content 
+        lv_count += 1
 
-# Main Program
+    print("Formatted Document Search Result - ")
+    print(lv_document_context)
+
+    lv_qa_formatted_prompt = lv_qa_prompt.format(  
+                                            question=mv_user_question,
+                                            context=lv_document_context
+                                         )
+    
+    if mv_selected_model == 'Google Gemini-pro':
+        lv_model = genai.GenerativeModel('gemini-pro')
+
+        print("Step4: Generating LLM response")
+        fn_display_user_messages("Step4: Generating LLM response","Info", mv_processing_message)
+    
+        lv_llm_response = lv_model.generate_content(lv_qa_formatted_prompt).text
+    
+        print("Step5: LLM response generated")
+        fn_display_user_messages("Step5: LLM response generated","Info", mv_processing_message)
+    
+        return lv_llm_response
+    
+# Main Function
 def main():
+    
     # -- Streamlit Settings
-    st.set_page_config("Chat With Your Product User Manual")
-    st.header("Chat With Your Product User Manual💁")
-    st.text("")
+    st.set_page_config(layout='wide')
+    col1, col2, col3 = st.columns(3)
+    col2.title("Chat with PDF")
     st.text("")
+
+    # -- Initialize chat history
+    if "messages" not in st.session_state:
+        st.session_state.messages = []
+
+    # -- Display Supported Models
+    col1, col2, col3 = st.columns(3)
+    mv_selected_model = col3.selectbox('Select Model',
+                                        [
+                                            'microsoft/phi-2',
+                                            'google/gemma-2b',
+                                            'mistralai/Mistral-7B-Instruct-v0.2',
+                                            'Google Gemini-pro'
+                                        ]
+                                      )
+    
+    # -- Display Supported Vector Stores
+    col1, col2, col3 = st.columns(3)
+    mv_selected_vector_db = col3.selectbox('Select Vector DB', ['FAISS'])
     st.text("")
 
+    # -- Reading PDF File
+    col1, col2, col3 = st.columns(3)
+    mv_pdf_input_file = col2.file_uploader("Choose a PDF file:", type=["pdf"])
+
     # -- Display Processing Details
-    mv_processing_message = st.empty()
     st.text("")
+    col1, col2, col3 = st.columns(3)
+    mv_processing_message = col2.empty()
     st.text("")
 
-    # -- Setting Chat History
-    if "messages" not in st.session_state:
-        st.session_state["messages"] = []
+    # -- Downloading Model Files
+    if mv_selected_model != "Google Gemini-pro":
+        fn_download_llm_models(mv_selected_model, mv_processing_message)
+    else:
+        print("Call Google API")
+
+    # -- Processing PDF
+    if (mv_pdf_input_file is not None):
+
+        # -- Upload PDF
+        fn_upload_pdf(mv_pdf_input_file, mv_processing_message)
+
+        # -- Create Vector Index
+        lv_vector_store = fn_create_vector_db(mv_pdf_input_file, mv_processing_message)
 
-    # -- Read User Manuals for Q&A
-    with st.sidebar:
-        mv_pdf_input_file = st.file_uploader("Choose a UM PDF file:", type=["pdf"])
+        # -- Perform RAG
+        col1, col2, col3 = st.columns(3)
         st.text("")
+        lv_chat_history = col2.chat_message
         st.text("")
+
+        if mv_user_question := col2.chat_input("Chat on PDF Data"):
+           # -- Add user message to chat history
+           st.session_state.messages.append({"role": "user", "content": mv_user_question})
+
+           # -- Generating LLM response
+           if mv_selected_model != "Google Gemini-pro":
+               lv_response = fn_generate_QnA_response(mv_selected_model, mv_user_question, lv_vector_store, mv_processing_message)
+           else:
+               lv_response = fn_generate_API_QnA_response(mv_selected_model, mv_user_question, lv_vector_store, mv_processing_message)
+           
+           # -- Adding assistant response to chat history
+           st.session_state.messages.append({"role": "assistant", "content": lv_response})
         
-        # -- Process Uploaded User Manual PDF
-        col1, col2, col3 = st.columns(3)
-        if col1.button("Submit"):
-            if mv_pdf_input_file is not None:
-                fn_upload_pdf(mv_pdf_input_file, mv_processing_message)
-                lv_pdf_page_content = fn_process_pf_data(mv_pdf_input_file, mv_processing_message)
-            else:
-                fn_display_user_messages("Upload PDF file before clicking on Submit", "Error", mv_processing_message)
-
-        # -- Clear Chat History
-        if col2.button("Reset"):
-            st.session_state["messages"] = []
-
-    # -- Creating Chat Details
-    mv_user_question = st.chat_input("Pass your input here")
-
-    # -- Recording Chat Input and Generating Response
-    if mv_user_question:
-        # -- Saving User Input
-        st.session_state.messages.append({"role": "user", "content": mv_user_question})
-
-        # -- Generating LLM Response
-        lv_response = fn_generate_QnA_response(mv_user_question, mv_pdf_input_file, mv_processing_message)
-
-        # -- Saving LLM Response
-        st.session_state.messages.append(
-            {"role": "agent", "content": lv_response}
-        )
-
-        # -- Display chat messages from history
-        for message in st.session_state.messages:
-            with st.chat_message(message["role"]):
-                st.markdown(message["content"])
-
-# Loading Main
-if __name__ == "__main__":
+           # -- Display chat messages from history on app rerun
+           for message in st.session_state.messages:
+               with lv_chat_history(message["role"]):
+                   st.markdown(message["content"])
+
+        # -- Validate Data
+
+        # -- Get Web Response
+
+# Calling Main Function
+if __name__ == '__main__':
     main()