app.py

import gradio as gr
import requests
import os
#
import json
import random
from elo import update_elo_ratings  # Custom function for ELO ratings
enable_btn = gr.Button.update(interactive=True)

# Load chatbot URLs and model names from a JSON file
with open('chatbot_urls.json', 'r') as file:
    chatbots = json.load(file)
def clear_chat(state):
    if state is not None:
        state = {}
    return state, None,None,gr.Button.update(interactive=False),gr.Button.update(interactive=False)





from datasets import load_dataset
import requests
import os

def get_user_elo_ratings():
    dataset = load_dataset("rwitz/mistral-elo-ratings", streaming=True)
    elo_ratings = dataset['train']  # or the relevant split
    return elo_ratings

def update_elo_rating(player_id, new_rating):
    # Fetch the current dataset
    elo_ratings = get_user_elo_ratings()

    # Function to update the rating of a specific player
    def update_rating(example):
        return {'rating': new_rating}

    # Update the dataset
    updated_dataset = elo_ratings.map(update_rating)

    # Convert updated dataset to a dictionary for pushing
    updated_data = updated_dataset.to_dict()

    # Push the updated dataset back to Hugging Face
    response = requests.post(
        "https://huggingface.co/datasets/rwitz/mistral-elo-ratings",
        headers={"Authorization": f"Bearer {os.environ.get('HUGGINGFACE_TOKEN')}"},
        json=updated_data
    )

    if response.status_code == 200:
        print("Successfully updated the dataset")
    else:
        print("Failed to update the dataset. Response code:", response.status_code)

# Function to get bot response
def format_alpaca_prompt(state):
    alpaca_prompt = "Below is an instruction that describes a task. Write a response that appropriately completes the request."
    alpaca_prompt2 = "Below is an instruction that describes a task. Write a response that appropriately completes the request."
    for message in state["history"][0]:
        j=""
        if message['role']=='user':
            j="### Instruction:\n"
        else:
            j="### Response:\n"
        alpaca_prompt += j+ message['content']+"\n\n"
    for message in state["history"][1]:
        j=""
        if message['role']=='user':
            j="### Instruction:\n"
        else:
            j="### Response:\n"
        alpaca_prompt2 += j+ message['content']+"\n\n"
    return [alpaca_prompt+"### Response:\n",alpaca_prompt2+"### Response:\n"]
def get_bot_response(url, prompt,state,bot_index):
    alpaca_prompt = format_alpaca_prompt(state)
    payload = {
        "input": {
            "prompt": alpaca_prompt[bot_index],
            "sampling_params": {
                "max_new_tokens": 50,
                "temperature": 0.7,
                "top_p":0.95
            }
        }
    }
    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "authorization": os.environ.get("RUNPOD_TOKEN")
    }
    response = requests.post(url, json=payload, headers=headers)
    return response.json()['output'].split('### Instruction')[0]

def chat_with_bots(user_input, state):
    bot_names = list(chatbots.keys())
    random.shuffle(bot_names)
    bot1_url, bot2_url = chatbots[bot_names[0]], chatbots[bot_names[1]]
    
    # Update the state with the names of the last bots
    state.update({'last_bots': [bot_names[0], bot_names[1]]})

    bot1_response = get_bot_response(bot1_url, user_input,state,0)
    bot2_response = get_bot_response(bot2_url, user_input,state,1)

    return bot1_response, bot2_response

def update_ratings(state, winner_index):
    elo_ratings = get_user_elo_ratings()
    bot_names = list(chatbots.keys())
    winner = state['last_bots'][winner_index]
    loser = state['last_bots'][1 - winner_index]
    
    elo_ratings = update_elo_ratings(elo_ratings, winner, loser)
    update_elo_rating(elo_ratings)
    return [('Winner: ',state['last_bots'][winner_index]),('Loser: ',state['last_bots'][1 - winner_index])]

def vote_up_model(state, chatbot,chatbot2):
    update_message = update_ratings(state, 0)
    chatbot.append(update_message[0])
    chatbot2.append(update_message[1])
    return chatbot, chatbot2,gr.Button.update(interactive=False),gr.Button.update(interactive=False),gr.Textbox.update(interactive=False),gr.Button.update(interactive=False)  # Disable voting buttons
def vote_down_model(state, chatbot,chatbot2):
    update_message = update_ratings(state, 1)
    chatbot2.append(update_message[0])
    chatbot.append(update_message[1])
    return chatbot,chatbot2, gr.Button.update(interactive=False),gr.Button.update(interactive=False),gr.Textbox.update(interactive=False),gr.Button.update(interactive=False)  # Disable voting buttons

def user_ask(state, chatbot1, chatbot2, textbox):
    global enable_btn
    user_input = textbox
    if len(user_input) > 200:
        user_input = user_input[:200]  # Limit user input to 200 characters

    # Updating state with the current ELO ratings
    state["elo_ratings"] = get_user_elo_ratings()
    if "history" not in state:
        state.update({'history': [[],[]]})
    state["history"][0].extend([
        {"role": "user", "content": user_input}])
    state["history"][1].extend([
        {"role": "user", "content": user_input}])
    # Chat with bots
    bot1_response, bot2_response = chat_with_bots(user_input, state)
    state["history"][0].extend([
        {"role": "bot1", "content": bot1_response},
    ])
    state["history"][1].extend([
        {"role": "bot2", "content": bot2_response},
    ])
    
    chatbot1.append((user_input,bot1_response))
    chatbot2.append((user_input,bot2_response))

    # Keep only the last 10 messages in history
    state["history"] = state["history"][-10:]

    # Format the conversation in ChatML format

    return state, chatbot1, chatbot2, textbox,enable_btn,enable_btn
import pandas as pd

# Function to generate leaderboard data

def generate_leaderboard():
    elo_ratings_dataset = get_user_elo_ratings()  # Returns a Hugging Face dataset

    # Convert the Hugging Face dataset to a pandas DataFrame
    leaderboard_data = pd.DataFrame(elo_ratings_dataset)

    # Assuming the dataset has columns 'bot_name' and 'elo_rating'
    leaderboard_data.columns = ['Chatbot', 'ELO Score']

    # Sort the DataFrame based on the ELO Score in descending order
    leaderboard_data = leaderboard_data.sort_values('ELO Score', ascending=False)

    return leaderboard_data


# Gradio interface setup
with gr.Blocks() as demo:
    state = gr.State({})
    with gr.Tab("Chatbot Arena"):
        with gr.Row():
            with gr.Column():
                chatbot1 = gr.Chatbot(label='Model A').style(height=600)
                upvote_btn_a = gr.Button(value="👍 Upvote A",interactive=False)
            
            with gr.Column():
                chatbot2 = gr.Chatbot(label='Model B').style(height=600)
                upvote_btn_b = gr.Button(value="👍 Upvote B",interactive=False)        
    
        textbox = gr.Textbox(placeholder="Enter your prompt (up to 200 characters)", max_chars=200)
        with gr.Row():
            submit_btn = gr.Button(value="Send")
            reset_btn = gr.Button(value="Reset")
        reset_btn.click(clear_chat, inputs=[state], outputs=[state, chatbot1, chatbot2, upvote_btn_a, upvote_btn_b])
        textbox.submit(user_ask, inputs=[state, chatbot1, chatbot2, textbox], outputs=[state, chatbot1, chatbot2, textbox,upvote_btn_a,upvote_btn_b])
        submit_btn.click(user_ask, inputs=[state, chatbot1, chatbot2, textbox], outputs=[state, chatbot1, chatbot2, textbox,upvote_btn_a,upvote_btn_b])
        upvote_btn_a.click(vote_up_model, inputs=[state, chatbot1,chatbot2], outputs=[chatbot1,chatbot2,upvote_btn_a,upvote_btn_b,textbox,submit_btn])
        upvote_btn_b.click(vote_down_model, inputs=[state, chatbot1,chatbot2], outputs=[chatbot1,chatbot2,upvote_btn_a,upvote_btn_b,textbox,submit_btn])
    with gr.Tab("Leaderboard"):
        leaderboard = gr.Dataframe(generate_leaderboard())
        refresh_btn = gr.Button("Refresh Leaderboard")

    # Function to refresh leaderboard
    def refresh_leaderboard():
        return generate_leaderboard()

    # Event handler for the refresh button
    refresh_btn.click(refresh_leaderboard, inputs=[], outputs=[leaderboard])

    # Launch the Gradio interface

    demo.launch()