import os import streamlit as st from dotenv import load_dotenv from PyPDF2 import PdfReader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import UnstructuredPDFLoader from langchain.text_splitter import CharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationalRetrievalChain from htmlTemplates import css, bot_template, user_template from langchain.llms import HuggingFaceHub from langchain.vectorstores import Chroma from gpt4all import GPT4All # set this key as an environment variable os.environ["HUGGINGFACEHUB_API_TOKEN"] = st.secrets['huggingface_token'] def add_logo(): st.markdown( f""" """, unsafe_allow_html=True, ) def get_pdf_text(pdf_docs : list) -> str: text = "" for pdf in pdf_docs: pdf_reader = PdfReader(pdf) for page in pdf_reader.pages: text += page.extract_text() return text def get_pdf_pages(pdf_docs): """ Extract text from a list of PDF documents. Parameters ---------- pdf_docs : list List of PDF documents to extract text from. Returns ------- str Extracted text from all the PDF documents. """ pages = [] import tempfile with tempfile.TemporaryDirectory() as tmpdirname: for pdf in pdf_docs: pdf_path=os.path.join(tmpdirname,pdf.name) with open(pdf_path, "wb") as f: f.write(pdf.getbuffer()) pdf_loader = UnstructuredPDFLoader(pdf_path) pdf_pages = pdf_loader.load_and_split() pages=pages+pdf_pages return pages #def get_text_chunks(text:str) ->list: # text_splitter = CharacterTextSplitter( # separator="\n", chunk_size=1500, chunk_overlap=300, length_function=len # ) # chunks = text_splitter.split_text(text) # return chunks def get_text_chunks(pages): """ Split the input text into chunks. Parameters ---------- text : str The input text to be split. Returns ------- list List of text chunks. """ text_splitter = RecursiveCharacterTextSplitter( chunk_size=1024, chunk_overlap=64 ) texts = text_splitter.split_documents(pages) print(str(len(texts))) return texts #def get_vectorstore(text_chunks : list) -> FAISS: # model = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2" # encode_kwargs = { # "normalize_embeddings": True # } # set True to compute cosine similarity # embeddings = HuggingFaceBgeEmbeddings( # model_name=model, encode_kwargs=encode_kwargs, model_kwargs={"device": "cpu"} # ) # vectorstore = FAISS.from_texts(texts=text_chunks, embedding=embeddings) # return vectorstore def get_vectorstore(text_chunks): """ Generate a vector store from a list of text chunks using HuggingFace BgeEmbeddings. Parameters ---------- text_chunks : list List of text chunks to be embedded. Returns ------- FAISS A FAISS vector store containing the embeddings of the text chunks. """ MODEL_NAME = "WhereIsAI/UAE-Large-V1" MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2" hf_embeddings = HuggingFaceEmbeddings(model_name=MODEL_NAME) vectorstore = Chroma.from_documents(text_chunks, hf_embeddings, persist_directory="db") return vectorstore def get_conversation_chain(vectorstore:FAISS) -> ConversationalRetrievalChain: # llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0613") #llm = HuggingFaceHub( # repo_id="mistralai/Mixtral-8x7B-Instruct-v0.1", # #repo_id="clibrain/lince-mistral-7b-it-es", # #repo_id="TheBloke/Mixtral-8x7B-Instruct-v0.1-GGUF" # model_kwargs={"temperature": 0.5, "max_length": 2096},#1048 #) #llm = HuggingFaceHub( # repo_id="mistralai/Mixtral-8x7B-Instruct-v0.1", # model_kwargs={"temperature": 0.5, "max_new_tokens": 1024, "max_length": 1048, "top_k": 3, "trust_remote_code": True, "torch_dtype": "auto"}, #) llm = GPT4All("TheBloke/Orca-2-13B-GGUF") # llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0613") memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) conversation_chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=vectorstore.as_retriever(), memory=memory ) return conversation_chain #def handle_userinput(user_question:str): # response = st.session_state.conversation({"pregunta": user_question}) # st.session_state.chat_history = response["chat_history"] # # for i, message in enumerate(st.session_state.chat_history): # if i % 2 == 0: # st.write(" Usuario: " + message.content) # else: # st.write("🤖 ChatBot: " + message.content) def handle_userinput(user_question): """ Handle user input and generate a response using the conversational retrieval chain. Parameters ---------- user_question : str The user's question. """ response = st.session_state.conversation({"question": user_question}) st.session_state.chat_history = response["chat_history"] for i, message in enumerate(st.session_state.chat_history): if i % 2 == 0: st.write("//_^ User: " + message.content) else: st.write("🤖 ChatBot: " + message.content) def main(): st.set_page_config( page_title="Chat with a Bot that tries to answer questions about multiple PDFs", page_icon=":books:", ) #st.markdown("# Charla con TedCasBot") #st.markdown("Este Bot será tu aliado a la hora de buscar información en múltiples documentos pdf. Déjanos ayudarte! 🙏🏾") st.markdown("# Chat with TedCasBot") st.markdown("This Bot is a powerful AI tool designed to simplify the process of extracting information from PDF documents") 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.header("Charla con un Bot 🤖🦾 que te ayudará a responder preguntas sobre tus pdfs:") st.header("Chat with the TedCasBot. He will help you with any doubt you may have with your documents:") user_question = st.text_input("Ask what you need!:") if user_question: handle_userinput(user_question) with st.sidebar: add_logo() st.subheader("Your documents") pdf_docs = st.file_uploader( "Upload your documents and ress 'Process'", accept_multiple_files=True ) if st.button("Process"): with st.spinner("Processing"): # get pdf text raw_text = get_pdf_text(pdf_docs) pages = get_pdf_pages(pdf_docs) # get the text chunks #text_chunks = get_text_chunks(raw_text) text_chunks = get_text_chunks(pages) # create vector store vectorstore = get_vectorstore(text_chunks) # create conversation chain st.session_state.conversation = get_conversation_chain(vectorstore) if __name__ == "__main__": main()