Initial app setup

.ipynb_checkpoints/README-checkpoint.md

@@ -0,0 +1,12 @@
+---
+title: Chat With Pdf
+emoji: 💬
+colorFrom: yellow
+colorTo: purple
+sdk: gradio
+app_file: app.py
+pinned: false
+license: mit
+---
+
+An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).

.ipynb_checkpoints/app-checkpoint.py

@@ -0,0 +1,32 @@
+import gradio as gr
+import core
+
+def process_pdf_and_text(pdf_file_path, user_text):
+    print(f"[INFO] The pdf file is in the {pdf_file_path}")
+    if not hasattr(process_pdf_and_text,"_called"):
+        core.process_pdf(pdf_file_path)
+        process_pdf_and_text._called = True
+    
+    result = core.process_query(user_text)
+    return result
+
+def main():
+    # input components
+    pdf_input = gr.File(label="Upload PDF File")
+    text_input = gr.TextArea(label="Enter the query")
+    # output component
+    output_text = gr.TextArea()
+    
+    # app interface
+    demo = gr.Interface(
+        fn=process_pdf_and_text,
+        inputs=[pdf_input, text_input],
+        outputs=output_text,
+        title="Chat With PDF",
+        description="RAG based Chat with pdf"
+    )
+
+    demo.launch()
+    
+if __name__ == "__main__":
+    main()

.ipynb_checkpoints/core-checkpoint.py

@@ -0,0 +1,14 @@
+from embeddings import Embeddings
+from rag import RAG
+
+rag_ = None
+
+def process_pdf(file:str):
+    emb = Embeddings(file)
+    emb.save_the_embeddings()
+    global rag_
+    rag_ = RAG()
+
+def process_query(user_text:str):
+    global rag_
+    return rag_.query(user_text)

.ipynb_checkpoints/embeddings-checkpoint.py

@@ -0,0 +1,120 @@
+# This file contains all the functionalities from the pdf extraction to the embeddings
+import os
+import re
+
+from tqdm import tqdm
+from spacy.lang.en import English
+import fitz
+import pandas as pd
+
+import torch
+from sentence_transformers import SentenceTransformer
+
+class Embeddings:
+
+    def __init__(self,pdf_file_path : str):
+        self.pdf_file_path = pdf_file_path
+        self.embedding_model_name = "all-mpnet-base-v2"
+        self.device = self.get_device()
+            
+    def get_device(self) -> str:
+        """ Returns the device """
+        device = 'cuda' if torch.cuda.is_available() else 'cpu'
+        return device
+
+    def text_formatter(self,text : str) -> str:
+        """ Convert the text that contains the /n with the space"""
+        formatted_text = text.replace('\n',' ').strip()
+    
+        return formatted_text
+
+    def count_and_split_sentence(self,text : str) -> (int,list[str]):
+        """To count and split the sentences from the given text """
+        nlp = English()
+        nlp.add_pipe("sentencizer")
+
+        list_of_sentences = list(nlp(text).sents)
+        list_of_sentences = [str(sentence) for sentence in list_of_sentences]
+
+        return len(list_of_sentences),list_of_sentences
+
+    def open_pdf(self):
+        """convert the pdf into dict dtype"""
+        doc = fitz.open(self.pdf_file_path)
+        data = []
+
+        print("[INFO] Converting the pdf into dict dtype")
+        for page_number,page in tqdm(enumerate(doc)):
+            text = page.get_text()
+            text = self.text_formatter(text = text)
+
+            sentence_count,sentences = self.count_and_split_sentence(text)
+
+            data.append(
+                {
+                    "page_number" : page_number,
+                    "char_count" : len(text),
+                    "word_count" : len(text.split(" ")),
+                    "sentence_count" : sentence_count,
+                    "token_count" : len(text) / 4,
+                    "sentence" : sentences,
+                    "text" : text
+                }
+            )
+
+        return data
+
+    def split_the_array(self,array_list : list,
+                    chunk_length : int) -> list[list[str]]:
+        """Split the array of sentences into groups of chunks"""
+        return [array_list[i:i+chunk_length] for i in range(0,len(array_list),chunk_length)]
+
+    def convert_to_chunk(self,chunk_size : int = 10) -> list[dict]:
+        """ Convert the sentences into chunks """
+        pages_and_texts = self.open_pdf()
+        pages_and_chunks = []
+
+        # splitting the chunks 
+        print("[INFO] Splitting the sentences ")
+        for item in tqdm(pages_and_texts):
+            item["sentence_chunks"] = self.split_the_array(item["sentence"],chunk_size)
+            item["chunk_count"] = len(item["sentence_chunks"])
+    
+        # splitting the chunks
+        print("[INFO] Splitting into chunks ")
+        for item in tqdm(pages_and_texts):
+            for chunks in item["sentence_chunks"]:
+                d = {}
+
+                joined_sentence = "".join(chunks).replace("  "," ").strip()
+                joined_sentence = re.sub(r'\.([A-Z])', r'. \1',joined_sentence) # .A -> . A it is used to provide a space after a sentence ends
+
+                if len(joined_sentence) / 4 > 30:
+                    d["page_number"] = item["page_number"]
+                    d["sentence_chunk"] = joined_sentence
+                    # stats
+                    d["char_count"] = len(joined_sentence)
+                    d["word_count"] = len(list(joined_sentence.split(" ")))
+                    d["token_count"] = len(joined_sentence) / 4 # 4 tokens ~ 1 word
+        
+                    pages_and_chunks.append(d)
+    
+        return pages_and_chunks
+
+    def convert_to_embedds(self,chunk_size = 10) -> list[dict] :
+    
+        data = self.convert_to_chunk(chunk_size)
+        
+        embedding_model = SentenceTransformer(model_name_or_path = self.embedding_model_name,device = self.device)
+        print("[INFO] Converting into embeddings ")
+        for item in tqdm(data):
+            item["embeddings"] = embedding_model.encode(item["sentence_chunk"], convert_to_tensor = True)
+    
+        return data
+
+    def save_the_embeddings(self,filename : str = "embeddings.csv",data : list[dict] = None):
+        embedd_file = filename
+        if data is None:
+            data = self.convert_to_embedds()
+        dataframe = pd.DataFrame(data)
+        dataframe.to_csv(embedd_file,index = False)

.ipynb_checkpoints/rag-checkpoint.py

@@ -0,0 +1,91 @@
+# this python file contains all steps from the retrieval to generation code
+import torch
+import numpy as np
+import pandas as pd
+from sentence_transformers import SentenceTransformer,util
+from transformers import AutoTokenizer , AutoModelForCausalLM
+
+
+class RAG:
+
+    def __init__(self):
+        self.model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+
+        self.embedding_model_name = "all-mpnet-base-v2"
+        self.embeddings_filename = "embeddings.csv"
+        self.data_pd = pd.read_csv(self.embeddings_filename)
+        self.data_dict = pd.read_csv(self.embeddings_filename).to_dict(orient='records')
+        self.data_embeddings = self.get_embeddings()
+        
+        self.embedding_model = SentenceTransformer(model_name_or_path = self.embedding_model_name,device = self.device)
+        # Tokenizer
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_id)
+        
+        # LLM
+        self.llm_model = AutoModelForCausalLM.from_pretrained(pretrained_model_name_or_path=self.model_id,
+                                                         torch_dtype=torch.float16).to(self.device)
+
+    def get_embeddings(self) -> list:
+        """Returns the embeddings from the csv file"""
+        data_embeddings = []
+    
+        for tensor_str in self.data_pd["embeddings"]:
+            values_str = tensor_str.split("[")[1].split("]")[0]
+            values_list = [float(val) for val in values_str.split(",")]
+            tensor_result = torch.tensor(values_list)
+            data_embeddings.append(tensor_result)
+    
+        data_embeddings = torch.stack(data_embeddings).to(self.device)
+        return data_embeddings
+
+
+    def retrieve_relevant_resource(self,user_query : str , k = 5):
+        """Function to retrieve relevant resource"""
+        query_embedding = self.embedding_model.encode(user_query, convert_to_tensor = True).to(self.device)
+        dot_score = util.dot_score( a = query_embedding, b = self.data_embeddings)[0]
+        score , idx = torch.topk(dot_score,k=k)
+        return score,idx
+
+    def prompt_formatter(self,query: str, context_items: list[dict]) -> str:
+        """
+        Augments query with text-based context from context_items.
+        """
+        # Join context items into one dotted paragraph
+        context = "- " + "\n- ".join([item["sentence_chunk"] for item in context_items])
+    
+        base_prompt = """Based on the following context items, please answer the query.
+    \nNow use the following context items to answer the user query:
+    {context}
+    \nRelevant passages: <extract relevant passages from the context here>
+    User query: {query}
+    Answer:"""
+    
+        # Update base prompt with context items and query   
+        base_prompt = base_prompt.format(context=context, query=query)
+    
+        # Create prompt template for instruction-tuned model
+        dialogue_template = [
+            {"role": "user",
+            "content": base_prompt}
+        ]
+    
+        # Apply the chat template
+        prompt = self.tokenizer.apply_chat_template(conversation=dialogue_template,
+                                              tokenize=False,
+                                              add_generation_prompt=True)
+        return prompt
+
+    def query(self,user_text : str):
+        scores, indices = self.retrieve_relevant_resource(user_text)
+        context_items = [self.data_dict[i] for i in indices]
+        prompt = self.prompt_formatter(query=user_text,context_items=context_items)
+        input_ids = self.tokenizer(prompt, return_tensors="pt").to(self.device)
+        outputs = self.llm_model.generate(**input_ids,max_new_tokens=256) 
+        output_text = self.tokenizer.decode(outputs[0])
+        output_text = output_text.split("<|assistant|>")
+        output_text = output_text[1].split("</s>")[0]
+        
+        return output_text
+
+    

.ipynb_checkpoints/requirements-checkpoint.txt

@@ -0,0 +1,9 @@
+numpy
+pandas
+spacy
+tqdm
+PyMuPDF
+torch
+sentence_transformers
+transformers
+gradio

app.py

@@ -1,63 +1,32 @@
 import gradio as gr
-from huggingface_hub import InferenceClient
-
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
-
-
-def respond(
-    message,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-
-        response += token
-        yield response
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
-
-
+import core
+
+def process_pdf_and_text(pdf_file_path, user_text):
+    print(f"[INFO] The pdf file is in the {pdf_file_path}")
+    if not hasattr(process_pdf_and_text,"_called"):
+        core.process_pdf(pdf_file_path)
+        process_pdf_and_text._called = True
+    
+    result = core.process_query(user_text)
+    return result
+
+def main():
+    # input components
+    pdf_input = gr.File(label="Upload PDF File")
+    text_input = gr.TextArea(label="Enter the query")
+    # output component
+    output_text = gr.TextArea()
+    
+    # app interface
+    demo = gr.Interface(
+        fn=process_pdf_and_text,
+        inputs=[pdf_input, text_input],
+        outputs=output_text,
+        title="Chat With PDF",
+        description="RAG based Chat with pdf"
+    )
+
+    demo.launch()
+    
 if __name__ == "__main__":
-    demo.launch()
+    main()

core.py

@@ -0,0 +1,14 @@
+from embeddings import Embeddings
+from rag import RAG
+
+rag_ = None
+
+def process_pdf(file:str):
+    emb = Embeddings(file)
+    emb.save_the_embeddings()
+    global rag_
+    rag_ = RAG()
+
+def process_query(user_text:str):
+    global rag_
+    return rag_.query(user_text)

embeddings.py

@@ -0,0 +1,120 @@
+# This file contains all the functionalities from the pdf extraction to the embeddings
+import os
+import re
+
+from tqdm import tqdm
+from spacy.lang.en import English
+import fitz
+import pandas as pd
+
+import torch
+from sentence_transformers import SentenceTransformer
+
+class Embeddings:
+
+    def __init__(self,pdf_file_path : str):
+        self.pdf_file_path = pdf_file_path
+        self.embedding_model_name = "all-mpnet-base-v2"
+        self.device = self.get_device()
+            
+    def get_device(self) -> str:
+        """ Returns the device """
+        device = 'cuda' if torch.cuda.is_available() else 'cpu'
+        return device
+
+    def text_formatter(self,text : str) -> str:
+        """ Convert the text that contains the /n with the space"""
+        formatted_text = text.replace('\n',' ').strip()
+    
+        return formatted_text
+
+    def count_and_split_sentence(self,text : str) -> (int,list[str]):
+        """To count and split the sentences from the given text """
+        nlp = English()
+        nlp.add_pipe("sentencizer")
+
+        list_of_sentences = list(nlp(text).sents)
+        list_of_sentences = [str(sentence) for sentence in list_of_sentences]
+
+        return len(list_of_sentences),list_of_sentences
+
+    def open_pdf(self):
+        """convert the pdf into dict dtype"""
+        doc = fitz.open(self.pdf_file_path)
+        data = []
+
+        print("[INFO] Converting the pdf into dict dtype")
+        for page_number,page in tqdm(enumerate(doc)):
+            text = page.get_text()
+            text = self.text_formatter(text = text)
+
+            sentence_count,sentences = self.count_and_split_sentence(text)
+
+            data.append(
+                {
+                    "page_number" : page_number,
+                    "char_count" : len(text),
+                    "word_count" : len(text.split(" ")),
+                    "sentence_count" : sentence_count,
+                    "token_count" : len(text) / 4,
+                    "sentence" : sentences,
+                    "text" : text
+                }
+            )
+
+        return data
+
+    def split_the_array(self,array_list : list,
+                    chunk_length : int) -> list[list[str]]:
+        """Split the array of sentences into groups of chunks"""
+        return [array_list[i:i+chunk_length] for i in range(0,len(array_list),chunk_length)]
+
+    def convert_to_chunk(self,chunk_size : int = 10) -> list[dict]:
+        """ Convert the sentences into chunks """
+        pages_and_texts = self.open_pdf()
+        pages_and_chunks = []
+
+        # splitting the chunks 
+        print("[INFO] Splitting the sentences ")
+        for item in tqdm(pages_and_texts):
+            item["sentence_chunks"] = self.split_the_array(item["sentence"],chunk_size)
+            item["chunk_count"] = len(item["sentence_chunks"])
+    
+        # splitting the chunks
+        print("[INFO] Splitting into chunks ")
+        for item in tqdm(pages_and_texts):
+            for chunks in item["sentence_chunks"]:
+                d = {}
+
+                joined_sentence = "".join(chunks).replace("  "," ").strip()
+                joined_sentence = re.sub(r'\.([A-Z])', r'. \1',joined_sentence) # .A -> . A it is used to provide a space after a sentence ends
+
+                if len(joined_sentence) / 4 > 30:
+                    d["page_number"] = item["page_number"]
+                    d["sentence_chunk"] = joined_sentence
+                    # stats
+                    d["char_count"] = len(joined_sentence)
+                    d["word_count"] = len(list(joined_sentence.split(" ")))
+                    d["token_count"] = len(joined_sentence) / 4 # 4 tokens ~ 1 word
+        
+                    pages_and_chunks.append(d)
+    
+        return pages_and_chunks
+
+    def convert_to_embedds(self,chunk_size = 10) -> list[dict] :
+    
+        data = self.convert_to_chunk(chunk_size)
+        
+        embedding_model = SentenceTransformer(model_name_or_path = self.embedding_model_name,device = self.device)
+        print("[INFO] Converting into embeddings ")
+        for item in tqdm(data):
+            item["embeddings"] = embedding_model.encode(item["sentence_chunk"], convert_to_tensor = True)
+    
+        return data
+
+    def save_the_embeddings(self,filename : str = "embeddings.csv",data : list[dict] = None):
+        embedd_file = filename
+        if data is None:
+            data = self.convert_to_embedds()
+        dataframe = pd.DataFrame(data)
+        dataframe.to_csv(embedd_file,index = False)

rag.py

@@ -0,0 +1,91 @@
+# this python file contains all steps from the retrieval to generation code
+import torch
+import numpy as np
+import pandas as pd
+from sentence_transformers import SentenceTransformer,util
+from transformers import AutoTokenizer , AutoModelForCausalLM
+
+
+class RAG:
+
+    def __init__(self):
+        self.model_id = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+
+        self.embedding_model_name = "all-mpnet-base-v2"
+        self.embeddings_filename = "embeddings.csv"
+        self.data_pd = pd.read_csv(self.embeddings_filename)
+        self.data_dict = pd.read_csv(self.embeddings_filename).to_dict(orient='records')
+        self.data_embeddings = self.get_embeddings()
+        
+        self.embedding_model = SentenceTransformer(model_name_or_path = self.embedding_model_name,device = self.device)
+        # Tokenizer
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_id)
+        
+        # LLM
+        self.llm_model = AutoModelForCausalLM.from_pretrained(pretrained_model_name_or_path=self.model_id,
+                                                         torch_dtype=torch.float16).to(self.device)
+
+    def get_embeddings(self) -> list:
+        """Returns the embeddings from the csv file"""
+        data_embeddings = []
+    
+        for tensor_str in self.data_pd["embeddings"]:
+            values_str = tensor_str.split("[")[1].split("]")[0]
+            values_list = [float(val) for val in values_str.split(",")]
+            tensor_result = torch.tensor(values_list)
+            data_embeddings.append(tensor_result)
+    
+        data_embeddings = torch.stack(data_embeddings).to(self.device)
+        return data_embeddings
+
+
+    def retrieve_relevant_resource(self,user_query : str , k = 5):
+        """Function to retrieve relevant resource"""
+        query_embedding = self.embedding_model.encode(user_query, convert_to_tensor = True).to(self.device)
+        dot_score = util.dot_score( a = query_embedding, b = self.data_embeddings)[0]
+        score , idx = torch.topk(dot_score,k=k)
+        return score,idx
+
+    def prompt_formatter(self,query: str, context_items: list[dict]) -> str:
+        """
+        Augments query with text-based context from context_items.
+        """
+        # Join context items into one dotted paragraph
+        context = "- " + "\n- ".join([item["sentence_chunk"] for item in context_items])
+    
+        base_prompt = """Based on the following context items, please answer the query.
+    \nNow use the following context items to answer the user query:
+    {context}
+    \nRelevant passages: <extract relevant passages from the context here>
+    User query: {query}
+    Answer:"""
+    
+        # Update base prompt with context items and query   
+        base_prompt = base_prompt.format(context=context, query=query)
+    
+        # Create prompt template for instruction-tuned model
+        dialogue_template = [
+            {"role": "user",
+            "content": base_prompt}
+        ]
+    
+        # Apply the chat template
+        prompt = self.tokenizer.apply_chat_template(conversation=dialogue_template,
+                                              tokenize=False,
+                                              add_generation_prompt=True)
+        return prompt
+
+    def query(self,user_text : str):
+        scores, indices = self.retrieve_relevant_resource(user_text)
+        context_items = [self.data_dict[i] for i in indices]
+        prompt = self.prompt_formatter(query=user_text,context_items=context_items)
+        input_ids = self.tokenizer(prompt, return_tensors="pt").to(self.device)
+        outputs = self.llm_model.generate(**input_ids,max_new_tokens=256) 
+        output_text = self.tokenizer.decode(outputs[0])
+        output_text = output_text.split("<|assistant|>")
+        output_text = output_text[1].split("</s>")[0]
+        
+        return output_text
+
+    

requirements.txt

@@ -1 +1,9 @@
-huggingface_hub==0.22.2
+numpy
+pandas
+spacy
+tqdm
+PyMuPDF
+torch
+sentence_transformers
+transformers
+gradio