自定义 MinDiffModel

在 TensorFlow.org 上查看 在 Google Colab 中运行 在 GitHub 上查看源码 下载笔记本

入门

在大多数情况下，直接使用 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 方法，您需要一个同时继承自 MinDiffModel 和 CustomModel 的自定义类。

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，则需要更改默认的解包 (unpacking) 行为以正确处理您的数据。

当将数据打包为自定义格式时，可能会出现如下情况。

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_data 和 original_inputs 的字典，y 是 original_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 文档中公共方法的完整列表）。

只有当这些方法在模型实例上被调用时（而不是在其他方法内部调用），这才会成为问题。虽然这种情况极不可能发生，但如果您确实处于这种情况，则必须重写并重命名某些方法，或者如果无法重命名，则可能需要考虑按照该指南在不使用 MinDiffModel 的情况下集成 MinDiff。

其他资源