政安晨:【Keras机器学习示例演绎】(二十七)—— 利用 NNCLR 进行自我监督对比学习

本文主要是介绍政安晨:【Keras机器学习示例演绎】(二十七)—— 利用 NNCLR 进行自我监督对比学习,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

目录

简介

自我监督学习

对比学习

NNCLR

设置

超参数

加载数据集

增强

准备扩增模块

编码器结构

用于对比预训练的 NNCLR 模型

预训练 NNCLR


政安晨的个人主页:政安晨

欢迎 👍点赞✍评论⭐收藏

收录专栏: TensorFlow与Keras机器学习实战

希望政安晨的博客能够对您有所裨益,如有不足之处,欢迎在评论区提出指正!

本文目标:计算机视觉自监督学习方法 NNCLR 的实现。

简介


自我监督学习


自我监督表示学习旨在从原始数据中获取稳健的样本表示,而无需昂贵的标签或注释。这一领域的早期方法侧重于定义预训练任务,这些任务涉及在一个有大量弱监督标签的领域中的代用任务。为解决此类任务而训练的编码器有望学习到一般特征,这些特征可能对其他需要昂贵注释的下游任务(如图像分类)有用。

对比学习


自监督学习技术的一大类别是使用对比损失的技术,这些技术已被广泛应用于图像相似性、降维(DrLIM)和人脸验证/识别等计算机视觉应用中。这些方法学习一个潜在空间,将正样本聚类在一起,同时将负样本推开。

NNCLR


在本示例中,我们实现了论文 With a Little Help from My Friends 中提出的 NNCLR:谷歌研究院和 DeepMind 共同发表的论文《视觉表征的最近邻对比学习》(With Little Help from My Friends: Nearest-Neighbor Contrastive Learning of Visual Representations)中提出的 NNCLR。

NNCLR 学习的是自我监督表征,它超越了单一实例的正向性,可以学习到更好的特征,这些特征不受不同视角、变形甚至类内变化的影响。基于聚类的方法提供了一种超越单一实例正向性的好方法,但假设整个聚类都是正向性,可能会由于早期过度泛化而影响性能。取而代之的是,NNCLR 将所学表示空间中的近邻作为正例。此外,NNCLR 还提高了 SimCLR(Keras 示例)等现有对比学习方法的性能,并减少了自监督方法对数据增强策略的依赖。

下面是论文作者提供的一个很好的可视化演示,展示了 NNCLR 如何以 SimCLR 的理念为基础:

我们可以看到,SimCLR 使用同一图像的两个视图作为正对。这两个视图是使用随机数据增强生成的,通过编码器得到正向嵌入对,我们最终使用了两个增强。而 NNCLR 则保留了一个代表完整数据分布的嵌入支持集,并使用最近邻方法形成正对。在训练过程中,支持集被用作内存,类似于 MoCo 中的队列(即先进先出)。

此示例需要使用 tensorflow_datasets,可通过此命令安装:

!pip install tensorflow-datasets

设置

import matplotlib.pyplot as plt
import tensorflow as tf
import tensorflow_datasets as tfds
import osos.environ["KERAS_BACKEND"] = "tensorflow"
import keras
import keras_cv
from keras import ops
from keras import layers

超参数


如原始论文所示,更大的队列规模很可能意味着更好的性能,但会带来巨大的计算开销。作者指出,NNCLR 的最佳结果是在队列大小为 98,304 时(这是他们实验过的最大队列大小)取得的。我们在这里使用 10,000 来展示一个工作示例。

AUTOTUNE = tf.data.AUTOTUNE
shuffle_buffer = 5000
# The below two values are taken from https://www.tensorflow.org/datasets/catalog/stl10
labelled_train_images = 5000
unlabelled_images = 100000temperature = 0.1
queue_size = 10000
contrastive_augmenter = {"brightness": 0.5,"name": "contrastive_augmenter","scale": (0.2, 1.0),
}
classification_augmenter = {"brightness": 0.2,"name": "classification_augmenter","scale": (0.5, 1.0),
}
input_shape = (96, 96, 3)
width = 128
num_epochs = 5  # Use 25 for better results
steps_per_epoch = 50  # Use 200 for better results

加载数据集


我们从 TensorFlow 数据集加载 STL-10 数据集,这是一个用于开发无监督特征学习、深度学习和自学算法的图像识别数据集。它受到 CIFAR-10 数据集的启发,并做了一些修改。

dataset_name = "stl10"def prepare_dataset():unlabeled_batch_size = unlabelled_images // steps_per_epochlabeled_batch_size = labelled_train_images // steps_per_epochbatch_size = unlabeled_batch_size + labeled_batch_sizeunlabeled_train_dataset = (tfds.load(dataset_name, split="unlabelled", as_supervised=True, shuffle_files=True).shuffle(buffer_size=shuffle_buffer).batch(unlabeled_batch_size, drop_remainder=True))labeled_train_dataset = (tfds.load(dataset_name, split="train", as_supervised=True, shuffle_files=True).shuffle(buffer_size=shuffle_buffer).batch(labeled_batch_size, drop_remainder=True))test_dataset = (tfds.load(dataset_name, split="test", as_supervised=True).batch(batch_size).prefetch(buffer_size=AUTOTUNE))train_dataset = tf.data.Dataset.zip((unlabeled_train_dataset, labeled_train_dataset)).prefetch(buffer_size=AUTOTUNE)return batch_size, train_dataset, labeled_train_dataset, test_datasetbatch_size, train_dataset, labeled_train_dataset, test_dataset = prepare_dataset()

增强


其他自监督技术,如 SimCLR、BYOL、SwAV 等,在很大程度上依赖于精心设计的数据扩增管道,以获得最佳性能。然而,NNCLR 对复杂扩增的依赖性较低,因为近邻数据已经提供了丰富的样本变化。扩增管道通常包括以下几种常见技术:

随机调整作物大小
多种颜色变形
高斯模糊

由于 NNCLR 对复杂增强的依赖性较低,我们将只使用随机裁剪和随机亮度来增强输入图像。

准备扩增模块

def augmenter(brightness, name, scale):return keras.Sequential([layers.Input(shape=input_shape),layers.Rescaling(1 / 255),layers.RandomFlip("horizontal"),keras_cv.layers.RandomCropAndResize(target_size=(input_shape[0], input_shape[1]),crop_area_factor=scale,aspect_ratio_factor=(3 / 4, 4 / 3),),keras_cv.layers.RandomBrightness(factor=brightness, value_range=(0.0, 1.0)),],name=name,)

编码器结构


使用 ResNet-50 作为编码器结构是文献中的标准结构。在原论文中,作者使用 ResNet-50 作为编码器架构,并对 ResNet-50 的输出进行空间平均。不过,请记住,功能更强大的模型不仅会增加训练时间,还会需要更多内存,并限制您可以使用的最大批次规模。在本例中,我们只使用四个卷积层。

def encoder():return keras.Sequential([layers.Input(shape=input_shape),layers.Conv2D(width, kernel_size=3, strides=2, activation="relu"),layers.Conv2D(width, kernel_size=3, strides=2, activation="relu"),layers.Conv2D(width, kernel_size=3, strides=2, activation="relu"),layers.Conv2D(width, kernel_size=3, strides=2, activation="relu"),layers.Flatten(),layers.Dense(width, activation="relu"),],name="encoder",)

用于对比预训练的 NNCLR 模型


我们在无标签图像上训练一个有对比损失的编码器。编码器顶部安装了一个非线性投影头,因为它能提高编码器的表征质量。

class NNCLR(keras.Model):def __init__(self, temperature, queue_size,):super().__init__()self.probe_accuracy = keras.metrics.SparseCategoricalAccuracy()self.correlation_accuracy = keras.metrics.SparseCategoricalAccuracy()self.contrastive_accuracy = keras.metrics.SparseCategoricalAccuracy()self.probe_loss = keras.losses.SparseCategoricalCrossentropy(from_logits=True)self.contrastive_augmenter = augmenter(**contrastive_augmenter)self.classification_augmenter = augmenter(**classification_augmenter)self.encoder = encoder()self.projection_head = keras.Sequential([layers.Input(shape=(width,)),layers.Dense(width, activation="relu"),layers.Dense(width),],name="projection_head",)self.linear_probe = keras.Sequential([layers.Input(shape=(width,)), layers.Dense(10)], name="linear_probe")self.temperature = temperaturefeature_dimensions = self.encoder.output_shape[1]self.feature_queue = keras.Variable(keras.utils.normalize(keras.random.normal(shape=(queue_size, feature_dimensions)),axis=1,order=2,),trainable=False,)def compile(self, contrastive_optimizer, probe_optimizer, **kwargs):super().compile(**kwargs)self.contrastive_optimizer = contrastive_optimizerself.probe_optimizer = probe_optimizerdef nearest_neighbour(self, projections):support_similarities = ops.matmul(projections, ops.transpose(self.feature_queue))nn_projections = ops.take(self.feature_queue, ops.argmax(support_similarities, axis=1), axis=0)return projections + ops.stop_gradient(nn_projections - projections)def update_contrastive_accuracy(self, features_1, features_2):features_1 = keras.utils.normalize(features_1, axis=1, order=2)features_2 = keras.utils.normalize(features_2, axis=1, order=2)similarities = ops.matmul(features_1, ops.transpose(features_2))batch_size = ops.shape(features_1)[0]contrastive_labels = ops.arange(batch_size)self.contrastive_accuracy.update_state(ops.concatenate([contrastive_labels, contrastive_labels], axis=0),ops.concatenate([similarities, ops.transpose(similarities)], axis=0),)def update_correlation_accuracy(self, features_1, features_2):features_1 = (features_1 - ops.mean(features_1, axis=0)) / ops.std(features_1, axis=0)features_2 = (features_2 - ops.mean(features_2, axis=0)) / ops.std(features_2, axis=0)batch_size = ops.shape(features_1)[0]cross_correlation = (ops.matmul(ops.transpose(features_1), features_2) / batch_size)feature_dim = ops.shape(features_1)[1]correlation_labels = ops.arange(feature_dim)self.correlation_accuracy.update_state(ops.concatenate([correlation_labels, correlation_labels], axis=0),ops.concatenate([cross_correlation, ops.transpose(cross_correlation)], axis=0),)def contrastive_loss(self, projections_1, projections_2):projections_1 = keras.utils.normalize(projections_1, axis=1, order=2)projections_2 = keras.utils.normalize(projections_2, axis=1, order=2)similarities_1_2_1 = (ops.matmul(self.nearest_neighbour(projections_1), ops.transpose(projections_2))/ self.temperature)similarities_1_2_2 = (ops.matmul(projections_2, ops.transpose(self.nearest_neighbour(projections_1)))/ self.temperature)similarities_2_1_1 = (ops.matmul(self.nearest_neighbour(projections_2), ops.transpose(projections_1))/ self.temperature)similarities_2_1_2 = (ops.matmul(projections_1, ops.transpose(self.nearest_neighbour(projections_2)))/ self.temperature)batch_size = ops.shape(projections_1)[0]contrastive_labels = ops.arange(batch_size)loss = keras.losses.sparse_categorical_crossentropy(ops.concatenate([contrastive_labels,contrastive_labels,contrastive_labels,contrastive_labels,],axis=0,),ops.concatenate([similarities_1_2_1,similarities_1_2_2,similarities_2_1_1,similarities_2_1_2,],axis=0,),from_logits=True,)self.feature_queue.assign(ops.concatenate([projections_1, self.feature_queue[:-batch_size]], axis=0))return lossdef train_step(self, data):(unlabeled_images, _), (labeled_images, labels) = dataimages = ops.concatenate((unlabeled_images, labeled_images), axis=0)augmented_images_1 = self.contrastive_augmenter(images)augmented_images_2 = self.contrastive_augmenter(images)with tf.GradientTape() as tape:features_1 = self.encoder(augmented_images_1)features_2 = self.encoder(augmented_images_2)projections_1 = self.projection_head(features_1)projections_2 = self.projection_head(features_2)contrastive_loss = self.contrastive_loss(projections_1, projections_2)gradients = tape.gradient(contrastive_loss,self.encoder.trainable_weights + self.projection_head.trainable_weights,)self.contrastive_optimizer.apply_gradients(zip(gradients,self.encoder.trainable_weights + self.projection_head.trainable_weights,))self.update_contrastive_accuracy(features_1, features_2)self.update_correlation_accuracy(features_1, features_2)preprocessed_images = self.classification_augmenter(labeled_images)with tf.GradientTape() as tape:features = self.encoder(preprocessed_images)class_logits = self.linear_probe(features)probe_loss = self.probe_loss(labels, class_logits)gradients = tape.gradient(probe_loss, self.linear_probe.trainable_weights)self.probe_optimizer.apply_gradients(zip(gradients, self.linear_probe.trainable_weights))self.probe_accuracy.update_state(labels, class_logits)return {"c_loss": contrastive_loss,"c_acc": self.contrastive_accuracy.result(),"r_acc": self.correlation_accuracy.result(),"p_loss": probe_loss,"p_acc": self.probe_accuracy.result(),}def test_step(self, data):labeled_images, labels = datapreprocessed_images = self.classification_augmenter(labeled_images, training=False)features = self.encoder(preprocessed_images, training=False)class_logits = self.linear_probe(features, training=False)probe_loss = self.probe_loss(labels, class_logits)self.probe_accuracy.update_state(labels, class_logits)return {"p_loss": probe_loss, "p_acc": self.probe_accuracy.result()}

预训练 NNCLR


我们按照论文中的建议,使用 0.1 的温度和前面解释过的 10,000 的队列大小来训练网络。我们使用 Adam 作为对比和探测优化器。在本例中,我们只对模型进行了 30 个历元的训练,但为了获得更好的性能,我们应该对模型进行更多历元的训练。

以下两个指标可用于监控预训练性能,我们也会记录这些指标(摘自 Keras 示例):

对比准确度:自监督指标,即图像表示与其不同增强版本的图像表示更相似的情况比与当前批次中任何其他图像的表示更相似的情况的比率。即使在没有标注示例的情况下,自监督指标也可用于超参数调整。


线性探测准确率:线性探测是评估自监督分类器的常用指标。它的计算方法是在编码器特征的基础上训练逻辑回归分类器的准确率。在我们的例子中,是通过在冻结编码器上训练单个密集层来实现的。需要注意的是,与在预训练阶段后训练分类器的传统方法不同,在本例中,我们在预训练阶段就对分类器进行了训练。这可能会略微降低分类器的准确性,但这样我们就可以在训练过程中监控其值,从而有助于实验和调试。

model = NNCLR(temperature=temperature, queue_size=queue_size)
model.compile(contrastive_optimizer=keras.optimizers.Adam(),probe_optimizer=keras.optimizers.Adam(),jit_compile=False,
)
pretrain_history = model.fit(train_dataset, epochs=num_epochs, validation_data=test_dataset
)

正如 SEER、SimCLR、SwAV 等以前的方法所显示的那样,当你只能获得非常有限的标注训练数据,但却能设法建立大量未标注数据的语料库时,自我监督学习就特别有用。


这篇关于政安晨:【Keras机器学习示例演绎】(二十七)—— 利用 NNCLR 进行自我监督对比学习的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/953132

相关文章

Python调用Orator ORM进行数据库操作

《Python调用OratorORM进行数据库操作》OratorORM是一个功能丰富且灵活的PythonORM库,旨在简化数据库操作,它支持多种数据库并提供了简洁且直观的API,下面我们就... 目录Orator ORM 主要特点安装使用示例总结Orator ORM 是一个功能丰富且灵活的 python O

Nginx设置连接超时并进行测试的方法步骤

《Nginx设置连接超时并进行测试的方法步骤》在高并发场景下,如果客户端与服务器的连接长时间未响应,会占用大量的系统资源,影响其他正常请求的处理效率,为了解决这个问题,可以通过设置Nginx的连接... 目录设置连接超时目的操作步骤测试连接超时测试方法:总结:设置连接超时目的设置客户端与服务器之间的连接

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》:本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.

Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

《Android悬浮窗开发示例((动态权限请求|前台服务和通知|悬浮窗创建)》本文介绍了Android悬浮窗的实现效果,包括动态权限请求、前台服务和通知的使用,悬浮窗权限需要动态申请并引导... 目录一、悬浮窗 动态权限请求1、动态请求权限2、悬浮窗权限说明3、检查动态权限4、申请动态权限5、权限设置完毕后

在 Spring Boot 中使用 @Autowired和 @Bean注解的示例详解

《在SpringBoot中使用@Autowired和@Bean注解的示例详解》本文通过一个示例演示了如何在SpringBoot中使用@Autowired和@Bean注解进行依赖注入和Bean... 目录在 Spring Boot 中使用 @Autowired 和 @Bean 注解示例背景1. 定义 Stud

使用 sql-research-assistant进行 SQL 数据库研究的实战指南(代码实现演示)

《使用sql-research-assistant进行SQL数据库研究的实战指南(代码实现演示)》本文介绍了sql-research-assistant工具,该工具基于LangChain框架,集... 目录技术背景介绍核心原理解析代码实现演示安装和配置项目集成LangSmith 配置(可选)启动服务应用场景

oracle DBMS_SQL.PARSE的使用方法和示例

《oracleDBMS_SQL.PARSE的使用方法和示例》DBMS_SQL是Oracle数据库中的一个强大包,用于动态构建和执行SQL语句,DBMS_SQL.PARSE过程解析SQL语句或PL/S... 目录语法示例注意事项DBMS_SQL 是 oracle 数据库中的一个强大包,它允许动态地构建和执行

如何通过海康威视设备网络SDK进行Java二次开发摄像头车牌识别详解

《如何通过海康威视设备网络SDK进行Java二次开发摄像头车牌识别详解》:本文主要介绍如何通过海康威视设备网络SDK进行Java二次开发摄像头车牌识别的相关资料,描述了如何使用海康威视设备网络SD... 目录前言开发流程问题和解决方案dll库加载不到的问题老旧版本sdk不兼容的问题关键实现流程总结前言作为

SpringBoot中使用 ThreadLocal 进行多线程上下文管理及注意事项小结

《SpringBoot中使用ThreadLocal进行多线程上下文管理及注意事项小结》本文详细介绍了ThreadLocal的原理、使用场景和示例代码,并在SpringBoot中使用ThreadLo... 目录前言技术积累1.什么是 ThreadLocal2. ThreadLocal 的原理2.1 线程隔离2

Python中顺序结构和循环结构示例代码

《Python中顺序结构和循环结构示例代码》:本文主要介绍Python中的条件语句和循环语句,条件语句用于根据条件执行不同的代码块,循环语句用于重复执行一段代码,文章还详细说明了range函数的使... 目录一、条件语句(1)条件语句的定义(2)条件语句的语法(a)单分支 if(b)双分支 if-else(