Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| from utils import * | |
| from save_data import add_new_data, get_sheet_service | |
| from instructions import * | |
| from constants import SDG_DETAILS, GPT_PROMPT_parallel, GPT_PROMPT_sequential, GPT_PROMPT_reverse_sequential | |
| class SessionManager: | |
| def __init__(self): | |
| self.sessions = [] | |
| def add_session(self, cooperate_style, task): | |
| if cooperate_style == "sequential": | |
| session = { | |
| "user_identification_code": None, | |
| "task": task, | |
| "cooperate_style": cooperate_style, | |
| "human_initial_answer": None, | |
| "ai_modificated_output": None, | |
| "evaluation": None | |
| } | |
| elif cooperate_style == "reverse_sequential": | |
| session = { | |
| "user_identification_code": None, | |
| "task": task, | |
| "cooperate_style": cooperate_style, | |
| "ai_initial_answer": None, | |
| "human_modifications": None, | |
| "final_answer": None, | |
| "evaluation": None | |
| } | |
| elif cooperate_style == "parallel": | |
| session = { | |
| "user_identification_code": None, | |
| "task": task, | |
| "cooperate_style": cooperate_style, | |
| "ai_initial_answer": None, | |
| "human_initial_answer": None, | |
| "merged_final_answer": None, | |
| "evaluation": None | |
| } | |
| self.sessions.append(session) | |
| return len(self.sessions) - 1 | |
| def update(self, index, output_content, key='final_output'): | |
| self.sessions[index][key] = output_content | |
| def get_session(self, index): | |
| return self.sessions[index] | |
| def save_session_to_sheet(self, index, service, SHEET_ID): | |
| session = self.sessions[index] | |
| new_row = list(session.values()) | |
| add_new_data(new_row, service, SHEET_ID, num_of_columns=len(new_row)) # ε¨ζεζ° | |
| def handle_create_sequential(task, human_input, session_manager, api_key, identification_code): | |
| cooperate_style = "sequential" | |
| session_index = session_manager.add_session(task=task, cooperate_style=cooperate_style) | |
| session_manager.update(session_index, human_input, 'human_initial_answer') | |
| output = merge_texts_sequential(task, human_input, api_key) | |
| session_manager.update(session_index, output, 'ai_modificated_output') | |
| session_manager.update(session_index, identification_code, 'user_identification_code') | |
| return output, session_index | |
| def handle_create_parallel(task, human_input, session_manager, api_key, identification_code): | |
| cooperate_style = "parallel" | |
| session_index = session_manager.add_session(task=task, cooperate_style=cooperate_style) | |
| ai_initial_answer = generate_ai_initial_answer(task, api_key) | |
| session_manager.update(session_index, human_input, 'human_initial_answer') | |
| session_manager.update(session_index, ai_initial_answer, 'ai_initial_answer') | |
| final_answer = merge_texts_parallel(task, human_input, ai_initial_answer, api_key) | |
| session_manager.update(session_index, final_answer, 'merged_final_answer') | |
| session_manager.update(session_index, identification_code, 'user_identification_code') | |
| return ai_initial_answer, final_answer, session_index | |
| def handle_create_reverse_sequential(task, session_manager, api_key, identification_code): | |
| cooperate_style = "reverse_sequential" | |
| session_index = session_manager.add_session(task=task, cooperate_style=cooperate_style) | |
| ai_initial_answer = generate_ai_initial_answer(task, api_key) | |
| session_manager.update(session_index, ai_initial_answer, 'ai_initial_answer') | |
| session_manager.update(session_index, identification_code, 'user_identification_code') | |
| return ai_initial_answer, session_index | |
| def handle_modify_reverse_sequential(session_index, modification_suggestions, session_manager, api_key): | |
| session = session_manager.get_session(session_index) | |
| session_manager.update(session_index, modification_suggestions, 'human_modifications') | |
| final_answer = modify_with_suggestion(session['task'], modification_suggestions, api_key) | |
| session_manager.update(session_index, final_answer, 'final_answer') | |
| return final_answer, session_index | |
| def evaluate_interaction(session_index, session_manager, api_key): | |
| session = session_manager.get_session(session_index) | |
| if session['cooperate_style'] == "sequential": | |
| evaluation = get_evaluation_with_gpt(session['task'], session['ai_modificated_output'], api_key) | |
| elif session['cooperate_style'] == "reverse_sequential": | |
| evaluation = get_evaluation_with_gpt(session['task'], session['final_answer'], api_key) | |
| elif session['cooperate_style'] == "parallel": | |
| evaluation = get_evaluation_with_gpt(session['task'], session['merged_final_answer'], api_key) | |
| session['evaluation'] = evaluation | |
| return evaluation | |
| def save_data(session_index, session_manager, service, SHEET_ID): | |
| session_manager.save_session_to_sheet(session_index, service, SHEET_ID) | |
| return "Data has been saved to Google Sheets." | |
| def login(identification_code): | |
| groups = ["A", "B", "C"] | |
| if not identification_code: | |
| return update_content(None) | |
| user_group_id = int(identification_code)//1000 | |
| if user_group_id in range(3): | |
| return update_content(groups[user_group_id]) | |
| else: | |
| return update_content(None) | |
| if __name__ == "__main__": | |
| api_key = get_api_key(local=False) | |
| service, SHEET_IDs = get_sheet_service(local=False) | |
| SHEET_ID1, SHEET_ID2, SHEET_ID3 = SHEET_IDs | |
| session_manager = SessionManager() | |
| with gr.Blocks() as app: | |
| with gr.Row(): | |
| identification_code = gr.Textbox(label="Enter your identification code") | |
| login_button = gr.Button("Login") | |
| login_status = gr.Textbox(label="Next Tasks", interactive=False) | |
| group = gr.State() | |
| with gr.Column(visible=False) as task: | |
| description = gr.Textbox(label="Task Description", | |
| value = default_task_discription(), | |
| lines=10) | |
| with gr.Accordion(label = "17 Sustainable Development Goals", | |
| open=False): | |
| gr.Markdown(SDG_DETAILS) | |
| # initialization of different group contents | |
| group_a_content = gr.Group(visible=False) | |
| group_b_content = gr.Group(visible=False) | |
| group_c_content = gr.Group(visible=False) | |
| def update_content(group): | |
| if group == "A": | |
| return gr.update(visible=True), gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), group_a_instructions() | |
| elif group == "B": | |
| return gr.update(visible=True), gr.update(visible=False), gr.update(visible=True), gr.update(visible=False), group_b_instructions() | |
| elif group == "C": | |
| return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), group_c_instructions() | |
| else: | |
| return gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), invalid_group() | |
| login_button.click(login, inputs=identification_code, outputs=[task, group_a_content, group_b_content, group_c_content, login_status]) | |
| with group_a_content: | |
| with gr.Row(): | |
| human_input = gr.Textbox(label="Human Input") | |
| with gr.Row(): | |
| submit_btn = gr.Button("Create") | |
| with gr.Row(): | |
| ai_output = gr.Textbox(label="AI modified output") | |
| session_index = gr.Number(label="Session Index", visible=False) | |
| submit_btn.click( | |
| fn=lambda task, human_input, id: handle_create_sequential(task, human_input, session_manager, api_key, id), | |
| inputs=[task, human_input, identification_code], | |
| outputs=[ai_output, session_index] | |
| ) | |
| # evaluate same for every group | |
| evaluate_btn = gr.Button("Evaluate") | |
| evaluation_result = gr.Textbox(label="Evaluation Result") | |
| evaluate_btn.click( | |
| fn=lambda session_index: evaluate_interaction(session_index, session_manager, api_key), | |
| inputs=[session_index], | |
| outputs=[evaluation_result] | |
| ) | |
| save_btn = gr.Button("Save Data") | |
| save_result = gr.Label() | |
| save_btn.click( | |
| fn=lambda session_index: save_data(session_index, session_manager, service, SHEET_ID1), | |
| inputs=[session_index], | |
| outputs=[save_result] | |
| ) | |
| with gr.Accordion(label="Appendix: AI instructions", open=False): | |
| gr.Markdown(GPT_PROMPT_sequential) | |
| with group_b_content: | |
| with gr.Row(): | |
| human_input = gr.Textbox(label="Human Input") | |
| with gr.Row(): | |
| create_btn = gr.Button("Create") | |
| with gr.Row(): | |
| ai_initial_output = gr.Textbox(label="AI Initial Output") | |
| final_output = gr.Textbox(label="Final Merged Output") | |
| session_index = gr.Number(label="Session Index", visible=False) | |
| create_btn.click( | |
| fn=lambda task, human_input, id: handle_create_parallel(task, human_input, session_manager, api_key, id), | |
| inputs=[task, human_input, identification_code], | |
| outputs=[ai_initial_output, final_output, session_index] | |
| ) | |
| evaluate_btn = gr.Button("Evaluate") | |
| evaluation_result = gr.Textbox(label="Evaluation Result") | |
| evaluate_btn.click( | |
| fn=lambda session_index: evaluate_interaction(session_index, session_manager, api_key), | |
| inputs=[session_index], | |
| outputs=[evaluation_result] | |
| ) | |
| save_btn = gr.Button("Save Data") | |
| save_result = gr.Label() | |
| save_btn.click( | |
| fn=lambda session_index: save_data(session_index, session_manager, service, SHEET_ID2), | |
| inputs=[session_index], | |
| outputs=[save_result] | |
| ) | |
| with gr.Accordion(label="Appendix: AI instructions", open=False): | |
| gr.Markdown(GPT_PROMPT_parallel) | |
| with group_c_content: | |
| with gr.Row(): | |
| create_initial_btn = gr.Button("Create") | |
| with gr.Row(): | |
| initial_answer = gr.Textbox(label="AI Initial Answer") | |
| modification_suggestions = gr.Textbox(label="Modification Suggestions") | |
| with gr.Row(): | |
| create_final_btn = gr.Button("Modify") | |
| with gr.Row(): | |
| final_answer = gr.Textbox(label="Final Answer") | |
| session_index = gr.Number(label="Session Index", visible=False) | |
| create_initial_btn.click( | |
| fn=lambda task, id: handle_create_reverse_sequential(task, session_manager, api_key, id), | |
| inputs=[task, identification_code], | |
| outputs=[initial_answer, session_index] | |
| ) | |
| create_final_btn.click( | |
| fn=lambda session_index, modification_suggestions: handle_modify_reverse_sequential(session_index, modification_suggestions, session_manager, api_key), | |
| inputs=[session_index, modification_suggestions], | |
| outputs=[final_answer, session_index] | |
| ) | |
| evaluate_btn = gr.Button("Evaluate") | |
| evaluation_result = gr.Textbox(label="Evaluation Result") | |
| evaluate_btn.click( | |
| fn=lambda session_index: evaluate_interaction(session_index, session_manager, api_key), | |
| inputs=[session_index], | |
| outputs=[evaluation_result] | |
| ) | |
| save_btn = gr.Button("Save Data") | |
| save_result = gr.Label() | |
| save_btn.click( | |
| fn=lambda session_index: save_data(session_index, session_manager, service, SHEET_ID3), | |
| inputs=[session_index], | |
| outputs=[save_result] | |
| ) | |
| with gr.Accordion(label="Appendix: AI instructions", open=False): | |
| gr.Markdown(GPT_PROMPT_reverse_sequential) | |
| app.launch(share=True) | |