具有批次数量循环的张量构造

Question

我想创建一个张量，它是某种变换矩阵（例如旋转矩阵）

我的模型预测 2 个参数：x1 和 x2

所以当 B 是批次数时，输出是 (B, 2) 的张量。

但是，当我写损失时，我必须知道这个“B”，因为我想迭代它：

def get_rotation_tensor(x):
    roll_mat = K.stack([ [[1, 0, 0],
                          [0, K.cos(x[i, 0]), -K.sin(x[i, 0])],
                          [0, K.sin(x[i, 0]), K.cos(x[i, 0])]] for i in range(BATCH_SIZE)])
    pitch_mat = K.stack([ [[K.cos(x[i, 1]), 0, K.sin(x[i, 1])],
                           [0, 1, 0],
                           [-K.sin(x[i, 1]), 0, K.cos(x[i, 1])]] for i in range(BATCH_SIZE)])
    return K.batch_dot(pitch_mat, roll_mat)

我能想到的唯一解决方案是提前预定义 BATCH_SIZE。但是有没有办法编写一个适用于每个批次大小的通用损失函数？

谢谢

Answer 1

我找到了解决办法

def get_rotation_tensor(x):
    ones = K.ones_like(x[:, 0])
    zeros = K.zeros_like(x[:, 0])
    roll_mat = K.stack([[ones, zeros, zeros],
                          [zeros, K.cos(x[:, 0]), -K.sin(x[:, 0])],
                          [zeros, K.sin(x[:, 0]), K.cos(x[:, 0])]])
    pitch_mat = K.stack([[K.cos(x[:, 1]), zeros, K.sin(x[:, 1])],
                           [zeros, ones, zeros],
                           [-K.sin(x[:, 1]), zeros, K.cos(x[:, 1])]])
    return K.batch_dot(K.permute_dimensions(pitch_mat, (2, 0, 1)), 
                       K.permute_dimensions(roll_mat, (2, 0, 1)))

Answer 2

也许我没有完全理解您的问题，但您不能仅通过传递给损失函数的张量的形状来确定批量大小。下面是一个展示这个想法的例子。我希望这会有所帮助。

# Install TensorFlow
try:
  # %tensorflow_version only exists in Colab.
  %tensorflow_version 2.x
except Exception:
  pass

import tensorflow as tf
print(tf.__version__)
print(tf.executing_eagerly())

# Setup repro section from Keras FAQ with TF1 to TF2 adjustments

import numpy as np
import random as rn

# The below is necessary for starting Numpy generated random numbers
# in a well-defined initial state.

np.random.seed(42)

# The below is necessary for starting core Python generated random numbers
# in a well-defined state.

rn.seed(12345)

# Force TensorFlow to use single thread.
# Multiple threads are a potential source of non-reproducible results.
# For further details, see: https://stackoverflow.com/questions/42022950/

session_conf = tf.compat.v1.ConfigProto(intra_op_parallelism_threads=1,
                                        inter_op_parallelism_threads=1)

# The below tf.set_random_seed() will make random number generation
# in the TensorFlow backend have a well-defined initial state.
# For further details, see:
# https://www.tensorflow.org/api_docs/python/tf/set_random_seed

tf.compat.v1.set_random_seed(1234)

sess = tf.compat.v1.Session(graph=tf.compat.v1.get_default_graph(), config=session_conf)
tf.compat.v1.keras.backend.set_session(sess)

# Rest of code follows ...

# Custom Loss
def my_custom_loss(y_true, y_pred):

    tf.print('inside my_custom_loss:')
    tf.print('y_true:')
    tf.print(y_true)
    tf.print('y_true column 0:')
    tf.print(y_true[:,0])
    tf.print('y_true column 1:')
    tf.print(y_true[:,1])
    tf.print('y_pred:')
    tf.print(y_pred)

# get length/batch size

    batch_size=tf.shape(y_pred)[0]
    tf.print('batch_size:')
    tf.print(batch_size)

    y_zeros = tf.zeros_like(y_pred)
    y_mask = tf.math.greater(y_pred, y_zeros)
    res = tf.boolean_mask(y_pred, y_mask)
    logres = tf.math.log(res)
    finres = tf.math.reduce_sum(logres)

    return finres

# Define model
model = tf.keras.models.Sequential()
model.add(tf.keras.layers.Dense(1, activation='linear', input_dim=1, name="Dense1"))
model.compile(optimizer='rmsprop', loss=my_custom_loss)
print('model.summary():')
print(model.summary())

# Generate dummy data
data = np.array([[2.0],[1.0],[1.0],[3.0],[4.0]])
labels = np.array([[[2.0],[1.0]],
                   [[0.0],[3.0]],
                   [[0.0],[3.0]],
                   [[0.0],[3.0]],
                   [[0.0],[3.0]]])

# Train the model.
print('training the model:')
print('-----')
model.fit(data, labels, epochs=1, batch_size=3)
print('done training the model.')

print(data.shape)
print(labels.shape)