keras Tensorflow回调作为CTC的自定义指标

kpbpu008 于 2022-11-30 发布在其他

关注(0)|答案(2)|浏览(191)

为了在我的模型（用TensorFlow 2.1.0版编写）的训练过程中生成更多指标，比如字符错误率（CER）和单词错误率（WER），我创建了一个回调函数传递给我的模型的fit函数，它能够在一个时期结束时生成CER和WER。
这是我的第二个选择，因为我想为此创建一个自定义指标，但你只能使用keras后端功能来自定义指标。有人对如何将下面的回调转换为自定义指标有什么建议吗（然后可以在验证和/或训练数据的训练过程中计算）？
我遇到的一些障碍包括：

无法将K.ctc_decode结果转换为稀疏Tensor
如何使用Keras后端计算像edit-distance这样的距离？

class Metrics(tf.keras.callbacks.Callback):
    def __init__(self, valid_data, steps):
        """
        valid_data is a TFRecordDataset with batches of 100 elements per batch, shuffled and repeated infinitely. 
        steps define the amount of batches per epoch
        """
        super(Metrics, self).__init__()
        self.valid_data = valid_data
        self.steps = steps

    def on_train_begin(self, logs={}):
        self.cer = []
        self.wer = []
        
    def on_epoch_end(self, epoch, logs={}):

        imgs = []
        labels = []
        for idx, (img, label) in enumerate(self.valid_data.as_numpy_iterator()):
            if idx >= self.steps:
                break
            imgs.append(img)
            labels.extend(label)

        imgs = np.array(imgs)
        labels = np.array(labels)

        out = self.model.predict((batch for batch in imgs))        
        input_length = len(max(out, key=len))

        out = np.asarray(out)
        out_len = np.asarray([input_length for _ in range(len(out))])

        decode, log = K.ctc_decode(out,
                                    out_len,
                                    greedy=True)

        decode = [[[int(p) for p in x if p != -1] for x in y] for y in decode][0]

        for (pred, lab) in zip(decode, labels):
        
            dist = editdistance.eval(pred, lab)
            self.cer.append(dist / (max(len(pred), len(lab))))
            self.wer.append(not np.array_equal(pred, lab))

        
        print("Mean CER: {}".format(np.mean([self.cer], axis=1)[0]))
        print("Mean WER: {}".format(np.mean([self.wer], axis=1)[0]))

keras

来源：https://stackoverflow.com/questions/60285167/tensorflow-callback-as-custom-metric-for-ctc

2条答案

按热度按时间

y3bcpkx11#

已在TF 2.3.1中解决，但也应适用于2.x的先前版本。
一些备注：

有关如何正确实现Tensorflow自定义指标的信息很少。该问题暗示使用回调来实现指标。因此，这需要更长的时间（由于度量on_epoch_end的显式额外计算），至少我是这么认为的。将其实现为tensorflow.keras.metrics.Metric的子类似乎是正确的方法，并且在时期正在进行时产生结果（如果verbose被正确设置）。
使用tf.edit_distance（使用稀疏Tensor）计算CER的编辑距离是相当容易的，这可以随后用于使用某种tf逻辑计算WER。
唉，我还没有找到如何在一个度量中同时实现CER和WER（因为它有相当多的重复代码），如果有人知道如何做到这一点，请与我联系。
自定义指标可以简单地添加到TF模型的编译中：self.model.compile(optimizer=opt, loss=loss, metrics=[CERMetric(), WERMetric()])

第一个

赞(0）回复(0）举报 2022-11-30

mspsb9vt2#

唉，我还没有找到如何在一个度量中同时实现CER和WER（因为它有相当多的重复代码），如果有人知道如何做到这一点，请与我联系。
嘿，这个解决方案真的帮了我很大的忙。到目前为止，有TensorFlow 2.10版本，所以对于这个版本，我写了一个WER和CER指标的组合，下面是最终的工作代码：

import tensorflow as tf

class CWERMetric(tf.keras.metrics.Metric):
    """ A custom TensorFlow metric to compute the Character Error Rate
    """
    def __init__(self, name='CWER', **kwargs):
        super(CWERMetric, self).__init__(name=name, **kwargs)
        self.cer_accumulator = tf.Variable(0.0, name="cer_accumulator", dtype=tf.float32)
        self.wer_accumulator = tf.Variable(0.0, name="wer_accumulator", dtype=tf.float32)
        self.counter = tf.Variable(0, name="counter", dtype=tf.int32)

    def update_state(self, y_true, y_pred, sample_weight=None):
        input_shape = tf.keras.backend.shape(y_pred)

        input_length = tf.ones(shape=input_shape[0], dtype='int32') * tf.cast(input_shape[1], 'int32')

        decode, log = tf.keras.backend.ctc_decode(y_pred, input_length, greedy=True)

        decode = tf.keras.backend.ctc_label_dense_to_sparse(decode[0], input_length)
        y_true_sparse = tf.cast(tf.keras.backend.ctc_label_dense_to_sparse(y_true, input_length), "int64")

        decode = tf.sparse.retain(decode, tf.not_equal(decode.values, -1))
        distance = tf.edit_distance(decode, y_true_sparse, normalize=True)

        correct_words_amount = tf.reduce_sum(tf.cast(tf.not_equal(distance, 0), tf.float32))

        self.wer_accumulator.assign_add(correct_words_amount)
        self.cer_accumulator.assign_add(tf.reduce_sum(distance))
        self.counter.assign_add(len(y_true))

    def result(self):
        return {
                "CER": tf.math.divide_no_nan(self.cer_accumulator, tf.cast(self.counter, tf.float32)),
                "WER": tf.math.divide_no_nan(self.wer_accumulator, tf.cast(self.counter, tf.float32))
        }

我仍然需要检查它是否计算正确的CER和WER，我会发现有些东西是失踪，我会更新这一点。

赞(0）回复(0）举报 2022-11-30

我来回答

keras Tensorflow回调作为CTC的自定义指标

2条答案

相关问题

热门标签

最新问答