import gradio as gr
import random
from datasets import load_dataset, Features, Value
import json
import os
import uuid
from huggingface_hub import HfApi
import time
# # Sample dataset with unique 10-digit IDs
# qa_dataset = {
# "1234567890": {
# "question": "What is the capital of France?",
# "choices": ["A. Berlin", "B. Madrid", "C. Paris", "D. Lisbon"],
# "answer": "C. Paris"
# },
# "0987654321": {
# "question": "What is the largest planet in our solar system?",
# "choices": ["A. Earth", "B. Jupiter", "C. Saturn", "D. Mars", "E. Venus"],
# "answer": "B. Jupiter"
# },
# # Add more questions with unique IDs as needed
# }
truth_data = load_dataset("commonsense-index-dev/commonsense-candidates", "iter7-0520", split="train")
LAST_LOG_UPDATE = time.time()
logs = None
qa_dataset = {}
for item in truth_data:
qa_dataset[item["id"]] = {
"question": item["task"],
"choices": item["choices"],
"answer": item["answer"]
}
if "metadata" in item:
qa_dataset[item["id"]]["reason"] = item["metadata"].get("reasoning", "N/A")
def update_logs():
global LAST_LOG_UPDATE
global logs
# if logs is None or time.time() - LAST_LOG_UPDATE > 3600:
if logs is None:
# update logs for every 60 minutes
logs = load_dataset("commonsense-index-dev/DemoFeedback", split="train")
LAST_LOG_UPDATE = time.time()
update_logs()
def get_random_question(user_name="Anonymous"):
global logs
update_logs()
# if user_name == "":
# user_name = "Anonymous"
# question_id = random.choice(list(qa_dataset.keys()))
# else:
# logs = load_dataset("commonsense-index-dev/DemoFeedback", split="train")
feedback_counts = {qid: 0 for qid in qa_dataset.keys()}
user_seen_data = set()
for item in logs:
feedback_counts[item["question_id"]] += 1
if True or item["user_name"] == user_name: # TODO: remove this when data is almost all annotated
user_seen_data.add(item["question_id"])
# sample a question that has the least feedback, and if there are multiple, sample randomly
min_feedback = min(feedback_counts.values())
question_ids = [k for k, v in feedback_counts.items() if v == min_feedback]
question_ids = list(set(question_ids) - user_seen_data)
question_id = random.choice(question_ids)
question_data = qa_dataset[question_id]
reasoning = question_data["reason"]
return question_id, question_data["question"], question_data["choices"], reasoning
def get_question_by_id(question_id):
if question_id in qa_dataset:
question_data = qa_dataset[question_id]
return question_id, question_data["question"], question_data["choices"]
else:
return None, "Invalid question ID", []
def check_answer(question_id, choice, reasoning):
correct_answer = qa_dataset[question_id]["answer"]
text = ""
if choice[3:] == correct_answer:
text += "### ✅ Correct!"
text += "\n### Reasoning: " + reasoning
else:
text += "### ❌ Incorrect. Try again!"
return text
def load_question(question_id=None, user_name="Anonymous"):
question_id, question, choices, reasoning = get_random_question(user_name)
question = f"---\n#### QID: {question_id}\n## {question} \n---"
choices_markdown = "\n".join(choices)
return question_id, question, choices_markdown, \
gr.update(value="", visible=True), reasoning, \
gr.update(value="", visible=True), \
gr.update(value="Submit your feedback! 🚀", interactive=True)
def show_buttons(choices_markdown):
choices = choices_markdown.split("\n")
visibility = [gr.update(visible=False)] * 10
for i in range(len(choices)):
# generate ABCDEFGHIJ labels
choices[i] = chr(65 + i) + ") " + choices[i]
visibility[i] = gr.update(visible=True, value=choices[i])
return visibility
def submit_feedback(question_id, user_reason, user_revision, example_quality, user_name_text):
if "N/A" in question_id or "N/A" in example_quality:
# send a message to the user to sample an example and select a choice first
return {
submit_button: {"interactive": True, "__type__": "update", "value": "Submit your feedback! 🚀 Please sample an example and select a choice!"},
}
# create a jsonl file and upload it to hf
if user_name_text == "":
user_name_text = "Anonymous"
feedback_item = {
"question_id": question_id,
"user_name": user_name_text,
"user_reason": user_reason,
"revision": user_revision,
"example_quality": example_quality,
}
jsonl_str = json.dumps(feedback_item)
api = HfApi()
token = os.getenv("HF_TOKEN")
if token is None:
raise ValueError("Hugging Face token not found. Ensure the HF_TOKEN environment variable is set.")
# Generate a random filename using UUID
filename = f"{uuid.uuid4()}.json"
# Define the repository
repo_id = "commonsense-index-dev/DemoFeedback"
# Upload the json_str as a file directly to the specified path in your dataset repository
api.upload_file(
token=token,
repo_id=repo_id,
repo_type="dataset",
path_or_fileobj=jsonl_str.encode("utf-8"), # Convert string to bytes
path_in_repo=filename,
commit_message=f"{user_name_text}'s feedback on {question_id}",
)
return {
submit_button: {"interactive": False, "__type__": "update", "value": "Submitted! ✅ \n Please sample the next one."}
}
def refresh_feedback(question_id):
return gr.update(value="", visible=True), gr.update(value="", visible=True), gr.update(value="", visible=True), gr.update(value="", visible=True)
with gr.Blocks() as app:
gr.Markdown("# Commonsense Index Data Viewer")
with gr.Row():
# question_id_input = gr.Textbox(label="Enter Question ID", placeholder="leave empty for random sampling")
random_button = gr.Button("🎲 Click here to randomly sample an example")
question_display = gr.Markdown(visible=True)
choices_markdown = gr.Markdown(visible=False)
choice_buttons = [gr.Button(visible=False) for _ in range(10)]
result_display = gr.Markdown(visible=True)
reasoning_display = gr.Markdown(visible=False)
question_id = gr.Textbox(label="Question ID:", interactive=False, visible=False)
with gr.Row():
with gr.Column(scale=2):
reason_textbox = gr.Textbox(label="Reason", placeholder="Please talk why the correct answer is correct and why the others are wrong. If you think this is a bad example, please explain too.", type="text", elem_classes="", max_lines=5, lines=3, show_copy_button=False, visible=True, scale=4, interactive=True)
revision_textbox = gr.Textbox(label="Revision", placeholder="Please suggest a revision to the question.", type="text", elem_classes="", max_lines=5, lines=3, show_copy_button=False, visible=True, scale=4, interactive=True)
with gr.Column():
example_quality = gr.Radio(label="Quality", choices=["Good", "Bad"], interactive=True, visible=True)
user_name = gr.Textbox(label="Your username", placeholder="Your username", type="text", elem_classes="", max_lines=1, show_copy_button=False, visible=True, interactive=True, show_label=False)
submit_button = gr.Button("Submit your feedback! 🚀", elem_classes="btn_boderline", visible=True, interactive=True)
random_button.click(fn=load_question, inputs=[user_name], outputs=[question_id, question_display, choices_markdown, result_display, reasoning_display, example_quality, submit_button])
choices_markdown.change(fn=show_buttons, inputs=choices_markdown, outputs=choice_buttons)
question_id.change(fn=refresh_feedback, inputs=[question_id], outputs=[reason_textbox, revision_textbox, example_quality])
submit_button.click(fn=submit_feedback, inputs=[question_id, reason_textbox, revision_textbox, example_quality, user_name], outputs=[submit_button])
for i, button in enumerate(choice_buttons):
button.click(fn=check_answer, inputs=[question_id, button, reasoning_display], outputs=result_display)
app.launch()