app.py

import urllib.request
import fitz
import re
import numpy as np
import tensorflow_hub as hub
import openai
import gradio as gr
import os
from sklearn.neighbors import NearestNeighbors

def download_pdf(url, output_path):
    urllib.request.urlretrieve(url, output_path)
    
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('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



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(openAI_key, prompt, model):
    openai.api_key = openAI_key
    temperature=0.7
    max_tokens=1500
    top_p=1
    frequency_penalty=0
    presence_penalty=0
    message = openai.ChatCompletion.create(
        model=model,
        messages=[
            {"role": "system", "content": "You are a question generator."},
            {"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,
    )
    print(message.choices[0])
    message = message.choices[0].message['content']
    return message

  
def generate_answer(question, openAI_key, model, difficulty):
    topn_chunks = recommender(question)
    prompt = 'search results:\n\n'
    for c in topn_chunks:
        prompt += c + '\n\n'
        
    prompt += "Create 4 " + difficulty +" level content complexity multiple-choice questions with 4 options each, providing the correct answer for each question-option pair based on the search results. \n"\
              "Cite each reference using [ Page Number] notation. Citation should be done at the end of each question."\
              "Only answer what is asked. The answer should be short and concise. \n\nQuery: "
    
    prompt += f"{question}\nAnswer:"
    answer = generate_text(openAI_key, prompt, model)
    answer = answer.replace("\n", "<br>")
    print(answer)
    return answer


def question_answer(chat_history, url, file, question, difficulty):
    openAI_key = "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"
    model = "gpt-4"
    try:
        if openAI_key.strip()=='':
            return '[ERROR]: Please enter your Open AI Key. Get your key here : https://platform.openai.com/account/api-keys'
        if url.strip() == '' and file is None:
            return '[ERROR]: Both URL and PDF is empty. Provide at least one.'
        if url.strip() != '' and file is not None:
            return '[ERROR]: Both URL and PDF is provided. Please provide only one (either URL or PDF).'
        if model is None or model =='':
            return '[ERROR]: You have not selected any model. Please choose an LLM model.'
        if url.strip() != '':
            glob_url = url
            download_pdf(glob_url, 'corpus.pdf')
            load_recommender('corpus.pdf')
        else:
            old_file_name = file.name
            file_name = file.name
            # file_name = file_name[:-12] + file_name[-4:]
            # os.rename(old_file_name, file_name)
            load_recommender(file_name)
        if question.strip() == '':
            return '[ERROR]: Question field is empty'
        answer = generate_answer(question, openAI_key, model, difficulty)
        chat_history.append([question, answer])
        print(chat_history)
        return chat_history
    except openai.error.InvalidRequestError as e:
        return f'[ERROR]: Either you do not have access to GPT4 or you have exhausted your quota!'



recommender = SemanticSearch()

title = 'Skillwise GAN'
description = """ Analyze your pdf to generate questions and answers. """

with gr.Blocks(css="""#chatbot { font-size: 14px; height: 780px!important; }""") as demo:

    gr.Markdown(f'<center><h3>{title}</h3></center>')
    gr.Markdown(description)

    with gr.Row():
        
        with gr.Group():
            with gr.Accordion(""):
                url = gr.Textbox(label='Enter PDF URL here   (Example: https://arxiv.org/pdf/1706.03762.pdf )')
                gr.Markdown("<center><h4>OR<h4></center>")
                file = gr.File(label='Upload your PDF/ Research Paper / Book here', file_types=['.pdf'])

            difficulty = gr.Textbox(label='Enter difficulty level')

            question = gr.Textbox(label='Enter your question title here')
            btn = gr.Button(value='Submit')

            btn.style(full_width=True)

        with gr.Group():
            chatbot = gr.Chatbot(placeholder="Chat History", label="Chat History", lines=500, elem_id="chatbot")



    # Bind the click event of the button to the question_answer function
    btn.click(
        question_answer,
        inputs=[chatbot, url, file, question, difficulty],
        outputs=[chatbot],
    )

demo.launch(debug=True)