TensorFlow2.0使用keras训练模型的实现

电玩女神 2022-10-30 15:28 11阅读 0赞

1.一般的模型构造、训练、测试流程  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
\# 模型构造  
inputs = keras.Input(shape=(784,), name='mnist\_input')  
h1 = layers.Dense(64, activation='relu')(inputs)  
h1 = layers.Dense(64, activation='relu')(h1)  
outputs = layers.Dense(10, activation='softmax')(h1)  
model = keras.Model(inputs, outputs)  
\# keras.utils.plot\_model(model, 'net001.png', show\_shapes=True)  
  
model.compile(optimizer=keras.optimizers.RMSprop(),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[keras.metrics.SparseCategoricalAccuracy()\])  
  
\# 载入数据  
(x\_train, y\_train), (x\_test, y\_test) = keras.datasets.mnist.load\_data()  
x\_train = x\_train.reshape(60000, 784).astype('float32') /255  
x\_test = x\_test.reshape(10000, 784).astype('float32') /255  
  
x\_val = x\_train\[-10000:\]  
y\_val = y\_train\[-10000:\]  
  
x\_train = x\_train\[:-10000\]  
y\_train = y\_train\[:-10000\]  
  
\# 训练模型  
history = model.fit(x\_train, y\_train, batch\_size=64, epochs=3,  
validation\_data=(x\_val, y\_val))  
print('history:')  
print(history.history)  
  
result = model.evaluate(x\_test, y\_test, batch\_size=128)  
print('evaluate:')  
print(result)  
pred = model.predict(x\_test\[:2\])  
print('predict:')  
print(pred)  
2.自定义损失和指标  
自定义指标只需继承Metric类， 并重写一下函数

\_init\_(self)，初始化。

update\_state(self，y\_true，y\_pred，sample\_weight = None)，它使用目标y\_true和模型预测y\_pred来更新状态变量。

result(self)，它使用状态变量来计算最终结果。

reset\_states(self)，重新初始化度量的状态。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
\# 这是一个简单的示例，显示如何实现CatgoricalTruePositives指标，该指标计算正确分类为属于给定类的样本数量  
  
class CatgoricalTruePostives(keras.metrics.Metric):  
def \_\_init\_\_(self, name='binary\_true\_postives', \*\*kwargs):  
super(CatgoricalTruePostives, self).\_\_init\_\_(name=name, \*\*kwargs)  
self.true\_postives = self.add\_weight(name='tp', initializer='zeros')  
  
def update\_state(self, y\_true, y\_pred, sample\_weight=None):  
y\_pred = tf.argmax(y\_pred)  
y\_true = tf.equal(tf.cast(y\_pred, tf.int32), tf.cast(y\_true, tf.int32))  
  
y\_true = tf.cast(y\_true, tf.float32)  
  
if sample\_weight is not None:  
sample\_weight = tf.cast(sample\_weight, tf.float32)  
y\_true = tf.multiply(sample\_weight, y\_true)  
  
return self.true\_postives.assign\_add(tf.reduce\_sum(y\_true))  
  
def result(self):  
return tf.identity(self.true\_postives)  
  
def reset\_states(self):  
self.true\_postives.assign(0.)  
  
  
model.compile(optimizer=keras.optimizers.RMSprop(1e-3),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[CatgoricalTruePostives()\])  
  
model.fit(x\_train, y\_train,  
batch\_size=64, epochs=3)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
\# 以定义网络层的方式添加网络loss  
class ActivityRegularizationLayer(layers.Layer):  
def call(self, inputs):  
self.add\_loss(tf.reduce\_sum(inputs) \* 0.1)  
return inputs  
  
inputs = keras.Input(shape=(784,), name='mnist\_input')  
h1 = layers.Dense(64, activation='relu')(inputs)  
h1 = ActivityRegularizationLayer()(h1)  
h1 = layers.Dense(64, activation='relu')(h1)  
outputs = layers.Dense(10, activation='softmax')(h1)  
model = keras.Model(inputs, outputs)  
\# keras.utils.plot\_model(model, 'net001.png', show\_shapes=True)  
  
model.compile(optimizer=keras.optimizers.RMSprop(),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[keras.metrics.SparseCategoricalAccuracy()\])  
model.fit(x\_train, y\_train, batch\_size=32, epochs=1)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
\# 也可以以定义网络层的方式添加要统计的metric  
class MetricLoggingLayer(layers.Layer):  
def call(self, inputs):  
self.add\_metric(keras.backend.std(inputs),  
name='std\_of\_activation',  
aggregation='mean')  
  
return inputs  
  
inputs = keras.Input(shape=(784,), name='mnist\_input')  
h1 = layers.Dense(64, activation='relu')(inputs)  
h1 = MetricLoggingLayer()(h1)  
h1 = layers.Dense(64, activation='relu')(h1)  
outputs = layers.Dense(10, activation='softmax')(h1)  
model = keras.Model(inputs, outputs)  
\# keras.utils.plot\_model(model, 'net001.png', show\_shapes=True)  
  
model.compile(optimizer=keras.optimizers.RMSprop(),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[keras.metrics.SparseCategoricalAccuracy()\])  
model.fit(x\_train, y\_train, batch\_size=32, epochs=1)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
\# 也可以直接在model上面加  
\# 也可以以定义网络层的方式添加要统计的metric  
class MetricLoggingLayer(layers.Layer):  
def call(self, inputs):  
self.add\_metric(keras.backend.std(inputs),  
name='std\_of\_activation',  
aggregation='mean')  
  
return inputs  
  
inputs = keras.Input(shape=(784,), name='mnist\_input')  
h1 = layers.Dense(64, activation='relu')(inputs)  
h2 = layers.Dense(64, activation='relu')(h1)  
outputs = layers.Dense(10, activation='softmax')(h2)  
model = keras.Model(inputs, outputs)  
  
model.add\_metric(keras.backend.std(inputs),  
name='std\_of\_activation',  
aggregation='mean')  
model.add\_loss(tf.reduce\_sum(h1)\*0.1)  
  
\# keras.utils.plot\_model(model, 'net001.png', show\_shapes=True)  
  
model.compile(optimizer=keras.optimizers.RMSprop(),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[keras.metrics.SparseCategoricalAccuracy()\])  
model.fit(x\_train, y\_train, batch\_size=32, epochs=1)  
处理使用validation\_data传入测试数据，还可以使用validation\_split划分验证数据

ps:validation\_split只能在用numpy数据训练的情况下使用

1  
model.fit(x\_train, y\_train, batch\_size=32, epochs=1, validation\_split=0.2)  
3.使用tf.data构造数据  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
def get\_compiled\_model():  
inputs = keras.Input(shape=(784,), name='mnist\_input')  
h1 = layers.Dense(64, activation='relu')(inputs)  
h2 = layers.Dense(64, activation='relu')(h1)  
outputs = layers.Dense(10, activation='softmax')(h2)  
model = keras.Model(inputs, outputs)  
model.compile(optimizer=keras.optimizers.RMSprop(),  
loss=keras.losses.SparseCategoricalCrossentropy(),  
metrics=\[keras.metrics.SparseCategoricalAccuracy()\])  
return model  
model = get\_compiled\_model()  
train\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_train, y\_train))  
train\_dataset = train\_dataset.shuffle(buffer\_size=1024).batch(64)  
  
val\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_val, y\_val))  
val\_dataset = val\_dataset.batch(64)  
  
\# model.fit(train\_dataset, epochs=3)  
\# steps\_per\_epoch 每个epoch只训练几步  
\# validation\_steps 每次验证，验证几步  
model.fit(train\_dataset, epochs=3, steps\_per\_epoch=100,  
validation\_data=val\_dataset, validation\_steps=3)  
4.样本权重和类权重  
“样本权重”数组是一个数字数组，用于指定批处理中每个样本在计算总损失时应具有多少权重。 它通常用于不平衡的分类问题（这个想法是为了给予很少见的类更多的权重）。 当使用的权重是1和0时，该数组可以用作损失函数的掩码（完全丢弃某些样本对总损失的贡献）。

“类权重”dict是同一概念的更具体的实例：它将类索引映射到应该用于属于该类的样本的样本权重。 例如，如果类“0”比数据中的类“1”少两倍，则可以使用class\_weight = \{0：1.，1：0.5\}。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
\# 增加第5类的权重  
import numpy as np  
\# 样本权重  
model = get\_compiled\_model()  
class\_weight = \{i:1.0 for i in range(10)\}  
class\_weight\[5\] = 2.0  
print(class\_weight)  
model.fit(x\_train, y\_train,  
class\_weight=class\_weight,  
batch\_size=64,  
epochs=4)  
\# 类权重  
model = get\_compiled\_model()  
sample\_weight = np.ones(shape=(len(y\_train),))  
sample\_weight\[y\_train == 5\] = 2.0  
model.fit(x\_train, y\_train,  
sample\_weight=sample\_weight,  
batch\_size=64,  
epochs=4)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
\# tf.data数据  
model = get\_compiled\_model()  
  
sample\_weight = np.ones(shape=(len(y\_train),))  
sample\_weight\[y\_train == 5\] = 2.0  
  
train\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_train, y\_train,  
sample\_weight))  
train\_dataset = train\_dataset.shuffle(buffer\_size=1024).batch(64)  
  
val\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_val, y\_val))  
val\_dataset = val\_dataset.batch(64)  
  
model.fit(train\_dataset, epochs=3, )  
5.多输入多输出模型  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
image\_input = keras.Input(shape=(32, 32, 3), name='img\_input')  
timeseries\_input = keras.Input(shape=(None, 10), name='ts\_input')  
  
x1 = layers.Conv2D(3, 3)(image\_input)  
x1 = layers.GlobalMaxPooling2D()(x1)  
  
x2 = layers.Conv1D(3, 3)(timeseries\_input)  
x2 = layers.GlobalMaxPooling1D()(x2)  
  
x = layers.concatenate(\[x1, x2\])  
  
score\_output = layers.Dense(1, name='score\_output')(x)  
class\_output = layers.Dense(5, activation='softmax', name='class\_output')(x)  
  
model = keras.Model(inputs=\[image\_input, timeseries\_input\],  
outputs=\[score\_output, class\_output\])  
keras.utils.plot\_model(model, 'multi\_input\_output\_model.png'  
, show\_shapes=True)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
\# 可以为模型指定不同的loss和metrics  
model.compile(  
optimizer=keras.optimizers.RMSprop(1e-3),  
loss=\[keras.losses.MeanSquaredError(),  
keras.losses.CategoricalCrossentropy()\])  
  
\# 还可以指定loss的权重  
model.compile(  
optimizer=keras.optimizers.RMSprop(1e-3),  
loss=\{'score\_output': keras.losses.MeanSquaredError(),  
'class\_output': keras.losses.CategoricalCrossentropy()\},  
metrics=\{'score\_output': \[keras.metrics.MeanAbsolutePercentageError(),  
keras.metrics.MeanAbsoluteError()\],  
'class\_output': \[keras.metrics.CategoricalAccuracy()\]\},  
loss\_weight=\{'score\_output': 2., 'class\_output': 1.\})  
  
\# 可以把不需要传播的loss置0  
model.compile(  
optimizer=keras.optimizers.RMSprop(1e-3),  
loss=\[None, keras.losses.CategoricalCrossentropy()\])  
  
\# Or dict loss version  
model.compile(  
optimizer=keras.optimizers.RMSprop(1e-3),  
loss=\{'class\_output': keras.losses.CategoricalCrossentropy()\})  
6.使用回 调  
Keras中的回调是在训练期间（在epoch开始时，batch结束时，epoch结束时等）在不同点调用的对象，可用于实现以下行为：

在培训期间的不同时间点进行验证（超出内置的每个时期验证）  
定期检查模型或超过某个精度阈值  
在训练似乎平稳时改变模型的学习率  
在训练似乎平稳时对顶层进行微调  
在培训结束或超出某个性能阈值时发送电子邮件或即时消息通知等等。  
可使用的内置回调有

ModelCheckpoint：定期保存模型。  
EarlyStopping：当训练不再改进验证指标时停止培训。  
TensorBoard：定期编写可在TensorBoard中显示的模型日志（更多细节见“可视化”）。  
CSVLogger：将丢失和指标数据流式传输到CSV文件。  
等等  
6.1回调使用  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
model = get\_compiled\_model()  
  
callbacks = \[  
keras.callbacks.EarlyStopping(  
\# Stop training when \`val\_loss\` is no longer improving  
monitor='val\_loss',  
\# "no longer improving" being defined as "no better than 1e-2 less"  
min\_delta=1e-2,  
\# "no longer improving" being further defined as "for at least 2 epochs"  
patience=2,  
verbose=1)  
\]  
model.fit(x\_train, y\_train,  
epochs=20,  
batch\_size=64,  
callbacks=callbacks,  
validation\_split=0.2)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
\# checkpoint模型回调  
model = get\_compiled\_model()  
check\_callback = keras.callbacks.ModelCheckpoint(  
filepath='mymodel\_\{epoch\}.h5',  
save\_best\_only=True,  
monitor='val\_loss',  
verbose=1  
)  
  
model.fit(x\_train, y\_train,  
epochs=3,  
batch\_size=64,  
callbacks=\[check\_callback\],  
validation\_split=0.2)  
1  
2  
3  
4  
5  
6  
7  
8  
9  
\# 动态调整学习率  
initial\_learning\_rate = 0.1  
lr\_schedule = keras.optimizers.schedules.ExponentialDecay(  
initial\_learning\_rate,  
decay\_steps=10000,  
decay\_rate=0.96,  
staircase=True  
)  
optimizer = keras.optimizers.RMSprop(learning\_rate=lr\_schedule)  
1  
2  
3  
4  
5  
6  
7  
\# 使用tensorboard  
tensorboard\_cbk = keras.callbacks.TensorBoard(log\_dir='./full\_path\_to\_your\_logs')  
model.fit(x\_train, y\_train,  
epochs=5,  
batch\_size=64,  
callbacks=\[tensorboard\_cbk\],  
validation\_split=0.2)