Spaces:
Runtime error
Runtime error
import urllib.request | |
import fitz | |
import re | |
import openai | |
import os | |
from semantic_search import SemanticSearch | |
recommender = SemanticSearch() | |
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 | |
# converts pdf to 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 | |
# converts a text into a list of chunks | |
def text_to_chunks(texts, word_length=150, start_page=1, file_number=1): | |
filtered_texts = [''.join(char for char in text if ord(char) < 128) for text in texts] | |
text_toks = [t.split(' ') for t in filtered_texts] | |
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'[PDF no. {file_number}] [Page no. {idx+start_page}]' + ' ' + '"' + chunk + '"' | |
chunks.append(chunk) | |
return chunks | |
# merges a list of pdfs into a list of chunks and fits the recommender | |
def load_recommender(paths, start_page=1): | |
global recommender | |
chunks = [] | |
for idx, path in enumerate(paths): | |
chunks += text_to_chunks(pdf_to_text(path, start_page=start_page), start_page=start_page, file_number=idx+1) | |
recommender.fit(chunks) | |
return 'Corpus Loaded.' | |
# calls the OpenAI API to generate a response for the given query | |
def generate_text(openAI_key, prompt, model="gpt-3.5-turbo"): | |
openai.api_key = openAI_key | |
temperature=0.7 | |
max_tokens=256 | |
top_p=1 | |
frequency_penalty=0 | |
presence_penalty=0 | |
if model == "text-davinci-003": | |
completions = openai.Completion.create( | |
engine=model, | |
prompt=prompt, | |
max_tokens=max_tokens, | |
n=1, | |
stop=None, | |
temperature=temperature, | |
) | |
message = completions.choices[0].text | |
else: | |
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 | |
# constructs the prompt for the given query | |
def construct_prompt(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 [PDF Number][Page Number] notation. "\ | |
"Only answer what is asked. The answer should be short and concise. \n\nQuery: " | |
prompt += f"{question}\nAnswer:" | |
return prompt | |
# main function that is called when the user clicks the submit button, generates an answer for the query | |
def question_answer(chat_history, url, files, question, openAI_key, model): | |
try: | |
if files == None: | |
files = [] | |
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 files == []: | |
return '[ERROR]: Both URL and PDF is empty. Provide at least one.' | |
if url.strip() != '' and files is not []: | |
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: | |
print(files) | |
filenames = [] | |
for file in files: | |
old_file_name = file.name | |
file_name = file.name | |
file_name = file_name[:-12] + file_name[-4:] | |
os.rename(old_file_name, file_name) | |
filenames.append(file_name) | |
load_recommender(filenames) | |
if question.strip() == '': | |
return '[ERROR]: Question field is empty' | |
prompt = construct_prompt(question) | |
answer = generate_text(openAI_key, prompt, model) | |
chat_history.append([question, answer]) | |
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!' | |