app.py

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

import sqlite3

from pymongo.mongo_client import MongoClient
from pymongo.server_api import ServerApi
async def direct_regenerate(model, user_input, chatbot):
    adapter = next(entry['adapter'] for entry in chatbots_data if entry['original_model'] == model)
    temp_state = {
        "history": [
            [{"role": "user", "content": user_input}]
        ]
    }
    response = await get_bot_response(adapter, user_input, temp_state, 0)
    chatbot[-1] = (user_input, response)
    return "", chatbot

    
password=os.environ.get("MONGODB")

def init_database():
    uri = f"mongodb+srv://new-user:{password}@cluster0.xb2urf6.mongodb.net/?retryWrites=true&w=majority&appName=Cluster0"
    client = MongoClient(uri)
    db = client["elo_ratings"]
    collection = db["elo_ratings"]
    return collection

def get_user_elo_ratings(collection):
    rows = list(collection.find())
    if rows:
        return {row['bot_name']: {'elo_rating': row['elo_rating'], 'games_played': row['games_played']} for row in rows}
    else:
        return {"default": {'elo_rating': 1200, 'games_played': 0}}

def update_elo_rating(collection, updated_ratings, winner, loser):
    collection.update_one({"bot_name": winner}, {"$set": {"elo_rating": updated_ratings[winner]['elo_rating'], "games_played": updated_ratings[winner]['games_played']}}, upsert=True)
    collection.update_one({"bot_name": loser}, {"$set": {"elo_rating": updated_ratings[loser]['elo_rating'], "games_played": updated_ratings[loser]['games_played']}}, upsert=True)

import json

with open('chatbots.txt', 'r') as file:
    chatbots_data = json.load(file)
    chatbots = [entry['adapter'] for entry in chatbots_data]
    
def clear_chat(state):
    # Reset state including the chatbot order
    state = {} if state is not None else state
    
    # Initialize the collection object
    collection = init_database()
    
    # Get the list of adapter names
    adapter_names = [entry['adapter'] for entry in chatbots_data]
    
    # Randomly select two new adapters
    selected_adapters = random.sample(adapter_names, 2)
    state['last_bots'] = selected_adapters
    
    # Reset other components
    return state, [], [], gr.Button.update(interactive=False), gr.Button.update(interactive=False), gr.Textbox.update(value='', interactive=True), gr.Button.update(interactive=True)

from datasets import load_dataset,DatasetDict,Dataset
import requests
import os


# Function to get bot response
def format_chatml_prompt(state):
    chatml_prompt = "<|im_start|>system You are a helpful assistant, who can think outside the box.<|im_end|>"
    chatml_prompt2 = "<|im_start|>system You are a helpful assistant, who can think outside the box.<|im_end|>"
    for message in state["history"][0]:
        if message['role'] == 'user':
            chatml_prompt += "\n<|im_start|>user " + message['content'] + "<|im_end|>"
        else:
            chatml_prompt += "\n<|im_start|>assistant " + message['content'] + "<|im_end|>"
    for message in state["history"][1]:
        if message['role'] == 'user':
            chatml_prompt2 += "\n<|im_start|>user " + message['content'] + "<|im_end|>"
        else:
            chatml_prompt2 += "\n<|im_start|>assistant " + message['content'] + "<|im_end|>"
    return [chatml_prompt + "\n<|im_start|>assistant", chatml_prompt2 + "\n<|im_start|>assistant"]
import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

async def get_bot_response(adapter_id, prompt, state, bot_index):
    chatml_prompt = format_chatml_prompt(state)
    print(chatml_prompt)
    payload = {
        "inputs": chatml_prompt[bot_index],
        "parameters": {
            "adapter_id": adapter_id,
            "adapter_source": "hub",
            "temperature": 0.7,
            "max_new_tokens": 100
        }
    }
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {os.environ.get('PREDIBASE_TOKEN')}"
    }
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post("https://serving.app.predibase.com/79957f/deployments/v2/llms/mistral-7b/generate",
                                    json=payload, headers=headers, timeout=30) as response:
                if response.status == 200:
                    response_data = await response.json()
                    response_text = response_data.get('generated_text', '')
                else:
                    error_text = await response.text()
                    print(error_text)
                    response_text = "Sorry, I couldn't generate a response."
        except (aiohttp.ClientError, asyncio.TimeoutError):
            response_text = "Sorry, I couldn't generate a response."
    return response_text.split('<|im_end|>')[0].strip()
async def chat_with_bots(user_input, state):
    # Use existing bot order from state if available, otherwise shuffle and initialize
    if 'last_bots' not in state or not state['last_bots']:
        random.shuffle(chatbots)
        state['last_bots'] = [chatbots[0], chatbots[1]]
    
    bot1_adapter, bot2_adapter = state['last_bots'][0], state['last_bots'][1]

    bot1_response, bot2_response = await asyncio.gather(
        get_bot_response(bot1_adapter, user_input, state, 0),
        get_bot_response(bot2_adapter, user_input, state, 1)
    )

    return bot1_response.replace("<|im_end|>",""), bot2_response.replace("<|im_end|>","")
def update_ratings(state, winner_index, collection):
    elo_ratings = get_user_elo_ratings(collection)
    winner_adapter = state['last_bots'][winner_index]
    loser_adapter = state['last_bots'][1 - winner_index]
    
    winner = next(entry['original_model'] for entry in chatbots_data if entry['adapter'] == winner_adapter)
    loser = next(entry['original_model'] for entry in chatbots_data if entry['adapter'] == loser_adapter)
    
    elo_ratings = update_elo_ratings(elo_ratings, winner_adapter, loser_adapter)
    update_elo_rating(collection, elo_ratings, winner_adapter, loser_adapter)
    return [('Winner: ', winner), ('Loser: ', loser)]
    
def vote_up_model(state, chatbot, chatbot2):
    collection = init_database()
    update_message = update_ratings(state, 0, collection)
    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)

def vote_down_model(state, chatbot, chatbot2):
    collection = init_database()
    update_message = update_ratings(state, 1, collection)
    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)

async 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

    collection = init_database()  # Initialize the collection object

    # Updating state with the current ELO ratings
    state["elo_ratings"] = get_user_elo_ratings(collection)
    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 = await 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, gr.update(value=''),enable_btn,enable_btn
import pandas as pd

# Function to generate leaderboard data

def generate_leaderboard(collection):
    rows = list(collection.find())
    leaderboard_data = pd.DataFrame(rows, columns=['bot_name', 'elo_rating', 'games_played'])
    leaderboard_data['original_model'] = leaderboard_data['bot_name'].apply(lambda x: next(entry['original_model'] for entry in chatbots_data if entry['adapter'] == x))
    leaderboard_data = leaderboard_data[['original_model', 'elo_rating', 'games_played']]
    leaderboard_data.columns = ['Chatbot', 'ELO Score', 'Games Played']
    leaderboard_data['ELO Score'] = leaderboard_data['ELO Score'].round().astype(int)
    leaderboard_data = leaderboard_data.sort_values('ELO Score', ascending=False)
    return leaderboard_data
    
def refresh_leaderboard():
    collection = init_database()
    return generate_leaderboard(collection)
async def direct_chat(model, user_input, chatbot):
    adapter = next(entry['adapter'] for entry in chatbots_data if entry['original_model'] == model)
    temp_state = {
        "history": [
            [{"role": "user", "content": user_input}],
            [{"role": "user", "content": user_input}]
        ]
    }
    response = await get_bot_response(adapter, user_input, temp_state, 0)
    chatbot.append((user_input, response))
    return "", chatbot
def reset_direct_chat():
    return "", [], gr.Dropdown.update(value=model_dropdown.value)
refresh_leaderboard()
# Gradio interface setup
# Gradio interface setup
with gr.Blocks() as demo:
    state = gr.State({})

    with gr.Tab("🤖 Chatbot Arena"):
        gr.Markdown("## 🥊 Let's see which chatbot wins!")
        
        with gr.Row():
            with gr.Column():
                chatbot1 = gr.Chatbot(label='🤖 Model A').style(height=500)
                upvote_btn_a = gr.Button(value="👍 Upvote A", interactive=False).style(full_width=True)

            with gr.Column():
                chatbot2 = gr.Chatbot(label='🤖 Model B').style(height=500)
                upvote_btn_b = gr.Button(value="👍 Upvote B", interactive=False).style(full_width=True)
        
        with gr.Row():
            with gr.Column(scale=5):
                textbox = gr.Textbox(placeholder="🎤 Enter your prompt (up to 200 characters)")
            submit_btn = gr.Button(value="Submit")

        with gr.Row():
            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_btn])
        submit_btn.click(user_ask, inputs=[state, chatbot1, chatbot2, textbox], outputs=[state, chatbot1, chatbot2, textbox, upvote_btn_a, upvote_btn_b], queue=True)

        collection = init_database()

        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])
        textbox.submit(user_ask, inputs=[state, chatbot1, chatbot2, textbox], outputs=[state, chatbot1, chatbot2, textbox, upvote_btn_a, upvote_btn_b], queue=True)

    with gr.Tab("💬 Direct Chat"):
        gr.Markdown("## 🗣️ Chat directly with a model!")
        
        with gr.Row():
            model_dropdown = gr.Dropdown(choices=[entry['original_model'] for entry in chatbots_data], value=chatbots_data[0]['original_model'], label="🤖 Select a model")
        with gr.Row():
            direct_chatbot = gr.Chatbot(label="💬 Direct Chat").style(height=500)
        
        with gr.Row():
            with gr.Column(scale=5):
                direct_textbox = gr.Textbox(placeholder="💭 Enter your message")
            direct_submit_btn = gr.Button(value="Submit")

        with gr.Row():
            direct_regenerate_btn = gr.Button(value="🔄 Regenerate")
            direct_reset_btn = gr.Button(value="🗑️ Reset Chat")

        # ...

        direct_regenerate_btn.click(direct_regenerate, inputs=[model_dropdown, direct_textbox, direct_chatbot], outputs=[direct_textbox, direct_chatbot])
        direct_textbox.submit(direct_chat, inputs=[model_dropdown, direct_textbox, direct_chatbot], outputs=[direct_textbox, direct_chatbot])
        direct_submit_btn.click(direct_chat, inputs=[model_dropdown, direct_textbox, direct_chatbot], outputs=[direct_textbox, direct_chatbot])
        direct_reset_btn.click(reset_direct_chat, None, [direct_textbox, direct_chatbot, model_dropdown])


    with gr.Tab("🏆 Leaderboard"):
        gr.Markdown("## 📊 Check out the top-performing models!")
        
        try:
            leaderboard = gr.Dataframe(refresh_leaderboard())
        except:
            leaderboard = gr.Dataframe()

        with gr.Row():
            refresh_btn = gr.Button("🔄 Refresh Leaderboard")
        
        refresh_btn.click(refresh_leaderboard, outputs=[leaderboard])

# Launch the Gradio interface
if __name__ == "__main__":
    demo.launch(share=False)