import gradio as gr import os # PERSISTENT DATA STORAGE: this code is used to make commits import json from huggingface_hub import hf_hub_download, file_exists, HfApi from random import shuffle from markdown import markdown # Read-only reference variables qIDs = ["mbe_46", "mbe_132", "mbe_287", "mbe_326", "mbe_334", "mbe_389", "mbe_563", "mbe_614", "mbe_642", "mbe_747", "mbe_779", "mbe_826", "mbe_845", "mbe_1042", "mbe_1134"] mode_options = ["e5", "colbert"] with open("question_data.json", "r") as f: all_questions = json.load(f) """ # State variables which interact with loading and unloading user_data = {} current_response = {} current_question = {} # read-only within gradio blocks user_id = "no_id" # Control global variables step = 0 mode = 1 def load_user_data(id): global user_data filename = id.replace('@', '_AT_').replace('.', '_DOT_') if file_exists(filename = "users/" + filename + ".json", repo_id = "ebrowne/test-data", repo_type = "dataset", token = os.getenv("HF_TOKEN")): print("File exists, downloading data.") # If the ID exists, download the file from HuggingFace path = hf_hub_download(repo_id = "ebrowne/test-data", token = os.getenv("HF_TOKEN"), filename = "users/" + filename + ".json", repo_type = "dataset") # Add their current status to user_data with open(path, "r") as f: user_data = json.load(f) else: # If the ID doesn't exist, create a format for the file and upload it to HuggingFace print("File does not exist, creating user.") shuffle(qIDs) modes = [] for i in range(len(qIDs)): temp = mode_options[:] shuffle(temp) modes.append(temp) # This is the format for a user's file on HuggingFace user_data = { "user_id": id, # original in email format, which was passed here "order": qIDs, # randomized order for each user "modes": modes, # randomized order for each user "current": 0, # user starts on first question "responses": [] # formatted as a list of current_responses } # Run the update method to upload the new JSON file to HuggingFace update_huggingface(id) def update_huggingface(id): global user_data print("Updating data...") filename = id.replace('@', '_AT_').replace('.', '_DOT_') # Create a local file that will be uploaded to HuggingFace with open(filename + ".json", "w") as f: json.dump(user_data, f) # Upload to hub (overwriting existing files...) api = HfApi() api.upload_file( path_or_fileobj=filename + ".json", path_in_repo="users/" + filename + ".json", repo_id="ebrowne/test-data", repo_type="dataset", token = os.getenv("HF_TOKEN") ) def reset_current_response(qid): global current_response current_response = { "user_id": user_id, "question_id": qid, "user_answer": 0, "e5_scores": [], # list of ten [score, score, score, score] "e5_set": [], # two values "e5_generation": [], # two values "colbert_scores": [], "colbert_set": [], "colbert_generation": [], "gold_set": [], "gold_generation": [] } with open("question_data.json", "r") as f: all_questions = json.load(f) # Loads the user's current question — this is the first question that the user has not made any progress on. def load_current_question(): global current_question q_index = user_data["current"] if q_index >= len(all_questions): print("Done") gr.Info("You've finished — thank you so much! There are no more questions. :)") current_question = {"question": "You're done! Thanks so much for your help.", "answers": ["I want to log out now.", "I want to keep answering questions.","I want to keep answering questions.", "I want to keep answering questions."], "correct_answer_index": 0, "top10_e5": ["You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!"], "generation_e5": "I don't know how to exit this code right now, so you're in an endless loop of this question until you quit.", "top10_colbert": ["You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!"], "generation_colbert": "I don't know how to exit this code right now, so you're in an endless loop of this question until you quit.", "top10_contains_gold_passage": False, "gold_passage": "GOLD PASSAGE: LOG OFF!", "gold_passage_generation": "what do you gain"} reset_current_response("USER FINISHED") else: qid = user_data["order"][q_index] current_question = all_questions[qid] reset_current_response(user_data["order"][q_index]) """ # THEMING: colors and styles (Gradio native) theme = gr.themes.Soft( primary_hue="sky", secondary_hue="sky", neutral_hue="slate", font=[gr.themes.GoogleFont('Inter'), 'ui-sans-serif', 'system-ui', 'sans-serif'], ) # BLOCKS: main user interface with gr.Blocks(theme = theme) as user_eval: # ALL VARIABLES AND LOADING # State variables which interact with loading and unloading user_data = gr.State({}) current_response = gr.State({}) current_question = gr.State({}) # read-only within gradio blocks user_id = gr.State("no_id") # Control global variables step = gr.State(0) mode = gr.State(1) # mode is always 1 for now def load_user_data(id): filename = id.replace('@', '_AT_').replace('.', '_DOT_') if file_exists(filename = "users/" + filename + ".json", repo_id = "ebrowne/test-data", repo_type = "dataset", token = os.getenv("HF_TOKEN")): print("File exists, downloading data.") # If the ID exists, download the file from HuggingFace path = hf_hub_download(repo_id = "ebrowne/test-data", token = os.getenv("HF_TOKEN"), filename = "users/" + filename + ".json", repo_type = "dataset") # Add their current status to user_data with open(path, "r") as f: return json.load(f) else: # If the ID doesn't exist, create a format for the file and upload it to HuggingFace print("File does not exist, creating user.") shuffle(qIDs) modes = [] for i in range(len(qIDs)): temp = mode_options[:] shuffle(temp) modes.append(temp) # This is the format for a user's file on HuggingFace return { "user_id": id, # original in email format, which was passed here "order": qIDs, # randomized order for each user "modes": modes, # randomized order for each user "current": 0, # user starts on first question "responses": [] # formatted as a list of current_responses } # No longer uploading after first creation: user must answer question for that. def update_huggingface(data): print("Updating data...") id = data["user_id"] print(id) print(data) filename = id.replace('@', '_AT_').replace('.', '_DOT_') # Create a local file that will be uploaded to HuggingFace with open(filename + ".json", "w") as f: json.dump(data, f) # Upload to hub (overwriting existing files...) api = HfApi() api.upload_file( path_or_fileobj=filename + ".json", path_in_repo="users/" + filename + ".json", repo_id="ebrowne/test-data", repo_type="dataset", token = os.getenv("HF_TOKEN") ) def reset_current_response(qid, uid): return { current_response : { "user_id": uid, "question_id": qid, "user_answer": 0, "e5_scores": [], # list of ten [score, score, score, score] "e5_set": [], # two values "e5_generation": [], # two values "colbert_scores": [], "colbert_set": [], "colbert_generation": [], "gold_set": [], "gold_generation": [] } } # Loads the user's current question — this is the first question that the user has not made any progress on. def load_current_question(user_data): q_index = user_data["current"] if q_index >= len(all_questions): print("Done") gr.Info("You've finished — thank you so much! There are no more questions. :)") new_response = reset_current_response("USER FINISHED", user_data["user_id"]) return {"question": "You're done! Thanks so much for your help.", "answers": ["I want to log out now.", "I want to keep answering questions.","I want to keep answering questions.", "I want to keep answering questions."], "correct_answer_index": 0, "top10_e5": ["You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!"], "generation_e5": "I don't know how to exit this code right now, so you're in an endless loop of this question until you quit.", "top10_colbert": ["You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!", "You're done; thank you!"], "generation_colbert": "I don't know how to exit this code right now, so you're in an endless loop of this question until you quit.", "top10_contains_gold_passage": False, "gold_passage": "GOLD PASSAGE: LOG OFF!", "gold_passage_generation": "what do you gain"}, new_response else: qid = user_data["order"][q_index] new_response = reset_current_response(user_data["order"][q_index], user_data["user_id"]) return all_questions[qid], new_response # Title text introducing study forward_btn = gr.Textbox("unchanged", visible = False, elem_id = "togglebutton") # used for toggling windows gr.HTML("""

Legal Retriever Evaluation Study

Score the passages based on the question and provided answer choices. Detailed instructions are found here.

""") gr.Markdown("---") # Passages and user evaluations thereof with gr.Row(equal_height = False, visible = False) as evals: # Passage text with gr.Column(scale = 2) as passages: selection = gr.HTML() """ selection = gr.HTML("

Retrieved Passage

" + current_question["top10_" + user_data["modes"][user_data["current"]][mode]][0] + "

") """ print(step) line = gr.Markdown("---") # New answers is able to render the Q and A with formatting. It doesn't change the contents of the answers. # new_answers = current_question["answers"].copy() # new_answers[current_question["correct_answer_index"]] = "**" + current_question["answers"][current_question["correct_answer_index"]] + "** ✅" passage_display = gr.Markdown() temp = """ ## Question and Answer """ # Scoring box with gr.Column(scale = 1) as scores_p: desc_0 = gr.Markdown("Does the passage describe **a legal rule or principle?**") eval_0 = gr.Radio(["Yes", "No"], label = "Legal Rule?") desc_1 = gr.Markdown("How **relevant** is this passage to the question?") eval_1 = gr.Slider(1, 5, step = 0.5, label = "Relevance", value = 3) desc_2 = gr.Markdown("How would you rate the passage's **quality** in terms of detail, clarity, and focus?") eval_2 = gr.Slider(1, 5, step = 0.5, label = "Quality", value = 3) desc_3 = gr.Markdown("How effectively does the passage **lead you to the correct answer?**") eval_3 = gr.Slider(-2, 2, step = 0.5, label = "Helpfulness", value = 0) btn_p = gr.Button("Next", interactive = False) # Users must enter in a yes/no value before moving on in the radio area def sanitize_score(rad): if rad == None: return {btn_p: gr.Button(interactive = False)} else: return {btn_p: gr.Button(interactive = True)} eval_0.change(fn = sanitize_score, inputs = [eval_0], outputs = [btn_p]) with gr.Column(scale = 1, visible = False) as scores_g: helps = gr.Markdown("Does this information **help answer** the question?") eval_helps = gr.Slider(-2, 2, step = 0.5, label = "Helpfulness", value = 0) satisfied = gr.Markdown("How **satisfied** are you by this answer?") eval_satisfied = gr.Slider(1, 5, step = 0.5, label = "User Satisfaction", value = 3) btn_g = gr.Button("Next") def next_p(e0, e1, e2, e3, cur_step, mode, cr, cq): cur_step += 1 # Add user data to the current response cr["e5_scores"].append([e0, e1, e2, e3]) # Next item if cur_step >= len(cq["top10_e5"]): # should always be 10 (DEBUG: >= to avoid out of bounds) # Step 10: all sources collapsible_string = "

Set of Passages

\n" for i, passage in enumerate(cq["top10_e5"]): collapsible_string += """ Passage """ + str(i + 1) + """

""" + passage + """

""" return { selection: gr.HTML(collapsible_string), scores_p: gr.Column(visible = False), scores_g: gr.Column(visible = True), eval_0: gr.Radio(value = None), eval_1: gr.Slider(value = 3), eval_2: gr.Slider(value = 3), eval_3: gr.Slider(value = 0), step: cur_step, # mode: 1, current_response: cr } else: return { selection: gr.HTML("""

Retrieved Passage

""" + cq["top10_e5"][cur_step] + "

"), eval_0: gr.Radio(value = None), eval_1: gr.Slider(value = 3), eval_2: gr.Slider(value = 3), eval_3: gr.Slider(value = 0), step: cur_step, # mode: 1, current_response: cr } def next_g(e_h, e_s, cur_step, mode, ud, cr, cq): cur_step += 1 if cur_step == 11: # Step 11: guaranteed to be generation # Add user data to the current response as SET evaluation, which comes before the generation # CHANGED FROM user_data["modes"][user_data["current"]] + "_set", as are all direct references to top10_e5 cr["e5_set"] = [e_h, e_s] return { selection: gr.HTML("""

Autogenerated Response

""" + markdown(cq["generation_e5"]) + "

"), eval_helps: gr.Slider(value = 0), eval_satisfied: gr.Slider(value = 3), step: cur_step, # mode: mode, user_data: ud, current_response: cr } # Steps 12 and 13 are gold passage + gold passage generation IF it is applicable if cur_step > 11: # and not current_question["top10_contains_gold_passage"] # When mode is 0 -> reset with mode = 1 """ if mode == 0: # The user just evaluated a generation for mode 0 current_response[user_data["modes"][user_data["current"]][mode] + "_generation"] = [e_h, e_s] return { selection: gr.HTML(\"""

Retrieved Passage

\""" + current_question["top10_" + user_data["modes"][user_data["current"]][1]][0] + "

"), # hard coded: first passage (0) of mode 2 (1), forward_btn: gr.Textbox("load new data"), eval_helps: gr.Slider(value = 0), eval_satisfied: gr.Slider(value = 3) } """ # When mode is 1 -> display GP and GP generation, then switch if cur_step == 12: # The user just evaluated a generation for mode 1 # CHANGE from user_data["modes"][user_data["current"]] + "_generation" cr["e5_generation"] = [e_h, e_s] return { selection: gr.HTML("""

Retrieved Passage

""" + cq["gold_passage"] + "

"), forward_btn: gr.Textbox(), eval_helps: gr.Slider(value = 0), eval_satisfied: gr.Slider(value = 3), step: cur_step, # mode: mode, user_data: ud, current_response: cr } elif cur_step == 13: # The user just evaluated the gold passage cr["gold_set"] = [e_h, e_s] return { selection: gr.HTML("""

Autogenerated Response

""" + markdown(cq["gold_passage_generation"]) + "

"), forward_btn: gr.Textbox(), eval_helps: gr.Slider(value = 0), eval_satisfied: gr.Slider(value = 3), step: cur_step, # mode: mode, user_data: ud, current_response: cr } else: # step = 14 # The user just evaluated the gold passage generation cr["gold_generation"] = [e_h, e_s] ud["current"] += 1 ud["responses"].append(cr) # adds new answers to current list of responses update_huggingface(ud) # persistence — update progress online, save answers cq, cr = load_current_question(ud) return { selection: gr.Markdown("Advancing to the next question..."), forward_btn: gr.Textbox("changed" + str(ud["current"])), # current forces event to trigger always eval_helps: gr.Slider(value = 0), eval_satisfied: gr.Slider(value = 3), step: 0, # mode: mode, user_data: ud, current_response: cr, current_question: cq } btn_p.click(fn = next_p, inputs = [eval_0, eval_1, eval_2, eval_3, step, mode, current_response, current_question], outputs = [selection, scores_p, scores_g, eval_0, eval_1, eval_2, eval_3, step, mode, current_response]) btn_g.click(fn = next_g, inputs = [eval_helps, eval_satisfied, step, mode, user_data, current_response, current_question], outputs = [selection, forward_btn, eval_helps, eval_satisfied, step, mode, user_data, current_response, current_question]) # Question and answering dynamics with gr.Row(equal_height = False, visible = False) as question: with gr.Column(): gr.Markdown("**Question**") q_text = gr.Markdown("Question") a = gr.Button("A") b = gr.Button("B") c = gr.Button("C") d = gr.Button("D") # I know this is inefficient... def answer_a(cr): # cr = current_response cr = list(cr.values())[0] cr["user_answer"] = 0 return { question: gr.Row(visible = False), evals: gr.Row(visible = True), current_response: cr } def answer_b(cr): # cr = current_response cr = list(cr.values())[0] cr["user_answer"] = 1 print(cr) return { question: gr.Row(visible = False), evals: gr.Row(visible = True), current_response: cr } def answer_c(cr): # cr = current_response cr = list(cr.values())[0] cr["user_answer"] = 2 return { question: gr.Row(visible = False), evals: gr.Row(visible = True), current_response: cr } def answer_d(cr): # cr = current_response cr = list(cr.values())[0] cr["user_answer"] = 3 return { question: gr.Row(visible = False), evals: gr.Row(visible = True), current_response: cr } a.click(fn = answer_a, inputs = [current_response], outputs = [question, evals, current_response]) b.click(fn = answer_b, inputs = [current_response], outputs = [question, evals, current_response]) c.click(fn = answer_c, inputs = [current_response], outputs = [question, evals, current_response]) d.click(fn = answer_d, inputs = [current_response], outputs = [question, evals, current_response]) def toggle(s, mode, cq): # s was step... # step = 0 if mode == 0: # temporarily disabled — will never be mode 0 mode = 1 # update mode to 1, will restart with same Q, next set of Ps print("Next set of passages for same question") return { scores_p: gr.Column(visible = True), scores_g: gr.Column(visible = False), evals: gr.Row(visible = True), question: gr.Row(visible = False), # step: step, # mode: mode } else: # reset mode to 0, will restart with new Q (set up new Q), first set of Ps # mode = 0 -> RIGHT NOW, ALWAYS 1 print("New question") new_answers = cq["answers"].copy() new_answers[cq["correct_answer_index"]] = "**" + cq["answers"][cq["correct_answer_index"]] + "** ✅" return { scores_p: gr.Column(visible = True), scores_g: gr.Column(visible = False), evals: gr.Row(visible = False), question: gr.Row(visible = True), q_text: gr.Markdown(cq["question"]), a: gr.Button(cq["answers"][0]), b: gr.Button(cq["answers"][1]), c: gr.Button(cq["answers"][2]), d: gr.Button(cq["answers"][3]), passage_display: gr.Markdown(""" ## Question and Answer *""" + cq["question"] + """* \n + """ + new_answers[0] + """ \n + """ + new_answers[1] + """ \n + """ + new_answers[2] + """ \n + """ + new_answers[3]), selection: gr.HTML("""

Retrieved Passage

""" + cq["top10_e5"][0] + "

"), step: 0, # mode: mode } # note change from "top10_" + user_data["modes"][user_data["current"]][mode]][0] forward_btn.change(fn = toggle, inputs = [step, mode, current_question], outputs = [scores_p, scores_g, evals, question, q_text, a, b, c, d, passage_display, selection, step, mode]) with gr.Row() as login: with gr.Column(): gr.Markdown("# Enter email to start") gr.Markdown("Thank you so much for your participation in our study! We're using emails to keep track of which questions you've answered and which you haven't seen. Use the same email every time to keep your progress saved. :)") email = gr.Textbox(label = "Email", placeholder = "you@email.com") s = gr.Button("Start!", interactive = False) def sanitize_login(text): if text == "": return {s: gr.Button(interactive = False)} else: return {s: gr.Button(interactive = True)} email.change(fn = sanitize_login, inputs = [email], outputs = [s]) def submit_email(email): loaded_data = load_user_data(email) # calls login, downloads data, initializes session # After loading user data, update with current question new_q, new_r = load_current_question(loaded_data) new_answers = new_q["answers"].copy() new_answers[new_q["correct_answer_index"]] = "**" + new_q["answers"][new_q["correct_answer_index"]] + "** ✅" return { question: gr.Row(visible = True), login: gr.Row(visible = False), selection: gr.HTML("""

Retrieved Passage

""" + new_q["top10_e5"][0] + "

"), # when testing both ColBERT and E5, this was "top10_" + user_data["modes"][user_data["current"]][mode]][0] passage_display: gr.Markdown(""" ## Question and Answer *""" + new_q["question"] + """* \n + """ + new_answers[0] + """ \n + """ + new_answers[1] + """ \n + """ + new_answers[2] + """ \n + """ + new_answers[3]), q_text: gr.Markdown(new_q["question"]), a: gr.Button(new_q["answers"][0]), b: gr.Button(new_q["answers"][1]), c: gr.Button(new_q["answers"][2]), d: gr.Button(new_q["answers"][3]), user_id: email, user_data: loaded_data, current_question: new_q, current_response: new_r } s.click(fn = submit_email, inputs = [email], outputs = [question, login, selection, passage_display, q_text, a, b, c, d, user_id, user_data, current_question, current_response]) # Starts on question, switches to evaluation after the user answers user_eval.launch() # https://github.com/gradio-app/gradio/issues/5791