app.py

import spaces
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
import torch
import gradio as gr
import logging
from huggingface_hub import login
import os
import traceback

from threading import Thread
from random import shuffle, choice

import json
import gspread
from google.oauth2.service_account import Credentials

logging.basicConfig(level=logging.DEBUG)

SPACER = '\n' + '*' * 40 + '\n'

SCOPES = ['https://www.googleapis.com/auth/spreadsheets', 'https://www.googleapis.com/auth/drive'] #spread scopes

HF_TOKEN = os.environ.get("HF_TOKEN", None)
login(token=HF_TOKEN)

system_prompts = {
    "English": "You are a helpful chatbot that answers user input in a concise and witty way.",
    "German": "Du bist ein hilfreicher Chatbot, der Usereingaben knapp und originell beantwortet.",
    "French": "Tu es un chatbot utile qui répond aux questions des utilisateurs de manière concise et originale.",
    "Spanish": "Eres un chatbot servicial que responde a las entradas de los usuarios de forma concisa y original."
}

htmL_info = "<center><h1>⚔️ Pharia Bot Battle</h1><p><big>Let the games begin: In this arena, the <a href='https://huggingface.co/Aleph-Alpha/Pharia-1-LLM-7B-control-hf'>Pharia 1 model</a> competes against secret challengers of comparable size.</p><ul><li>Try a prompt in a language you want to explore</li><li>Set the parameters and vote for the best answers</li><li>After casting your vote, both bots reveal their identity</li><p>Please note that inputs, outputs and votes are logged anonymously. Feel free to use the bot if you’re cool with that!</p></big></center>"

model_info = [{"id": "Aleph-Alpha/Pharia-1-LLM-7B-control-hf",
                "name": "Pharia 1 LLM 7B control hf"}]

challenger_models = [{"id": "NousResearch/Meta-Llama-3.1-8B-Instruct",
                    "name": "Meta Llama 3.1 8B Instruct"},
                    {"id": "mistralai/Mistral-7B-Instruct-v0.3",
                    "name": "Mistral 7B Instruct v0.3"}]

challenger_model = choice(challenger_models)
model_info.append(challenger_model)
shuffle(model_info)
chatbot_a_name = model_info[0]['name']
chatbot_b_name = model_info[1]['name']

device = "cuda" 

try: 
    tokenizer_a = AutoTokenizer.from_pretrained(model_info[0]['id'])
    model_a = AutoModelForCausalLM.from_pretrained(
        model_info[0]['id'],
        torch_dtype=torch.float16,
        device_map="auto",
        trust_remote_code=True,
    )

    tokenizer_b = AutoTokenizer.from_pretrained(model_info[1]['id'])
    model_b = AutoModelForCausalLM.from_pretrained(
        model_info[1]['id'],
        torch_dtype=torch.float16,
        device_map="auto",
        trust_remote_code=True,
    )

except Exception as e:
    logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')


def get_google_credentials():
    """Sets credentials for remote sheet"""
    service_account_info = {
        "type": "service_account",
        "project_id": os.environ.get("GOOGLE_PROJECT_ID"),
        "private_key_id": os.environ.get("GOOGLE_PRIVATE_KEY_ID"),
        "private_key": os.environ.get("GOOGLE_PRIVATE_KEY").replace('\\n', '\n'),
        "client_email": os.environ.get("GOOGLE_CLIENT_EMAIL"),
        "client_id": os.environ.get("GOOGLE_CLIENT_ID"),
        "auth_uri": os.environ.get("GOOGLE_AUTH_URI"),
        "token_uri": os.environ.get("GOOGLE_TOKEN_URI"),
        "auth_provider_x509_cert_url": os.environ.get("GOOGLE_AUTH_PROVIDER_CERT_URL"),
        "client_x509_cert_url": os.environ.get("GOOGLE_CLIENT_CERT_URL")
    }

    credentials = Credentials.from_service_account_info(service_account_info,scopes=SCOPES)
    return credentials


def get_google_sheet():
    """Intits auth, gets and returns instance of remote sheet"""
    credentials = get_google_credentials()
    client = gspread.authorize(credentials)
    sheet = client.open("pharia_bot_battle_logs").sheet1  # Open your Google Sheet
    return sheet


def apply_pharia_template(messages, add_generation_prompt=False):
    """Chat template not defined in Pharia model configs. 
    Adds chat template for Pharia. Expects a list of messages. 
    add_generation_prompt:bool extends tmplate for generation.
    """

    pharia_template = """<|begin_of_text|>"""
    role_map = {
        "system": "<|start_header_id|>system<|end_header_id|>\n",
        "user": "<|start_header_id|>user<|end_header_id|>\n",
        "assistant": "<|start_header_id|>assistant<|end_header_id|>\n",
    }
    
    for message in messages:
        role = message["role"]
        content = message["content"]
        pharia_template += role_map.get(role, "") + content + "<|eot_id|>\n"
    
    if add_generation_prompt:
        pharia_template += "<|start_header_id|>assistant<|end_header_id|>\n"
    
    return pharia_template


@spaces.GPU()
def generate_both(system_prompt, input_text, 
                    chatbot_a, chatbot_b, 
                    max_new_tokens=2048, temperature=0.2, 
                    top_p=0.9, repetition_penalty=1.1):
    try: 
        text_streamer_a = TextIteratorStreamer(tokenizer_a, skip_prompt=True)
        text_streamer_b = TextIteratorStreamer(tokenizer_b, skip_prompt=True)

        system_prompt_list = [{"role": "system", "content": system_prompt}] if system_prompt else []
        input_text_list = [{"role": "user", "content": input_text}]

        chat_history_a = []
        for user, assistant in chatbot_a:
            chat_history_a.append({"role": "user", "content": user})
            chat_history_a.append({"role": "assistant", "content": assistant})

        chat_history_b = []
        for user, assistant in chatbot_b:
            chat_history_b.append({"role": "user", "content": user})
            chat_history_b.append({"role": "assistant", "content": assistant})
        
        new_messages_a = system_prompt_list + chat_history_a + input_text_list
        new_messages_b = system_prompt_list + chat_history_b + input_text_list

        logging.debug(f'{SPACER}\nNew message bot A: \n{new_messages_a}\n{SPACER}')
        logging.debug(f'{SPACER}\nNew message bot B: \n{new_messages_b}\n{SPACER}')

        if "Pharia" in model_info[0]['id']:
            formatted_conversation = apply_pharia_template(messages=new_messages_a, add_generation_prompt=True)
            tokenized = tokenizer_a(formatted_conversation, return_tensors="pt").to(device)
            #logging.debug(tokenized) #attention_mask
            input_ids_a = tokenized.input_ids
            tokenizer_a.eos_token = "<|endoftext|>" # not set für Pharia
            tokenizer_a.pad_token = "<|padding|>" # not set für Pharia
        else: 
            input_ids_a = tokenizer_a.apply_chat_template(
                new_messages_a,
                add_generation_prompt=True,
                dtype=torch.float16,
                return_tensors="pt"
            ).to(device)

        if "Pharia" in model_info[1]['id']:
            formatted_conversation = apply_pharia_template(messages=new_messages_a, add_generation_prompt=True)
            tokenized = tokenizer_b(formatted_conversation, return_tensors="pt").to(device)
            #logging.debug(tokenized)
            input_ids_b = tokenized.input_ids
            tokenizer_b.eos_token = "<|endoftext|>" # not set für Pharia
            tokenizer_b.pad_token = "<|padding|>" # not set für Pharia
        else:
            input_ids_b = tokenizer_b.apply_chat_template(
                new_messages_b,
                add_generation_prompt=True,
                dtype=torch.float16,
                return_tensors="pt"
            ).to(device)
        

        generation_kwargs_a = dict(
            input_ids=input_ids_a,
            streamer=text_streamer_a,
            max_new_tokens=max_new_tokens,
            pad_token_id=tokenizer_a.eos_token_id,
            do_sample=True,
            temperature=temperature,
            top_p=top_p,
            repetition_penalty=repetition_penalty,
        )

        generation_kwargs_b = dict(
            input_ids=input_ids_b,
            streamer=text_streamer_b,
            max_new_tokens=max_new_tokens,
            pad_token_id=tokenizer_b.eos_token_id,
            do_sample=True,
            temperature=temperature,
            top_p=top_p,
            repetition_penalty=repetition_penalty,
        )

        thread_a = Thread(target=model_a.generate, kwargs=generation_kwargs_a)
        thread_b = Thread(target=model_b.generate, kwargs=generation_kwargs_b)

        thread_a.start()
        thread_b.start()

        chatbot_a.append([input_text, ""])
        chatbot_b.append([input_text, ""])

        finished_a = False
        finished_b = False
    except Exception as e:
        logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

    while not (finished_a and finished_b):
        if not finished_a:
            try:
                text_a = next(text_streamer_a)
                if tokenizer_a.eos_token in text_a:
                    eot_location = text_a.find(tokenizer_a.eos_token)
                    text_a = text_a[:eot_location]
                    finished_a = True
                chatbot_a[-1][-1] += text_a
                yield chatbot_a, chatbot_b
            except StopIteration:
                finished_a = True
            except Exception as e:
                logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

        if not finished_b:
            try:
                text_b = next(text_streamer_b)
                if tokenizer_b.eos_token in text_b:
                    eot_location = text_b.find(tokenizer_b.eos_token)
                    text_b = text_b[:eot_location]
                    finished_b = True
                chatbot_b[-1][-1] += text_b
                yield chatbot_a, chatbot_b
            except StopIteration:
                finished_b = True
            except Exception as e:
                logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

    try:
        
        # chatbot_a[-1][1] Second index of last in list

        sheet_row = [system_prompt, input_text, max_new_tokens, temperature, top_p, repetition_penalty, chatbot_a_name, chatbot_a[-1][1], chatbot_b_name, chatbot_b[-1][1], "None", "None"]
        logging.debug(f'{SPACER}\nOutput row: {sheet_row}') 
        sheet = get_google_sheet()
        sheet.append_row(sheet_row, table_range="A1:L1")

    except Exception as e:
        logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

    return chatbot_a, chatbot_b

def clear():
    return [], []

def handle_vote(selection, chatbot_a, chatbot_b):
    if selection == "Bot A kicks ass!":
        chatbot_a.append(["🏆", f"Thanks, man. I am {chatbot_a_name}"])
        chatbot_b.append(["💩", f"Pffff … I am {chatbot_b_name}"])
        chatbot_a_vote = "Winner"
        chatbot_b_vote = "Looser"
    elif selection == "Bot B crushes it!":
        chatbot_a.append(["🤡", f"Rigged … I am {chatbot_a_name}"])
        chatbot_b.append(["🥇", f"Well deserved! I am {chatbot_b_name}"])
        chatbot_a_vote = "Looser"
        chatbot_b_vote = "Winner"
    else:
        chatbot_a.append(["🤝", f"Lame … I am {chatbot_a_name}"])
        chatbot_b.append(["🤝", f"Dunno. I am {chatbot_b_name}"])
        chatbot_a_vote = "Draw"
        chatbot_b_vote = "Draw"

    try:
        # chatbot_a[-1][1] Second index of last in list
        sheet_row = ["None", "None", 0, 0, 0, 0, chatbot_a_name, "None", chatbot_b_name, "None", chatbot_a_vote, chatbot_b_vote]
        logging.debug(f'{SPACER}\nOutput row: {sheet_row}') 
        sheet = get_google_sheet()
        sheet.append_row(sheet_row, table_range="A1:L1")
    except Exception as e:
        logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

    return chatbot_a, chatbot_b

with gr.Blocks() as demo:
    try:
        with gr.Column():
            gr.HTML(htmL_info)
            gr.HTML("<h2>Set Parameters</h2>")

            with gr.Row(variant="compact"):
                with gr.Column(scale=0):
                    language_dropdown = gr.Dropdown(choices=["English", "German", "French", "Spanish"], label="Select Language for System Prompt",value="English")
                with gr.Column():
                    system_prompt = gr.Textbox(lines=1, label="System Prompt", value=system_prompts["English"], show_copy_button=True)
            with gr.Row(variant="compact"):
                with gr.Column(scale=1):
                    submit_btn = gr.Button(value="Generate", variant="primary")
                    clear_btn = gr.Button(value="Clear", variant="secondary")
                input_text = gr.Textbox(lines=1, label="Prompt", value="Write a Nike style ad headline about the shame of being second best.", scale=3, show_copy_button=True)
            with gr.Accordion(label="Generation Configurations", open=False):
                max_new_tokens = gr.Slider(minimum=128, maximum=4096, value=512, label="Max new tokens", step=128)
                temperature = gr.Slider(minimum=0.0, maximum=1.0, value=0.7, label="Temperature", step=0.01)
                top_p = gr.Slider(minimum=0.0, maximum=1.0, value=0.97, label="Top_p", step=0.01)
                repetition_penalty = gr.Slider(minimum=0.1, maximum=2.0, value=1.1, label="Repetition Penalty", step=0.1)

            gr.HTML("<h2>Check outputs</h2>")
            with gr.Row(variant="panel"):
                with gr.Column():
                    chatbot_a = gr.Chatbot(label="Model A", show_copy_button=True, height=500)
                with gr.Column():
                    chatbot_b = gr.Chatbot(label="Model B", show_copy_button=True, height=500)

            gr.HTML("<h2>Vote!</h2>")
            with gr.Row(variant="panel"):
                    better_bot = gr.Radio(["Bot A kicks ass!", "Bot B crushes it!", "It's a draw."], label="Rate the output!")
            
        language_dropdown.change(
            lambda lang: system_prompts[lang], 
            inputs=[language_dropdown], 
            outputs=[system_prompt]
        )

        better_bot.select(handle_vote, inputs=[better_bot, chatbot_a, chatbot_b], outputs=[chatbot_a, chatbot_b]) 
        input_text.submit(generate_both, inputs=[system_prompt, input_text, chatbot_a, chatbot_b, max_new_tokens, temperature, top_p, repetition_penalty], outputs=[chatbot_a, chatbot_b])
        submit_btn.click(generate_both, inputs=[system_prompt, input_text, chatbot_a, chatbot_b, max_new_tokens, temperature, top_p, repetition_penalty], outputs=[chatbot_a, chatbot_b])
        clear_btn.click(clear, outputs=[chatbot_a, chatbot_b])
    except Exception as e:
        logging.error(f'{SPACER} Error: {e}, Traceback {traceback.format_exc()}')

if __name__ == "__main__":
    demo.queue().launch()