import subprocess import streamlit as st from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.vectorstores import Chroma, FAISS from langchain.embeddings import FastEmbedEmbeddings # General embeddings from HuggingFace models. from langchain.memory import ConversationBufferMemory from langchain.callbacks.manager import CallbackManager from langchain.callbacks import StreamlitCallbackHandler from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler from htmlTemplates import css, bot_template, user_template from langchain.llms import LlamaCpp, OpenAI, GooglePalm # For loading transformer models. from langchain.document_loaders import PyPDFLoader, TextLoader, CSVLoader from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain import hub import tempfile import os import glob import shutil import time # TEXT LOADERS def get_pdf_text(pdf_docs): """ Purpose: A hypothetical loader for PDF files in Python. Usage: Used to extract text or other information from PDF documents. Load Function: A load_pdf function might be used to read and extract data from a PDF file. input : pdf document path returns : extracted text """ temp_dir = tempfile.TemporaryDirectory() temp_filepath = os.path.join(temp_dir.name, pdf_docs.name) with open(temp_filepath, "wb") as f: f.write(pdf_docs.getvalue()) pdf_loader = PyPDFLoader(temp_filepath) pdf_doc = pdf_loader.load() return pdf_doc def get_text_file(text_docs): """ """ temp_dir = tempfile.TemporaryDirectory() temp_filepath = os.path.join(temp_dir.name, text_docs.name) with open(temp_filepath, "wb") as f: f.write(text_docs.getvalue()) text_loader = TextLoader(temp_filepath) text_doc = text_loader.load() return text_doc def get_csv_file(csv_docs): temp_dir = tempfile.TemporaryDirectory() temp_filepath = os.path.join(temp_dir.name, csv_docs.name) with open(temp_filepath, "wb") as f: f.write(csv_docs.getvalue()) csv_loader = CSVLoader(temp_filepath) csv_doc = csv_loader.load() return csv_doc # Break the documents into chunks def get_text_chunks(documents): """ For the compute purpose we will split the document into multiple smaller chunks. IMPORTANT : If the chunks too small we will miss the context and if its too large we will have longer compute time """ text_splitter = RecursiveCharacterTextSplitter( chunk_size= 1000, chunk_overlap=200, ) text_chunks = text_splitter.split_documents(documents) return text_chunks # Save chunks to vector store def get_vectorstore(text_chunks): """ Load our vectors into chroma DB, Googles Vector Store """ vectorstore = Chroma.from_documents(documents= text_chunks, embedding= st.session_state.embeddings, persist_directory= "./vectordb/") return vectorstore # Bind the Vector DB, Large Language models and Embedding models all into one container def get_conversation_chain(vectorstore): """ This is a langchain model where we will be binding the runner to infer data from LLM """ model_path = st.session_state.model callback_manager = CallbackManager([StreamingStdOutCallbackHandler()]) if st.session_state.model == "Google_PaLm" : llm = GooglePalm(google_api_key = "Add your google palm API", max_output_tokens = 4000, callback_manager=callback_manager) elif st.session_state.model == "Open_AIGPT-3.5-Turbo": llm = OpenAI(api_key = "add your openAI Key", callback_manager = callback_manager, max_tokens= 4000 ) else: llm = LlamaCpp(model_path= model_path, n_ctx= 4000, max_tokens= 4000, f16_kv = True, callback_manager = callback_manager, verbose=True) prompt_template = """You are a personal HR Bot assistant for answering any questions about Companies policies You are given a question and a set of documents. If the user's question requires you to provide specific information from the documents, give your answer based only on the examples provided below. DON'T generate an answer that is NOT written in the provided examples. If you don't find the answer to the user's question with the examples provided to you below, answer that you didn't find the answer in the documentation and propose him to rephrase his query with more details. Use bullet points if you have to make a list, only if necessary. Use 'DOCUMENTS' as a reference point, to understand and give a consciese output in 3 or 5 sentences. QUESTION: {question} DOCUMENTS: ========= {context} ========= Finish by proposing your help for anything else. """ rag_prompt_custom = PromptTemplate.from_template(prompt_template) # prompt = hub.pull("rlm/rag-prompt") prompt = hub.pull("rlm/rag-prompt-mistral") conversation_chain = RetrievalQA.from_chain_type( llm, retriever= vectorstore.as_retriever(), chain_type_kwargs={"prompt": prompt}, ) conversation_chain.callback_manager = callback_manager conversation_chain.memory = ConversationBufferMemory() return conversation_chain # an stream lit interface to handle and save our chats def handle_userinput(): clear = False # Add clear chat button if st.button("Clear Chat history"): clear = True st.session_state.messages = [] # initialise our stream lit chat interface if "messages" not in st.session_state: st.session_state.messages = [{"role": "assistant", "content": "How can I help you?"}] for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"]) # Clear the cash memory if clear: st.session_state.conversation.memory.clear() clear = False if prompt := st.chat_input(): with st.chat_message("user"): st.markdown(prompt) # add user question to chat history st.session_state.messages.append( {"role": "user", "content": prompt}) with st.chat_message("assistant"): # set up a call back handler st_callback = StreamlitCallbackHandler(st.container()) message_holder = st.empty() full_response = "" # streamlit call back manager st.session_state.conversation.callback_manager = st_callback msg = st.session_state.conversation.run(prompt) #st.markdown(msg) for chunk in msg.split(): full_response += chunk + " " time.sleep(0.09) # add a blinking cursor to simulate typing message_holder.markdown(full_response + "✏️ ") # Display the responce message_holder.info(full_response) # add responce to session state st.session_state.messages.append({"role": "assistant", "content": full_response}) # Function to apply rounded edges using CSS def add_rounded_edges(image_path="./randstad_featuredimage.png", radius=30): st.markdown( f'', unsafe_allow_html=True,) st.image(image_path, use_column_width=True, output_format='auto') def main(): st.set_page_config(page_title="RANDSTAD", page_icon=":books:") st.write(css, unsafe_allow_html=True) if "conversation" not in st.session_state: st.session_state.conversation = None if "chat_history" not in st.session_state: st.session_state.chat_history = None st.title("💬 Randstad HR Chatbot") st.subheader("🚀 A HR powered by Generative AI") # default model st.session_state.model = "Google_PaLm" # user_question = st.text_input("Ask a question about your documents:") st.session_state.embeddings = FastEmbedEmbeddings( model_name= "BAAI/bge-base-en-v1.5", cache_dir="./embedding_model/") if len(glob.glob("./vectordb/*.sqlite3")) > 0 : vectorstore = Chroma(persist_directory="./vectordb/", embedding_function=st.session_state.embeddings) st.session_state.conversation = get_conversation_chain(vectorstore) handle_userinput() # side bar information with st.sidebar: # calling a add_rounded_edges() st.subheader("Select Your Embedding Model Model") LLM = list( glob.glob('./models/*.gguf') ) LLM.extend(["Open_AIGPT-3.5-Turbo", "Google_PaLm"]) st.session_state.model = st.selectbox( 'Models', LLM ) st.subheader("Your documents") docs = st.file_uploader( "Upload File (pdf,text,csv...) and click 'Process'", accept_multiple_files=True) if st.button("Process"): with st.spinner("Processing"): # get pdf text doc_list = [] # using the helper function below lets load our dependencies # Step 1 : Load the documents for file in docs: print('file - type : ', file.type) if file.type == 'text/plain': # file is .txt doc_list.extend(get_text_file(file)) elif file.type in ['application/octet-stream', 'application/pdf']: # file is .pdf doc_list.extend(get_pdf_text(file)) elif file.type == 'text/csv': # file is .csv doc_list.extend(get_csv_file(file)) # Step 2 : Break them into Chunks text_chunks = get_text_chunks(doc_list) # Step 3 : Create Embeddings and save them to Vector DB vectorstore = get_vectorstore(text_chunks) # Step 4 : Get our conversation chain st.session_state.conversation = get_conversation_chain(vectorstore) if __name__ == '__main__': command = 'CMAKE_ARGS="-DLLAMA_CUBLAS=on" FORCE_CMAKE=1 pip install llama-cpp-python --no-cache-dir' # Run the command using subprocess try: subprocess.run(command, shell=True, check=True) print("Command executed successfully.") except subprocess.CalledProcessError as e: print(f"Error: {e}") main()