Create app.py

app.py

@@ -0,0 +1,177 @@
+from pypdf import PdfReader
+import torch
+import PyPDF2
+from io import BytesIO
+from langchain.prompts import PromptTemplate
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.embeddings import HuggingFaceEmbeddings
+from langchain.vectorstores import FAISS
+from langchain.chains import RetrievalQA
+import gradio as gr
+import time
+
+from langchain.memory import ConversationBufferMemory
+
+
+from langchain.llms.huggingface_pipeline import HuggingFacePipeline
+from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline, BitsAndBytesConfig
+from langchain.document_loaders import PyPDFDirectoryLoader
+
+CHUNK_SIZE = 1000
+# Using HuggingFaceEmbeddings with the chosen embedding model
+embeddings = HuggingFaceEmbeddings(
+    model_name="sentence-transformers/all-mpnet-base-v2",model_kwargs = {"device": "cuda"})
+
+# transformer model configuration
+quant_config = BitsAndBytesConfig(
+    bnb_4bit_compute_dtype=torch.bfloat16
+)
+
+
+def load_llm():
+
+    model_id = "Deci/DeciLM-6b-instruct"
+    tokenizer = AutoTokenizer.from_pretrained(model_id)
+    model = AutoModelForCausalLM.from_pretrained(model_id,
+                                            trust_remote_code=True,
+                                            device_map = "auto",
+                                            quantization_config=quant_config)
+    pipe = pipeline("text-generation",
+                    model=model,
+                    tokenizer=tokenizer,
+                    temperature=0,
+                    num_beams=5,
+                    no_repeat_ngram_size=4,
+                    early_stopping=True,
+                    max_new_tokens=50,
+                )
+    
+    llm = HuggingFacePipeline(pipeline=pipe)
+
+    return llm
+
+def add_text(history, text):
+    if not text:
+        raise gr.Error('Enter text')
+    history = history + [(text, '')]
+
+    return history
+
+def upload_file(file):
+    # file_path = [file.name for file in files]
+    print(type(file))
+    return file
+
+def process_file(files):
+
+   
+    
+    # loader = PyPDFLoader(file_path= file.name)
+    # document = loader.load()
+
+    pdf_text = ""
+    for file in files:
+        # pdf_stream = BytesIO(file.name.content)
+        pdf = PyPDF2.PdfReader(file.name)
+        for page in pdf.pages:
+            pdf_text += page.extract_text()
+
+    
+
+
+
+    
+
+        
+    # split into smaller chunks
+    text_splitter = RecursiveCharacterTextSplitter(chunk_size=CHUNK_SIZE, chunk_overlap=200)
+
+    splits = text_splitter.create_documents([pdf_text])
+
+    # create a FAISS vector store db 
+
+    # embedd the chunks and store in the db
+    vectorstore_db = FAISS.from_documents(splits, embeddings)
+
+    #create a custom prompt
+    custom_prompt_template = """Given the uploaded files, generate a pecise answer to the question asked by the user.
+    If you don't know the answer, just say that you don't know, don't try to make up an answer.
+    Context= {context}
+    History = {history}
+    Question= {question}
+    Helpful Answer:
+    """
+    prompt = PromptTemplate(template=custom_prompt_template, input_variables=["question", "context", "history"])
+    
+
+    # set QA chain with memory
+    qa_chain_with_memory = RetrievalQA.from_chain_type(llm=load_llm(),
+                                                       chain_type='stuff',
+                                                       return_source_documents=True,
+                                                       retriever=vectorstore_db.as_retriever(),
+                                                       chain_type_kwargs={"verbose": True,
+                                                                          "prompt": prompt,
+                                                                          "memory": ConversationBufferMemory(
+                                                                              input_key="question",
+                                                                              memory_key="history",
+                                                                              return_messages=True) })
+    
+    # get answers
+    return qa_chain_with_memory
+
+
+def generate_bot_response(history,query, btn):
+
+    if not btn:
+        raise gr.Error(message='Upload a PDF')
+    
+    qa_chain_with_memory = process_file(btn)
+
+
+    bot_response = qa_chain_with_memory({"query": query})
+
+    # return bot_response["result"]
+    for char in bot_response['result']:
+           history[-1][-1] += char
+           time.sleep(0.05)
+           yield history,''
+
+with gr.Blocks() as demo:
+    with gr.Row():
+            with gr.Row():                 
+                chatbot = gr.Chatbot(label="DeciLM-6b-instruct bot", value=[], elem_id='chatbot')
+            with gr.Row():
+                 file_output = gr.File(label="Your PDFs")
+                 with gr.Column():
+                    btn = gr.UploadButton("📁 Upload a PDF(s)", file_types=[".pdf"], file_count="multiple")
+                    
+
+    with gr.Column():        
+        with gr.Column():
+            txt = gr.Text(show_label=False, placeholder="Enter question")
+ 
+        with gr.Column():
+            submit_btn = gr.Button('Ask')
+
+
+# Event handler for uploading a PDF
+    btn.upload(fn=upload_file, inputs=[btn], outputs=[file_output])
+
+
+    submit_btn.click(
+            fn= add_text,
+            inputs=[chatbot, txt],
+            outputs=[chatbot],
+            queue=False
+        ).success(
+            fn=generate_bot_response,
+            inputs=[chatbot, txt, btn],
+            outputs=[chatbot, txt]
+        ).success(
+            fn=upload_file,
+            inputs=[btn],
+            outputs=[file_output]
+        )
+
+if __name__ == "__main__":
+    demo.launch()