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.
