Строительство более умных моделей с наиболее недооцененной функцией Tensorflow - типами разжигания

28 июля 2025 г.

Обзор контента

• Tensor API отправка
• Отправка на один API
• Отправка для всех Unary Elementwise API
• Отправка для бинарных All ElementWise API
• ПАКТЕРНЫЕ EXTENSIONTYPES
• BatchableExtensionType Пример: сеть
• API -интерфейсы TensorFlow, которые поддерживают extensionTypes
• @tf.function
• Управление потоком
• Поток управления автографами
• Керас
• Сохраняйте модель
• Наборы данных

Tensor API отправка

Типы расширения могут быть «тензорными», в том смысле, что они специализируют или расширяют интерфейс, определяемыйtf.Tensorтип. Примеры типов расширения, подобных тензору, включаютRaggedTensorВSparseTensor, иMaskedTensorПолемОтправка декораторовМожет использоваться для переопределения поведения по умолчанию операций Tensorflow при применении к тензороподобным типам расширения. Tensorflow в настоящее время определяет три декоратора диспетчеризации:

• @tf.experimental.dispatch_for_api(tf_api)
• @tf.experimental.dispatch_for_unary_elementwise_apis(x_type)
• @tf.experimental.dispatch_for_binary_elementwise_apis(x_type, y_type)

Отправка на один API

Аtf.experimental.dispatch_for_apiДекоратор переопределяет поведение по умолчанию указанной операции TensorFlow, когда она вызывается с указанной подписью. Например, вы можете использовать этот декоратор, чтобы указать, какtf.stackдолжен обработатьMaskedTensorценности:

@tf.experimental.dispatch_for_api(tf.stack)
def masked_stack(values: List[MaskedTensor], axis = 0):
  return MaskedTensor(tf.stack([v.values for v in values], axis),
                      tf.stack([v.mask for v in values], axis))

Это переопределяет реализацию по умолчанию дляtf.stackВсякий раз, когда это вызывается со спискомMaskedTensorЗначения (так какvaluesаргумент аннотирован сtyping.List[MaskedTensor]):

x = MaskedTensor([1, 2, 3], [True, True, False])
y = MaskedTensor([4, 5, 6], [False, True, True])
tf.stack([x, y])

Разрешитьtf.stackобрабатывать списки смешанныхMaskedTensorиTensorзначения, вы можете уточнить аннотацию типа дляvaluesпараметр и соответствующим образом обновите корпус функции:

tf.experimental.unregister_dispatch_for(masked_stack)

def convert_to_masked_tensor(x):
  if isinstance(x, MaskedTensor):
    return x
  else:
    return MaskedTensor(x, tf.ones_like(x, tf.bool))

@tf.experimental.dispatch_for_api(tf.stack)
def masked_stack_v2(values: List[Union[MaskedTensor, tf.Tensor]], axis = 0):
  values = [convert_to_masked_tensor(v) for v in values]
  return MaskedTensor(tf.stack([v.values for v in values], axis),
                      tf.stack([v.mask for v in values], axis))
x = MaskedTensor([1, 2, 3], [True, True, False])
y = tf.constant([4, 5, 6])
tf.stack([x, y, x])

Список API, которые могут быть переопределены, см. Документацию API дляtf.experimental.dispatch_for_apiПолем

Отправка для всех Unary Elementwise API

Аtf.experimental.dispatch_for_unary_elementwise_apisдекоратор переопределяет поведение по умолчаниювсеUnarary Elementwise Ops (напримерtf.math.cos) всякий раз, когда значение для первого аргумента (обычно называетсяx) соответствует аннотации типаx_typeПолем Украшенная функция должна принимать два аргумента:

• api_func: Функция, которая принимает один параметр и выполняет операцию ElementWise (например,tf.abs)
• x: Первый аргумент в отношении операции ElementWise.

Следующий пример обновляет все операции Unary ElementWise для обработкиMaskedTensorтип:

@tf.experimental.dispatch_for_unary_elementwise_apis(MaskedTensor)
 def masked_tensor_unary_elementwise_api_handler(api_func, x):
   return MaskedTensor(api_func(x.values), x.mask)

Эта функция теперь будет использоваться всякий раз, когда вызовая уникальная операция по элементуMaskedTensorПолем

x = MaskedTensor([1, -2, -3], [True, False, True])
 print(tf.abs(x))

print(tf.ones_like(x, dtype=tf.float32))

Отправка для бинарных All ElementWise API

Сходным образом,tf.experimental.dispatch_for_binary_elementwise_apisможно использовать для обновления всех бинарных элементов ElementWise для обработкиMaskedTensorтип:

@tf.experimental.dispatch_for_binary_elementwise_apis(MaskedTensor, MaskedTensor)
def masked_tensor_binary_elementwise_api_handler(api_func, x, y):
  return MaskedTensor(api_func(x.values, y.values), x.mask & y.mask)

x = MaskedTensor([1, -2, -3], [True, False, True])
y = MaskedTensor([[4], [5]], [[True], [False]])
tf.math.add(x, y)

Для списка API ElementWise, которые переопределяются, перейдите в документацию API дляtf.experimental.dispatch_for_unary_elementwise_apisиtf.experimental.dispatch_for_binary_elementwise_apisПолем

ПартийныйExtensionTypeс

АнонцаExtensionTypeявляетсяпартийныйЕсли один экземпляр может быть использован для представления партии значений. Как правило, это достигается путем добавления партийных размеров ко всем вложеннымTensorс Следующие API -интерфейсы TensorFlow требуют, чтобы любые входы типа расширения были пакетными:

• tf.data.Dataset (batchВunbatchВfrom_tensor_slices)
• tf.keras (fitВevaluateВpredict)
• tf.map_fn

По умолчанию,BatchableExtensionTypeсоздает пакетные значения, оставляя любые вложенныеTensorс,CompositeTensorпесокExtensionTypeс Если это не подходит для вашего класса, вам нужно будет использоватьtf.experimental.ExtensionTypeBatchEncoderЧтобы переопределить это поведение по умолчанию. Например, было бы неуместно создать партиюtf.SparseTensorЗначения, просто укладывая отдельные скудные тензоры 'valuesВindices, иdense_shapeПоля - в большинстве случаев вы не можете сложить эти тензоры, поскольку они имеют несовместимые формы; И даже если бы вы могли, результат не будет действительнымSparseTensorПолем

BatchableExtensionTypeпример:Network

В качестве примера рассмотрим простоNetworkКласс, используемый для балансировки нагрузки, который отслеживает, сколько работы остается в каждом узле, и сколько пропускной способности доступно для перемещения работы между узлами:

class Network(tf.experimental.ExtensionType):  # This version is not batchable.
  work: tf.Tensor       # work[n] = work left to do at node n
  bandwidth: tf.Tensor  # bandwidth[n1, n2] = bandwidth from n1->n2

net1 = Network([5., 3, 8], [[0., 2, 0], [2, 0, 3], [0, 3, 0]])
net2 = Network([3., 4, 2], [[0., 2, 2], [2, 0, 2], [2, 2, 0]])

Чтобы сделать этот тип партии, измените базовый тип наBatchableExtensionTypeи отрегулируйте форму каждого поля, чтобы включить дополнительные размеры партии. Следующий пример также добавляетshapeПоле, чтобы отслеживать форму партии. ЭтотshapeПоле не требуетсяtf.data.Datasetилиtf.map_fn, но этоявляетсятребуетсяtf.kerasПолем

class Network(tf.experimental.BatchableExtensionType):
  shape: tf.TensorShape  # batch shape. A single network has shape=[].
  work: tf.Tensor        # work[*shape, n] = work left to do at node n
  bandwidth: tf.Tensor   # bandwidth[*shape, n1, n2] = bandwidth from n1->n2

  def __init__(self, work, bandwidth):
    self.work = tf.convert_to_tensor(work)
    self.bandwidth = tf.convert_to_tensor(bandwidth)
    work_batch_shape = self.work.shape[:-1]
    bandwidth_batch_shape = self.bandwidth.shape[:-2]
    self.shape = work_batch_shape.merge_with(bandwidth_batch_shape)

  def __repr__(self):
    return network_repr(self)

def network_repr(network):
  work = network.work
  bandwidth = network.bandwidth
  if hasattr(work, 'numpy'):
    work = ' '.join(str(work.numpy()).split())
  if hasattr(bandwidth, 'numpy'):
    bandwidth = ' '.join(str(bandwidth.numpy()).split())
  return (f"<Network shape={network.shape} work={work} bandwidth={bandwidth}>")

net1 = Network([5., 3, 8], [[0., 2, 0], [2, 0, 3], [0, 3, 0]])
net2 = Network([3., 4, 2], [[0., 2, 2], [2, 0, 2], [2, 2, 0]])
batch_of_networks = Network(
    work=tf.stack([net1.work, net2.work]),
    bandwidth=tf.stack([net1.bandwidth, net2.bandwidth]))
print(f"net1={net1}")
print(f"net2={net2}")
print(f"batch={batch_of_networks}")

Затем вы можете использоватьtf.data.DatasetИтерация через партию сетей:

dataset = tf.data.Dataset.from_tensor_slices(batch_of_networks)
for i, network in enumerate(dataset):
  print(f"Batch element {i}: {network}")

И вы также можете использоватьmap_fnЧтобы применить функцию к каждому элементу партии:

def balance_work_greedy(network):
  delta = (tf.expand_dims(network.work, -1) - tf.expand_dims(network.work, -2))
  delta /= 4
  delta = tf.maximum(tf.minimum(delta, network.bandwidth), -network.bandwidth)
  new_work = network.work + tf.reduce_sum(delta, -1)
  return Network(new_work, network.bandwidth)

tf.map_fn(balance_work_greedy, batch_of_networks)

API -интерфейс tensorflow, которые поддерживаютExtensionTypeс

@tf.function

tf.functionявляется декоратором, который предварительно считывает графики тензора для функций Python, что может существенно повысить производительность вашего кода TensorFlow. Значения типа расширения могут использоваться прозрачно с помощью@tf.function-Корированные функции.

class Pastry(tf.experimental.ExtensionType):
  sweetness: tf.Tensor  # 2d embedding that encodes sweetness
  chewiness: tf.Tensor  # 2d embedding that encodes chewiness

@tf.function
def combine_pastry_features(x: Pastry):
  return (x.sweetness + x.chewiness) / 2

cookie = Pastry(sweetness=[1.2, 0.4], chewiness=[0.8, 0.2])
combine_pastry_features(cookie)

Если вы хотите явно указатьinput_signatureдляtf.function, тогда вы можете сделать это с помощью типа расширенияTypeSpecПолем

pastry_spec = Pastry.Spec(tf.TensorSpec([2]), tf.TensorSpec(2))

@tf.function(input_signature=[pastry_spec])
def increase_sweetness(x: Pastry, delta=1.0):
  return Pastry(x.sweetness + delta, x.chewiness)

increase_sweetness(cookie)

Конкретные функции

Бетонные функции инкапсулируют отдельные прослеженные графики, которые построеныtf.functionПолем Типы удлинения могут быть прозрачно использованы с помощью конкретных функций.

cf = combine_pastry_features.get_concrete_function(pastry_spec)
cf(cookie)

Управление потоком

Типы расширения подтверждаются операциями управления TensorFlow:

• tf.cond
• tf.case
• tf.while_loop
• tf.identity

# Example: using tf.cond to select between two MaskedTensors. Note that the
# two MaskedTensors don't need to have the same shape.
a = MaskedTensor([1., 2, 3], [True, False, True])
b = MaskedTensor([22., 33, 108, 55], [True, True, True, False])
condition = tf.constant(True)
print(tf.cond(condition, lambda: a, lambda: b))

# Example: using tf.while_loop with MaskedTensor.
cond = lambda i, _: i < 10
def body(i, mt):
  return i + 1, mt.with_values(mt.values + 3 / 7)
print(tf.while_loop(cond, body, [0, b])[1])

Поток управления автографами

Типы расширения также подтверждаются операторами потока управления вtf.function(Используя автограф). В следующем примереifзаявление иforоператоры автоматически преобразуются вtf.condиtf.while_loopОперации, которые поддерживают типы расширения.

@tf.function
def fn(x, b):
  if b:
    x = MaskedTensor(x, tf.less(x, 0))
  else:
    x = MaskedTensor(x, tf.greater(x, 0))
  for i in tf.range(5 if b else 7):
    x = x.with_values(x.values + 1 / 2)
  return x

print(fn(tf.constant([1., -2, 3]), tf.constant(True)))
print(fn(tf.constant([1., -2, 3]), tf.constant(False)))

Керас

tf.kerasэто API высокого уровня Tensorflow для создания и обучения моделей глубокого обучения. Типы расширения могут передаваться в качестве входных данных в модель кераса, передаваемые между слоями кераса и возвращаются моделями Keras. Керас в настоящее время ставит два требования к типам расширения:

• Они должны быть партиями (перейти к "ExtensionTypeS "выше).
• У них должно быть поле или имущество названоshapeПолемshape[0]Предполагается, что является партийным измерением.

В следующих двух подразделах приведены примеры, показывающие, как типы расширения можно использовать с керами.

Керас Пример:Network

Для первого примера рассмотритеNetworkкласс, определяемый в «БэктерномExtensionTypeS "Раздел выше, который можно использовать для работы с балансировкой нагрузки между узлами. Его определение повторяется здесь:

class Network(tf.experimental.BatchableExtensionType):
  shape: tf.TensorShape  # batch shape. A single network has shape=[].
  work: tf.Tensor        # work[*shape, n] = work left to do at node n
  bandwidth: tf.Tensor   # bandwidth[*shape, n1, n2] = bandwidth from n1->n2

  def __init__(self, work, bandwidth):
    self.work = tf.convert_to_tensor(work)
    self.bandwidth = tf.convert_to_tensor(bandwidth)
    work_batch_shape = self.work.shape[:-1]
    bandwidth_batch_shape = self.bandwidth.shape[:-2]
    self.shape = work_batch_shape.merge_with(bandwidth_batch_shape)

  def __repr__(self):
    return network_repr(self)

single_network = Network(  # A single network with 4 nodes.
    work=[8.0, 5, 12, 2],
    bandwidth=[[0.0, 1, 2, 2], [1, 0, 0, 2], [2, 0, 0, 1], [2, 2, 1, 0]])

batch_of_networks = Network(  # Batch of 2 networks, each w/ 2 nodes.
    work=[[8.0, 5], [3, 2]],
    bandwidth=[[[0.0, 1], [1, 0]], [[0, 2], [2, 0]]])

Вы можете определить новый слой кераса, который обрабатываетNetworkс

class BalanceNetworkLayer(tf.keras.layers.Layer):
  """Layer that balances work between nodes in a network.

  Shifts work from more busy nodes to less busy nodes, constrained by bandwidth.
  """
  def call(self, inputs):
    # This function is defined above in the "Batchable `ExtensionType`s" section.
    return balance_work_greedy(inputs)

Затем вы можете использовать эти слои для создания простой модели. КормитьExtensionTypeв модель, вы можете использоватьtf.keras.layer.Inputслой сtype_specустановить на тип расширенияTypeSpecПолем Если модель Keras будет использоваться для обработки партий, тоtype_specдолжен включать в себя партийное измерение.

input_spec = Network.Spec(shape=None,
                          work=tf.TensorSpec(None, tf.float32),
                          bandwidth=tf.TensorSpec(None, tf.float32))
model = tf.keras.Sequential([
    tf.keras.layers.Input(type_spec=input_spec),
    BalanceNetworkLayer(),
    ])

Наконец, вы можете применить модель к одной сети и к партии сетей.

model(single_network)

model(batch_of_networks)

Керас Пример: Маскедтензор

В этом примере,MaskedTensorраспространяется на поддержкуKerasПолемshapeопределяется как свойство, которое рассчитывается изvaluesполе. Керас требует, чтобы вы добавили это свойство как к расширению, так и к егоTypeSpecПолемMaskedTensorтакже определяет а__name__переменная, которая потребуется дляSavedModelсериализация (ниже).

class MaskedTensor(tf.experimental.BatchableExtensionType):
  # __name__ is required for serialization in SavedModel; see below for details.
  __name__ = 'extension_type_colab.MaskedTensor'

  values: tf.Tensor
  mask: tf.Tensor

  shape = property(lambda self: self.values.shape)
  dtype = property(lambda self: self.values.dtype)

  def with_default(self, default):
    return tf.where(self.mask, self.values, default)

  def __repr__(self):
    return masked_tensor_str(self.values, self.mask)

  class Spec:
    def __init__(self, shape, dtype=tf.float32):
      self.values = tf.TensorSpec(shape, dtype)
      self.mask = tf.TensorSpec(shape, tf.bool)

    shape = property(lambda self: self.values.shape)
    dtype = property(lambda self: self.values.dtype)

    def with_shape(self):
      return MaskedTensor.Spec(tf.TensorSpec(shape, self.values.dtype),
                               tf.TensorSpec(shape, self.mask.dtype))

Затем декораторы диспетчеры используются для переопределения поведения по умолчанию нескольких API -интерфейсов TensorFlow. Поскольку эти API используются стандартными слоями кераса (например,Denseслой), переопределение их позволит нам использовать эти слои сMaskedTensorПолем Для целей этого примераmatmulДля маскированных тензоров определяется для обработки значений в масках как нулей (то есть не включать их в продукт).

@tf.experimental.dispatch_for_unary_elementwise_apis(MaskedTensor)
def unary_elementwise_op_handler(op, x):
 return MaskedTensor(op(x.values), x.mask)

@tf.experimental.dispatch_for_binary_elementwise_apis(
    Union[MaskedTensor, tf.Tensor],
    Union[MaskedTensor, tf.Tensor])
def binary_elementwise_op_handler(op, x, y):
  x = convert_to_masked_tensor(x)
  y = convert_to_masked_tensor(y)
  return MaskedTensor(op(x.values, y.values), x.mask & y.mask)

@tf.experimental.dispatch_for_api(tf.matmul)
def masked_matmul(a: MaskedTensor, b,
                  transpose_a=False, transpose_b=False,
                  adjoint_a=False, adjoint_b=False,
                  a_is_sparse=False, b_is_sparse=False,
                  output_type=None):
  if isinstance(a, MaskedTensor):
    a = a.with_default(0)
  if isinstance(b, MaskedTensor):
    b = b.with_default(0)
  return tf.matmul(a, b, transpose_a, transpose_b, adjoint_a,
                   adjoint_b, a_is_sparse, b_is_sparse, output_type)

Затем вы можете построить модель кераса, которая принимаетMaskedTensorВходы, используя стандартные слои кераса:

input_spec = MaskedTensor.Spec([None, 2], tf.float32)

masked_tensor_model = tf.keras.Sequential([
    tf.keras.layers.Input(type_spec=input_spec),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1)])
masked_tensor_model.compile(loss='binary_crossentropy', optimizer='rmsprop')

a = MaskedTensor([[1., 2], [3, 4], [5, 6]],
                  [[True, False], [False, True], [True, True]])
masked_tensor_model.fit(a, tf.constant([[1], [0], [1]]), epochs=3)
print(masked_tensor_model(a))

Сохраняйте модель

АСохраняйте модельявляется сериализованной программой Tensorflow, включая веса и вычисления. Он может быть построен из модели Keras или из пользовательской модели. В любом случае типы расширения могут использоваться прозрачно с помощью функций и методов, определяемых SavedModel.

SaveedModel может сохранять модели, слои и функции, которые обрабатывают типы расширения, если типы расширения имеют__name__поле. Это имя используется для регистрации типа расширения, поэтому оно может быть расположено при загрузке модели.

Пример: сохранение модели кераса

Модели кераса, которые используют типы расширения, могут быть сохранены с помощьюSavedModelПолем

masked_tensor_model_path = tempfile.mkdtemp()
tf.saved_model.save(masked_tensor_model, masked_tensor_model_path)
imported_model = tf.saved_model.load(masked_tensor_model_path)
imported_model(a)

Пример: сохранение пользовательской модели

SaveedModel также может использоваться для сохранения пользовательскогоtf.ModuleПодклассы с функциями, которые обрабатывают типы расширения.

class CustomModule(tf.Module):
  def __init__(self, variable_value):
    super().__init__()
    self.v = tf.Variable(variable_value)

  @tf.function
  def grow(self, x: MaskedTensor):
    """Increase values in `x` by multiplying them by `self.v`."""
    return MaskedTensor(x.values * self.v, x.mask)

module = CustomModule(100.0)

module.grow.get_concrete_function(MaskedTensor.Spec(shape=None,
                                                    dtype=tf.float32))
custom_module_path = tempfile.mkdtemp()
tf.saved_model.save(module, custom_module_path)
imported_model = tf.saved_model.load(custom_module_path)
imported_model.grow(MaskedTensor([1., 2, 3], [False, True, False]))

Загрузка сохраненной модели, когдаExtensionTypeнедоступен

Если вы загрузитеSavedModelкоторый используетExtensionType, но этоExtensionTypeнедоступен (то есть он не был импортирован), тогда вы получите предупреждение, а TensorFlow вернется к использованию объекта «анонимный расширение». Этот объект будет иметь те же поля, что и исходный тип, но не будет никакой дальнейшей настройки, которую вы добавили для типа, такие как пользовательские методы или свойства.

С использованиемExtensionTypeS с TensorFlow Aerding

В настоящее время,Tensorflow Aerding(и другие потребители словаря «подписи» сохранения модели) требуют, чтобы все входы и выходы были необработанными тензорами. Если вы хотите использовать TensorFlow Ared с моделью, которая использует типы расширения, вы можете добавить методы обертки, которые составляют или разлагают значения типа расширения из тензоров. Например:

class CustomModuleWrapper(tf.Module):
  def __init__(self, variable_value):
    super().__init__()
    self.v = tf.Variable(variable_value)

  @tf.function
  def var_weighted_mean(self, x: MaskedTensor):
    """Mean value of unmasked values in x, weighted by self.v."""
    x = MaskedTensor(x.values * self.v, x.mask)
    return (tf.reduce_sum(x.with_default(0)) /
            tf.reduce_sum(tf.cast(x.mask, x.dtype)))

  @tf.function()
  def var_weighted_mean_wrapper(self, x_values, x_mask):
    """Raw tensor wrapper for var_weighted_mean."""
    return self.var_weighted_mean(MaskedTensor(x_values, x_mask))

module = CustomModuleWrapper([3., 2., 8., 5.])

module.var_weighted_mean_wrapper.get_concrete_function(
    tf.TensorSpec(None, tf.float32), tf.TensorSpec(None, tf.bool))
custom_module_path = tempfile.mkdtemp()
tf.saved_model.save(module, custom_module_path)
imported_model = tf.saved_model.load(custom_module_path)
x = MaskedTensor([1., 2., 3., 4.], [False, True, False, True])
imported_model.var_weighted_mean_wrapper(x.values, x.mask)

Datasetс

tf.dataэто API, который позволяет вам создавать сложные входные трубопроводы из простых многоразовых кусочков. Его основная структура данныхtf.data.Dataset, который представляет последовательность элементов, в которой каждый элемент состоит из одного или нескольких компонентов.

ЗданиеDatasetS с типами расширения

Наборы данных могут быть построены из значений типа расширения, используяDataset.from_tensorsВDataset.from_tensor_slices, илиDataset.from_generator:

ds = tf.data.Dataset.from_tensors(Pastry(5, 5))
iter(ds).next()

mt = MaskedTensor(tf.reshape(range(20), [5, 4]), tf.ones([5, 4]))
ds = tf.data.Dataset.from_tensor_slices(mt)
for value in ds:
  print(value)

def value_gen():
  for i in range(2, 7):
    yield MaskedTensor(range(10), [j%i != 0 for j in range(10)])

ds = tf.data.Dataset.from_generator(
    value_gen, output_signature=MaskedTensor.Spec(shape=[10], dtype=tf.int32))
for value in ds:
  print(value)

Партии и непредвзятыеDatasetS с типами расширения

Наборы данных с типами расширения могут быть смещены и не снимаются с использованиемDataset.batchиDataset.unbatchПолем

batched_ds = ds.batch(2)
for value in batched_ds:
  print(value)

unbatched_ds = batched_ds.unbatch()
for value in unbatched_ds:
  print(value)

---

Оригинал