init

app.py

@@ -0,0 +1,106 @@
+import urllib.request 
+import fitz
+import re
+import numpy as np
+import tensorflow_hub as hub
+import openai
+import gradio as gr
+import os
+import shutil
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from sklearn.neighbors import NearestNeighbors
+import huggingface_hub
+
+openai.base_url = "https://api-inference.huggingface.co/models/mistralai/Mixtral-8x7B-Instruct-v0.1/v1/"
+openai.api_key = huggingface_hub.get_token()
+
+from util import pdf_to_text, text_to_chunks, SemanticSearch
+
+recommender = SemanticSearch()
+def load_recommender(path, start_page=1):
+    global recommender
+    texts = pdf_to_text(path, start_page=start_page)
+    chunks = text_to_chunks(texts, start_page=start_page)
+    recommender.fit(chunks)
+    return 'Corpus Loaded.'
+
+
+def generate_text(prompt, model = "gpt-3.5-turbo-16k-0613"):
+    
+    model="mistralai/Mixtral-8x7B-Instruct-v0.1"
+    
+    temperature=0.7
+    max_tokens=256
+    top_p=1
+    frequency_penalty=0
+    presence_penalty=0
+    message = openai.ChatCompletion.create(
+        model=model,
+        messages=[
+            {"role": "system", "content": "You are a helpful assistant."},
+            {"role": "assistant", "content": "Here is some initial assistant message."},
+            {"role": "user", "content": prompt}
+        ],
+        temperature=.3,
+        max_tokens=max_tokens,
+        top_p=top_p,
+        frequency_penalty=frequency_penalty,
+        presence_penalty=presence_penalty,
+    ).choices[0].message['content']
+    return message
+
+def generate_answer(question):
+    topn_chunks = recommender(question)
+    prompt = 'search results:\n\n'
+    for c in topn_chunks:
+        prompt += c + '\n\n'
+        
+    prompt += "Instructions: Compose a comprehensive reply to the query using the search results given. "\
+              "Cite each reference using [ Page Number] notation. "\
+              "Only answer what is asked. The answer should be short and concise. \n\nQuery: "
+    
+    prompt += f"{question}\nAnswer:"
+    answer = generate_text(prompt)
+    return answer
+
+import google.generativeai as genai
+
+def question_anwser(chat_history, file, question):
+    suffix = Path(file.name).suffix
+    with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+        shutil.copyfile(file.name, tmp.name)
+        tmp_path = Path(tmp.name)
+
+    load_recommender(str(tmp_path))
+    answer = generate_answer(question)
+    chat_history.append([question, answer])
+    return chat_history
+
+title = 'PDF GPT '
+description = """ PDF GPT """
+
+with gr.Blocks(css="""#chatbot { font-size: 14px; min-height: 1200; }""") as demo:
+
+    gr.Markdown(f'<center><h3>{title}</h3></center>')
+    gr.Markdown(description)
+
+    with gr.Row():
+        
+        with gr.Group():
+            with gr.Accordion("URL or pdf file"):
+                file = gr.File(label='Upload your PDF/ Research Paper / Book here', file_types=['.pdf'])
+            question = gr.Textbox(label='Enter your question here')
+            btn = gr.Button(value='Submit')
+
+        with gr.Group():
+            chatbot = gr.Chatbot(label="Chat History", elem_id="chatbot")
+
+    btn.click(
+        question_anwser,
+        inputs=[chatbot, file, question],
+        outputs=[chatbot],
+        api_name="predict",
+    )
+
+demo.launch(server_name="0.0.0.0")

requirements.txt

@@ -0,0 +1,7 @@
+openai
+PyMuPDF
+numpy
+scikit-learn
+tensorflow-cpu
+tensorflow-hub
+gradio

util.py

@@ -0,0 +1,82 @@
+import re
+import fitz
+import numpy as np
+from sklearn.neighbors import NearestNeighbors
+import tensorflow_hub as hub
+
+def preprocess(text):
+    text = text.replace('\n', ' ')
+    text = re.sub('\s+', ' ', text)
+    return text
+
+
+def pdf_to_text(path, start_page=1, end_page=None):
+    doc = fitz.open(path)
+    total_pages = doc.page_count
+
+    if end_page is None:
+        end_page = total_pages
+
+    text_list = []
+
+    for i in range(start_page-1, end_page):
+        text = doc.load_page(i).get_text("text")
+        text = preprocess(text)
+        text_list.append(text)
+
+    doc.close()
+    return text_list
+
+
+def text_to_chunks(texts, word_length=150, start_page=1):
+    text_toks = [t.split(' ') for t in texts]
+    page_nums = []
+    chunks = []
+    
+    for idx, words in enumerate(text_toks):
+        for i in range(0, len(words), word_length):
+            chunk = words[i:i+word_length]
+            if (i+word_length) > len(words) and (len(chunk) < word_length) and (
+                len(text_toks) != (idx+1)):
+                text_toks[idx+1] = chunk + text_toks[idx+1]
+                continue
+            chunk = ' '.join(chunk).strip()
+            chunk = f'[Page no. {idx+start_page}]' + ' ' + '"' + chunk + '"'
+            chunks.append(chunk)
+    return chunks
+
+class SemanticSearch:
+    
+    def __init__(self):
+        # self.use = hub.load('./tf_encoder_model/')
+        self.use = hub.load('https://tfhub.dev/google/universal-sentence-encoder/4')
+        self.fitted = False
+    
+    def fit(self, data, batch=1000, n_neighbors=5):
+        self.data = data
+        self.embeddings = self.get_text_embedding(data, batch=batch)
+        n_neighbors = min(n_neighbors, len(self.embeddings))
+        self.nn = NearestNeighbors(n_neighbors=n_neighbors)
+        self.nn.fit(self.embeddings)
+        self.fitted = True
+    
+    
+    def __call__(self, text, return_data=True):
+        inp_emb = self.use([text])
+        neighbors = self.nn.kneighbors(inp_emb, return_distance=False)[0]
+        
+        if return_data:
+            return [self.data[i] for i in neighbors]
+        else:
+            return neighbors
+    
+    
+    def get_text_embedding(self, texts, batch=1000):
+        embeddings = []
+        for i in range(0, len(texts), batch):
+            text_batch = texts[i:(i+batch)]
+            emb_batch = self.use(text_batch)
+            embeddings.append(emb_batch)
+        embeddings = np.vstack(embeddings)
+        return embeddings
+