来也科技智能自动化平台_数智化转型_财务RPA_政务RPA

@torch.no_grad()def _momentum_update_key_encoder(self):"""    Momentum update of the key encoder    """for param_q, param_k in zip(self.encoder_q.parameters(), self.encoder_k.parameters()):        param_k.data = param_k.data * self.m + param_q.data * (1. - self.m)

'''将原始的序列特征 shape 为 B x SeqLen x FeatureDim 的输入特征转成 shape为 B*ceil(SeqLen/5) x FeatureDim 的输出特征相当增加Batch Size， 去自监督对比维度为FeatureDim 的特征'''def feature_flat(self, feature):    dim = tf.shape(feature)[2]    feature = tf.keras.layers.AvgPool1D(pool_size=5, padding="same")(feature)    feature = tf.reshape(feature, [-1, dim])return feature

#预测模型self.predictor_model = tf.keras.Sequential([    tf.keras.layers.Dense(self.LATENT_DIM,        use_bias=False,        kernel_regularizer=tf.keras.regularizers.l2(0.001),    ),    tf.keras.layers.LeakyReLU(alpha=0.18),    tf.keras.layers.BatchNormalization(),    tf.keras.layers.Dense(self.PROJECT_DIM),])# 投影模型self.projector_model = tf.keras.Sequential([    tf.keras.layers.Dense(self.PROJECT_DIM,        use_bias=False,        kernel_regularizer=tf.keras.regularizers.l2(0.001),    ),# 模型崩溃，加入batchnorm    tf.keras.layers.BatchNormalization(),    tf.keras.layers.LeakyReLU(alpha=0.18),    tf.keras.layers.Dense(self.PROJECT_DIM),    tf.keras.layers.BatchNormalization(),])#simsiam 的前向传播def forward_sim_siam(self, inputs, training, compute_err, valid):    parallel = 1if self.multi_gpu:        parallel = self.strategy.num_replicas_in_sync    img = inputs["image"]    img1, img2 = self.get_aug_img(img, training)_, _, feature1 = self.model(img1, training=training)_, _, feature2 = self.model(img2, training=training)    z1, z2 = self.projector_model(feature1), self.projector_model(feature2)    p1, p2 = self.predictor_model(z1), self.predictor_model(z2)    p1, z1, p2, z2 = list(map(self.feature_flat, [p1, z1, p2, z2]))    loss = self.loss_simsiam(p1, z2) / 2 + self.loss_simsiam(p2, z1) / 2    result = {"total_loss": loss}return {k: v / parallel for k, v in result.items()}
#simsiam 的loss计算    def loss_simsiam(self, p, z):'''     对z要停止梯度回传    '''    z = tf.stop_gradient(z)    p = tf.math.l2_normalize(p, axis=1)    z = tf.math.l2_normalize(z, axis=1)'''    希望的是p和z相似度尽可能大    '''return -tf.reduce_mean(tf.reduce_sum((p * z), axis=1))

# clr 投影模型self.predictor_model_clr = tf.keras.Sequential([    tf.keras.layers.Dense(        self.PROJECT_DIM,        use_bias=False,    ),    tf.keras.layers.LeakyReLU(alpha=0.18),    tf.keras.layers.Dense(self.PROJECT_DIM),])
def forward_simclr(self, inputs, training, compute_err, valid):    parallel = 1    if self.multi_gpu:        parallel = self.strategy.num_replicas_in_sync    img1, img2 = self.get_aug_img(inputs["image"], training)    _, _, feature1 = self.model(img1, training=training)    _, _, feature2 = self.model(img2, training=training)    z1, z2 = self.predictor_model_clr(feature1), self.predictor_model_clr(feature2)    '''     将sequence feature 转为 frame的feature    '''    z1, z2 = self.feature_flat(z1), self.feature_flat(z2)    loss = self.info_nce_loss(z1, z2)    result = {"total_loss": loss}    return {k: v / parallel for k, v in result.items()}  def info_nce_loss(self, projections_1, projections_2):    projections_1 = tf.math.l2_normalize(projections_1, axis=1)    projections_2 = tf.math.l2_normalize(projections_2, axis=1)    similarities = (            tf.matmul(projections_1, projections_2, transpose_b=True) / 0.1    )
    batch_size = tf.shape(projections_1)[0]    contrastive_labels = tf.range(batch_size)    '''    希望正样本的图片特征相似度足够大，负样本对相似度足够小，采用交叉熵的方式    正样本的label=1，负样本=0    '''    loss_1_2 = tf.keras.losses.sparse_categorical_crossentropy(        contrastive_labels, similarities, from_logits=True    )    loss_2_1 = tf.keras.losses.sparse_categorical_crossentropy(        contrastive_labels, tf.transpose(similarities), from_logits=True    )    '''    交叉预测    '''    loss = (loss_1_2 + loss_2_1) / 2    return tf.reduce_mean(loss)