1
$\begingroup$

I am trying to build a CNN model based on the concepts of Contrastive Learning. In specific based on Triplet loss.

I have 5 different class labels and I create triplets such that in a triplet, two images are from the same class and the third one is from another class.

I have a CNN model which takes one input from a triplet at a time and generates its corresponding embedding in 128 dimensions. All three embedding embeddings from a triplet are used for calculating loss. The loss is based on the Triplet loss.

Further, the loss is backpropagated and training is carried out stochastically.

The idea is to use the trained model to generate one embedding for an input image which can be further used for multi-class classification problems.

My question is, is this method of 3 forward passes and 1 backward pass valid in Tensorflow?

Here is a fragment of my code that I am using for training:

def cnn():
    model_input = layers.Input(shape=(112, 112, 3))
    x = layers.Conv2D(filters=16, kernel_size=3, padding='same', name='Conv1')(model_input)
    x = layers.MaxPool2D()(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)

    x = layers.Conv2D(filters=32, kernel_size=3, padding='same', name='Conv2')(x)
    x = layers.MaxPool2D()(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)

    x = layers.Conv2D(filters=64, kernel_size=3, padding='same', name='Conv3')(x)
    x = layers.MaxPool2D()(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)

    x = layers.Conv2D(filters=128, kernel_size=3, padding='same', name='Conv4')(x)
    x = layers.MaxPool2D()(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)

    x = layers.Conv2D(filters=256, kernel_size=3, padding='same', name='Conv5')(x)
    x = layers.MaxPool2D()(x)
    x = layers.BatchNormalization()(x)
    x = layers.ReLU()(x)

    x = layers.GlobalAvgPool2D(name='GAP')(x)
    output = layers.Dense(128, activation='tanh', name='Dense1')(x)
    origin = tf.zeros_like(output, dtype=float)
    unit_vector = tf.divide(output, tf.sqrt(tf.reduce_sum(tf.square(output-origin)))) # Normalize vector(L2_Norm)
    shared_model = Model(inputs=model_input, outputs=unit_vector)
    shared_model.summary()
    return shared_model
 
def triplets_loss(anchor_sample, positive_sample, negative_sample, alpha=0.2):
    anchor_pos_dist = tf.sqrt(tf.reduce_sum(tf.square(anchor_sample - positive_sample))) # distance between positive pairs
    anchor_neg_dist = tf.sqrt(tf.reduce_sum(tf.square(anchor_sample - negative_sample)))# distance between negative pairs
    triplet_loss = tf.maximum(((anchor_pos_dist - anchor_neg_dist) + alpha), 0.000001) # triplet loss
    return triplet_loss
 
def train(train_data_dir, training_batch=4, lr=1e-4, epochs=100,margin=0.2):
    model = cnn()
    ### creating triplet data loader object ###
    train_data_util_instance = TripletFormulator(data_path_dictionary=train_data_dir, 
    batch=training_batch)
    train_data_array_dict, data_count = train_data_util_instance.data_loader()
    majority_class = max(data_count, key=data_count.get)
    ######
    optimizer = tf.keras.optimizers.Adam(learning_rate=lr)
    for epoch in range(epochs):
        total_train_loss = 0
        start_time = time.time()
        batch_no = 0
        for majority_class_batch in train_data_array_dict[majority_class]:
            batch_loss = 0
            train_batch_dict = {}
            train_batch_dict['A'] = next(iter(train_data_array_dict['A']))
            train_batch_dict['B'] = next(iter(train_data_array_dict['B']))
            train_batch_dict['C'] = majority_class_batch
            train_batch_dict['D'] = next(iter(train_data_array_dict['D']))
            train_batch_dict['E'] = next(iter(train_data_array_dict['E']))
            train_triplets = train_data_util_instance.triplet_generator(train_batch_dict)
            for triplets in train_triplets:
                with tf.GradientTape() as tape:
                     anchor = model(tf.reshape(triplets[0], [-1, 112, 112, 3]))
                     positive = model(tf.reshape(triplets[1], [-1, 112, 112, 3]))
                     negative = model(tf.reshape(triplets[2], [-1, 112, 112, 3]))
                     if np.isnan(anchor).any() or np.isnan(positive).any() or np.isnan(negative).any():
                         print('NAN FOUND')
                     else:
                         loss = triplets_loss(anchor_sample=anchor, positive_sample=positive,negative_sample=negative, alpha=self.alpha)
                     total_train_loss += loss
                     batch_loss += loss
                grads = tape.gradient(loss, model.trainable_weights)
                optimizer.apply_gradients(zip(grads, model.trainable_weights))
             print(epoch, batch_no, batch_loss)
             batch_no += 1
        end_time = time.time()
        print('Training_loss: ', total_train_loss, 'Time_taken: ', end_time - start_time)

When I start training, I can see the training loss converging. But the overall idea is confusing about the number of forward passes per backward pass. Also, is this method a concept of weight sharing?

I would be very eager to discuss this topic as I do not see many problems similar to this.

In most cases, the CNN model takes multiple input and generate multiple outputs and further, the embeddings are used for binary classification that is if the inputs are the same or not.

I would be waiting for your comments and suggestions.

$\endgroup$

2 Answers 2

0
$\begingroup$

In triplet loss you basically have one encoder and the loss is minimised according to enter image description here

In this sense, the weights are shared between the encoding calculated on the negative, anchor and positive parts of the sample.

I'd recommend however to not hard pick negative and positive samples for every anchor, but to let your code pick a convenient triplet instead: a negative and positive maximising the distance.

Tensorflow has this implemented directly: https://www.tensorflow.org/addons/api_docs/python/tfa/losses/TripletSemiHardLoss

$\endgroup$
0
$\begingroup$

I'm not exactly sure what triplet loss is and not sure if I follow your explanation.

If you have a single loss function (i.e. a single scalar number), you have one forward pass and one backward pass. It doesn't matter if there are certain layers that are used multiple times (3 times, presumably) in the forward pass, that just means that layer will also be used 3 times in the backward pass.

$\endgroup$
2
  • $\begingroup$ Hello Taw, Thanks for the reply. Actually there is one single model which I am using three times to get three embeddings. and then I calculate loss on the basis of these embeddings. If you look at the code, in the function ´def train´, under the section of ´with tf.GradientTape() as tape: ´, I have tried to implement this concept. Triplet loss - en.wikipedia.org/wiki/Triplet_loss $\endgroup$ Dec 25, 2021 at 22:50
  • $\begingroup$ Correct me if wrong. I believe that while backpropogating, in my code, unwantedly i have created 3 different models and the model associated with negative samples is only getting updated. $\endgroup$ Dec 25, 2021 at 23:10

You must log in to answer this question.

Not the answer you're looking for? Browse other questions tagged .