import urllib.request import fitz import re import numpy as np import tensorflow as tf 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=200, 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(api_type, engine, openAI_key, openAI_base, openAI_API_version, temperature, prompt): openai.api_type = api_type openai.api_base = openAI_base openai.api_version = openAI_API_version openai.api_key = openAI_key completions = openai.ChatCompletion.create(engine=engine, max_tokens=2048, n=1, stop=None, temperature=temperature, messages=[{ "role": "user", "content": prompt }]) print(completions) message = completions['choices'][0]['message']['content'] return message def generate_answer(api_type, engine, openAI_key, openAI_base, openAI_API_version, temperature, user_prompt, question): topn_chunks = recommender(question) prompt = "" prompt += 'search results:\n\n' for c in topn_chunks: prompt += c + '\n\n' prompt += f"Instructions: {user_prompt}"\ f"\n\nQuery: {question}\nAnswer: " print(prompt) answer = generate_text(api_type, engine, openAI_key, openAI_base, openAI_API_version, temperature, prompt) return answer def question_answer(api_type, engine, openAI_key, openAI_base, openAI_API_version, url, file, temperature, user_prompt, question): if openAI_key.strip() == '': return '[ERROR]: Please enter you Open AI Key. Get your key here : https://platform.openai.com/account/api-keys' if url.strip() == '' and file == None: return '[ERROR]: Both URL and PDF is empty. Provide at least one.' if url.strip() != '' and file != None: return '[ERROR]: Both URL and PDF is provided. Please provide only one (eiter URL or PDF).' 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' return generate_answer(api_type, engine, openAI_key, openAI_base, openAI_API_version, temperature, user_prompt, question) def chatbot_respond(api_type, engine, openAI_key, openAI_base, openAI_API_version, url, file, temperature, user_prompt, question, chatbot): bot_message = question_answer(api_type, engine, openAI_key, openAI_base, openAI_API_version, url, file, temperature, user_prompt, question) chatbot.append((question, bot_message)) return "", chatbot recommender = SemanticSearch() title = 'HKU PDF GPT Azure' description = """ PDF GPT allows you to chat with your PDF file using Universal Sentence Encoder and Open AI. It gives hallucination free response than other tools as the embeddings are better than OpenAI. The returned response can even cite the page number in square brackets([]) where the information is located,adding credibility to the responses and helping to locate pertinent information quickly. """ def save_settings(): return gr.Tabs.update(selected=1) with gr.Blocks() as demo: gr.Markdown(f'