from prompt import Prompt from openai import OpenAI from fuzzywuzzy import fuzz from fuzzywuzzy import process import gradio as gr import pandas as pd import os class Backend: def __init__(self): self.agent = OpenAI() self.prompt = Prompt() def read_file_single(self, file): # read the file if file is not None: with open(file.name, 'r') as f: text = f.read() else: raise gr.Error("You need to upload a file first") return text def phrase_pdf(self, file_path): from langchain.document_loaders import UnstructuredPDFLoader loader = UnstructuredPDFLoader(file_path, model = 'elements') file = loader.load() return file[0].page_content def read_file(self, files): # read the file text_list = [] self.filename_list = [] if files is not None: for file in files: if file.name.split('.')[-1] == 'pdf': # convert pdf to txt text = self.phrase_pdf(file.name) else: with open(file.name, 'r', encoding='utf-8') as f: text = f.read() text_list.append(text) self.filename_list.append(file.name.split('\\')[-1]) else: raise gr.Error("You need to upload a file first") return text_list def highlight_text(self, text, highlight_list): # Find the original sentences # Split the passage into sentences sentences_in_passage = text.split('.') sentences_in_passage = [i.split('\n') for i in sentences_in_passage] new_sentences_in_passage = [] for i in sentences_in_passage: new_sentences_in_passage =new_sentences_in_passage + i # hightlight the reference for hl in highlight_list: # Find the best match using fuzzy matching best_match = process.extractOne(hl, new_sentences_in_passage, scorer=fuzz.partial_ratio) text = text.replace(best_match[0], f'{best_match[0]} (match score:{best_match[1]})') # add line break text = text.replace('\n', f"
") # add scroll bar text = f'
{text}
' return text def process_file(self, file, questions, openai_key, progress = gr.Progress()): # record the questions self.questions = questions # get the text_list self.text_list = self.read_file(file) # make the prompt prompt_list = [self.prompt.get(text, questions, 'v3') for text in self.text_list] # interact with openai self.res_list = [] for prompt in progress.tqdm(prompt_list, desc = 'Generating answers...'): res = self.agent(prompt, with_history = False, temperature = 0.1, model = 'gpt-3.5-turbo-16k', api_key = openai_key) res = self.prompt.process_result(res, 'v3') self.res_list.append(res) # Use the first file as default # Use the first question for multiple questions gpt_res = self.res_list[0] self.gpt_result = gpt_res self.current_question = 0 self.totel_question = len(res.keys()) self.current_passage = 0 self.total_passages = len(self.res_list) # make a dataframe to record everything self.ori_answer_df = pd.DataFrame() self.answer_df = pd.DataFrame() for i, res in enumerate(self.res_list): tmp = pd.DataFrame(res).T tmp = tmp.reset_index() tmp = tmp.rename(columns={"index":"question_id"}) tmp['filename'] = self.filename_list[i] tmp['question'] = self.questions self.ori_answer_df = pd.concat([tmp, self.ori_answer_df]) self.answer_df = pd.concat([tmp, self.answer_df]) # default fist question res = res['Question 1'] question = self.questions[self.current_question] self.answer = res['answer'] self.text = self.text_list[0] self.highlighted_out = res['original sentences'] highlighted_out_html = self.highlight_text(self.text, self.highlighted_out) self.highlighted_out = '\n'.join(self.highlighted_out) file_name = self.filename_list[self.current_passage] return file_name, question, self.answer, highlighted_out_html, self.answer, self.highlighted_out def process_results(self, answer_correct, correct_answer, reference_correct, correct_reference): if not hasattr(self, 'clicked_correct_answer'): raise gr.Error("You need to judge whether the generated answer is correct first") if not hasattr(self, 'clicked_correct_reference'): raise gr.Error("You need to judge whether the highlighted reference is correct first") if not hasattr(self, 'answer_df'): raise gr.Error("You need to submit the document first") if self.current_question >= self.totel_question or self.current_question < 0: raise gr.Error("No more questions, please return back") # record the answer condition = (self.answer_df['question_id'] == f'Question {self.current_question + 1}' ) & \ (self.answer_df['filename'] == self.filename_list[self.current_passage]) self.answer_df.loc[condition, 'answer_correct'] = answer_correct self.answer_df.loc[condition, 'reference_correct'] = reference_correct # self.answer_df.loc[f'Question {self.current_question + 1}', 'answer_correct'] = answer_correct # self.answer_df.loc[f'Question {self.current_question + 1}', 'reference_correct'] = reference_correct if self.clicked_correct_answer == True: if hasattr(self, 'answer'): self.answer_df.loc[condition, 'correct_answer'] = self.answer else: raise gr.Error("You need to submit the document first") else: # self.answer_df.loc[f'Question {self.current_question + 1}', 'correct_answer'] = correct_answer self.answer_df.loc[condition, 'correct_answer'] = correct_answer if self.clicked_correct_reference == True: if hasattr(self, 'highlighted_out'): self.answer_df.loc[condition, 'correct_reference'] = self.highlighted_out else: raise gr.Error("You need to submit the document first") else: self.answer_df.loc[condition, 'correct_reference'] = correct_reference gr.Info('Results saved!') return "Results saved!" def process_next(self): self.current_question += 1 if hasattr(self, 'clicked_correct_answer'): del self.clicked_correct_answer if hasattr(self, 'clicked_correct_reference'): del self.clicked_correct_reference if self.current_question >= self.totel_question: # self.current_question -= 1 return "No more questions!", "No more questions!", "No more questions!", "No more questions!", 'No more questions!', 'No more questions!', 'Still need to click the button above to save the results', None, None else: res = self.gpt_result[f'Question {self.current_question + 1}'] question = self.questions[self.current_question] self.answer = res['answer'] self.highlighted_out = res['original sentences'] highlighted_out_html = self.highlight_text(self.text, self.highlighted_out) self.highlighted_out = '\n'.join(self.highlighted_out) file_name = self.filename_list[self.current_passage] return file_name, question, self.answer, highlighted_out_html, 'Please judge on the generated answer', 'Please judge on the generated answer', 'Still need to click the button above to save the results', None, None def process_last(self): self.current_question -= 1 # To make sure to correct the answer first if hasattr(self, 'clicked_correct_answer'): del self.clicked_correct_answer if hasattr(self, 'clicked_correct_reference'): del self.clicked_correct_reference # check question boundary if self.current_question < 0: # self.current_question += 1 return "No more questions!", "No more questions!", "No more questions!", "No more questions!", 'No more questions!', 'No more questions!', 'Still need to click the button above to save the results', None, None else: res = self.gpt_result[f'Question {self.current_question + 1}'] question = self.questions[self.current_question] self.answer = res['answer'] self.highlighted_out = res['original sentences'] highlighted_out_html = self.highlight_text(self.text, self.highlighted_out) self.highlighted_out = '\n'.join(self.highlighted_out) file_name = self.filename_list[self.current_passage] return file_name, question, self.answer, highlighted_out_html, 'Please judge on the generated answer', 'Please judge on the generated answer', 'Still need to click the button above to save the results', None, None def switch_next_passage(self): self.current_question = 0 # To make sure to correct the answer first if hasattr(self, 'clicked_correct_answer'): del self.clicked_correct_answer if hasattr(self, 'clicked_correct_reference'): del self.clicked_correct_reference self.current_passage += 1 if self.current_passage >= self.total_passages: # self.current_passage -= 1 return "No more passages!", "No more passages!", "No more passages!", "No more passages!", 'No more passages!', 'No more passages!', 'Still need to click the button above to save the results', None, None else: self.text = self.text_list[self.current_passage] gpt_res = self.res_list[self.current_passage] self.gpt_result = gpt_res res = self.gpt_result[f'Question {self.current_question + 1}'] question = self.questions[self.current_question] self.answer = res['answer'] self.highlighted_out = res['original sentences'] highlighted_out_html = self.highlight_text(self.text, self.highlighted_out) self.highlighted_out = '\n'.join(self.highlighted_out) file_name = self.filename_list[self.current_passage] return file_name, question, self.answer, highlighted_out_html, 'Please judge on the generated answer', 'Please judge on the generated answer', 'Still need to click the button above to save the results', None, None def switch_last_passage(self): self.current_question = 0 # To make sure to correct the answer first if hasattr(self, 'clicked_correct_answer'): del self.clicked_correct_answer if hasattr(self, 'clicked_correct_reference'): del self.clicked_correct_reference self.current_passage -= 1 if self.current_passage < 0: # self.current_passage += 1 return "No more passages!", "No more passages!", "No more passages!", "No more passages!", 'No more passages!', 'No more passages!', 'Still need to click the button above to save the results', None, None else: self.text = self.text_list[self.current_passage] gpt_res = self.res_list[self.current_passage] self.gpt_result = gpt_res res = self.gpt_result[f'Question {self.current_question + 1}'] question = self.questions[self.current_question] self.answer = res['answer'] self.highlighted_out = res['original sentences'] highlighted_out_html = self.highlight_text(self.text, self.highlighted_out) self.highlighted_out = '\n'.join(self.highlighted_out) file_name = self.filename_list[self.current_passage] return file_name, question, self.answer, highlighted_out_html, 'Please judge on the generated answer', 'Please judge on the generated answer', 'Still need to click the button above to save the results', None, None def download_answer(self, path = './tmp', name = 'answer.xlsx'): os.makedirs(path, exist_ok = True) path = os.path.join(path, name) # self.ori_answer_df['questions'] = self.questions self.ori_answer_df.to_excel(path, index = False) return path def download_corrected(self, path = './tmp', name = 'corrected_answer.xlsx'): os.makedirs(path, exist_ok = True) path = os.path.join(path, name) # self.answer_df['questions'] = self.questions self.answer_df.to_excel(path, index = False) return path def change_correct_answer(self, correctness): if correctness == "Correct": self.clicked_correct_answer = True return "No need to change" else: if hasattr(self, 'answer'): self.clicked_correct_answer = False return self.answer else: return "No answer yet, you need to submit the document first" def change_correct_reference(self, correctness): if correctness == "Correct": self.clicked_correct_reference = True return "No need to change" else: if hasattr(self, 'highlighted_out'): self.clicked_correct_reference = False return self.highlighted_out else: return "No answer yet, you need to submit the document first"