encapsulates metric logic and state args name optional string name of the metric instance dtype optional data type of the metric result standalone usage python m somemetric for input in m updatestateinput print final result m result usage with compile api python model keras sequential model addkeras layers dense64 activation relu model addkeras layers dense64 activation relu model addkeras layers dense10 activation softmax model compileoptimizerkeras optimizers rmsprop0 01 losskeras losses categoricalcrossentropy metricskeras metrics categoricalaccuracy data np random random1000 32 labels np random random1000 10 model fitdata labels epochs10 to be implemented by subclasses init all state variables should be created in this method by calling self addvariable like self var self addvariable updatestate has all updates to the state variables like self var assign result computes and returns a scalar value or a dict of scalar values for the metric from the state variables example subclass implementation python class binarytruepositivesmetric def initself name binarytruepositives kwargs super initnamename kwargs self truepositives self addvariable shape initializer zeros name truepositives def updatestateself ytrue ypred sampleweightnone ytrue ops castytrue bool ypred ops castypred bool values ops logicaland ops equalytrue true ops equalypred true values ops castvalues self dtype if sampleweight is not none sampleweight ops castsampleweight self dtype sampleweight ops broadcastto sampleweight ops shapevalues values ops multiplyvalues sampleweight self truepositives assignself truepositives ops sumvalues def resultself return self truepositives reset all of the metric state variables this function is called between epochssteps when a metric is evaluated during training accumulate statistics for the metric raise notimplementederror def statelessupdatestateself metricvariables args kwargs if lenmetricvariables lenself variables raise valueerror argument metricvariables must be a list of tensors fcorresponding 1 1 to self class name variables freceived list with length lenmetricvariables but fexpected lenself variables variables gather variable mapping mapping listzipself variables metricvariables call in stateless scope with backend statelessscopestatemappingmapping as scope self updatestateargs kwargs gather updated variables metricvariables for v in self variables newv scope getcurrentvaluev if newv is not none metricvariables appendnewv else metricvariables appendv return metricvariables def resultself raise notimplementederror def statelessresultself metricvariables if lenmetricvariables lenself variables raise valueerror argument metricvariables must be a list of tensors fcorresponding 1 1 to self class name variables freceived list with length lenmetricvariables but fexpected lenself variables variables gather variable mapping mapping listzipself variables metricvariables call in stateless scope with backend statelessscopestatemappingmapping res self result return res property def dtypeself return self dtype def addvariableself shape initializer dtypenone namenone self checksupercalled with backend namescopeself name callerself initializer initializers getinitializer variable backend variable initializerinitializer shapeshape dtypedtype trainablefalse namename prevent doubletracking self tracker addtostorevariables variable return variable def addweightself shape initializernone dtypenone namenone backwards compatibility alias return self addvariable shapeshape initializerinitializer dtypedtype namename property def variablesself variables self variables for metric in self metrics variables extendmetric variables return variables def callself args kwargs self checksupercalled self updatestateargs kwargs return self result def getconfigself track variables layers metrics encapsulates metric logic and state args name optional string name of the metric instance dtype optional data type of the metric result standalone usage python m somemetric for input in m update_state input print final result m result usage with compile api python model keras sequential model add keras layers dense 64 activation relu model add keras layers dense 64 activation relu model add keras layers dense 10 activation softmax model compile optimizer keras optimizers rmsprop 0 01 loss keras losses categoricalcrossentropy metrics keras metrics categoricalaccuracy data np random random 1000 32 labels np random random 1000 10 model fit data labels epochs 10 to be implemented by subclasses __init__ all state variables should be created in this method by calling self add_variable like self var self add_variable update_state has all updates to the state variables like self var assign result computes and returns a scalar value or a dict of scalar values for the metric from the state variables example subclass implementation python class binarytruepositives metric def __init__ self name binary_true_positives kwargs super __init__ name name kwargs self true_positives self add_variable shape initializer zeros name true_positives def update_state self y_true y_pred sample_weight none y_true ops cast y_true bool y_pred ops cast y_pred bool values ops logical_and ops equal y_true true ops equal y_pred true values ops cast values self dtype if sample_weight is not none sample_weight ops cast sample_weight self dtype sample_weight ops broadcast_to sample_weight ops shape values values ops multiply values sample_weight self true_positives assign self true_positives ops sum values def result self return self true_positives reset all of the metric state variables this function is called between epochs steps when a metric is evaluated during training accumulate statistics for the metric gather variable mapping call in stateless scope gather updated variables compute the current metric value returns a scalar tensor or a dictionary of scalar tensors gather variable mapping call in stateless scope prevent double tracking backwards compatibility alias return the serializable config of the metric track variables layers metrics
from keras import backend
from keras import initializers
from keras import ops
from keras.api_export import keras_export
from keras.utils.naming import auto_name
from keras.utils.tracking import Tracker


@keras_export(["keras.Metric", "keras.metrics.Metric"])
class Metric:
    

    def __init__(self, dtype=None, name=None):
        self.name = name or auto_name(self.__class__.__name__)
        self._dtype = dtype
        self._metrics = []
        self._variables = []
        self._tracker = Tracker(
            {
                "variables": (
                    lambda x: isinstance(x, backend.Variable),
                    self._variables,
                ),
                "metrics": (lambda x: isinstance(x, Metric), self._metrics),
            }
        )

    def reset_state(self):
        
        for v in self.variables:
            v.assign(ops.zeros(v.shape, dtype=v.dtype))

    def update_state(self, *args, **kwargs):
        
        raise NotImplementedError

    def stateless_update_state(self, metric_variables, *args, **kwargs):
        if len(metric_variables) != len(self.variables):
            raise ValueError(
                "Argument `metric_variables` must be a list of tensors "
                f"corresponding 1:1 to {self.__class__.__name__}().variables. "
                f"Received list with length {len(metric_variables)}, but "
                f"expected {len(self.variables)} variables."
            )
        
        mapping = list(zip(self.variables, metric_variables))

        
        with backend.StatelessScope(state_mapping=mapping) as scope:
            self.update_state(*args, **kwargs)

        
        metric_variables = []
        for v in self.variables:
            new_v = scope.get_current_value(v)
            if new_v is not None:
                metric_variables.append(new_v)
            else:
                metric_variables.append(v)
        return metric_variables

    def result(self):
        
        raise NotImplementedError

    def stateless_result(self, metric_variables):
        if len(metric_variables) != len(self.variables):
            raise ValueError(
                "Argument `metric_variables` must be a list of tensors "
                f"corresponding 1:1 to {self.__class__.__name__}().variables. "
                f"Received list with length {len(metric_variables)}, but "
                f"expected {len(self.variables)} variables."
            )
        
        mapping = list(zip(self.variables, metric_variables))

        
        with backend.StatelessScope(state_mapping=mapping):
            res = self.result()
        return res

    @property
    def dtype(self):
        return self._dtype

    def add_variable(self, shape, initializer, dtype=None, name=None):
        self._check_super_called()
        with backend.name_scope(self.name, caller=self):
            initializer = initializers.get(initializer)
            variable = backend.Variable(
                initializer=initializer,
                shape=shape,
                dtype=dtype,
                trainable=False,
                name=name,
            )
        
        self._tracker.add_to_store("variables", variable)
        return variable

    def add_weight(self, shape=(), initializer=None, dtype=None, name=None):
        
        return self.add_variable(
            shape=shape, initializer=initializer, dtype=dtype, name=name
        )

    @property
    def variables(self):
        variables = self._variables[:]
        for metric in self._metrics:
            variables.extend(metric._variables)
        return variables

    def __call__(self, *args, **kwargs):
        self._check_super_called()
        self.update_state(*args, **kwargs)
        return self.result()

    def get_config(self):
        
        return {"name": self.name, "dtype": self.dtype}

    @classmethod
    def from_config(cls, config):
        return cls(**config)

    def __setattr__(self, name, value):
        
        if hasattr(self, "_tracker"):
            value = self._tracker.track(value)
        return super().__setattr__(name, value)

    def _check_super_called(self):
        if not hasattr(self, "_tracker"):
            raise RuntimeError(
                "You forgot to call `super().__init__()` "
                "in the `__init__()` method. Go add it!"
            )