TFMA Types

tensorflow_model_analysis.types

Types.

Attributes

AddMetricsCallbackType module-attribute

AddMetricsCallbackType = Callable[
    [
        TensorTypeMaybeDict,
        TensorTypeMaybeDict,
        TensorTypeMaybeDict,
    ],
    Dict[str, Tuple[TensorType, TensorType]],
]

ConcreteStructuredMetricValue module-attribute

ConcreteStructuredMetricValue = TypeVar(
    "ConcreteStructuredMetricValue",
    bound="StructuredMetricValue",
)

DictOfFetchedTensorValues module-attribute

DictOfFetchedTensorValues = Dict[
    FPLKeyType, Dict[str, TensorValue]
]

DictOfTensorType module-attribute

DictOfTensorType = Dict[str, TensorType]

DictOfTensorTypeMaybeDict module-attribute

DictOfTensorTypeMaybeDict = Dict[str, TensorTypeMaybeDict]

DictOfTensorValue module-attribute

DictOfTensorValue = Dict[str, TensorValue]

DictOfTensorValueMaybeDict module-attribute

DictOfTensorValueMaybeDict = Dict[str, TensorValueMaybeDict]

DictOfTypeSpec module-attribute

DictOfTypeSpec = Dict[str, TypeSpec]

DictOfTypeSpecMaybeDict module-attribute

DictOfTypeSpecMaybeDict = Dict[str, TypeSpecMaybeDict]

Extracts module-attribute

Extracts = MutableMapping[str, Any]

FPLKeyType module-attribute

FPLKeyType = Union[str, Tuple[str, ...]]

FeaturesPredictionsLabels module-attribute

FeaturesPredictionsLabels = NamedTuple(
    "FeaturesPredictionsLabels",
    [
        ("input_ref", int),
        ("features", DictOfFetchedTensorValues),
        ("predictions", DictOfFetchedTensorValues),
        ("labels", DictOfFetchedTensorValues),
    ],
)

MaterializedColumn module-attribute

MaterializedColumn = NamedTuple(
    "MaterializedColumn",
    [
        ("name", str),
        (
            "value",
            Union[
                List[bytes],
                List[int],
                List[float],
                bytes,
                int,
                float,
            ],
        ),
    ],
)

MaybeMultipleEvalSharedModels module-attribute

MaybeMultipleEvalSharedModels = Union[
    EvalSharedModel,
    List[EvalSharedModel],
    Dict[str, EvalSharedModel],
]

MetricValueType module-attribute

MetricValueType = Union[
    PrimitiveMetricValueType, ndarray, StructuredMetricValue
]

MetricVariablesType module-attribute

MetricVariablesType = List[Any]

PrimitiveMetricValueType module-attribute

PrimitiveMetricValueType = Union[float, int, number]

TensorOrOperationType module-attribute

TensorOrOperationType = Union[TensorType, Operation]

TensorType module-attribute

TensorType = Union[Tensor, SparseTensor, RaggedTensor]

TensorTypeMaybeDict module-attribute

TensorTypeMaybeDict = Union[TensorType, DictOfTensorType]

TensorTypeMaybeMultiLevelDict module-attribute

TensorTypeMaybeMultiLevelDict = Union[
    TensorTypeMaybeDict, DictOfTensorTypeMaybeDict
]

TensorValue module-attribute

TensorValue = Union[
    ndarray,
    SparseTensorValue,
    RaggedTensorValue,
    SparseTensorValue,
]

TensorValueMaybeDict module-attribute

TensorValueMaybeDict = Union[TensorValue, DictOfTensorValue]

TensorValueMaybeMultiLevelDict module-attribute

TensorValueMaybeMultiLevelDict = Union[
    TensorValueMaybeDict, DictOfTensorValueMaybeDict
]

TypeSpecMaybeDict module-attribute

TypeSpecMaybeDict = Union[TypeSpec, DictOfTypeSpec]

TypeSpecMaybeMultiLevelDict module-attribute

TypeSpecMaybeMultiLevelDict = Union[
    TypeSpecMaybeDict, DictOfTypeSpecMaybeDict
]

Classes

EvalSharedModel

Bases: NamedTuple('EvalSharedModel', [('model_path', str), ('add_metrics_callbacks', List[Callable]), ('include_default_metrics', bool), ('example_weight_key', Union[str, Dict[str, str]]), ('additional_fetches', List[str]), ('model_loader', ModelLoader), ('model_name', str), ('model_type', str), ('rubber_stamp', bool), ('is_baseline', bool), ('resource_hints', Optional[Dict[str, Any]]), ('backend_config', Optional[Any])])

Shared model used during extraction and evaluation.

ATTRIBUTE	DESCRIPTION
model_path	Path to EvalSavedModel (containing the saved_model.pb file).
add_metrics_callbacks	Optional list of callbacks for adding additional metrics to the graph. The names of the metrics added by the callbacks should not conflict with existing metrics. See below for more details about what each callback should do. The callbacks are only used during evaluation.
include_default_metrics	True to include the default metrics that are part of the saved model graph during evaluation.
example_weight_key	Example weight key (single-output model) or dict of example weight keys (multi-output model) keyed by output_name.
additional_fetches	Prefixes of additional tensors stored in signature_def.inputs that should be fetched at prediction time. The "features" and "labels" tensors are handled automatically and should not be included in this list.
model_loader	Model loader.
model_name	Model name (should align with ModelSpecs.name).
model_type	Model type (tfma.TF_KERAS, tfma.TF_LITE, tfma.TF_ESTIMATOR, ..).
rubber_stamp	True if this model is being rubber stamped. When a model is rubber stamped diff thresholds will be ignored if an associated baseline model is not passed.
is_baseline	The model is the baseline for comparison or not.
resource_hints	The beam resource hints to apply to the PTransform which runs inference for this model.
backend_config	The backend config for running model inference.

More details on add_metrics_callbacks:

Each add_metrics_callback should have the following prototype: def add_metrics_callback(features_dict, predictions_dict, labels_dict):

Note that features_dict, predictions_dict and labels_dict are not necessarily dictionaries - they might also be Tensors, depending on what the model's eval_input_receiver_fn returns.

It should create and return a metric_ops dictionary, such that metric_ops['metric_name'] = (value_op, update_op), just as in the Trainer.

Short example:

def add_metrics_callback(features_dict, predictions_dict, labels): metrics_ops = {} metric_ops['mean_label'] = tf.metrics.mean(labels) metric_ops['mean_probability'] = tf.metrics.mean(tf.slice( predictions_dict['probabilities'], [0, 1], [2, 1])) return metric_ops

ModelLoader

ModelLoader(
    construct_fn: Callable[[], Any],
    tags: Optional[List[str]] = None,
)

Model loader is responsible for loading shared model types.

ATTRIBUTE	DESCRIPTION
construct_fn	A callable which creates the model instance. The callable should take no args as input (typically a closure is used to capture necessary parameters).
tags	Optional model tags (e.g. 'serve' for serving or 'eval' for EvalSavedModel).

Source code in tensorflow_model_analysis/api/types.py

def __init__(
    self, construct_fn: Callable[[], Any], tags: Optional[List[str]] = None
):
  self.construct_fn = construct_fn
  self.tags = tags
  self._shared_handle = shared.Shared()

Attributes

construct_fn instance-attribute

construct_fn = construct_fn

tags instance-attribute

tags = tags

Functions

load

load(
    model_load_time_callback: Optional[
        Callable[[int], None]
    ] = None,
) -> Any

Returns loaded model.

PARAMETER	DESCRIPTION
model_load_time_callback	Optional callback to track load time. TYPE: Optional[Callable[[int], None]] DEFAULT: None

Source code in tensorflow_model_analysis/api/types.py

def load(
    self, model_load_time_callback: Optional[Callable[[int], None]] = None
) -> Any:
  """Returns loaded model.

  Args:
    model_load_time_callback: Optional callback to track load time.
  """
  if model_load_time_callback:
    construct_fn = self._construct_fn_with_load_time(model_load_time_callback)
  else:
    construct_fn = self.construct_fn
  return self._shared_handle.acquire(construct_fn)

RaggedTensorValue

Bases: NamedTuple('RaggedTensorValue', [('values', ndarray), ('nested_row_splits', List[ndarray])])

RaggedTensorValue encapsulates a batch of ragged tensor values.

ATTRIBUTE	DESCRIPTION
values	A np.ndarray of values.
nested_row_splits	A list of np.ndarray values representing the row splits (one per dimension including the batch dimension).

SparseTensorValue

Bases: NamedTuple('SparseTensorValue', [('values', ndarray), ('indices', ndarray), ('dense_shape', ndarray)])

SparseTensorValue encapsulates a batch of sparse tensor values.

ATTRIBUTE	DESCRIPTION
values	A np.ndarray of values.
indices	A np.ndarray of indices.
dense_shape	A np.ndarray representing the dense shape.

StructuredMetricValue

Bases: ABC

The base class for all structured metrics used within TFMA.

This class allows custom metrics to control how proto serialization happens, and how to handle basic algebraic operations used in computing confidence intervals and model diffs. By implementing the _apply_binary_op methods, subclasses can then be treated like primitive numeric types.

Functions

to_proto abstractmethod

to_proto() -> MetricValue

Source code in tensorflow_model_analysis/api/types.py

@abc.abstractmethod
def to_proto(self) -> metrics_for_slice_pb2.MetricValue:
  ...

ValueWithTDistribution

Bases: NamedTuple('ValueWithTDistribution', [('sample_mean', MetricValueType), ('sample_standard_deviation', MetricValueType), ('sample_degrees_of_freedom', int), ('unsampled_value', MetricValueType)])

Represents the t-distribution value.

It includes sample_mean, sample_standard_deviation, sample_degrees_of_freedom. And also unsampled_value is also stored here to record the value calculated without bootstrapping. The sample_standard_deviation is calculated as: \sqrt{ \frac{1}{N-1} \sum_{i=1}^{N}{(x_i - \bar{x})^2} }

VarLenTensorValue

Bases: NamedTuple('VarLenTensorValue', [('values', ndarray), ('indices', ndarray), ('dense_shape', ndarray)])

VarLenTensorValue encapsulates a batch of varlen dense tensor values.

ATTRIBUTE	DESCRIPTION
values	A np.ndarray of values.
indices	A np.ndarray of indices.
dense_shape	A np.ndarray representing the dense shape of the entire tensor. Note that each row (i.e. set of values sharing the same value for the first / batch dimension) is considered to have its own shape based on the presence of values.

Classes

DenseRowIterator

DenseRowIterator(tensor)

An Iterator over rows of a VarLenTensorValue as dense np.arrays.

Because the VarLenTensorValue was created from a set of variable length (dense) arrays, we can invert this process to turn a VarLenTensorValue back into the original dense arrays.

Source code in tensorflow_model_analysis/api/types.py

def __init__(self, tensor):
  self._tensor = tensor
  self._offset = 0

Functions

Functions

dense_rows

dense_rows()

Source code in tensorflow_model_analysis/api/types.py

def dense_rows(self):
  return self.DenseRowIterator(self)

from_dense_rows classmethod

from_dense_rows(
    dense_rows: Iterable[ndarray],
) -> VarLenTensorValue

Converts a collection of variable length dense arrays into a tensor.

PARAMETER	DESCRIPTION
dense_rows	A sequence of possibly variable length 1D arrays. TYPE: Iterable[ndarray]

RETURNS	DESCRIPTION
VarLenTensorValue	A new VarLenTensorValue containing the sparse representation of the
VarLenTensorValue	vertically stacked dense rows. The dense_shape attribute on the result
VarLenTensorValue	will be (num_rows, max_row_len).

Source code in tensorflow_model_analysis/api/types.py

@classmethod
def from_dense_rows(
    cls, dense_rows: Iterable[np.ndarray]
) -> 'VarLenTensorValue':
  """Converts a collection of variable length dense arrays into a tensor.

  Args:
    dense_rows: A sequence of possibly variable length 1D arrays.

  Returns:
    A new VarLenTensorValue containing the sparse representation of the
    vertically stacked dense rows. The dense_shape attribute on the result
    will be (num_rows, max_row_len).
  """
  rows = []
  index_arrays = []
  max_row_len = 0
  num_rows = 0
  for i, row in enumerate(dense_rows):
    num_rows += 1
    if row.size:
      if row.ndim <= 1:
        # Add a dimension for unsized numpy array. This will solve the problem
        # where scalar numpy arrays like np.array(None), np.array(0) can not
        # be merged with other numpy arrays.
        row = row.reshape(-1)
        rows.append(row)
      else:
        raise ValueError(
            'Each non-empty dense row should be 1D or scalar but'
            f' found row with shape {row.shape}.'
        )
      index_arrays.append(np.array([[i, j] for j in range(len(row))]))
    max_row_len = max(max_row_len, row.size)
  if index_arrays:
    values = np.concatenate(rows, axis=0)
    indices = np.concatenate(index_arrays, axis=0)
  else:
    # empty case
    values = np.array([])
    indices = np.empty((0, 2))
  dense_shape = np.array([num_rows, max_row_len])
  return cls.__new__(
      cls, values=values, indices=indices, dense_shape=dense_shape
  )