在训练TensorFlow模型（!!不是Keras模型）时，如何获取模型中间层（op）的输入和输出？

Question

在训练 TensorFlow 模型（!!不是 Keras 模型）时，是否可以获得模型中间层（op）的输入和输出？

我使用来自Tensorflow tutorial 的示例作为演示：

为了更清楚地解释，我在不改变其目的的情况下对模型进行了一些更改。

张量流版本：2.8.0

class MyModel(Model):
  def __init__(self):
    super(MyModel, self).__init__()
    self.conv1 = Conv2D(32, 3, activation='relu')
    self.flatten = Flatten()
    self.d1 = Dense(128, activation='relu')
    self.d2 = Dense(10)

  def call(self, x):
    # x = self.conv1(x)
    # x = self.flatten(x)
    # x = self.d1(x)
    # return self.d2(x)
    x1 = self.conv1(x) 
    x2 = self.flatten(x1)
    x3 = self.d1(x2)
    return self.d2(x3)

是否可以得到模型中的x1、x2、x3或self.conv1的输入输出？

Answer 1

我建议在训练期间使用自定义 Keras 回调将数据提供给模型，然后保存权重和输出。您可以向回调提供您的训练数据或其他数据，例如您的测试数据：

import tensorflow as tf
import numpy as np

class MyModel(tf.keras.Model):
  def __init__(self):
    super(MyModel, self).__init__()
    self.conv1 = tf.keras.layers.Conv2D(32, 3, activation='relu')
    self.flatten = tf.keras.layers.Flatten()
    self.d1 = tf.keras.layers.Dense(128, activation='relu')
    self.d2 = tf.keras.layers.Dense(10)

  def call(self, x):
    x1 = self.conv1(x) 
    x2 = self.flatten(x1)
    x3 = self.d1(x2)
    return self.d2(x3)

class CustomCallback(tf.keras.callbacks.Callback):
   def __init__(self, data):
        self.data = data
   def on_epoch_end(self, epoch, logs=None):
        #if epoch == some_value: <-- can also define a condition
        conv_layer = self.model.layers[0]
        outputs = conv_layer(self.data)
        np.save('conv_outputs', np.array(outputs)) 
        np.save('conv_weights', np.array(conv_layer.weights))
        tf.print('Saved Conv2D outputs and weights')

model = MyModel()
x_train = tf.random.normal((10, 32, 32, 3))
x_test = tf.random.normal((10, 32, 32, 3))
model.compile(optimizer='adam', loss = tf.keras.losses.SparseCategoricalCrossentropy(True))
model.fit(x_train, tf.random.uniform((10, 1), maxval=10), epochs=2, callbacks=[CustomCallback(x_test)], batch_size=2)

Answer 2

有很多方法，但我考虑了模型学习的目的，当您从自定义类开始时，您可以为模型分配一些值，并且您可以以相同的方式读取层权重或模型（X）或模型的输出。预测(X) 值。

[ 样本 ] ：

import os
from os.path import exists

import gym
import ale_py

import tensorflow as tf
import tensorflow_io as tfio

import matplotlib.pyplot as plt
import matplotlib.animation as animation

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
[PhysicalDevice(name='/physical_device:GPU:0', device_type='GPU')]
None
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
physical_devices = tf.config.experimental.list_physical_devices('GPU')
assert len(physical_devices) > 0, "Not enough GPU hardware devices available"
config = tf.config.experimental.set_memory_growth(physical_devices[0], True)
print(physical_devices)
print(config)

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Games Environments
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
env = gym.make("ALE/ChopperCommand-v5")
n_outputs = env.action_space.n
obs = env.reset()
observation, reward, done, info = env.step(1)

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Variables
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
global step
global action
step = 1
action = 1
CROP_SIZE = [ 210, 160 ]
IMAGE_SIZE = [ 210, 160, 3 ] 
BATCH_SIZE = 1
NUM_BOXES = 1
LONG_STEPS = 100000000000

boxes = tf.constant([ 0.26, 0.05, 0.8, 1.0 ], shape=(1, 4))
box_indices = tf.constant([ 0 ], shape=(1, ))

fig = plt.figure()
image = plt.imread( "F:\datasets\downloads\cats_name\train\Symbols\01.jpg" )
im = plt.imshow(image)

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Class / Definition
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
class MyModel(tf.keras.Model):
    def __init__(self):
        super().__init__()
        self.optimizer = tf.keras.optimizers.Nadam( learning_rate=0.00001, beta_1=0.9, beta_2=0.999, epsilon=1e-07, name='Nadam')
        self.lossfn = tf.keras.losses.SparseCategoricalCrossentropy( from_logits=False, reduction=tf.keras.losses.Reduction.AUTO, name='sparse_categorical_crossentropy' )
        
        self.input1 = tf.keras.layers.InputLayer(input_shape=( 210, 160, 1 ), name="input_01")
        self.normalize1 = tf.keras.layers.Normalization(mean=3., variance=2., name="normalize_01"),
        self.normalize2 = tf.keras.layers.Normalization(mean=4., variance=6., name="normalize_01"),
        self.conv2d = tf.keras.layers.Conv2D(32, (3, 3), activation='relu')
        self.maxpool2d = tf.keras.layers.MaxPooling2D((2, 2))
        self.dense1 = tf.keras.layers.Dense(4, activation=tf.nn.relu, name="Dense_01")
        self.dense2 = tf.keras.layers.Dense(5, activation=tf.nn.softmax, name="Dense_02")
        self.dense3 = tf.keras.layers.Dense(16, activation=tf.nn.softmax, name="Dense_03")
        self.flattern = tf.keras.layers.Flatten(name="flattern")
        self.model = tf.keras.models.Sequential([
            self.input1,
            tf.keras.layers.Normalization(mean=3., variance=2.),
            tf.keras.layers.Normalization(mean=4., variance=6.),
            tf.keras.layers.Conv2D(32, (3, 3), activation='relu'),
            tf.keras.layers.MaxPooling2D((2, 2)),
            tf.keras.layers.Flatten(name="flattern"),
            self.dense3,
        ])
    
    def call(self, inputs, training=None):
        result = self.model( inputs, training  )
        
        return result

def animate( i ):
    global step
    global action

    step = step + 1
    
    observation, reward, done, info = env.step(action)
    env.render()
    im.set_array( observation )
    
    image_array = tf.keras.preprocessing.image.img_to_array( observation )
    image_cropped = tf.image.crop_and_resize( tf.expand_dims(image_array, axis=0), boxes, box_indices, CROP_SIZE )
    image_cropped = tf.reshape( image_cropped[0], IMAGE_SIZE )
    # grey scales
    image_greyscales = tf.image.rgb_to_grayscale( image_cropped ).numpy()
    
    prediction_result = model.predict(tf.constant(tf.cast(image_greyscales, dtype=tf.int64), shape=(1, 210, 160, 1), dtype=tf.int64))
    action = tf.constant(tf.math.argmax( prediction_result[0] ), shape=(1, 1, 1), dtype=tf.int64)[0][0][0].numpy()
    
    if reward > 0 :
        input_dataset = tf.constant(tf.cast(image_greyscales, dtype=tf.int64 ), shape=(1, 1, 210, 160, 1), dtype=tf.int64)
        label_dataset = tf.constant( action, shape=(1, 1, 1), dtype=tf.int64 )
        dataset = tf.data.Dataset.from_tensor_slices(( input_dataset, label_dataset ))
        history = model.fit( dataset, batch_size=100, epochs=10, callbacks=[custom_callback] )
    else :

        if step % 8 == 0 :
            action = random_action( action )
            observation, reward, done, info = env.step(action)
            env.render()
            im.set_array( observation )
            
            image_array = tf.keras.preprocessing.image.img_to_array( observation )
            image_cropped = tf.image.crop_and_resize( tf.expand_dims(image_array, axis=0), boxes, box_indices, CROP_SIZE )
            image_cropped = tf.reshape( image_cropped[0], IMAGE_SIZE )
            image_greyscales = tf.image.rgb_to_grayscale( image_cropped ).numpy()
            
            input_dataset = tf.constant(tf.cast(image_greyscales, dtype=tf.int64 ), shape=(1, 1, 210, 160, 1), dtype=tf.int64)
            label_dataset = tf.constant( action, shape=(1, 1, 1), dtype=tf.int64 )
            dataset = tf.data.Dataset.from_tensor_slices(( input_dataset, label_dataset ))
            history = model.fit( dataset, batch_size=100, epochs=10, callbacks=[custom_callback] )
        else :
            pass

    plt.xlabel( str(step) + ": action = " + str( action ) )
    plt.show()
    return im,
    
def random_action(action): 

    temp = tf.random.normal([n_outputs], 1, 0.2, tf.float32)
    # temp_2 = tf.constant( tf.ones(( n_outputs )) * 48 )
    action = tf.math.argmax(temp).numpy()

    return action

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Callback
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
class custom_callback(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs={}):
        if( logs['accuracy'] >= 0.97 ):
            self.model.stop_training = True
            return
        else :
            output_layer = self.model.get_layer( name="Dense_03" )  #   <keras.layers.core.dense.Dense object at 0x000002CB9D3267F0>
            # you may utilize the model training weight here
    
custom_callback = custom_callback()
    
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: DataSet
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
image_array = tf.keras.preprocessing.image.img_to_array( observation )
image_cropped = tf.image.crop_and_resize( tf.expand_dims(image_array, axis=0), boxes, box_indices, CROP_SIZE )
image_cropped = tf.reshape( image_cropped[0], IMAGE_SIZE )
# grey scales
image_greyscales = tf.image.rgb_to_grayscale( image_cropped ).numpy()

input_dataset = tf.constant(tf.cast(image_greyscales, dtype=tf.int64), shape=(1, 1, 210, 160, 1), dtype=tf.int64)
label_dataset = tf.constant([0], shape=(1, 1, 1), dtype=tf.int64)
dataset = tf.data.Dataset.from_tensor_slices(( input_dataset, label_dataset ))

model = MyModel( )
optimizer = tf.keras.optimizers.Nadam( learning_rate=0.00001, beta_1=0.9, beta_2=0.999, epsilon=1e-07, name='Nadam')
lossfn = tf.keras.losses.SparseCategoricalCrossentropy( from_logits=False, reduction=tf.keras.losses.Reduction.AUTO, name='sparse_categorical_crossentropy' )
model.compile(optimizer=optimizer, loss=lossfn, metrics=['accuracy'])
model.build(input_shape=( None, 210, 160, 1 ))
model.summary()
history = model.fit( dataset, batch_size=100, epochs=5, callbacks=[custom_callback] )

prediction_result = model.predict(tf.constant(tf.cast(image_greyscales, dtype=tf.int64), shape=(1, 210, 160, 1), dtype=tf.int64))
print( prediction_result )

"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
: Tasks
"""""""""""""""""""""""""""""""""""""""""""""""""""""""""
while LONG_STEPS > 0:
    ani = animation.FuncAnimation(fig, animate, interval=50, blit=True)
    plt.show()

plt.close()