(5)keras函数式API

作者: 纵春水东流 | 来源:发表于2021-05-10 21:26 被阅读0次

(5)keras函数式API
keras定义模型的两种方法
构建高级模型(05)
Keras函数式 API
keras
DL4J中文文档/Keras模型导入/函数模型
Keras多输出模型构建
keras函数式编程如何使用BN还有RELU
用Keras的函数式API创建模型
TensorFlow2.0教程-keras 函数api

一、目录
1、多输入模型
2、多输出模型
3、模块
inception
残差链接
共享参数
4、自定义模块
5、模型集成
6、回调函数

二、代码
1、多输入模型
输入问答模型

from keras.models import Model
from keras import layers
from keras import Input

text_vocabulary_size = 10000
question_vocabulary_size = 10000
answer_vocabulary_size = 500

text_input = Input(shape=(None,), dtype='int32', name='text')
embedded_text = layers.Embedding(text_vocabulary_size, 64)(text_input)
encoded_text = layers.LSTM(32)(embedded_text)

question_input = Input(shape=(None,),dtype='int32',name='question')
embedded_question = layers.Embedding(question_vocabulary_size, 32)(question_input)
encoded_question = layers.LSTM(16)(embedded_question)

concatenated = layers.concatenate([encoded_text, encoded_question],axis=-1)
answer = layers.Dense(answer_vocabulary_size,activation='softmax')(concatenated)
model = Model([text_input, question_input], answer)
model.compile(optimizer='rmsprop',loss='categorical_crossentropy',metrics=['acc'])

2、多输出模型
如社交文章预测人性别年龄

from keras import layers
from keras import Input
from keras.models import Model
vocabulary_size = 50000
num_income_groups = 10
posts_input = Input(shape=(None,), dtype='int32', name='posts')
embedded_posts = layers.Embedding(256, vocabulary_size)(posts_input)
x = layers.Conv1D(128, 5, activation='relu')(embedded_posts)
x = layers.MaxPooling1D(5)(x)
x = layers.Conv1D(256, 5, activation='relu')(x)
x = layers.Conv1D(256, 5, activation='relu')(x)
x = layers.MaxPooling1D(5)(x)
x = layers.Conv1D(256, 5, activation='relu')(x)
x = layers.Conv1D(256, 5, activation='relu')(x)
x = layers.Conv1D(256, 5, activation='relu')(x)
x = layers.Dense(128, activation='relu')(x)
age_prediction = layers.Dense(1, name='age')(x)
income_prediction = layers.Dense(num_income_groups,activation='softmax',name='income')(x)
gender_prediction = layers.Dense(1, activation='sigmoid', name='gender')(x)
model = Model(posts_input,[age_prediction, income_prediction, gender_prediction])
#损失加权
model.compile(optimizer='rmsprop',
loss={'age': 'mse','income': 'categorical_crossentropy','gender': 'binary_crossentropy'},
loss_weights={'age': 0.25,'income': 1.,'gender': 10.})

model.fit(posts, [age_targets, income_targets, gender_targets],epochs=10, batch_size=64)

3、模块
(1)inception

from keras import layers
branch_a = layers.Conv2D(128, 1,
activation='relu', strides=2)(x)
branch_b = layers.Conv2D(128, 1, activation='relu')(x)
branch_b = layers.Conv2D(128, 3, activation='relu', strides=2)(branch_b)
branch_c = layers.AveragePooling2D(3, strides=2)(x)
branch_c = layers.Conv2D(128, 3, activation='relu')(branch_c)
branch_d = layers.Conv2D(128, 1, activation='relu')(x)
branch_d = layers.Conv2D(128, 3, activation='relu')(branch_d)
branch_d = layers.Conv2D(128, 3, activation='relu', strides=2)(branch_d)
output = layers.concatenate([branch_a, branch_b, branch_c, branch_d], axis=-1)

(2)残差链接

#如果特征图尺寸相同，直接相加
from keras import layers
x = ...
y = layers.Conv2D(128, 3, activation='relu', padding='same')(x)
y = layers.Conv2D(128, 3, activation='relu', padding='same')(y)
y = layers.Conv2D(128, 3, activation='relu', padding='same')(y)
y = layers.add([y, x])

#如果特征图尺寸不相同，
x=...
y=layers.Conv2D(128, 3, activation='relu', padding='same')(x)
y=layers.Conv2D(128, 3, activation='relu', padding='same')(y)
y=layers.MaxPooling2D(2, strides=2)(y)
residual = layers.Conv2D(128, 1, strides=2, padding='same')(x)
y = layers.add([y, residual])

(3)共享权重层

from keras import layers
from keras import Input
from keras.models import Model

lstm = layers.LSTM(32)#实例化一次
left_input = Input(shape=(None, 128))
left_output = lstm(left_input)

right_input = Input(shape=(None, 128))
right_output = lstm(right_input)#重复权重

merged = layers.concatenate([left_output, right_output], axis=-1)
predictions = layers.Dense(1, activation='sigmoid')(merged)#在上面构建一个分类器

model = Model([left_input, right_input], predictions)
model.fit([left_data, right_data], targets)

4、自定义模块

class MyDense(tf.keras.Model):
    def __init__(self,units):
        super().__init__()
        self.units = units
    
    def build(self,X_shape):#参数初始化
        self.weight = self.add_weight(
            name='weight',
            shape=[X_shape[-1],self.units],
            initializer=tf.random_normal_initializer())
        
        self.bias = self.add_weight(
            name='bias',
            shape=[self.units],
            initializer=tf.zeros_initializer())
    
    def call(self,X):#返回值
        linear = tf.matmul(X,self.weight) + self.bias
        return tf.nn.relu(linear)

dense = MyDense(3)
dense(tf.random.uniform((2, 5)))
dense.get_weights()

5、模型集成

preds_a = model_a.predict(x_val)
preds_b = model_b.predict(x_val)
preds_c = model_c.predict(x_val)
preds_d = model_d.predict(x_val)
final_preds = 0.25 * (preds_a + preds_b + preds_c + preds_d)

6、回调函数

常用
keras.callbacks.ModelCheckpoint
keras.callbacks.EarlyStopping
keras.callbacks.LearningRateScheduler
keras.callbacks.ReduceLROnPlateau
keras.callbacks.CSVLogger

#当模型10轮内没有改善，学习率下降为1/10
callbacks_list = [keras.callbacks.ReduceLROnPlateau(monitor='val_loss',factor=0.1,patience=10,)]

callbacks_list = [
keras.callbacks.EarlyStopping(monitor='acc',patience=1,),
keras.callbacks.ModelCheckpoint(filepath='my_model.h5',monitor='val_loss',save_best_only=True,)]
model.compile(optimizer='rmsprop',loss='binary_crossentropy',metrics=['acc'])
model.fit(x, y,epochs=10,batch_size=32,callbacks=callbacks_list,validation_data=(x_val, y_val))

#tensorBoard回调函数
callbacks = [keras.callbacks.TensorBoard(log_dir='my_log_dir',histogram_freq=1,embeddings_freq=1,)]
history = model.fit(x_train, y_train,epochs=20,batch_size=128,
validation_split=0.2,callbacks=callbacks)

tensorboard --logdir=my_log_dir

编写你自己的回调函数

on_epoch_begin
on_epoch_en
on_batch_begin
on_batch_end
on_train_begin
on_train_end

self.model :调回函数模型实例
self.validation_data:传入fit作为验证损失的值
import keras
import numpy as np
class ActivationLogger(keras.callbacks.Callback):
  def set_model(self, model):
  self.model = model
  layer_outputs = [layer.output for layer in model.layers]
  self.activations_model = keras.models.Model(model.input,layer_outputs)

  def on_epoch_end(self, epoch, logs=None):
  if self.validation_data is None:
    raise RuntimeError('Requires validation_data.')
   validation_sample = self.validation_data[0][0:1]
   activations = self.activations_model.predict(validation_sample)
   f = open('activations_at_epoch_' + str(epoch) + '.npz', 'w')
   np.savez(f, activations)
   f.close()