app.py

# -*- coding: utf-8 -*-
"""GradioAppTest.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1QhxoNhhM_kcaoQOyz5hsNWLcf2m2L225
"""

# !pip install gradio
# !pip install transformers

import gradio as gr
from transformers import pipeline

"""## JSON"""

# Define the process that the models will be trained for
trainedProcess = "praksa"
trainedProcessJSON = "Praksa"

json = [
    {
        "name": "Praksa",
        "phases": [
            {
                "name": "Odabir preferencija",
                "alias": ["Prijava prakse", "Odabir zadatka", "Prvi korak"],
                "description": "Odabir preferencija je prvi korak u procesu polaganja prakse. Zahtjeva da student odabere zadatak sa popisa...",
                "duration": "1 mjesec",
            },
            {
                "name": "Ispunjavanje prijavnice",
                "description": "Ispunjavanje prijavnice je drugi korak u procesu polaganja prakse. Student mora ispuniti prijavnicu koja se nalazi na stranici kolegija...",
                "duration": "1 tjedan",
            },
            {
                "name": "Predaja dnevnika prakse",
                "alias": ["Završetak prakse", "Dnevnik"],
                "description": "Predaja dnevnika prakse zadnji je korak u procesu polaganja prakse. S završetkom rada, student predaje dnevnik prakse na stranicu kolegija...",
                "duration": "3 dana",
            },
        ],
        "duration": "2 mjeseca",
    },
    {
        "name": "Izrada završnog rada",
        "phases": [
            {
                "name": "Prijava teme",
                "alias": ["Prvi korak"],
                "description": "Prvi korak u procesu izrade završnog rada je prijava teme. Zahtjeva da student odabere mentora te prijavi temu sa popisa...",
                "duration": "5 dana",
            },
            {
                "name": "Ispuna obrasca",
                "description": "Student ispunjava obrazac sa prijavljenom temom...",
                "duration": "4 dana",
            },
            {
                "name": "Obrana rada",
                "description": "Student brani svoj rad pred komosijom...",
                "duration": "1 sat",
            },
        ],
        "duration": "3 mjeseca",
    },
]

# If tasks do not contain alias propery, assign an empty one to them
for process in json:
    for task in process["phases"]:
        if "alias" not in task:
            task["alias"] = []

"""## User intent recognition model

CPU ~6m

GPU ~3m
"""

# Define training epochs
training_epochs = 10
label_size = 6

# Define dataset URL for training
UIDatasetURL = 'https://docs.google.com/spreadsheets/d/e/2PACX-1vSPR-FPTMBcYRynP4JdwYQQ8dAhSx1x8i1LPckUcuIUUlrWT82b5Thqb1bBNnPeGJPxxX1CJAlFSd6F/pub?output=xlsx'

# Will require runetime restart on Google colab (sometimes, idk)
# !pip install tensorflow_text

# !pip install text-hr

"""### Data loading"""

import tensorflow as tf
import tensorflow_text as tft
import tensorflow_hub as tfh
import pandas as pd
import numpy as np
import seaborn as sns

# Text preprocessor for bert based models
preprocessor = tfh.KerasLayer('https://tfhub.dev/google/universal-sentence-encoder-cmlm/multilingual-preprocess/2')

# Language Agnostic BERT sentence encoder
model = tfh.KerasLayer('https://tfhub.dev/google/LaBSE/2')

# Read the data
import pandas as pd
data = pd.read_excel(UIDatasetURL)

columns = ['text', 'intent', 'process']
data.columns = columns

# data = data[data["process"] == trainedProcess].drop(columns="process")
data = data.drop(columns="process")

"""#### Category merging"""

# Convert categories to codes
data['intent'] = data['intent'].astype('category')
data['intent_codes'] = data['intent'].cat.codes

"""#### Normalize data

### Text preprocessing

1. Remove punctuation
2. Lowercase the text
3. Apply tokenization
4. Remove stopwords
5. Apply lemmatizer
"""

import string
import re
import nltk
import text_hr

nltk.download('stopwords')
nltk.download('wordnet')
nltk.download('omw-1.4')
from nltk.stem.porter import PorterStemmer
from nltk.stem import WordNetLemmatizer

def remove_punctuation(text):
    return "".join([i for i in text if i not in string.punctuation])

def tokenization(text):
    return re.split(r"\s+",text)

stopwords = nltk.corpus.stopwords.words('english')
def remove_stopwords(text):
    return [i for i in text if i not in stopwords]

porter_stemmer = PorterStemmer()
def stemming(text):
    return [porter_stemmer.stem(word) for word in text]

wordnet_lemmatizer = WordNetLemmatizer()
def lemmatizer(text):
    return [wordnet_lemmatizer.lemmatize(word) for word in text]

data['text'] = data['text']\
    .apply(lambda x: remove_punctuation(x))\
    .apply(lambda x: x.lower())\
    .apply(lambda x: tokenization(x))\
    .apply(lambda x: lemmatizer(x))

stop_words_list_hr = []
for word_base, l_key, cnt, _suff_id, wform_key, wform in text_hr.get_all_std_words():
    if word_base is not None: stop_words_list_hr.append(word_base)
    if wform is not None: stop_words_list_hr.append(wform)

stop_words_list_hr = list(dict.fromkeys(stop_words_list_hr))

def remove_stopwords_hr(text):
    output = [i for i in text if i not in stop_words_list_hr]
    return output

data['text'] = data['text'].apply(lambda x: remove_stopwords_hr(x))

data['text'] = data['text'].str.join(" ")

"""### Split validation and training data

Train 75%, validation 25%
"""

codes = data['intent_codes'].unique()

# Variable to understand the meaning behind codes
CODES_REPR = data[["intent_codes", "intent"]].drop_duplicates().sort_values("intent_codes")


def codeToIntent(prediction) -> str:
    """ Returns the intent of the prediction, not the code """
    return CODES_REPR[CODES_REPR["intent_codes"] == prediction.argmax()].iloc[0]["intent"]

preprocessed_validation_data = pd.DataFrame(columns=data.columns)
preprocessed_train_data = pd.DataFrame(columns=data.columns)

for c in codes:
    sample = data[data['intent_codes'] == c]
    sample = sample.sample(frac=1)
    # val = sample.sample(frac=0.25)
    val = sample.sample(frac=0)
    train = pd.concat([sample, val]).drop_duplicates(keep=False)
    preprocessed_validation_data = preprocessed_validation_data.append(val, ignore_index=True)
    preprocessed_train_data = preprocessed_train_data.append(train, ignore_index=True)

# Preprocessed google translation data
train_data_eng = preprocessed_train_data[['text', 'intent_codes']]
train_data_eng.columns = ['text', 'intent_codes']

validation_data_eng = preprocessed_validation_data[['text', 'intent_codes']]
validation_data_eng.columns = ['text', 'intent_codes']

def df_to_dataset(df, shuffle=True, batch_size=16):
    df = df.copy()
    labels = df.pop('intent_codes')
    lables_cat = tf.keras.utils.to_categorical(labels, label_size)
    dataset = tf.data.Dataset.from_tensor_slices((dict(df), lables_cat))
    if shuffle:
        dataset = dataset.shuffle(buffer_size=len(df))
    dataset = dataset.batch(batch_size).prefetch(batch_size)
    return dataset

_validation = train_data_eng
train_data_eng = df_to_dataset(train_data_eng)

# validation_data_eng = df_to_dataset(validation_data_eng)
validation_data_eng = df_to_dataset(_validation)

"""### Model definition and training

2 epochs training (testing purposes)
"""

# Model builder
def model_build():
    inputs = tf.keras.layers.Input(shape=(), dtype=tf.string, name='text')
    encoded_input = preprocessor(inputs)
    encoder_outputs = model(encoded_input)

    x = encoder_outputs['pooled_output']
    x = tf.keras.layers.Dropout(0.1)(x)
    x = tf.keras.layers.Dense(128, activation='relu')(x)
    x = tf.keras.layers.Dropout(0.7)(x)
    outputs = tf.keras.layers.Dense(label_size, activation='softmax', name='classifier')(x)
    
    return tf.keras.Model(inputs, outputs)

# Build a model with preprocessed data 
model_eng = model_build()
model_eng.compile(
    optimizer = tf.keras.optimizers.Adam(0.001),
    loss = tf.keras.losses.CategoricalCrossentropy(from_logits=True),
    metrics = tf.keras.metrics.CategoricalAccuracy()
)

eng_history = model_eng.fit(
    train_data_eng,
    epochs = training_epochs,
    batch_size = 16,
    validation_data = validation_data_eng,
)

"""## Data extraction pipeline"""

# !pip install transformers

from transformers import pipeline

pipe = pipeline("token-classification", model="rkrstacic/bpmn-task-extractor")

"""## Sentence similarity"""

# !pip install -U sentence-transformers

import numpy as np
from typing import List, Dict

# Function that shows the result
def predictNER(text: str) -> Dict:
    currentString = "".join([x["word"] for x in pipe(text) if x["entity"] != "LABEL_0"])
                
    # Return dictionary without empty values
    return { "Task": currentString.replace("▁", " ")[1:] }

from sentence_transformers import SentenceTransformer, util

model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

from typing import List
import torch

def getTaskSimilarityIndex(flatIndex: int, tasks) -> int:
    """ Get task index based on the flatten task list """
    for index, task in enumerate(tasks):
        if flatIndex <= len(task["alias"]):
            return index
        
        flatIndex -= len(task["alias"]) + 1
        
    return -1

def getFlattenTasks(tasks) -> List[str]:
    """ Returns the flatten version of task names and their aliases """
    resTasks = []

    for task in tasks:
        resTasks.append(task["name"])
        resTasks = resTasks + task["alias"]
    
    return resTasks

def taskSimilarity(text: str, tasks) -> int:
    """ Returns the task index which is the most similar to the text """
    return getTaskSimilarityIndex(torch.argmax(util.pytorch_cos_sim(
        model.encode(predictNER(text)["Task"], convert_to_tensor=True),
        model.encode(getFlattenTasks(tasks), convert_to_tensor=True)
    )).item(), tasks)

"""## Using the user intent model"""

def preprocessText(text: str) -> str:
    """ Do the same preprocessing as the UI model training input data """
    text = remove_punctuation(text)
    text = text.lower()
    text = tokenization(text)
    text = lemmatizer(text)
    text = remove_stopwords_hr(text)

    return " ".join(text)

def predict_intent(text: str) -> str:
    """ Predict the text intent based on the abovetrained model """
    return codeToIntent(model_eng.predict([preprocessText(text)], verbose=False))

def getPhases(phases) -> str:
    """ P1: Returns the formatted phases """
    phases = [phase["name"].lower() for phase in phases]
    return ', '.join(phases[:-1]) + ' i ' + phases[-1]

# Define functions that handle output text formatting

def getP1String(process) -> str:
    return f"Faze procesa za proces '{process['name']}' su: {getPhases(process['phases'])}"

def getP2String(process) -> str:
    return f"Proces '{process['name']}' traje {process['duration']}"

def getP3String(taskName: str, task) -> str:
    return f"Kratki opis '{taskName}': {task['description']}"

def getP4String(taskName: str, task) -> str:
    return f"Proces '{taskName}' traje {task['duration']}"

def getP5String(taskIndex: int, taskName: str, process) -> str:
    if len(process["phases"]) <= taskIndex + 1:
        return f"'{taskName}' je zadnji korak u procesu '{process['name']}'"
    
    return f"Nakon '{taskName}' je '{process['phases'][taskIndex + 1]['name'].lower()}'"

def getP6String() -> str:
    return "Nažalost, ne razumijem Vaše pitanje"

def print_result(text: str, process) -> None:
    """ Chatbot output messages based on intent """
    intent = predict_intent(text)
    taskIndex = taskSimilarity(text, process["phases"])
    task = process["phases"][taskIndex]
    taskName = task["name"].lower()

    # P1: Koje su faze
    if intent == 'P1':
        return(getP1String(process))

    # P2: Koliko traje cijeli proces
    elif intent == 'P2':
        return(getP2String(process))

    # P3: Kako ide odabir preferencija?
    elif intent == 'P3':
        return(getP3String(taskName, task))

    # P4: Koliko traje {task}
    elif intent == 'P4':
        return(getP4String(taskName, task))

    # P5: Što je nakon {task}
    elif intent == 'P5':
        return(getP5String(taskIndex, taskName, process))
    
    # Ništa od navedenog
    else:
        return(getP6String())

def chatbot(Question, Process="Default") -> None:
    """ By: Rafael Krstačić """
    currentProcess = None

    for process in json:
        if process["name"] == Process:
            currentProcess = process
            break
    else:
        raise KeyError("Process does not exist in json")

    return print_result(Question, currentProcess)
    
"""## Gradio app"""

iface = gr.Interface(
    fn=chatbot,
    inputs=["text", gr.inputs.Dropdown(["Praksa", "Izrada završnog rada"])],
    outputs=["text"],
    title="Software module for answering questions on processes"
)

iface.launch()