计算 Tensorflow 2 中每个时期后每个 class 的召回率
Calculate recall for each class after each epoch in Tensorflow 2
我正在尝试在使用 Tensorflow 的模型中的每个时期之后计算二进制和多 class(一个热编码)classification 场景的召回率2 的 Keras API。例如对于二进制 classification 我希望能够做类似
的事情
import tensorflow as tf
model = tf.keras.Sequential()
model.add(...)
model.add(tf.keras.layers.Dense(1))
model.compile(metrics=[binary_recall(label=0), binary_recall(label=1)], ...)
history = model.fit(...)
plt.plot(history.history['binary_recall_0'])
plt.plot(history.history['binary_recall_1'])
plt.show()
或者在多个 class 场景中,我想做类似
的事情
model = tf.keras.Sequential()
model.add(...)
model.add(tf.keras.layers.Dense(3))
model.compile(metrics=[recall(label=0), recall(label=1), recall(label=2)], ...)
history = model.fit(...)
plt.plot(history.history['recall_0'])
plt.plot(history.history['recall_1'])
plt.plot(history.history['recall_2'])
plt.show()
我正在为一个不平衡的数据集开发一个 class 化器,并希望能够看到我的少数 class(s) 的召回率在什么时候开始下降。
我在此处 的 multi-class classifier 中找到了针对特定 class 的精度实现。我正在尝试将其适应我的需要,但 keras.backend
对我来说仍然很陌生,因此非常感谢任何帮助。
我也不清楚我是否可以使用 Keras metrics
(因为它们是在每批结束时计算的,然后取平均值)或者我是否需要使用 Keras callbacks
(哪个可以 运行 在每个纪元的末尾)。在我看来它不应该对召回产生影响(例如 8/10 == (3/5 + 5/5) / 2
)但这就是召回在 Keras 2 中被删除的原因所以也许我遗漏了一些东西(https://github.com/keras-team/keras/issues/5794)
编辑-部分解决方案(多classclass化)
@mujjiga 的解决方案适用于二进制 class 化和多 class class 化,但正如@P-Gn 指出的那样,tensorflow 2 的 Recall metric 支持开箱即用多class class化。例如
from tensorflow.keras.metrics import Recall
model = ...
model.compile(loss='categorical_crossentropy', metrics=[
Recall(class_id=0, name='recall_0')
Recall(class_id=1, name='recall_1')
Recall(class_id=2, name='recall_2')
])
history = model.fit(...)
plt.plot(history.history['recall_2'])
plt.plot(history.history['val_recall_2'])
plt.show()
有多种方法可以做到这一点,但使用 callback
似乎是最好和最 kerasy 的方法。一个 side-note 在我告诉你如何之前:
I am also not clear on if I can use Keras metrics (as they are
calculated at the end of each batch and then averaged) or if I need to
use Keras callbacks (which can run at the end of each epoch).
这不是真的。 Keras' callbacks可以使用以下方法:
- on_epoch_begin:在每个纪元开始时调用。
- on_epoch_end:在每个纪元结束时调用。
- on_batch_begin:在每批开始时调用。
- on_batch_end:在每批结束时调用。
- on_train_begin:模型训练开始时调用
- on_train_end: 模型训练结束时调用
无论您使用 keras
还是 tf.keras
都是如此。
您可以在下面找到我对自定义回调的实现。
class RecallHistory(keras.callbacks.Callback):
def on_train_begin(self, logs={}):
self.recall = {}
def on_epoch_end(self, epoch, logs={}):
# Compute and store recall for each class here.
self.recall[...] = 42
history = RecallHistory()
model.fit(..., callbacks=[history])
print(history.recall)
在 TF2 中,tf.keras.metrics.Recall
获得了一个能够做到这一点的 class_id
成员。使用 FashionMNIST 的示例:
import tensorflow as tf
(x_train, y_train), _ = tf.keras.datasets.fashion_mnist.load_data()
x_train = x_train[..., None].astype('float32') / 255
y_train = tf.keras.utils.to_categorical(y_train)
input_shape = x_train.shape[1:]
model = tf.keras.Sequential([
tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=input_shape),
tf.keras.layers.MaxPool2D(pool_size=2),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'),
tf.keras.layers.MaxPool2D(pool_size=2),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(units=256, activation='relu'),
tf.keras.layers.Dropout(0.5),
tf.keras.layers.Dense(units=10, activation='softmax')])
model.compile(loss='categorical_crossentropy', optimizer='Adam',
metrics=[tf.keras.metrics.Recall(class_id=i) for i in range(10)])
model.fit(x_train, y_train, batch_size=128, epochs=50)
在 TF 1.13 中,tf.keras.metric.Recall
没有这个 class_id
参数,但它可以通过子类化添加(有些令人惊讶的是,这在 TF2 的 alpha 版本中似乎是不可能的)。
class Recall(tf.keras.metrics.Recall):
def __init__(self, *, class_id, **kwargs):
super().__init__(**kwargs)
self.class_id= class_id
def update_state(self, y_true, y_pred, sample_weight=None):
y_true = y_true[:, self.class_id]
y_pred = tf.cast(tf.equal(
tf.math.argmax(y_pred, axis=-1), self.class_id), dtype=tf.float32)
return super().update_state(y_true, y_pred, sample_weight)
我们可以使用sklearn的classification_report
和keras的Callback
来实现。
工作代码示例(带注释)
import tensorflow as tf
import keras
from tensorflow.python.keras.layers import Dense, Input
from tensorflow.python.keras.models import Sequential
from tensorflow.python.keras.callbacks import Callback
from sklearn.metrics import recall_score, classification_report
from sklearn.datasets import make_classification
import numpy as np
import matplotlib.pyplot as plt
# Model -- Binary classifier
binary_model = Sequential()
binary_model.add(Dense(16, input_shape=(2,), activation='relu'))
binary_model.add(Dense(8, activation='relu'))
binary_model.add(Dense(1, activation='sigmoid'))
binary_model.compile('adam', loss='binary_crossentropy')
# Model -- Multiclass classifier
multiclass_model = Sequential()
multiclass_model.add(Dense(16, input_shape=(2,), activation='relu'))
multiclass_model.add(Dense(8, activation='relu'))
multiclass_model.add(Dense(3, activation='softmax'))
multiclass_model.compile('adam', loss='categorical_crossentropy')
# callback to find metrics at epoch end
class Metrics(Callback):
def __init__(self, x, y):
self.x = x
self.y = y if (y.ndim == 1 or y.shape[1] == 1) else np.argmax(y, axis=1)
self.reports = []
def on_epoch_end(self, epoch, logs={}):
y_hat = np.asarray(self.model.predict(self.x))
y_hat = np.where(y_hat > 0.5, 1, 0) if (y_hat.ndim == 1 or y_hat.shape[1] == 1) else np.argmax(y_hat, axis=1)
report = classification_report(self.y,y_hat,output_dict=True)
self.reports.append(report)
return
# Utility method
def get(self, metrics, of_class):
return [report[str(of_class)][metrics] for report in self.reports]
# Generate some train data (2 class) and train
x, y = make_classification(n_features=2, n_redundant=0, n_informative=2,
random_state=1, n_clusters_per_class=1)
metrics_binary = Metrics(x,y)
binary_model.fit(x, y, epochs=30, callbacks=[metrics_binary])
# Generate some train data (3 class) and train
x, y = make_classification(n_features=2, n_redundant=0, n_informative=2,
random_state=1, n_clusters_per_class=1, n_classes=3)
y = keras.utils.to_categorical(y,3)
metrics_multiclass = Metrics(x,y)
multiclass_model.fit(x, y, epochs=30, callbacks=[metrics_multiclass])
# Plotting
plt.close('all')
plt.plot(metrics_binary.get('recall',0), label='Class 0 recall')
plt.plot(metrics_binary.get('recall',1), label='Class 1 recall')
plt.plot(metrics_binary.get('precision',0), label='Class 0 precision')
plt.plot(metrics_binary.get('precision',1), label='Class 1 precision')
plt.plot(metrics_binary.get('f1-score',0), label='Class 0 f1-score')
plt.plot(metrics_binary.get('f1-score',1), label='Class 1 f1-score')
plt.legend(loc='lower right')
plt.show()
plt.close('all')
for m in ['recall', 'precision', 'f1-score']:
for c in [0,1,2]:
plt.plot(metrics_multiclass.get(m,c), label='Class {0} {1}'.format(c,m))
plt.legend(loc='lower right')
plt.show()
输出
优点:
classification_report
提供大量指标
- 可以通过将训练数据传递给
Metrics
构造函数来计算关于训练数据的验证数据的指标。
我正在尝试在使用 Tensorflow 的模型中的每个时期之后计算二进制和多 class(一个热编码)classification 场景的召回率2 的 Keras API。例如对于二进制 classification 我希望能够做类似
的事情import tensorflow as tf
model = tf.keras.Sequential()
model.add(...)
model.add(tf.keras.layers.Dense(1))
model.compile(metrics=[binary_recall(label=0), binary_recall(label=1)], ...)
history = model.fit(...)
plt.plot(history.history['binary_recall_0'])
plt.plot(history.history['binary_recall_1'])
plt.show()
或者在多个 class 场景中,我想做类似
的事情model = tf.keras.Sequential()
model.add(...)
model.add(tf.keras.layers.Dense(3))
model.compile(metrics=[recall(label=0), recall(label=1), recall(label=2)], ...)
history = model.fit(...)
plt.plot(history.history['recall_0'])
plt.plot(history.history['recall_1'])
plt.plot(history.history['recall_2'])
plt.show()
我正在为一个不平衡的数据集开发一个 class 化器,并希望能够看到我的少数 class(s) 的召回率在什么时候开始下降。
我在此处 keras.backend
对我来说仍然很陌生,因此非常感谢任何帮助。
我也不清楚我是否可以使用 Keras metrics
(因为它们是在每批结束时计算的,然后取平均值)或者我是否需要使用 Keras callbacks
(哪个可以 运行 在每个纪元的末尾)。在我看来它不应该对召回产生影响(例如 8/10 == (3/5 + 5/5) / 2
)但这就是召回在 Keras 2 中被删除的原因所以也许我遗漏了一些东西(https://github.com/keras-team/keras/issues/5794)
编辑-部分解决方案(多classclass化) @mujjiga 的解决方案适用于二进制 class 化和多 class class 化,但正如@P-Gn 指出的那样,tensorflow 2 的 Recall metric 支持开箱即用多class class化。例如
from tensorflow.keras.metrics import Recall
model = ...
model.compile(loss='categorical_crossentropy', metrics=[
Recall(class_id=0, name='recall_0')
Recall(class_id=1, name='recall_1')
Recall(class_id=2, name='recall_2')
])
history = model.fit(...)
plt.plot(history.history['recall_2'])
plt.plot(history.history['val_recall_2'])
plt.show()
有多种方法可以做到这一点,但使用 callback
似乎是最好和最 kerasy 的方法。一个 side-note 在我告诉你如何之前:
I am also not clear on if I can use Keras metrics (as they are calculated at the end of each batch and then averaged) or if I need to use Keras callbacks (which can run at the end of each epoch).
这不是真的。 Keras' callbacks可以使用以下方法:
- on_epoch_begin:在每个纪元开始时调用。
- on_epoch_end:在每个纪元结束时调用。
- on_batch_begin:在每批开始时调用。
- on_batch_end:在每批结束时调用。
- on_train_begin:模型训练开始时调用
- on_train_end: 模型训练结束时调用
无论您使用 keras
还是 tf.keras
都是如此。
您可以在下面找到我对自定义回调的实现。
class RecallHistory(keras.callbacks.Callback):
def on_train_begin(self, logs={}):
self.recall = {}
def on_epoch_end(self, epoch, logs={}):
# Compute and store recall for each class here.
self.recall[...] = 42
history = RecallHistory()
model.fit(..., callbacks=[history])
print(history.recall)
在 TF2 中,tf.keras.metrics.Recall
获得了一个能够做到这一点的 class_id
成员。使用 FashionMNIST 的示例:
import tensorflow as tf
(x_train, y_train), _ = tf.keras.datasets.fashion_mnist.load_data()
x_train = x_train[..., None].astype('float32') / 255
y_train = tf.keras.utils.to_categorical(y_train)
input_shape = x_train.shape[1:]
model = tf.keras.Sequential([
tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=input_shape),
tf.keras.layers.MaxPool2D(pool_size=2),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'),
tf.keras.layers.MaxPool2D(pool_size=2),
tf.keras.layers.Dropout(0.3),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(units=256, activation='relu'),
tf.keras.layers.Dropout(0.5),
tf.keras.layers.Dense(units=10, activation='softmax')])
model.compile(loss='categorical_crossentropy', optimizer='Adam',
metrics=[tf.keras.metrics.Recall(class_id=i) for i in range(10)])
model.fit(x_train, y_train, batch_size=128, epochs=50)
在 TF 1.13 中,tf.keras.metric.Recall
没有这个 class_id
参数,但它可以通过子类化添加(有些令人惊讶的是,这在 TF2 的 alpha 版本中似乎是不可能的)。
class Recall(tf.keras.metrics.Recall):
def __init__(self, *, class_id, **kwargs):
super().__init__(**kwargs)
self.class_id= class_id
def update_state(self, y_true, y_pred, sample_weight=None):
y_true = y_true[:, self.class_id]
y_pred = tf.cast(tf.equal(
tf.math.argmax(y_pred, axis=-1), self.class_id), dtype=tf.float32)
return super().update_state(y_true, y_pred, sample_weight)
我们可以使用sklearn的classification_report
和keras的Callback
来实现。
工作代码示例(带注释)
import tensorflow as tf
import keras
from tensorflow.python.keras.layers import Dense, Input
from tensorflow.python.keras.models import Sequential
from tensorflow.python.keras.callbacks import Callback
from sklearn.metrics import recall_score, classification_report
from sklearn.datasets import make_classification
import numpy as np
import matplotlib.pyplot as plt
# Model -- Binary classifier
binary_model = Sequential()
binary_model.add(Dense(16, input_shape=(2,), activation='relu'))
binary_model.add(Dense(8, activation='relu'))
binary_model.add(Dense(1, activation='sigmoid'))
binary_model.compile('adam', loss='binary_crossentropy')
# Model -- Multiclass classifier
multiclass_model = Sequential()
multiclass_model.add(Dense(16, input_shape=(2,), activation='relu'))
multiclass_model.add(Dense(8, activation='relu'))
multiclass_model.add(Dense(3, activation='softmax'))
multiclass_model.compile('adam', loss='categorical_crossentropy')
# callback to find metrics at epoch end
class Metrics(Callback):
def __init__(self, x, y):
self.x = x
self.y = y if (y.ndim == 1 or y.shape[1] == 1) else np.argmax(y, axis=1)
self.reports = []
def on_epoch_end(self, epoch, logs={}):
y_hat = np.asarray(self.model.predict(self.x))
y_hat = np.where(y_hat > 0.5, 1, 0) if (y_hat.ndim == 1 or y_hat.shape[1] == 1) else np.argmax(y_hat, axis=1)
report = classification_report(self.y,y_hat,output_dict=True)
self.reports.append(report)
return
# Utility method
def get(self, metrics, of_class):
return [report[str(of_class)][metrics] for report in self.reports]
# Generate some train data (2 class) and train
x, y = make_classification(n_features=2, n_redundant=0, n_informative=2,
random_state=1, n_clusters_per_class=1)
metrics_binary = Metrics(x,y)
binary_model.fit(x, y, epochs=30, callbacks=[metrics_binary])
# Generate some train data (3 class) and train
x, y = make_classification(n_features=2, n_redundant=0, n_informative=2,
random_state=1, n_clusters_per_class=1, n_classes=3)
y = keras.utils.to_categorical(y,3)
metrics_multiclass = Metrics(x,y)
multiclass_model.fit(x, y, epochs=30, callbacks=[metrics_multiclass])
# Plotting
plt.close('all')
plt.plot(metrics_binary.get('recall',0), label='Class 0 recall')
plt.plot(metrics_binary.get('recall',1), label='Class 1 recall')
plt.plot(metrics_binary.get('precision',0), label='Class 0 precision')
plt.plot(metrics_binary.get('precision',1), label='Class 1 precision')
plt.plot(metrics_binary.get('f1-score',0), label='Class 0 f1-score')
plt.plot(metrics_binary.get('f1-score',1), label='Class 1 f1-score')
plt.legend(loc='lower right')
plt.show()
plt.close('all')
for m in ['recall', 'precision', 'f1-score']:
for c in [0,1,2]:
plt.plot(metrics_multiclass.get(m,c), label='Class {0} {1}'.format(c,m))
plt.legend(loc='lower right')
plt.show()
输出
优点:
classification_report
提供大量指标- 可以通过将训练数据传递给
Metrics
构造函数来计算关于训练数据的验证数据的指标。