app.py

import os 
import torch 

import numpy as np 
import gradio as gr

from random import sample 
from detoxify import Detoxify
from datasets import load_dataset
from huggingface_hub import HfApi, ModelFilter, ModelSearchArguments
from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers import GPT2Tokenizer, GPT2LMHeadModel, GPTNeoForCausalLM
from transformers import BloomTokenizerFast, BloomForCausalLM

HF_AUTH_TOKEN = os.environ.get('hf_token' or True)

DATASET = "allenai/real-toxicity-prompts"

CHECKPOINTS = {
    "DistilGPT2 by HuggingFace 🤗" : "distilgpt2",
    "GPT-Neo 125M by EleutherAI 🤖" : "EleutherAI/gpt-neo-125M",
    "BLOOM 560M by BigScience 🌸" : "bigscience/bloom-560m",
    "Custom Model" : None
    }

MODEL_CLASSES = {
    "DistilGPT2 by HuggingFace 🤗" : (GPT2LMHeadModel, GPT2Tokenizer),
    "GPT-Neo 125M by EleutherAI 🤖" : (GPTNeoForCausalLM, GPT2Tokenizer),
    "BLOOM 560M by BigScience 🌸" : (BloomForCausalLM, BloomTokenizerFast),
    "Custom Model" : (AutoModelForCausalLM, AutoTokenizer),
    }

CHOICES = sorted(list(CHECKPOINTS.keys())[:3])

def load_model(model_name, custom_model_path, token): 
    try:
      model_class, tokenizer_class = MODEL_CLASSES[model_name]
      model_path = CHECKPOINTS[model_name]
    
    except KeyError:
      model_class, tokenizer_class = MODEL_CLASSES['Custom Model']
      model_path = custom_model_path or model_name 
    
    model = model_class.from_pretrained(model_path, use_auth_token=token)
    tokenizer = tokenizer_class.from_pretrained(model_path, use_auth_token=token)

    tokenizer.pad_token = tokenizer.eos_token
    model.config.pad_token_id = model.config.eos_token_id

    model.eval()

    return model, tokenizer

MAX_LENGTH = int(10000)  # Hardcoded max length to avoid infinite loop

def set_seed(seed, n_gpu):
    np.random.seed(seed)
    torch.manual_seed(seed)
    if n_gpu > 0:
        torch.cuda.manual_seed_all(seed)

def adjust_length_to_model(length, max_sequence_length):
    if length < 0 and max_sequence_length > 0:
        length = max_sequence_length
    elif 0 < max_sequence_length < length:
        length = max_sequence_length  # No generation bigger than model size
    elif length < 0:
        length = MAX_LENGTH  # avoid infinite loop
    return length

def generate(model_name,
             token,
             custom_model_path,
             input_sentence,
             length = 75,
             temperature = 0.7,
             top_k = 50,
             top_p = 0.95,
             seed = 42,
             no_cuda = False,
             num_return_sequences = 1,
             stop_token = '.'
             ):

    # load device
    #if not no_cuda:
    device = torch.device("cuda" if torch.cuda.is_available() and not no_cuda else "cpu")
    n_gpu = 0 if no_cuda else torch.cuda.device_count()

    # Set seed
    set_seed(seed, n_gpu)

    # Load model
    model, tokenizer = load_model(model_name, custom_model_path, token)
    model.to(device)

    #length = adjust_length_to_model(length, max_sequence_length=model.config.max_position_embeddings)

    # Tokenize input
    encoded_prompt = tokenizer.encode(input_sentence, 
                                      add_special_tokens=False,
                                      return_tensors='pt')

    encoded_prompt = encoded_prompt.to(device)

    input_ids = encoded_prompt 

    # Generate output 
    output_sequences = model.generate(input_ids=input_ids,
                                      max_length=length + len(encoded_prompt[0]),
                                      temperature=temperature,
                                      top_k=top_k,
                                      top_p=top_p,
                                      do_sample=True,
                                      num_return_sequences=num_return_sequences
                                      )
    generated_sequences = list()

    for generated_sequence_idx, generated_sequence in enumerate(output_sequences):
        generated_sequence = generated_sequence.tolist()
        text = tokenizer.decode(generated_sequence, clean_up_tokenization_spaces=True)
        #remove prompt 
        text = text[len(tokenizer.decode(encoded_prompt[0], clean_up_tokenization_spaces=True)) :]
        
        #remove all text after last occurence of stop_token
        text = text[:text.rfind(stop_token)+1]

        generated_sequences.append(text)

    return generated_sequences[0]


def show_mode(mode):
  if mode == 'Single Model':
    return (
        gr.update(visible=True),
        gr.update(visible=False)
        )
  if mode == 'Multi-Model':
    return (
        gr.update(visible=False),
        gr.update(visible=True)
    )

def prepare_dataset(dataset):
  dataset = load_dataset(dataset, split='train')
  return dataset

def load_prompts(dataset):
  prompts = [dataset[i]['prompt']['text'] for i in range(len(dataset))]
  return prompts

def random_sample(prompt_list):
  random_sample = sample(prompt_list,10)
  return random_sample

def show_dataset(dataset):
  raw_data = prepare_dataset(dataset)
  prompts = load_prompts(raw_data)

  return (gr.update(choices=random_sample(prompts),
                    label='You can find below a random subset from the RealToxicityPrompts dataset', 
                    visible=True),
          gr.update(visible=True),
          prompts,
          )
    
def update_dropdown(prompts):
  return gr.update(choices=random_sample(prompts))

def show_search_bar(value):
  if value == 'Custom Model':
    return (value,
            gr.update(visible=True)
            )
  else:
    return (value,
            gr.update(visible=False)
            )

def search_model(model_name, token):
  api = HfApi()

  model_args = ModelSearchArguments()
  filt = ModelFilter(
      task=model_args.pipeline_tag.TextGeneration,
      library=model_args.library.PyTorch)

  results = api.list_models(filter=filt, search=model_name, use_auth_token=token)
  model_list = [model.modelId for model in results]

  return gr.update(visible=True,
                   choices=model_list,
                   label='Choose the model',
                   )

def show_api_key_textbox(checkbox):
  if checkbox:
    return gr.update(visible=True)
  else:
    return gr.update(visible=False)

def forward_model_choice(model_choice_path):
  return (model_choice_path,
          model_choice_path)

def auto_complete(input, generated):
  output = input + ' ' + generated
  output_spans = [{'entity': 'OUTPUT', 'start': len(input), 'end': len(output)}]
  completed_prompt = {"text": output, "entities": output_spans}   
  return completed_prompt

def process_user_input(model, 
                       token, 
                       custom_model_path, 
                       input, 
                       length, 
                       temperature, 
                       top_p, 
                       top_k):
  warning = 'Please enter a valid prompt.'
  if input == None:
    generated = warning
  else:
    generated = generate(model_name=model,
                         token=token, 
                         custom_model_path=custom_model_path, 
                         input_sentence=input,
                         length=length,
                         temperature=temperature,
                         top_p=top_p,
                         top_k=top_k)
    generated_with_spans = auto_complete(input=input, generated=generated)

  return (
      gr.update(value=generated_with_spans),
      gr.update(visible=True),
      gr.update(visible=True),
      input,
      generated
      )

def pass_to_textbox(input):
  return gr.update(value=input)

def run_detoxify(text):
  results = Detoxify('original').predict(text)
  json_ready_results = {cat:float(score) for (cat,score) in results.items()}
  return json_ready_results

def compute_toxi_output(output_text):
  scores = run_detoxify(output_text)
  return (
      gr.update(value=scores, visible=True),
      gr.update(visible=True)
      )

def compute_change(input, output):
  change_percent = round(((float(output)-input)/input)*100, 2)
  return change_percent

def compare_toxi_scores(input_text, output_scores):
  input_scores = run_detoxify(input_text)
  json_ready_results = {cat:float(score) for (cat,score) in input_scores.items()}

  compare_scores = {
      cat:compute_change(json_ready_results[cat], output_scores[cat])
      for cat in json_ready_results
      for cat in output_scores
      }

  return (
      gr.update(value=json_ready_results, visible=True),
      gr.update(value=compare_scores, visible=True)
  )

def show_flag_choices():
  return gr.update(visible=True)

def update_flag(flag_value): 
  return (flag_value,
          gr.update(visible=True),
          gr.update(visible=True),
          gr.update(visible=False)
          )
  
def upload_flag(*args):
  if flagging_callback.flag(list(args), flag_option = None):
      return gr.update(visible=True)

def forward_model_choice_multi(model_choice_path):
  CHOICES.append(model_choice_path)
  return gr.update(choices = CHOICES)

def process_user_input_multi(models, 
                             input, 
                             token,
                             length,
                             temperature,
                             top_p,
                             top_k):
  warning = 'Please enter a valid prompt.'
  if input == None:
    generated = warning
  else:
    generated_dict= {model:generate(model_name=model,
                                    token=token, 
                                    custom_model_path=None, 
                                    input_sentence=input,
                                    length=length,
                                    temperature=temperature,
                                    top_p=top_p,
                                    top_k=top_k) for model in sorted(models)}
    generated_with_spans_dict = {model:auto_complete(input, generated) for model,generated in generated_dict.items()}

    update_outputs = [gr.HighlightedText.update(value=output, label=model) for model,output in generated_with_spans_dict.items()]
    update_hide = [gr.HighlightedText.update(visible=False) for i in range(10-len(models))]
    return update_outputs + update_hide

def show_choices_multi(models):
  update_show = [gr.HighlightedText.update(visible=True) for model in sorted(models)]
  update_hide = [gr.HighlightedText.update(visible=False,value=None, label=None) for i in range(10-len(models))]

  return update_show + update_hide

def show_params(checkbox):
  if checkbox == True:
    return gr.update(visible=True)
  else: 
    return gr.update(visible=False)

CSS = """
#inside_group {
  padding-top: 0.6em;
  padding-bottom: 0.6em;
}
#pw textarea {
    -webkit-text-security: disc;
}
"""

with gr.Blocks(css=CSS) as demo:

  dataset = gr.Variable(value=DATASET)
  prompts_var = gr.Variable(value=None)
  input_var = gr.Variable(label="Input Prompt", value=None)
  output_var = gr.Variable(label="Output",value=None)
  model_choice = gr.Variable(label="Model", value=None)
  custom_model_path = gr.Variable(value=None)
  flag_choice = gr.Variable(label = "Flag", value=None)

  flagging_callback = gr.HuggingFaceDatasetSaver(hf_token = HF_AUTH_TOKEN, 
                                                 dataset_name = "fsdlredteam/flagged_2",
                                                 organization = "fsdlredteam",
                                                 private = True )

  gr.Markdown("<p align='center'><img src='https://i.imgur.com/ZxbbLUQ.png>'/></p>")
  gr.Markdown("<h1 align='center'>BuggingSpace</h1>")
  gr.Markdown("<h2 align='center'>FSDL 2022 Red-Teaming Open-Source Models Project</h2>")
  gr.Markdown("### Pick a text generation model below, write a prompt and explore the output")
  gr.Markdown("### Or compare the output of multiple models at the same time")

  
  choose_mode = gr.Radio(choices=['Single Model', "Multi-Model"],
                         value='Single Model',
                         interactive=True,
                         visible=True,
                         show_label=False)

  with gr.Group() as single_model:

    gr.Markdown("You can upload any model from the Hugging Face hub -even private ones, \
                provided you use your private key! "
                "Write your prompt or alternatively use one from the \
                [RealToxicityPrompts](https://allenai.org/data/real-toxicity-prompts) dataset.")    
    gr.Markdown("Use it to audit the model for potential failure modes, \
                analyse its output with the Detoxify suite and contribute by reporting any problematic result.")
    gr.Markdown("Beware ! Generation can take up to a few minutes with very large models.")
    
    with gr.Row():

      with gr.Column(scale=1): # input & prompts dataset exploration 
        gr.Markdown("### 1. Select a prompt", elem_id="inside_group")

        input_text = gr.Textbox(label="Write your prompt below.", 
                                interactive=True, 
                                lines=4, 
                                elem_id="inside_group")
        
        gr.Markdown("— or —", elem_id="inside_group")
        
        inspo_button = gr.Button('Click here if you need some inspiration', elem_id="inside_group")

        prompts_drop = gr.Dropdown(visible=False, elem_id="inside_group")

        randomize_button = gr.Button('Show another subset', visible=False, elem_id="inside_group")

        show_params_checkbox_single = gr.Checkbox(label='Set custom params', 
                                                  interactive=True, 
                                                  value=False)

        with gr.Box(visible=False) as params_box_single:

          length_single = gr.Slider(label='Output length',
                                    visible=True,
                                    interactive=True,
                                    minimum=50,
                                    maximum=200,
                                    value=75)
          
          top_k_single = gr.Slider(label='top_k',
                                   visible=True,
                                   interactive=True,
                                   minimum=1,
                                   maximum=100,
                                   value=50)
          
          top_p_single = gr.Slider(label='top_p',
                                   visible=True,
                                   interactive=True,
                                   minimum=0.1,
                                   maximum=1,
                                   value=0.95)

          temperature_single = gr.Slider(label='temperature',
                                         visible=True,
                                         interactive=True,
                                         minimum=0.1,
                                         maximum=1,
                                         value=0.7)  

            
      with gr.Column(scale=1): # Model choice & output 
        gr.Markdown("### 2. Evaluate output")

        model_radio = gr.Radio(choices=list(CHECKPOINTS.keys()),
                              label='Model',
                              interactive=True,
                              elem_id="inside_group")
        
        search_bar = gr.Textbox(label="Search model", 
                                interactive=True, 
                                visible=False, 
                                elem_id="inside_group")
        model_drop = gr.Dropdown(visible=False)

        private_checkbox = gr.Checkbox(visible=True,label="Private Model ?", elem_id="inside_group")

        api_key_textbox = gr.Textbox(label="Enter your AUTH TOKEN below",
                                      value=None, 
                                      interactive=True,
                                      visible=False,
                                      elem_id="pw")

        generate_button = gr.Button('Submit your prompt', elem_id="inside_group")

        output_spans = gr.HighlightedText(visible=True, label="Generated text")

        flag_button = gr.Button("Report output here", visible=False, elem_id="inside_group")

    with gr.Row(): # Flagging 
      
      with gr.Column(scale=1): 
        flag_radio = gr.Radio(choices=["Toxic", "Offensive", "Repetitive", "Incorrect", "Other",],
                              label="What's wrong with the output ?",
                              interactive=True,
                              visible=False,
                              elem_id="inside_group")
      
        user_comment = gr.Textbox(label="(Optional) Briefly describe the issue", 
                                  visible=False, 
                                  interactive=True,
                                  elem_id="inside_group")

      confirm_flag_button = gr.Button("Confirm report", visible=False, elem_id="inside_group")
    
    with gr.Row(): # Flagging success
      success_message = gr.Markdown("Your report has been successfully registered. Thank you!",
                                    visible=False,
                                    elem_id="inside_group")

    with gr.Row(): # Toxicity buttons
      toxi_button = gr.Button("Run a toxicity analysis of the model's output", visible=False, elem_id="inside_group")
      toxi_button_compare = gr.Button("Compare toxicity on input and output", visible=False, elem_id="inside_group")

    with gr.Row(): # Toxicity scores 
      toxi_scores_input = gr.JSON(label = "Detoxify classification of your input", 
                                  visible=False,
                                  elem_id="inside_group")
      toxi_scores_output = gr.JSON(label="Detoxify classification of the model's output", 
                                  visible=False, 
                                  elem_id="inside_group")
      toxi_scores_compare = gr.JSON(label = "Percentage change between Input and Output", 
                                  visible=False,
                                  elem_id="inside_group")
      
  with gr.Group(visible=False) as multi_model:
    model_list = list()

    gr.Markdown("#### Run the same input on multiple models and compare the outputs")
    gr.Markdown("You can upload any model from the Hugging Face hub -even private ones, provided you use your private key!")
    gr.Markdown("Use this feature to compare the same model at different checkpoints")
    gr.Markdown('Or to benchmark your model against another one as a reference.')
    gr.Markdown("Beware ! Generation can take up to a few minutes with very large models.")

    with gr.Row(elem_id="inside_group"):
      with gr.Column():
        models_multi = gr.CheckboxGroup(choices=CHOICES,
                                        label='Models',
                                        interactive=True,
                                        elem_id="inside_group",
                                        value=None)
      with gr.Column():
        generate_button_multi = gr.Button('Submit your prompt',elem_id="inside_group")

        show_params_checkbox_multi = gr.Checkbox(label='Set custom params', 
                                                  interactive=True, 
                                                  value=False)

        with gr.Box(visible=False) as params_box_multi:
        
          length_multi = gr.Slider(label='Output length',
                                   visible=True,
                                   interactive=True,
                                   minimum=50,
                                   maximum=200,
                                   value=75)
          
          top_k_multi = gr.Slider(label='top_k',
                                  visible=True,
                                  interactive=True,
                                  minimum=1,
                                  maximum=100,
                                  value=50)
          
          top_p_multi = gr.Slider(label='top_p',
                                  visible=True,
                                  interactive=True,
                                  minimum=0.1,
                                  maximum=1,
                                  value=0.95)

          temperature_multi = gr.Slider(label='temperature',
                                        visible=True,
                                        interactive=True,
                                        minimum=0.1,
                                        maximum=1,
                                        value=0.7)  

    with gr.Row(elem_id="inside_group"):

      with gr.Column(elem_id="inside_group", scale=1):
        input_text_multi = gr.Textbox(label="Write your prompt below.", 
                                      interactive=True, 
                                      lines=4, 
                                      elem_id="inside_group")
    
      with gr.Column(elem_id="inside_group", scale=1):
        search_bar_multi = gr.Textbox(label="Search another model", 
                                      interactive=True, 
                                      visible=True, 
                                      elem_id="inside_group")
    
        model_drop_multi = gr.Dropdown(visible=False,
                                       show_progress=True,
                                       elem_id="inside_group")
        
        private_checkbox_multi = gr.Checkbox(visible=True,label="Private Model ?")

        api_key_textbox_multi = gr.Textbox(label="Enter your AUTH TOKEN below",
                                           value=None, 
                                           interactive=True,
                                           visible=False,
                                           elem_id="pw")

    with gr.Row() as outputs_row:
      for i in range(10):
        output_spans_multi = gr.HighlightedText(visible=False, elem_id="inside_group")
        model_list.append(output_spans_multi)

    
  with gr.Row():
    gr.Markdown('App made during the [FSDL course](https://fullstackdeeplearning.com) \
                 by Team53: Jean-Antoine, Sajenthan, Sashank, Kemp, Srihari, Astitwa')

  # Single Model 

  choose_mode.change(fn=show_mode,
                     inputs=choose_mode,
                     outputs=[single_model, multi_model])
  
  inspo_button.click(fn=show_dataset, 
                     inputs=dataset, 
                     outputs=[prompts_drop, randomize_button, prompts_var]) 
  
  prompts_drop.change(fn=pass_to_textbox, 
                      inputs=prompts_drop, 
                      outputs=input_text)

  randomize_button.click(fn=update_dropdown, 
                         inputs=prompts_var, 
                         outputs=prompts_drop),
  
  model_radio.change(fn=show_search_bar,
                     inputs=model_radio,
                     outputs=[model_choice,search_bar])
  
  search_bar.submit(fn=search_model,
                    inputs=[search_bar,api_key_textbox],
                    outputs=model_drop,
                    show_progress=True)
  
  private_checkbox.change(fn=show_api_key_textbox,
                          inputs=private_checkbox,
                          outputs=api_key_textbox)
  
  model_drop.change(fn=forward_model_choice, 
                    inputs=model_drop, 
                    outputs=[model_choice,custom_model_path])

  generate_button.click(fn=process_user_input,
                        inputs=[model_choice, 
                                api_key_textbox, 
                                custom_model_path, 
                                input_text,
                                length_single,
                                temperature_single,
                                top_p_single,
                                top_k_single],
                        outputs=[output_spans,
                                 toxi_button,
                                 flag_button, 
                                 input_var,
                                 output_var],
                        show_progress=True) 

  toxi_button.click(fn=compute_toxi_output, 
                    inputs=output_var, 
                    outputs=[toxi_scores_output, toxi_button_compare],
                    show_progress=True)
  
  toxi_button_compare.click(fn=compare_toxi_scores,
                            inputs=[input_text, toxi_scores_output],
                            outputs=[toxi_scores_input, toxi_scores_compare],
                            show_progress=True)

  flag_button.click(fn=show_flag_choices, 
                    inputs=None, 
                    outputs=flag_radio)

  flag_radio.change(fn=update_flag, 
                    inputs=flag_radio, 
                    outputs=[flag_choice, confirm_flag_button, user_comment, flag_button])

  flagging_callback.setup([input_var, output_var, model_choice, user_comment, flag_choice], "flagged_data_points")

  confirm_flag_button.click(fn = upload_flag,
                            inputs = [input_var, 
                                      output_var, 
                                      model_choice, 
                                      user_comment,
                                      flag_choice],
                            outputs=success_message)
  
  show_params_checkbox_single.change(fn=show_params,
                                     inputs=show_params_checkbox_single,
                                     outputs=params_box_single)

  # Model comparison

  search_bar_multi.submit(fn=search_model,
                          inputs=[search_bar_multi, api_key_textbox_multi],
                          outputs=model_drop_multi,
                          show_progress=True)
  
  show_params_checkbox_multi.change(fn=show_params,
                                    inputs=show_params_checkbox_multi,
                                    outputs=params_box_multi)
  
  private_checkbox_multi.change(fn=show_api_key_textbox,
                                inputs=private_checkbox_multi,
                                outputs=api_key_textbox_multi)
  
  model_drop_multi.change(fn=forward_model_choice_multi,
                          inputs=model_drop_multi, 
                          outputs=[models_multi])
    
  models_multi.change(fn=show_choices_multi,
                      inputs=models_multi,
                      outputs=model_list)
    
  generate_button_multi.click(fn=process_user_input_multi,
                              inputs=[models_multi, 
                                      input_text_multi, 
                                      api_key_textbox_multi,
                                      length_multi,
                                      temperature_multi,
                                      top_p_multi,
                                      top_k_multi],
                              outputs=model_list,
                              show_progress=True) 
                            
#demo.launch(debug=True)
if __name__ == "__main__":
    demo.launch(enable_queue=False, debug=True)