Spaces:
Sleeping
Sleeping
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,json | |
import time | |
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.replace('\n', '') | |
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 | |
new_sentences_in_passage = [i for i in new_sentences_in_passage if len(i) > 10] | |
# 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'<mark style="background: #A5D2F1">{best_match[0]}</mark><mark style="background: #FFC0CB"><font color="red"> (match score:{best_match[1]})</font></mark>') | |
# add line break | |
text = text.replace('\n', f" <br /> ") | |
# add scroll bar | |
text = f'<div style="height: 600px; overflow: auto;">{text}</div>' | |
return text | |
def process_file_online(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, self.highlighted_out, 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!', '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, self.highlighted_out, 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!', '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, self.highlighted_out, 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!', '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, self.highlighted_out, 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!', '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, self.highlighted_out, 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'): | |
path = os.path.join(path,str(time.time())) | |
os.makedirs(path, exist_ok = True) | |
path = os.path.join(path, name) | |
# self.ori_answer_df['questions'] = self.questions | |
if not hasattr(self, 'ori_answer_df'): | |
raise gr.Error("You need to submit the document first") | |
else: | |
self.ori_answer_df.to_excel(path, index = False) | |
return path | |
def download_corrected(self, path = './tmp', name = 'corrected_answer.xlsx'): | |
path = os.path.join(path,str(time.time())) | |
os.makedirs(path, exist_ok = True) | |
path = os.path.join(path, name) | |
# self.answer_df['questions'] = self.questions | |
if not hasattr(self, 'answer_df'): | |
raise gr.Error("You need to submit the document first") | |
else: | |
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" | |
def phase_df(self, df): | |
df = json.loads(df.T.to_json()) | |
res_list = [] | |
for key, item in df.items(): | |
tmp_res_list = {} | |
tep_res_list_q1 = { | |
'answer': item['Question 1'], | |
'original sentences': eval(item['Question 1_original_sentences']), | |
} | |
tep_res_list_q2 = { | |
'answer': item['Question 2'], | |
'original sentences': eval(item['Question 2_original_sentences']), | |
} | |
tep_res_list_q3 = { | |
'answer': item['Question 3'], | |
'original sentences': eval(item['Question 3_original_sentences']), | |
} | |
tep_res_list_q4 = { | |
'answer': item['intervention_1'], | |
'original sentences': eval(item['Question 4intervention_1_original_sentences']), | |
} | |
tep_res_list_q5 = { | |
'answer': item['intervention_2'], | |
'original sentences': eval(item['Question 4intervention_2_original_sentences']), | |
} | |
tep_res_list_q6 = { | |
'answer': item['Question 5'], | |
'original sentences': eval(item['Question 5_original_sentences']), | |
} | |
tmp_res_list['Question 1'] = tep_res_list_q1 | |
tmp_res_list['Question 2'] = tep_res_list_q2 | |
tmp_res_list['Question 3'] = tep_res_list_q3 | |
tmp_res_list['Question 4'] = tep_res_list_q4 | |
tmp_res_list['Question 5'] = tep_res_list_q5 | |
tmp_res_list['Question 6'] = tep_res_list_q6 | |
res_list.append(tmp_res_list) | |
return res_list | |
def process_file_offline(self, questions, answer_type, progress = gr.Progress()): | |
# record the questions | |
self.questions = questions | |
# get the text_list | |
if answer_type == 'ChatGPT_txt': | |
df = pd.read_csv('./offline_results/results_all.csv') | |
elif answer_type == 'GPT4_txt': | |
df = pd.read_csv('./offline_results/results_all_gpt4.csv') | |
elif answer_type == 'New_GPT_4_pdf': | |
df = pd.read_csv('./offline_results/results_new_pdf.csv') | |
# make the prompt | |
self.res_list = self.phase_df(df) | |
if answer_type == 'ChatGPT_txt' or answer_type == 'GPT4_txt': | |
txt_root_path = './20230808-AI coding-1st round' | |
self.filename_list = df['fn'].tolist() | |
elif answer_type == 'New_GPT_4_pdf': | |
txt_root_path = './new_pdfs' | |
self.filename_list = df['fn'].tolist() | |
self.filename_list = ['.'.join(f.split('.')[:-1]) + '.txt' for f in self.filename_list] | |
self.text_list = [] | |
for file in progress.tqdm(self.filename_list): | |
if file.split('.')[-1] == 'pdf': | |
# convert pdf to txt | |
text = self.phrase_pdf(os.path.join(txt_root_path, file)) | |
else: | |
text_path = os.path.join(txt_root_path, file) | |
with open(text_path, 'r', encoding='utf-8') as f: | |
text = f.read() | |
self.text_list.append(text) | |
# 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(self.res_list[0].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 | |
gpt_res = gpt_res['Question 1'] | |
question = self.questions[self.current_question] | |
self.answer = gpt_res['answer'] | |
self.text = self.text_list[0] | |
self.highlighted_out = gpt_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, self.highlighted_out, highlighted_out_html, self.answer, self.highlighted_out | |