new-vsc/be/PIPE/keras_model.py

import sys
sys.path.append('../')
import tensorflow as tf
try:
    import tensorflow.python.keras as keras
    from tensorflow.python.keras import layers
    import tensorflow.python.keras.backend as K
except:
    import tensorflow.keras as keras
    from tensorflow.keras import layers
    import tensorflow.keras.backend as K
from typing import Optional
from PIPE.config import Config
from PIPE.keras_attention_layer import Attention_layer


class Code2VecModel():
    def __init__(self,config: Config):
        self.keras_train_model: Optional[keras.Model] = None
        self.config = config

##################################################搭建模型结构函数########################################################
    def _create_keras_model(self):
        path_source_token_input = layers.Input((self.config.MAX_CONTEXTS,), dtype=tf.int32)
        path_input = layers.Input((self.config.MAX_CONTEXTS,), dtype=tf.int32)
        path_target_token_input = layers.Input((self.config.MAX_CONTEXTS,), dtype=tf.int32)
        context_valid_mask = layers.Input((self.config.MAX_CONTEXTS,))

        # path embedding layer
        # (None, max_contents) -> (None,max_contents,path_embedding_size)
        paths_embedded = layers.Embedding(
            self.config.path_vocab_size, self.config.PATH_EMBEDDINGS_SIZE, name = 'path_embedding'
        )(path_input)

        # terminal embedding layer
        # (None, max_contents) -> (None,max_contents,token_embedding_size)
        token_embedding_shared_layer = layers.Embedding(
            self.config.token_vocab_size, self.config.TOKEN_EMBEDDINGS_SIZE, name = 'token_embedding'
        )

        path_source_token_embedded = token_embedding_shared_layer(path_source_token_input)
        path_target_token_embedded = token_embedding_shared_layer(path_target_token_input)

        # concatenate layer: paths -> [source, path, target]
        # [3 * (None,max_contents, token_embedding_size)] -> (None, max_contents,3*embedding_size)
        context_embedded = layers.Concatenate()([path_source_token_embedded, paths_embedded, path_target_token_embedded])
        context_embedded = layers.Dropout(1 - self.config.DROPOUT_KEEP_RATE)(context_embedded)

        # Dense layer: (None,max_contents,3*embedding_size) -> (None,max_contents, code_vector_size)
        context_after_dense = layers.TimeDistributed(
            layers.Dense(self.config.CODE_VECTOR_SIZE, use_bias=False, activation='tanh')
        )(context_embedded)

        # attention layer:  (None, max_contents,code_vector_size) -> (None,code_vector_size)
        code_vectors, attention_weights = Attention_layer(name='attention')(
            [context_after_dense, context_valid_mask]
        )

        """
            下面是用C2AE分类器进行分类的模型
        """
        Fx = layers.Dense(
           186 , use_bias=True,activation=None,name='Fx'
        )(code_vectors)

        Fx_relu = tf.tanh(Fx)
        Fx_dropout = layers.Dropout(0.5)(Fx_relu)

        targets_input = layers.Input((self.config.categories,), dtype=tf.float32)
        targets_hidden = layers.Dense(
            186, use_bias=True,activation=None,name='targets_hidden'
        )(targets_input)
        targets_hidden_relu = tf.tanh(targets_hidden)

        targets_hidden_dropout = layers.Dropout(0.5)(targets_hidden_relu)
        targets_output = layers.Dense(
            186, use_bias=True,activation=None,name='targets_embedding'
        )(targets_hidden_dropout)
        targets_output_relu = tf.tanh(targets_output)

        targets_output_dropout = layers.Dropout(0.5)(targets_output_relu)

        targets_loss = layers.subtract([targets_output_dropout,Fx_dropout],name='targets_loss')
        Fd1 = layers.Dense(
            186, use_bias=True,activation=None,name='Fd1'
        )(Fx_dropout)

        Fd_relu1 = tf.tanh(Fd1)

        Fd_dropout1 = layers.Dropout(0.5)(Fd_relu1)

        Fd = layers.Dense(
            186, use_bias=True, activation=None, name='Fd'
        )(Fd_dropout1)

        Fd_relu = tf.tanh(Fd)

        Fd_dropout = layers.Dropout(0.5)(Fd_relu)

        target_index_3 = layers.Dense(
            self.config.categories, use_bias=True,activation='sigmoid',name='target_index_3'
        )(Fd_dropout)

        inputs = [path_source_token_input, path_input, path_target_token_input, context_valid_mask, targets_input]
        self.keras_train_model = keras.Model(inputs = inputs, outputs = [target_index_3,targets_loss])

        print("------------------create_keras_model Done.-------------------------")

    @classmethod

    def _create_optimizer(self):
        return tf.keras.optimizers.Adam(lr=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-08, decay=0.0)

    def _comile_keras_model(self, optimizer=None):
        if optimizer is None:
            optimizer = self._create_optimizer()

        ################################评估指标##############################
        def exactMatch(y_true, y_pred):
            y_pred1 = y_pred
            row_dif = K.cast(K.sum(K.round(K.clip(y_true * (1 - y_pred1) + (1-y_true) * y_pred1,0,1)), axis=1) > K.epsilon(),'float32')
            dif = K.sum(K.round(row_dif))
            row_equ = K.cast(K.abs(K.sum(K.round(K.clip(y_true * y_pred1 + (1-y_true) * (1 - y_pred1),0,1)), axis=1) - self.config.categories) < K.epsilon(),'float32')
            equ = K.sum(K.round(row_equ))
            return equ / (equ + dif + K.epsilon())

        def micro_getPrecsion(y_true, y_pred):
            TP = tf.reduce_sum(y_true * tf.round(y_pred))
            TN = tf.reduce_sum((1 - y_true) * (1 - tf.round(y_pred)))
            FP = tf.reduce_sum((1 - y_true) * tf.round(y_pred))
            FN = tf.reduce_sum(y_true * (1 - tf.round(y_pred)))
            precision = TP / (TP + FP)
            return precision

        def micro_getRecall(y_true, y_pred):
            TP = tf.reduce_sum(y_true * tf.round(y_pred))
            TN = tf.reduce_sum((1 - y_true) * (1 - tf.round(y_pred)))
            FP = tf.reduce_sum((1 - y_true) * tf.round(y_pred))
            FN = tf.reduce_sum(y_true * (1 - tf.round(y_pred)))
            precision = TP / (TP + FP)
            recall = TP / (TP + FN)
            return recall

        # F1-score评价指标
        def micro_F1score(y_true, y_pred):
            TP = tf.reduce_sum(y_true * tf.round(y_pred))
            TN = tf.reduce_sum((1 - y_true) * (1 - tf.round(y_pred)))
            FP = tf.reduce_sum((1 - y_true) * tf.round(y_pred))
            FN = tf.reduce_sum(y_true * (1 - tf.round(y_pred)))
            precision = TP / (TP + FP)
            recall = TP / (TP + FN)
            F1score = 2 * precision * recall / (precision + recall)
            return F1score

        def macro_getPrecison(y_true, y_pred):
            col_TP = K.sum(y_true * K.round(y_pred), axis=0)
            col_TN = K.sum((1 - y_true) * (1 - K.round(y_pred)), axis=0)
            col_FP = K.sum((1 - y_true) * K.round(y_pred), axis=0)
            col_FN = K.sum(y_true * (1 - K.round(y_pred)), axis=0)
            precsion = K.mean(col_TP / (col_TP + col_FP + K.epsilon()))
            return precsion

        def macro_getRecall(y_true, y_pred):
            # print(y_true)
            row_TP = K.sum(y_true * K.round(y_pred), axis=0)
            row_TN = K.sum((1 - y_true) * (1 - K.round(y_pred)), axis=0)
            row_FP = K.sum((1 - y_true) * K.round(y_pred), axis=0)
            row_FN = K.sum(y_true * (1 - K.round(y_pred)), axis=0)
            recall = K.mean(row_TP / (row_TP + row_FN + K.epsilon()))
            return recall

        def macro_getF1score(y_true, y_pred):
            precision = macro_getPrecison(y_true, y_pred)
            recall = macro_getRecall(y_true, y_pred)
            F1score = 2 * precision * recall / (precision + recall)
            return F1score

        """
            C2AE损失函数：
                custom_loss: 
                    返回模型最后一层target_loss的平方和,这里y_true是随机设的
                a_cross_loss:
                    返回输出的二分类交叉熵
        """
        def custom_loss(y_true,y_pred):
            return 1*tf.reduce_mean(tf.square(y_pred))

        def a_cross_loss(y_true, y_pred):
            cross_loss = tf.add(tf.log(1e-10 + y_pred) * y_true, tf.log(1e-10 + (1 - y_pred)) * (1 - y_true))
            cross_entropy_label = -1 * tf.reduce_mean(tf.reduce_sum(cross_loss, 1))
            return 0.1*cross_entropy_label

        self.keras_train_model.compile(
            loss = {'target_index_3':a_cross_loss,'targets_loss':custom_loss},
            optimizer=optimizer,
            metrics={'target_index_3':[exactMatch, micro_getPrecsion, micro_getRecall,micro_F1score,macro_getPrecison,macro_getRecall,macro_getF1score]}
        )

if __name__ == "__main__":
    config = Config(set_defaults=True, load_from_args=True, verify=False)
    model = Code2VecModel(config)
    model._create_keras_model()