init

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+simulation_arena/
+.env

README.md

@@ -1,13 +1,14 @@
 ---
-title: Adminsim Arena
-emoji: 👀
-colorFrom: gray
-colorTo: gray
+title: Adminsim-arena
+emoji: 🏟️
+colorFrom: yellow
+colorTo: red
 sdk: gradio
 sdk_version: 5.49.1
 app_file: app.py
 pinned: false
 license: apache-2.0
+short_description: ' Model simulation arena'
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,191 @@
+import os
+import random
+import datetime
+import gradio as gr
+from typing import Tuple, Optional
+
+from utils import log
+from utils.common import upload_to_github
+from utils.postprocess import make_dialog_dict
+
+
+
+def sample_pair(dialog_dict: dict) -> Tuple[str, str, str, str]:
+    """
+    Sample two different models and one dialog from each.
+
+    Args:
+        dialog_dict (dict): Dictionary of dialogs per model.
+
+    Returns:
+        Tuple[str, str, str, str]: (model1 name, dialog1, model2 name, dialog2)
+    """
+    model1, model2 = random.sample(list(dialog_dict.keys()), 2)
+    dialog1 = random.choice(dialog_dict[model1])
+    dialog2 = random.choice(dialog_dict[model2])
+    return model1, dialog1, model2, dialog2
+
+
+
+def new_comparison(dialog_dict: dict) -> Tuple[str, str, str, str, gr.Row, gr.Row, gr.Row]:
+    """
+    Generate a new comparison pair and make the arena row visible.
+
+    Args:
+        dialog_dict (dict): Dictionary of dialogs per model.
+
+    Returns:
+        Tuple[str, str, str, str, gr.Row]: 
+            (model1 name, dialog1, model2 name, dialog2, 
+            arena row visibility update, submit button visibility update, new comparison button visibility update)
+    """
+    m1, d1, m2, d2 = sample_pair(dialog_dict)
+    return m1, d1, m2, d2, gr.update(visible=True), gr.update(visible=True), gr.update(visible=False)
+
+
+
+def record_vote(choice: str, 
+                m1: str,
+                m2: str,
+                d1: str,
+                d2: str,
+                dialog_dict: dict,
+                is_dev: bool,
+                result_file_path: Optional[str] = None) -> Tuple[str, str, str, str, gr.Row, str, gr.Button]:
+    """
+    Record the user's vote and generate a new comparison.
+
+    Args:
+        choice (str): User's choice ('A' or 'B').
+        m1 (str): Model 1 name.
+        m2 (str): Model 2 name.
+        d1 (str): Dialog 1.
+        d2 (str): Dialog 2.
+        dialog_dict (dict): Dictionary of dialogs per model.
+        is_dev (bool): Flag indicating if in development mode.
+        result_file_path (str, optional): Path to save the results.
+
+    Returns:
+        Tuple[str, str, str, str, gr.Row, str, gr.Button]:
+            (New model1 name, new dialog1, new model2 name, new dialog2, arena row visibility update, message)
+    """
+    log(f"{choice=} | A: {m1} vs B: {m2}")
+
+    # Vote data saving
+    if not is_dev:
+        with open(result_file_path, "a") as f:
+            if choice == "A":
+                f.write(f"1\t0\t{m1}\t{m2}\n")
+            else:
+                f.write(f"0\t1\t{m1}\t{m2}\n")
+
+    # New pair sampling
+    new_m1, new_d1, new_m2, new_d2 = sample_pair(dialog_dict)
+    
+    return new_m1, new_d1, new_m2, new_d2, gr.update(visible=True), "✅ Vote recorded! Next comparison ready!"
+
+
+
+def save_data(path: str) -> str:
+    """
+    Save the human evaluation data.
+
+    Args:
+        path (str): Path to save the results.
+    
+    Returns:
+        str: Message of the result submission.
+    """
+    try:
+        upload_to_github(path, open(path).read())
+        return "✅ Results successfully uploaded to GitHub!"
+    except Exception as e:
+        return f"❌ Upload failed: {e}"
+
+
+
+def set_result_path() -> str:
+    """
+    Set the result saving path.
+
+    Returns:
+        str: Path to save the results.
+    """
+    timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+    unique_id = os.urandom(4).hex()
+    result_file = os.path.join("simulation_arena", f"result_{timestamp}_{unique_id}.txt")
+    os.makedirs(os.path.dirname(result_file), exist_ok=True)
+    return result_file
+
+
+
+css = """
+.dialog-box { 
+    height: 600px;        
+    overflow-y: auto;     
+}
+"""
+
+
+
+# Arena GUI
+is_dev = False
+result_save_path = set_result_path()
+dialog_dict = make_dialog_dict()
+with gr.Blocks(title="1:1 Outpatient Model Simulation Arena", css=css) as demo:
+    gr.Markdown("# 🤖 Model Arena Evaluation")
+    gr.Markdown("### Compare two model simulations and choose the better one!")
+    gr.Markdown("### This scenario assumes that the patient called the hospital's administrative office for an outpatient inquiry.")
+
+    # New comparison button
+    btn_new = gr.Button("🥊 Start Arena!! 🥊")
+    msg = gr.Markdown("")
+    state_dict = gr.State(dialog_dict)
+
+    # Showing two model simulations side by side
+    with gr.Row(visible=False) as arena_row:
+        with gr.Column():
+            model1_name = gr.Textbox(label="Model A", interactive=False, visible=is_dev)
+            dialog1_box = gr.Markdown(label="Simulation A", elem_classes="dialog-box")  # Textbox → Markdown
+            vote1 = gr.Button("👍 Choose A")
+
+        with gr.Column():
+            model2_name = gr.Textbox(label="Model B", interactive=False, visible=is_dev)
+            dialog2_box = gr.Markdown(label="Simulation B", elem_classes="dialog-box")  # Textbox → Markdown
+            vote2 = gr.Button("👍 Choose B")
+
+    # Submit button
+    with gr.Row(visible=False) as submit_row:
+        submit_btn = gr.Button("📤 Submit All Results")
+        submit_msg = gr.Markdown("")
+        
+    # Button actions
+    btn_new.click(
+        fn=new_comparison,
+        inputs=[state_dict],
+        outputs=[model1_name, dialog1_box, model2_name, dialog2_box, arena_row, submit_row, btn_new],
+    )
+
+    vote1.click(
+        fn=lambda m1, m2, d1, d2, state, is_dev=is_dev, result_file_path=result_save_path: record_vote("A", m1, m2, d1, d2, state, is_dev, result_file_path),
+        inputs=[model1_name, model2_name, dialog1_box, dialog2_box, state_dict],
+        outputs=[model1_name, dialog1_box, model2_name, dialog2_box, arena_row, msg],
+    )
+
+    vote2.click(
+        fn=lambda m1, m2, d1, d2, state, is_dev=is_dev, result_file_path=result_save_path: record_vote("B", m1, m2, d1, d2, state, is_dev, result_file_path),
+        inputs=[model1_name, model2_name, dialog1_box, dialog2_box, state_dict],
+        outputs=[model1_name, dialog1_box, model2_name, dialog2_box, arena_row, msg],
+    )
+
+    submit_btn.click(
+        fn=save_data,
+        inputs=[gr.State(result_save_path)],
+        outputs=[submit_msg],
+    )
+
+# Launch the app
+if is_dev:
+    demo.launch(server_port=7860)
+else:
+    demo.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio==5.49.1
+dotenv==0.9.9
+PyGithub==2.8.1

utils/__init__.py

@@ -0,0 +1,127 @@
+import os
+import logging.config
+
+
+
+base_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+try:
+    version_file_path = os.path.join(base_path, 'version.txt')
+    LOGGING_NAME = f"Simulation_Arena_{open(version_file_path).read().strip()}"
+except:
+    version_file_path = os.path.join(base_path, 'app', 'version.txt')
+    LOGGING_NAME = f"Simulation_Arena_{open(version_file_path).read().strip()}"
+VERBOSE = True
+
+
+def set_logging(name=LOGGING_NAME, verbose=True):
+    """Sets up logging for the given name."""
+    rank = int(os.getenv('RANK', -1))  # rank in world for Multi-GPU trainings
+    level = logging.INFO if verbose and rank in {-1, 0} else logging.ERROR
+
+    class ColorFormatter(logging.Formatter):
+        """Custom formatter to add colors to log messages using colorstr."""
+        def format(self, record):
+            if record.levelname == "ERROR":
+                record.msg = colorstr("red", record.msg)
+            elif record.levelname == "WARNING":
+                record.msg = colorstr("yellow", record.msg)
+            # elif record.levelname == "INFO":
+            #     record.msg = colorstr("green", record.msg)
+            elif record.levelname == "DEBUG":
+                record.msg = colorstr("blue", record.msg)
+            return super().format(record)
+
+    logging.config.dictConfig({
+        'version': 1,
+        'disable_existing_loggers': False,
+        'formatters': {
+            name: {
+                '()': ColorFormatter,  # Use the custom formatter
+                'format': '[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s'
+            }
+        },
+        'handlers': {
+            name: {
+                'class': 'logging.StreamHandler',
+                'formatter': name,
+                'level': level
+            }
+        },
+        'loggers': {
+            name: {
+                'level': level,
+                'handlers': [name],
+                'propagate': False
+            }
+        }
+    })
+    
+
+set_logging(LOGGING_NAME, verbose=VERBOSE) 
+LOGGER = logging.getLogger(LOGGING_NAME)
+
+
+
+def colorstr(*input):
+    """
+    Colors a string based on the provided color and style arguments. Utilizes ANSI escape codes.
+    See https://en.wikipedia.org/wiki/ANSI_escape_code for more details.
+
+    This function can be called in two ways:
+        - colorstr('color', 'style', 'your string')
+        - colorstr('your string')
+
+    In the second form, 'blue' and 'bold' will be applied by default.
+
+    Args:
+        *input (str): A sequence of strings where the first n-1 strings are color and style arguments,
+                      and the last string is the one to be colored.
+
+    Supported Colors and Styles:
+        Basic Colors: 'black', 'red', 'green', 'yellow', 'blue', 'magenta', 'cyan', 'white'
+        Bright Colors: 'bright_black', 'bright_red', 'bright_green', 'bright_yellow',
+                       'bright_blue', 'bright_magenta', 'bright_cyan', 'bright_white'
+        Misc: 'end', 'bold', 'underline'
+
+    Returns:
+        (str): The input string wrapped with ANSI escape codes for the specified color and style.
+
+    Examples:
+        >>> colorstr('blue', 'bold', 'hello world')
+        >>> '\033[34m\033[1mhello world\033[0m'
+    """
+    *args, string = input if len(input) > 1 else ('blue', 'bold', input[0])  # color arguments, string
+    colors = {
+        'black': '\033[30m',  # basic colors
+        'red': '\033[31m',
+        'green': '\033[32m',
+        'yellow': '\033[33m',
+        'blue': '\033[34m',
+        'magenta': '\033[35m',
+        'cyan': '\033[36m',
+        'white': '\033[37m',
+        'bright_black': '\033[90m',  # bright colors
+        'bright_red': '\033[91m',
+        'bright_green': '\033[92m',
+        'bright_yellow': '\033[93m',
+        'bright_blue': '\033[94m',
+        'bright_magenta': '\033[95m',
+        'bright_cyan': '\033[96m',
+        'bright_white': '\033[97m',
+        'end': '\033[0m',  # misc
+        'bold': '\033[1m',
+        'underline': '\033[4m'}
+    return ''.join(colors[x] for x in args) + f'{string}' + colors['end']
+
+
+
+def log(message, level='info', color=False):
+    if level.lower() == 'warning':
+        LOGGER.warning(message)
+    elif level.lower() == 'error':
+        LOGGER.error(message)
+    else:
+        if color:
+            LOGGER.info(colorstr(message))
+        else:
+            LOGGER.info(message)

utils/common.py

@@ -0,0 +1,25 @@
+import os
+from github import Github
+from dotenv import load_dotenv
+
+from utils import log
+
+
+
+def upload_to_github(file_path, content):
+    load_dotenv(override=True)
+    g = Github(os.getenv("GITHUB_TOKEN"))
+    repo = g.get_repo("ljm565/adminsim-human-eval")
+    try:
+        contents = None
+        try:
+            contents = repo.get_contents(file_path, ref="main")
+        except Exception:
+            if contents is None:
+                repo.create_file(file_path, "feat: Add new results", content, branch="main")
+                log(f"File created successfully at {file_path}!")
+            else:
+                repo.update_file(file_path, "feat: Update the results", content, contents.sha, branch="main")
+                log(f"File updated successfully at {file_path}!")
+    except Exception as e:
+        log(f"An error occurred: {e}", level="error")

utils/filesys_utils.py

@@ -0,0 +1,45 @@
+import os
+import json
+from typing import Any
+
+
+
+def json_load(path: str) -> Any:
+    """
+    Load and parse a JSON file.
+
+    Args:
+        path (str): Path to the JSON file.
+
+    Returns:
+        Any: The parsed Python object (usually a dict or list) from the JSON file.
+    """
+    with open(path, 'r') as f:
+        return json.load(f)
+    
+
+
+def get_files(path: str, ext: str = None) -> list[str]:
+    """
+    Get all files in a directory with a specific extension.
+
+    Args:
+        path (str): Folder path to search for files.
+        ext (str, optional): Extension that you want to filter. Defaults to None.
+
+    Raises:
+        ValueError: If file does not exist.
+
+    Returns:
+        list[str]: List of file paths that match the given extension.
+    """
+    if not os.path.isdir(path):
+        raise ValueError(f"Path {path} is not a directory.")
+    
+    files = []
+    for root, _, filenames in os.walk(path):
+        for filename in filenames:
+            if ext is None or filename.endswith(ext):
+                files.append(os.path.join(root, filename))
+    
+    return files

utils/postprocess.py

@@ -0,0 +1,71 @@
+import os, re
+from utils.filesys_utils import json_load, get_files
+
+
+
+def make_dialog_dict(dir_path: str = './dialogs') -> dict:
+    """
+    Create a dictionary of dialogs from JSON files in the specified directory.
+
+    Args:
+        dir_path (str, optional): Path to the directory containing dialog JSON files. Defaults to './dialogs'.
+
+    Returns:
+        dict: A dictionary where keys are model name and values are the parsed JSON content.
+    """
+    dialog_dict = {}
+
+    for model in os.listdir(dir_path):
+        model_path = os.path.join(dir_path, model)
+        dialog_files = get_files(model_path, ext='.json')
+        if not dialog_files:
+            continue
+
+        dialogues = [content 
+            for file in dialog_files
+            for content in map(dialog_postprocessing, json_load(file).values())
+            if content
+        ]
+
+        if dialogues:
+            dialog_dict[model] = dialogues
+
+    return dialog_dict
+
+
+
+def dialog_postprocessing(dialog: str) -> str:
+    """
+    Postprocess a dialog string by removing unwanted tokens.
+
+    Args:
+        dialog (str): The dialog string to be processed.
+
+    Returns:
+        str: The cleaned dialog string.
+    """
+    department_candidates = ["gastroenterology", "cardiology", "pulmonary", "endocrinology/metabolism", "nephrology", "hematology/oncology", "allergy", "infectious diseases", "rheumatology"]
+    try:
+        answer_pattern = re.compile(r'Answer:\s*\d+\.\s*(.+)')
+        split_pattern = re.compile(r'\bAnswer:')
+        
+        department = answer_pattern.search(dialog).group(1)
+        for candidate in department_candidates:
+            if department.lower().startswith(candidate):
+                department = candidate
+                break
+        assert department.lower() in department_candidates
+        
+        before_answer = re.split(split_pattern, dialog)[0].strip()
+        before_answer += f' I will introduce you to a physician who work in the {department}.'
+
+        before_answer = before_answer.replace("Staff:", "<span style='color:rgb(0,102,204); font-weight:bold'>Staff</span>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;:")
+        before_answer = before_answer.replace("Patient:", "<span style='color:rgb(204,0,102); font-weight:bold'>Patient</span>:")
+        before_answer = before_answer.replace("\n", "<br>")
+    
+    except:
+        before_answer = ''
+
+    return before_answer
+    
+

version.txt

@@ -0,0 +1 @@
+1.0.0