app.py

#!/usr/bin/env python
# coding: utf-8

# In[ ]:


import numpy as np
import tensorflow as tf
import tensorflow_addons as tfa
from tensorflow.keras import layers
import transformers
import sentencepiece as spm
#show the version of the package imported with text instructions\
print("Tensorflow version: ", tf.__version__)
print("Tensorflow Addons version: ", tfa.__version__)
print("Transformers version: ", transformers.__version__)
print("Sentencepiece version: ", spm.__version__)
print("Numpy version: ", np.__version__)


# In[ ]:


class MeanPool(tf.keras.layers.Layer):
    def call(self, inputs, mask=None):
        broadcast_mask = tf.expand_dims(tf.cast(mask, "float32"), -1)
        embedding_sum = tf.reduce_sum(inputs * broadcast_mask, axis=1)
        mask_sum = tf.reduce_sum(broadcast_mask, axis=1)
        mask_sum = tf.math.maximum(mask_sum, tf.constant([1e-9]))
        return embedding_sum / mask_sum
class WeightsSumOne(tf.keras.constraints.Constraint):
    def __call__(self, w):
        return tf.nn.softmax(w, axis=0)


# In[ ]:


tokenizer = transformers.AutoTokenizer.from_pretrained("microsoft/deberta-v3-large"
)
tokenizer.save_pretrained('./tokenizer/')

cfg = transformers.AutoConfig.from_pretrained("microsoft/deberta-v3-large", output_hidden_states=True)
cfg.hidden_dropout_prob = 0
cfg.attention_probs_dropout_prob = 0
cfg.save_pretrained('./tokenizer/')


# In[ ]:


def deberta_encode(texts, tokenizer=tokenizer):
    input_ids = []
    attention_mask = []
    
    for text in texts:
        token = tokenizer(text, 
                          add_special_tokens=True, 
                          max_length=512, 
                          return_attention_mask=True, 
                          return_tensors="np", 
                          truncation=True, 
                          padding='max_length')
        input_ids.append(token['input_ids'][0])
        attention_mask.append(token['attention_mask'][0])
    
    return np.array(input_ids, dtype="int32"), np.array(attention_mask, dtype="int32")


# In[ ]:


MAX_LENGTH=512
BATCH_SIZE=8


# In[ ]:


def get_model():
    input_ids = tf.keras.layers.Input(
        shape=(MAX_LENGTH,), dtype=tf.int32, name="input_ids"
    )
    
    attention_masks = tf.keras.layers.Input(
        shape=(MAX_LENGTH,), dtype=tf.int32, name="attention_masks"
    )
   
    deberta_model = transformers.TFAutoModel.from_pretrained("microsoft/deberta-v3-large", config=cfg)
    
        
    REINIT_LAYERS = 1
    normal_initializer = tf.keras.initializers.GlorotUniform()
    zeros_initializer = tf.keras.initializers.Zeros()
    ones_initializer = tf.keras.initializers.Ones()

#     print(f'\nRe-initializing encoder block:')
    for encoder_block in deberta_model.deberta.encoder.layer[-REINIT_LAYERS:]:
#         print(f'{encoder_block}')
        for layer in encoder_block.submodules:
            if isinstance(layer, tf.keras.layers.Dense):
                layer.kernel.assign(normal_initializer(shape=layer.kernel.shape, dtype=layer.kernel.dtype))
                if layer.bias is not None:
                    layer.bias.assign(zeros_initializer(shape=layer.bias.shape, dtype=layer.bias.dtype))

            elif isinstance(layer, tf.keras.layers.LayerNormalization):
                layer.beta.assign(zeros_initializer(shape=layer.beta.shape, dtype=layer.beta.dtype))
                layer.gamma.assign(ones_initializer(shape=layer.gamma.shape, dtype=layer.gamma.dtype))

    deberta_output = deberta_model.deberta(
        input_ids, attention_mask=attention_masks
    )
    hidden_states = deberta_output.hidden_states
    
    #WeightedLayerPool + MeanPool of the last 4 hidden states
    stack_meanpool = tf.stack(
        [MeanPool()(hidden_s, mask=attention_masks) for hidden_s in hidden_states[-4:]], 
        axis=2)
    
    weighted_layer_pool = layers.Dense(1,
                                       use_bias=False,
                                       kernel_constraint=WeightsSumOne())(stack_meanpool)
    
    weighted_layer_pool = tf.squeeze(weighted_layer_pool, axis=-1)
    output=layers.Dense(15,activation='linear')(weighted_layer_pool)
    #x = layers.Dense(6, activation='linear')(x)
    
    #output = layers.Rescaling(scale=4.0, offset=1.0)(x)
    model = tf.keras.Model(inputs=[input_ids, attention_masks], outputs=output)
    
    #Compile model with Layer-wise Learning Rate Decay
    layer_list = [deberta_model.deberta.embeddings] + list(deberta_model.deberta.encoder.layer)
    layer_list.reverse()
    
    INIT_LR = 1e-5
    LLRDR = 0.9
    LR_SCH_DECAY_STEPS = 1600
    
    lr_schedules = [tf.keras.optimizers.schedules.ExponentialDecay(
        initial_learning_rate=INIT_LR * LLRDR ** i, 
        decay_steps=LR_SCH_DECAY_STEPS, 
        decay_rate=0.3) for i in range(len(layer_list))]
    lr_schedule_head = tf.keras.optimizers.schedules.ExponentialDecay(
        initial_learning_rate=1e-4, 
        decay_steps=LR_SCH_DECAY_STEPS, 
        decay_rate=0.3)
    
    optimizers = [tf.keras.optimizers.Adam(learning_rate=lr_sch) for lr_sch in lr_schedules]
    
    optimizers_and_layers = [(tf.keras.optimizers.Adam(learning_rate=lr_schedule_head), model.layers[-4:])] +\
        list(zip(optimizers, layer_list))
    
    optimizer = tfa.optimizers.MultiOptimizer(optimizers_and_layers)
    
    model.compile(optimizer=optimizer,
                 loss='mse',
                 metrics=[tf.keras.metrics.RootMeanSquaredError()],
                 )
    return model


# In[ ]:


tf.keras.backend.clear_session()
model = get_model()
model.load_weights('./best_model_fold2.h5')


# In[ ]:





# In[ ]:


# map the integer labels to their original string representation
label_mapping = {
    0: 'Greeting',
    1: 'Curiosity',
    2: 'Interest',
    3: 'Obscene',
    4: 'Annoyed',
    5: 'Openness',
    6: 'Anxious',
    7: 'Acceptance',
    8: 'Uninterested',
    9: 'Informative',
    10: 'Accusatory',
    11: 'Denial',
    12: 'Confused',
    13: 'Disapproval',
    14: 'Remorse'
}

#label_strings = [label_mapping[label] for label in labels]

#print(label_strings)


# In[ ]:


def inference(texts):
    prediction = model.predict(deberta_encode([texts]))
    labels = np.argmax(prediction, axis=1)
    label_strings = [label_mapping[label] for label in labels]
    return label_strings[0]


# # GPT

# In[ ]:


import openai
import os
import pandas as pd
import gradio as gr


# In[ ]:


openai.organization = os.environ['org_id']
openai.api_key = os.environ['openai_api']
model_version = "gpt-3.5-turbo"
model_token_limit = 10
model_temperature = 0.1


# In[ ]:


def generatePrompt () :
    labels = ["Openness", 
    "Anxious",
    "Confused",
    "Disapproval",
    "Remorse",
    "Uninterested",
    "Accusatory",
    "Annoyed",
    "Interest",
    "Curiosity",
    "Acceptance",
    "Obscene",
    "Denial",
    "Informative",
    "Greeting"]

    formatted_labels = ', '.join(labels[:-1]) + ', or ' + labels[-1] + '.'

    label_set = ["Openness", "Anxious", "Confused", "Disapproval", "Remorse", "Accusatory",
          "Denial", "Obscene", "Uninterested", "Annoyed", "Informative", "Greeting",
          "Interest", "Curiosity", "Acceptance"]

    formatted_labels = ', '.join(label_set[:-1]) + ', or ' + label_set[-1] + '.\n'

    # The basic task to assign GPT (in natural language)
    base_task = "Classify the following text messages into one of the following categories using one word: " + formatted_labels
    base_task += "Provide only a one word response. Use only the labels provided.\n"

    return base_task


# In[ ]:


def predict(message):
    
    prompt = [{"role": "user", "content": generatePrompt () + "Text: "+ message}]
    
    response = openai.ChatCompletion.create(
                    model=model_version,
                    temperature=model_temperature,
                    max_tokens=model_token_limit,
                    messages=prompt
                )
        
    return response["choices"][0]["message"]["content"]


# # Update

# In[ ]:


model_version = "gpt-3.5-turbo"
model_token_limit = 2000
model_temperature = 0.1


# In[ ]:


def revision(message):
    base_prompt = "Here is a conversation between a Caller and a Volunteer. The Volunteer is trying to be as non-accusatory as possible but also wants to get as much information about the caller as possible. What should the volunteer say next in this exchange? Proved 3 possible responses."

    prompt = [{"role": "user", "content": base_prompt + message}]
    
    response = openai.ChatCompletion.create(
                    model=model_version,
                    temperature=model_temperature,
                    max_tokens=model_token_limit,
                    messages=prompt
                )

    return response["choices"][0]["message"]["content"]


# In[ ]:


import gradio as gr

def combine(a):
    return a + "hello"




with gr.Blocks() as demo:
    gr.Markdown("## DeBERTa Sentiment Analysis")
    gr.Markdown("This is a custom DeBERTa model architecture for sentiment analysis with 15 labels: Openness, Anxiety, Confusion, Disapproval, Remorse, Accusation, Denial, Obscenity, Disinterest, Annoyance, Information, Greeting, Interest, Curiosity, or Acceptance.<br />Please enter your sentence(s) in the input box below and click the Submit button. The model will then process the input and provide the sentiment in one of the labels.<br/>The Test Example section below provides some input examples. Click on them and submit them to the model to see how it works.")

    txt = gr.Textbox(label="Input", lines=2)
    txt_1 = gr.Textbox(value="", label="Output")
    btn = gr.Button(value="Submit")
    btn.click(inference, inputs=txt, outputs= txt_1)

    demoExample = [
        "Hello, how are you?",
        "I am so happy to be here!",
        "i don't have time for u"
    ]

    gr.Markdown("## Text Examples")
    gr.Examples(
        demoExample,
        txt,
        txt_1,
        inference
    )

with gr.Blocks() as gptdemo:

    gr.Markdown("## GPT Sentiment Analysis")
    gr.Markdown("This a custom GPT model for sentiment analysis with 15 labels: Openness, Anxiety, Confusion, Disapproval, Remorse, Accusation, Denial, Obscenity, Disinterest, Annoyance, Information, Greeting, Interest, Curiosity, or Acceptance.<br />Please enter your sentence(s) in the input box below and click the Submit button. The model will then process the input and provide the sentiment in one of the labels.<br />The Test Example section below provides some input examples. Click on them and submit them to the model to see how it works.Please note that the input may be collected by service providers.")
    txt = gr.Textbox(label="Input", lines=2)
    txt_1 = gr.Textbox(value="", label="Output")
    btn = gr.Button(value="Submit")
    btn.click(predict, inputs=txt, outputs= txt_1)

    gptExample = [
        "Hello, how are you?",
        "Are you busy at the moment?",
        "I'm doing real good"
    ]

    gr.Markdown("## Text Examples")
    gr.Examples(
        gptExample,
        txt,
        txt_1,
        predict
    )


with gr.Blocks() as revisiondemo:
    gr.Markdown("## Conversation Revision")
    gr.Markdown("This is a custom GPT model designed to generate possible response texts based on previous contexts. You can input a conversation between a caller and a volunteer, and the model will provide three possible responses based on the input. <br />The Test Example section below provides some input examples. Click on them and submit them to the model to see how it works. Please note that the input may be collected by service providers.")
    txt = gr.Textbox(label="Input", lines=2)
    txt_1 = gr.Textbox(value="", label="Output",lines=4)
    btn = gr.Button(value="Submit")
    btn.click(revision, inputs=txt, outputs= txt_1)

    revisionExample = ["Caller: sup\nVolunteer: Hey, how's it going?\nCaller: not very well, actually\nVolunteer: What's the matter?\nCaller: it's my wife, don't worry about it"]

    with gr.Column():
        gr.Markdown("## Text Examples")
        gr.Examples(
            revisionExample,
            [txt],
            txt_1,
            revision
        )




gr.TabbedInterface([demo, gptdemo,revisiondemo], ["Model", "GPT","Text Revision"]
).launch(inline=False)