自定义 MinDiffModel

介绍

在大多数情况下,直接使用 MinDiffModel(如 "将 MinDiff 与 MinDiffModel 集成" 指南 中所述)就足够了。但是,您可能需要自定义行为。造成这种情况的两个主要原因是

  • 您正在使用的 keras.Model 具有您想要保留的自定义行为。
  • 您希望 MinDiffModel 的行为与默认行为不同。

在这两种情况下,您都需要对 MinDiffModel 进行子类化以实现所需的结果。

设置

pip install --upgrade tensorflow-model-remediation
import tensorflow as tf
tf.get_logger().setLevel('ERROR')  # Avoid TF warnings.
from tensorflow_model_remediation import min_diff
from tensorflow_model_remediation.tools.tutorials_utils import uci as tutorials_utils

首先,下载数据。为了简洁起见,输入准备逻辑已分解为辅助函数,如 输入准备指南 中所述。您可以阅读完整指南以了解有关此过程的详细信息。

# Original Dataset for training, sampled at 0.3 for reduced runtimes.
train_df = tutorials_utils.get_uci_data(split='train', sample=0.3)
train_ds = tutorials_utils.df_to_dataset(train_df, batch_size=128)

# Dataset needed to train with MinDiff.
train_with_min_diff_ds = (
    tutorials_utils.get_uci_with_min_diff_dataset(split='train', sample=0.3))

保留原始模型自定义项

tf.keras.Model 旨在通过子类化轻松自定义,如 此处 所述。如果您的模型具有您希望在应用 MinDiff 时保留的自定义实现,则需要对 MinDiffModel 进行子类化。

原始自定义模型

要了解如何保留自定义项,请创建一个自定义模型,该模型在调用其自定义 train_step 时将属性设置为 True。这不是一个有用的自定义项,但将用于说明行为。

class CustomModel(tf.keras.Model):

  # Customized train_step
  def train_step(self, *args, **kwargs):
    self.used_custom_train_step = True  # Marker that we can check for.
    return super(CustomModel, self).train_step(*args, **kwargs)

训练此类模型与训练普通 Sequential 模型相同。

model = tutorials_utils.get_uci_model(model_class=CustomModel)  # Use CustomModel.

model.compile(optimizer='adam', loss='binary_crossentropy')

_ = model.fit(train_ds.take(1), epochs=1, verbose=0)

# Model has used the custom train_step.
print('Model used the custom train_step:')
print(hasattr(model, 'used_custom_train_step'))  # True

对 MinDiffModel 进行子类化

如果您尝试直接使用 MinDiffModel,则模型将不会使用自定义 train_step

model = tutorials_utils.get_uci_model(model_class=CustomModel)
model = min_diff.keras.MinDiffModel(model, min_diff.losses.MMDLoss())

model.compile(optimizer='adam', loss='binary_crossentropy')

_ = model.fit(train_with_min_diff_ds.take(1), epochs=1, verbose=0)

# Model has not used the custom train_step.
print('Model used the custom train_step:')
print(hasattr(model, 'used_custom_train_step'))  # False

为了使用正确的 train_step 方法,您需要一个自定义类,该类同时对 MinDiffModelCustomModel 进行子类化。

class CustomMinDiffModel(min_diff.keras.MinDiffModel, CustomModel):
  pass  # No need for any further implementation.

训练此模型将使用 CustomModel 中的 train_step

model = tutorials_utils.get_uci_model(model_class=CustomModel)

model = CustomMinDiffModel(model, min_diff.losses.MMDLoss())

model.compile(optimizer='adam', loss='binary_crossentropy')

_ = model.fit(train_with_min_diff_ds.take(1), epochs=1, verbose=0)

# Model has used the custom train_step.
print('Model used the custom train_step:')
print(hasattr(model, 'used_custom_train_step'))  # True

自定义 MinDiffModel 的默认行为

在其他情况下,您可能希望更改 MinDiffModel 的特定默认行为。最常见的用例是更改默认解包行为,以便在不使用 pack_min_diff_data 的情况下正确处理您的数据。

将数据打包到自定义格式时,可能如下所示。

def _reformat_input(inputs, original_labels):
  min_diff_data = min_diff.keras.utils.unpack_min_diff_data(inputs)
  original_inputs = min_diff.keras.utils.unpack_original_inputs(inputs)

  return ({
      'min_diff_data': min_diff_data,
      'original_inputs': original_inputs}, original_labels)

customized_train_with_min_diff_ds = train_with_min_diff_ds.map(_reformat_input)

数据集 customized_train_with_min_diff_ds 返回由元组 (x, y) 组成的批次,其中 x 是包含 min_diff_dataoriginal_inputs 的字典,而 yoriginal_labels

for x, _ in customized_train_with_min_diff_ds.take(1):
  print('Type of x:', type(x))  # dict
  print('Keys of x:', x.keys())  # 'min_diff_data', 'original_inputs'

此数据格式不是 MinDiffModel 默认情况下所期望的格式,将 customized_train_with_min_diff_ds 传递给它会导致意外行为。要解决此问题,您需要创建自己的子类。

class CustomUnpackingMinDiffModel(min_diff.keras.MinDiffModel):

  def unpack_min_diff_data(self, inputs):
    return inputs['min_diff_data']

  def unpack_original_inputs(self, inputs):
    return inputs['original_inputs']

使用此子类,您可以像其他示例一样进行训练。

model = tutorials_utils.get_uci_model()
model = CustomUnpackingMinDiffModel(model, min_diff.losses.MMDLoss())

model.compile(optimizer='adam', loss='binary_crossentropy')

_ = model.fit(customized_train_with_min_diff_ds, epochs=1)

自定义 MinDiffModel 的局限性

创建自定义 MinDiffModel 为更复杂的用例提供了极大的灵活性。但是,它仍然不支持某些边缘情况。

call 之前对输入进行预处理或验证

MinDiffModel 进行子类化最大的局限性在于,它要求输入数据的 x 组件(即由 tf.data.Dataset 返回的批次中的第一个或唯一元素)在不进行预处理或验证的情况下传递给 call

这是因为 min_diff_data 被打包到输入数据的 x 组件中。任何预处理或验证都不会期望包含 min_diff_data 的额外结构,并且很可能会中断。

如果预处理或验证很容易自定义(例如,分解为其自身方法),则可以通过覆盖它来轻松解决此问题,以确保它能够正确处理额外的结构。

带有验证的示例可能如下所示

class CustomMinDiffModel(min_diff.keras.MinDiffModel, CustomModel):

  # Override so that it correctly handles additional `min_diff_data`.
  def validate_inputs(self, inputs):
    original_inputs = self.unpack_original_inputs(inputs)
    ...  # Optionally also validate min_diff_data
    # Call original validate method with correct inputs
    return super(CustomMinDiffModel, self).validate(original_inputs)

如果预处理或验证不容易自定义,那么使用 MinDiffModel 可能不适合你,你需要按照 本指南 中的说明集成 MinDiff。

方法名称冲突

你的模型可能包含方法,其名称与 MinDiffModel 中实现的方法冲突(请参阅 API 文档 中的公共方法完整列表)。

只有在模型实例上调用这些方法(而不是在其他方法内部调用)时才会出现问题。虽然不太可能,但如果你遇到这种情况,你将不得不覆盖并重命名一些方法,或者,如果不可能,你可能需要考虑按照 本指南 中的说明集成 MinDiff,而无需使用 MinDiffModel

其他资源

  • 有关公平性评估的深入讨论,请参阅 公平性指标指南
  • 有关修复和 MinDiff 的一般信息,请参阅 修复概述
  • 有关 MinDiff 周围要求的详细信息,请参阅 本指南
  • 要查看在 Keras 中使用 MinDiff 的端到端教程,请参阅 本教程