"""Advanced control of the model that most users won't need to use.

Attributes:

infer_prediction_signature: Instantiate the model graph after training. This

allows the model to be saved without specifying an input signature and

without calling "predict", "evaluate". Disabling this logic can be useful

in two situations: (1) When the exported signature is different from the

one used during training, (2) When using a fixed-shape pre-processing that

consume 1 dimensional tensors (as keras will automatically expend its

shape to rank 2). For example, when using tf.Transform.

yggdrasil_training_config: Yggdrasil Decision Forests training

configuration. Expose a few extra hyper-parameters.

yggdrasil_deployment_config: Configuration of the computing resources used

to train the model e.g. number of threads. Does not impact the model

quality.

fail_on_non_keras_compatible_feature_name: If true (default), training will

fail if one of the feature name is not compatible with part of the Keras

API. If false, a warning will be generated instead.

predict_single_probability_for_binary_classification: Only used for binary

classification. If true (default), the prediction of a binary class model

is a tensor of shape [None, 1] containing the probability of the positive

class (value=1). If false, the prediction of a binary class model is a

tensor of shape [None, num_classes=2] containing the probability of the

complementary classes.

metadata_framework: Metadata describing the framework used to train the

model.

metadata_owner: Metadata describing who trained the model.

populate_history_with_yggdrasil_logs: If false (default) and if a validation

dataset is provided, populate the model's history with the final

validation evaluation computed by the Keras metric (i.e. one evaluation).

If true or if no validation dataset is provided, populate the model's

history with the yggdrasil training logs. The yggdrasil training logs

contains more metrics, but those might not be comparable with other non

TF-DF models.

disable_categorical_integer_offset_correction: Yggdrasil Decision Forests

reserves the value 0 of categorical integer features to the OOV item, so

the value 0 cannot be used directly. If the

`disable_categorical_integer_offset_correction` is true, a +1 offset might

be applied before calling the inference code. This attribute should be

disabled when creating manually a model with categorical integer features.

Ultimately, Yggdrasil Decision Forests will support the value 0 as a

normal value and this parameter will be removed. If

`disable_categorical_integer_offset_correction` is false, this +1 offset

is never applied.

node_format: Yggdrasil Decision Forests node format for the saved model. If

not specified, uses the recommended format. The node format is visible in

the node summary. For models to be compatible with the open-source version

of TensorFlow Decision Forests and TensorFlow Serving, the node format

should be BLOB_SEQUENCE.

allow_slow_inference: If false, slow inference engines are not allowed. If

the model is only available with the slow engine, an error is raised. If

true, the fastest compatible inference engine (possibly the slow one) will

be used.

force_ydf_port: Socket port for YDF GRPC to use during distributed training

in addition to the TF GRPC. The chief and the workers should be able to

communicate thought this port. If not set, an available port is

automatically selected.

output_secondary_class_predictions: If true, in the case of a multi-task

model, the predictions of secondary tasks are exported in the model

predictions. If false, the model only outputs the primary tasks

predictions.

"""

def __init__(

self,

infer_prediction_signature: Optional[bool] = True,

yggdrasil_training_config: Optional[YggdrasilTrainingConfig] = None,

yggdrasil_deployment_config: Optional[YggdrasilDeploymentConfig] = None,

fail_on_non_keras_compatible_feature_name: Optional[bool] = True,

predict_single_probability_for_binary_classification: Optional[

bool

] = True,

metadata_framework: Optional[str] = "TF Keras",

metadata_owner: Optional[str] = None,

populate_history_with_yggdrasil_logs: bool = False,

disable_categorical_integer_offset_correction: bool = False,

node_format: Optional[NodeFormat] = None,

allow_slow_inference: bool = True,

force_ydf_port: Optional[int] = None,

output_secondary_class_predictions: bool = False,

):

self.infer_prediction_signature = infer_prediction_signature

self.yggdrasil_training_config = (

yggdrasil_training_config or abstract_learner_pb2.TrainingConfig()

)

self.yggdrasil_deployment_config = (

yggdrasil_deployment_config or abstract_learner_pb2.DeploymentConfig()

)

self.fail_on_non_keras_compatible_feature_name = (

fail_on_non_keras_compatible_feature_name

)

self.predict_single_probability_for_binary_classification = (

predict_single_probability_for_binary_classification

)

self.metadata_framework = metadata_framework

self.metadata_owner = metadata_owner

self.populate_history_with_yggdrasil_logs = (

populate_history_with_yggdrasil_logs

)

self.disable_categorical_integer_offset_correction = (

disable_categorical_integer_offset_correction

)

self.node_format = node_format

self.allow_slow_inference = allow_slow_inference

self.force_ydf_port = force_ydf_port

"""A single task in a multi-task configuration.

Models trained to predict label with primary=False are used to help the

training of models with primary=True. If primary is true for all the tasks,

each task will be trained independently.

Args:

label: Key of the label.

task: Task for the label.

primary: The predictions of primary tasks are returned by "model.predict()",

the output of non-primary tasks are not returned by "model.predict()".

Non-primary tasks can be used to improve the primary task quality.

output: If None, "model.predict" returns a dictionary of prediction indexed

by "label" values. If set, "model.predict" returns a dictionary of

predictions indexed by "output".

"""

label: str

task: TaskType = Task.CLASSIFICATION

primary: bool = True

output: Optional[str] = None

@property

def prediction_key(self) -> str:

"""Keras Model V2 wrapper around an Yggdrasil Model.

See "CoreModel" in "core.py" for the definition of the arguments.

"""

def __init__(

self,

task: Optional[TaskType] = Task.CLASSIFICATION,

ranking_group: Optional[str] = None,

verbose: int = 1,

advanced_arguments: Optional[AdvancedArguments] = None,

name: Optional[str] = None,

preprocessing: Optional[keras_internal.Functional] = None,

postprocessing: Optional[keras_internal.Functional] = None,

uplift_treatment: Optional[str] = None,

temp_directory: Optional[str] = None,

multitask: Optional[List[MultiTaskItem]] = None,

tuner: Optional[tuner_lib.Tuner] = None,

learner: Optional[str] = "RANDOM_FOREST",

):

super(InferenceCoreModel, self).__init__(name=name)

self._verbose = verbose

self._preprocessing = preprocessing

self._postprocessing = postprocessing

self._temp_directory = temp_directory

if advanced_arguments is None:

self._advanced_arguments = AdvancedArguments()

else:

self._advanced_arguments = copy.deepcopy(advanced_arguments)

# Training configuration

training_config = self._advanced_arguments.yggdrasil_training_config

# Label

if multitask is None:

self._multitask = [MultiTaskItem(label=_LABEL, task=task)]

self._is_multitask = False

training_config.label = tf_core.normalize_inputs_regexp(_LABEL, False)

training_config.task = task

else:

self._multitask = multitask

self._is_multitask = True

training_config.label = tf_core.normalize_inputs_regexp(

multitask[0].label, False

)

training_config.task = multitask[0].task

# Learner

if tuner is not None:

if self._is_multitask:

raise ValueError("Multi-task learning is not compatible with the tuner")

tuner.set_base_learner(learner)

training_config.MergeFrom(tuner.train_config())

elif self._is_multitask:

training_config.learner = "MULTITASKER"

multitasker_config = training_config.Extensions[

multitasker_pb2.multitasker_config

]

multitasker_config.base_learner.learner = learner

for sub_task in self._multitask:

multitasker_task = multitasker_config.subtasks.add()

multitasker_task.train_config.label = tf_core.normalize_inputs_regexp(

sub_task.label, False

)

multitasker_task.train_config.task = sub_task.task

multitasker_task.primary = sub_task.primary

else:

training_config.learner = learner

# Ranking group

if ranking_group:

training_config.ranking_group = ranking_group

# Uplift treatment

if uplift_treatment:

training_config.uplift_treatment = uplift_treatment

# Copy the metadata

if (

not training_config.metadata.HasField("framework")

and self._advanced_arguments.metadata_framework

):

training_config.metadata.framework = (

self._advanced_arguments.metadata_framework

)

if (

not training_config.metadata.HasField("owner")

and self._advanced_arguments.metadata_owner

):

training_config.metadata.owner = self._advanced_arguments.metadata_owner

for sub_task in self._multitask:

if (sub_task.task == Task.RANKING) != (ranking_group is not None):

raise ValueError(

"ranking_key is used iif. the task is RANKING or the loss is a "

"ranking loss"

)

# True iif. the model is trained.

self._is_trained = tf.Variable(False, trainable=False, name="is_trained")

# Unique ID to identify the model during training.

self._training_model_id = generate_training_id()

# The following fields contain the trained model. They are set during the

# graph construction and training process.

# The compiled Yggdrasil models. Indexed the same way as "_multitask".

self._models: Optional[List[tf_op.ModelV2]] = None

# Compiled Yggdrasil model specialized for returning the active leaves.

# This model is initialized at the first call to "call_get_leaves" or

# "predict_get_leaves". Indexed the same way as "_multitask".

self._models_get_leaves: Optional[List[tf_op.ModelV2]] = None

# Semantic of the input features.

# Also defines what are the input features of the model.

self._semantics: Optional[Dict[str, FeatureSemantic]] = None

# List of Yggdrasil feature identifiers i.e. feature seen by the Yggdrasil

# learner. Those are computed after the preprocessing, unfolding and

# casting.

self._normalized_input_keys: Optional[List[str]] = None

# Textual description of the model.

self._description: Optional[str] = None

def ranking_group(self) -> Optional[str]:

training_config = self._advanced_arguments.yggdrasil_training_config

if not training_config.HasField("ranking_group"):

return None

return training_config.ranking_group

def uplift_treatment(self) -> Optional[str]:

training_config = self._advanced_arguments.yggdrasil_training_config

if not training_config.HasField("uplift_treatment"):

return None

return training_config.uplift_treatment

@property

def task(self) -> Optional[TaskType]:

"""Task to solve (e.g. CLASSIFICATION, REGRESSION, RANKING)."""

if self._is_multitask:

raise ValueError(

"Cannot call .task() on a multitask model. Use .multitask() instead."

)

assert len(self._multitask) == 1

return self._multitask[0].task

@property

def multitask(self) -> List[MultiTaskItem]:

"""Tasks to solve."""

if not self._is_multitask:

raise ValueError(

"Cannot call .multitask() on a non-multitask model. Use .task()"

" instead."

)

return self._multitask

def make_inspector(self, index: int = 0) -> inspector_lib.AbstractInspector:

"""Creates an inspector to access the internal model structure.

Usage example:

```python

inspector = model.make_inspector()

print(inspector.num_trees())

print(inspector.variable_importances())

```

Args:

index: Index of the sub-model. Only used for multitask models.

Returns:

A model inspector.

"""

path = self.yggdrasil_model_path_tensor().numpy().decode("utf-8")

return inspector_lib.make_inspector(

path, file_prefix=self.yggdrasil_model_prefix(index)

)

def get_config(self):

"""Not supported by TF-DF, returning empty directory to avoid warnings."""

return {}

@tf.function(input_signature=[])

def yggdrasil_model_path_tensor(

self, multitask_model_index: int = 0

) -> Optional[tf.Tensor]:

"""Gets the path to yggdrasil model, if available.

The effective path can be obtained with:

```python

yggdrasil_model_path_tensor().numpy().decode("utf-8")

```

Args:

multitask_model_index: Index of the sub-model. Only used for multitask

models.

Returns:

Path to the Yggdrasil model.

"""

if multitask_model_index >= len(self._models):

raise ValueError(

f"Requesting sub-model {multitask_model_index} but only"

f" {len(self._models)} sub-models are available"

)

return self._models[

multitask_model_index

]._compiled_model._model_loader.get_model_path() # pylint: disable=protected-access

def yggdrasil_model_prefix(self, index: int = 0) -> str:

"""Gets the prefix of the internal yggdrasil model."""

if index >= len(self._models):

raise ValueError(

f"Requesting sub-model {index} but only {len(self._models)} "

"sub-models are available"

)

return self._models[index]._compiled_model._model_loader.get_model_prefix() # pylint: disable=protected-access

def make_predict_function(self): # pytype: disable=signature-mismatch # overriding-parameter-count-checks

"""Prediction of the model (!= evaluation)."""

@tf.function(reduce_retracing=True)

def predict_function_not_trained(iterator):

"""Prediction of a non-trained model. Returns "zeros"."""

data = next(iterator)

x, _, _ = keras_internal.unpack_x_y_sample_weight(data)

batch_size = _batch_size(x)

return tf.zeros([batch_size, 1])

@tf.function(reduce_retracing=True)

def predict_function_trained(iterator, model):

"""Prediction of a trained model.

The only difference with "super.make_predict_function()" is that

"self.predict_function" is not set and that the "distribute_strategy"

is not used.

Args:

iterator: Iterator over the dataset.

model: Model object.

Returns:

Model predictions.

"""

def run_step(data):

outputs = model.predict_step(data)

with tf.control_dependencies(_minimum_control_deps(outputs)):

model._predict_counter.assign_add(1) # pylint:disable=protected-access

return outputs

data = next(iterator)

return run_step(data)

if self._is_trained.value():

return partial(predict_function_trained, model=self)

else:

return predict_function_not_trained

def make_test_function(self): # pytype: disable=signature-mismatch # overriding-parameter-count-checks

"""Predictions for evaluation."""

@tf.function(reduce_retracing=True)

def test_function_not_trained(iterator):

"""Evaluation of a non-trained model."""

next(iterator)

return {}

@tf.function(reduce_retracing=True)

def step_function_trained(model, iterator):

"""Evaluation of a trained model.

The only difference with "super.make_test_function()" is that

"self.test_function" is not set.

Args:

model: Model object.

iterator: Iterator over dataset.

Returns:

Evaluation metrics.

"""

def run_step(data):

outputs = model.test_step(data)

with tf.control_dependencies(_minimum_control_deps(outputs)):

model._test_counter.assign_add(1) # pylint:disable=protected-access

return outputs

data = next(iterator)

outputs = model.distribute_strategy.run(run_step, args=(data,))

outputs = _reduce_per_replica(

outputs, self.distribute_strategy, reduction="first"

)

return outputs

if self._is_trained.value():

# Special case if steps_per_execution is one.

if (

self._steps_per_execution is None

or self._steps_per_execution.numpy().item() == 1

):

def test_function(iterator):

"""Runs a test execution with a single step."""

return step_function_trained(self, iterator)

if not self.run_eagerly:

test_function = tf.function(test_function, reduce_retracing=True)

if self._cluster_coordinator:

return lambda it: self._cluster_coordinator.schedule( # pylint: disable=g-long-lambda

test_function, args=(it,)

)

else:

return test_function

# If we're using a coordinator, use the value of self._steps_per_execution

# at the time the function is called/scheduled, and not when it is

# actually executed.

elif self._cluster_coordinator:

def test_function(iterator, steps_per_execution):

"""Runs a test execution with multiple steps."""

for _ in tf.range(steps_per_execution):

outputs = step_function_trained(self, iterator)

return outputs

if not self.run_eagerly:

test_function = tf.function(test_function, reduce_retracing=True)

return lambda it: self._cluster_coordinator.schedule( # pylint: disable=g-long-lambda

test_function, args=(it, self._steps_per_execution.value())

)

else:

def test_function(iterator):

"""Runs a test execution with multiple steps."""

for _ in tf.range(self._steps_per_execution):

outputs = step_function_trained(self, iterator)

return outputs

if not self.run_eagerly:

test_function = tf.function(test_function, reduce_retracing=True)

return test_function

else:

return test_function_not_trained

@tf.function(reduce_retracing=True)

def _build_normalized_inputs(self, inputs) -> Dict[str, tf_core.AnyTensor]:

"""Computes the normalized input of the model.

The normalized inputs are inputs compatible with the Yggdrasil model.

Args:

inputs: Input tensors.

Returns:

Normalized inputs.

"""

assert self._semantics is not None

assert self._models is not None

if self._preprocessing is not None:

inputs = self._preprocessing(inputs)

if isinstance(inputs, dict):

# Native format

pass

elif isinstance(inputs, tf.Tensor):

if len(self._semantics) != 1:

raise ValueError(

"Calling model with input shape different from the "

"input shape provided during training: Feeding a single array "

f"{inputs} while the model was trained on {self._semantics}."

)

inputs = {next(iter(self._semantics.keys())): inputs}

elif isinstance(inputs, list) or isinstance(inputs, tuple):

# Note: The name of a tensor (value.name) can change between the training

# and the inference.

inputs = {str(idx): value for idx, value in enumerate(inputs)}

else:

raise ValueError(

"The inference input tensor is expected to be a tensor, list of "

f"tensors or a dictionary of tensors. Got {inputs} instead"

)

# Normalize the input tensor to match Yggdrasil requirements.

semantic_inputs = tf_core.combine_tensors_and_semantics(

inputs, self._semantics

)

normalized_semantic_inputs = tf_core.normalize_inputs(

semantic_inputs,

categorical_integer_offset_correction=not self._advanced_arguments.disable_categorical_integer_offset_correction,

)

normalized_inputs, _ = tf_core.decombine_tensors_and_semantics(

normalized_semantic_inputs

)

return normalized_inputs

@tf.function(reduce_retracing=True)

def call(self, inputs, training=False):

"""Inference of the model.

This method is used for prediction and evaluation of a trained model.

Args:

inputs: Input tensors.

training: Is the model being trained. Always False.

Returns:

Model predictions.

"""

del training

if self._semantics is None:

tf_logging.warning(

(

"The model was called directly (i.e. using `model(data)` instead"

" of using `model.predict(data)`) before being trained. The model"

" will only return zeros until trained. The output shape might"

" change after training %s"

),

inputs,

)

return tf.zeros([_batch_size(inputs), 1])

normalized_inputs = self._build_normalized_inputs(inputs)

predictions = {}

has_secondary_tasks = any([not t.primary for t in self._multitask])

if has_secondary_tasks:

# The model contains two "layers" of DF models. The output of the first

# layer is used as input for the second layer.

secondary_outputs = {}

for item_idx, multitask_item in enumerate(self._multitask):

if multitask_item.primary:

continue

sub_pred = self._models[item_idx].apply(normalized_inputs)

finalized_pred = self._finalize_predictions(

multitask_item.task, sub_pred, like_engine=True

)

for pred_idx in range(finalized_pred.shape[1]):

secondary_outputs[f"{multitask_item.label}:{pred_idx}"] = (

finalized_pred[:, pred_idx]

)

if self._advanced_arguments.output_secondary_class_predictions:

predictions[multitask_item.prediction_key] = (

self._finalize_predictions(multitask_item.task, sub_pred)

)

# Add the output of the first layer (the secondary models) to the global

# input pool.

normalized_inputs = {**normalized_inputs, **secondary_outputs}

for item_idx, multitask_item in enumerate(self._multitask):

if not multitask_item.primary:

continue

sub_pred = self._models[item_idx].apply(normalized_inputs)

predictions[multitask_item.prediction_key] = self._finalize_predictions(

multitask_item.task, sub_pred

)

if not self._is_multitask:

predictions = predictions[self._multitask[0].label]

if self._postprocessing is not None:

predictions = self._postprocessing(predictions)

return predictions

@tf.function(reduce_retracing=True)

def _finalize_predictions(

self, task: TaskType, predictions, like_engine: bool = False

):

"""Finalizes the predictions before being sent back to the user.

Args:

task: The task of the predictions.

predictions: The model output.

like_engine: If set, ignore the model configuration and finalize the

predictions so they have the same shape as Yggdrasil engine output.

"""

if not like_engine and (

self._advanced_arguments.predict_single_probability_for_binary_classification

and task == Task.CLASSIFICATION

and predictions.dense_predictions.shape[1] == 2

):

# Yggdrasil returns the probably of both classes in binary classification.

# Keras expects only the value (logit or probability) of the "positive"

# class (value=1).

return predictions.dense_predictions[:, 1:2]

else:

return predictions.dense_predictions

@tf.function(reduce_retracing=True)

def call_get_leaves(self, inputs):

"""Computes the index of the active leaf in each tree.

The active leaf is the leave that that receive the example during inference.

The returned value "leaves[i,j]" is the index of the active leave for the

i-th example and the j-th tree. Leaves are indexed by depth first

exploration with the negative child visited before the positive one

(similarly as "iterate_on_nodes()" iteration). Leaf indices are also

available with LeafNode.leaf_idx.

Args:

inputs: Input tensors. Same signature as the model's "call(inputs)".

Returns:

Index of the active leaf for each tree in the model.

"""

if self._is_multitask:

raise ValueError(

"call_get_leaves is not compatible with multi-task models"

)

assert len(self._models_get_leaves) == 1

if self._semantics is None:

tf_logging.warning(

(

"The model was called directly using `call_get_leaves` before "

"being trained. This method will "

"only return zeros until trained. The output shape might change "

"after training %s"

),

inputs,

)

return tf.zeros([_batch_size(inputs), 1])

self._ensure_model_get_leaves_ready()

normalized_inputs = self._build_normalized_inputs(inputs)

return self._models_get_leaves[0].apply_get_leaves(normalized_inputs)

def predict_get_leaves(self, x):

"""Gets the index of the active leaf of each tree.

The active leaf is the leave that that receive the example during inference.

The returned value "leaves[i,j]" is the index of the active leave for the

i-th example and the j-th tree. Leaves are indexed by depth first

exploration with the negative child visited before the positive one

(similarly as "iterate_on_nodes()" iteration). Leaf indices are also

available with LeafNode.leaf_idx.

Args:

x: Input samples as a tf.data.Dataset.

Returns:

Index of the active leaf for each tree in the model.

"""

self._ensure_model_get_leaves_ready()

leaves = []

for row in x:

if isinstance(row, tuple):

# Remove the label and weight.

row = row[0]

leaves.append(self.call_get_leaves(row))

return tf.concat(leaves, axis=0).numpy()

def _ensure_model_get_leaves_ready(self):

"""Ensures that the model that generates the leaves is available."""

# TODO: Re-use "_models" if it supports the get-leaves inference.

assert not self._is_multitask

if self._models_get_leaves is None:

self._models_get_leaves = [

tf_op.ModelV2(

model_path=self.yggdrasil_model_path_tensor()

.numpy()

.decode("utf-8"),

file_prefix=self.yggdrasil_model_prefix(0),

verbose=False,

output_types=["LEAVES"],

)

]

def compile(self, metrics=None, weighted_metrics=None, **kwargs):

"""Configure the model for training.

Unlike for most Keras model, calling "compile" is optional before calling

"fit".

Args:

metrics: List of metrics to be evaluated by the model during training and

testing.

weighted_metrics: List of metrics to be evaluated and weighted by

`sample_weight` or `class_weight` during training and testing.

**kwargs: Other arguments passed to compile.

Raises:

ValueError: Invalid arguments.

"""

super(InferenceCoreModel, self).compile(

metrics=metrics, weighted_metrics=weighted_metrics, **kwargs

)

def summary(self, line_length=None, positions=None, print_fn=None): # pytype: disable=signature-mismatch # overriding-parameter-count-checks

"""Shows information about the model."""

super(InferenceCoreModel, self).summary(

line_length=line_length, positions=positions, print_fn=print_fn

)

if print_fn is None:

print_fn = print

if self._models is not None:

print_fn(self._description)

# TODO: Use Trace Protocol For TF DF custom types to avoid

# clearing the cache.

def _clear_function_cache(self):

"""Clear the @tf.function cache and force re-tracing."""

self.call = tf.function(self.call._python_function, reduce_retracing=True)

def _extract_sample(self, x):

"""Extracts a sample (e.g.

batch, row) from the training dataset.

Returns None is the sample cannot be extracted.

Args:

x: Training dataset in the same format as "fit".

Returns:

A sample.

"""

if isinstance(x, tf.data.Dataset):

return x.take(1)

if isinstance(x, input_lib.DistributedDatasetsFromFunction):

try:

dataset = x._dataset_fn(None) # pylint: disable=protected-access

# Extract the example here (instead of inside of "predict") to make

# sure this operation is done on the chief.

for row in dataset.take(1):

x, _, _ = keras_internal.unpack_x_y_sample_weight(row)

return x

except Exception: # pylint: disable=broad-except

pass

try:

# Work for numpy array and TensorFlow Tensors.

return tf.nest.map_structure(lambda v: v[0:1], x)

except Exception: # pylint: disable=broad-except

pass

try:

# Works for list of primitives.

if isinstance(x, list) and isinstance(

x[0], (int, float, str, bytes, bool)

):

return x[0:1]

except Exception: # pylint: disable=broad-except

pass

tf_logging.warning("Dataset sampling not implemented for %s", x)

return None

def _build(self, x):

"""Build the internal graph similarly as "build" for classical Keras models.

Compared to the classical build, supports features with dtypes != float32.

Args:

x: Training dataset in the same format as "fit".

"""

if self._verbose >= 1:

tf_logging.info("Compiling model...")

# Note: Build does not support dtypes other than float32.

super(InferenceCoreModel, self).build([])

# Force the creation of the graph.

# If a sample cannot be extracted, the graph will be built at the first call

# to "predict" or "evaluate".

if self._advanced_arguments.infer_prediction_signature:

sample = self._extract_sample(x)

if sample is not None:

self.predict(sample, verbose=0)

if self._verbose >= 1:

tf_logging.info("Model compiled.")

def _set_from_yggdrasil_model(

self,

inspector: inspector_lib.AbstractInspector,

path: str,

file_prefix: Optional[str] = None,

input_model_signature_fn: Optional[

tf_core.InputModelSignatureFn

] = tf_core.build_default_input_model_signature,

):

if not self._is_compiled:

self.compile()

features = inspector.features()

semantics = {

feature.name: tf_core.column_type_to_semantic(feature.type)

for feature in features

}

self._training_model_id = file_prefix

self._semantics = semantics

self._normalized_input_keys = sorted(list(semantics.keys()))

self._is_trained.assign(True)

if isinstance(inspector, inspector_lib._MultitaskerInspector): # pylint: disable=protected-access

assert inspector.model_type() == "MULTITASKER"

assert self._is_multitask

self._models = []

for task_idx in range(len(self._multitask)):

self._models.append(

tf_op.ModelV2(

model_path=path,

verbose=False,

file_prefix=inspector.submodel_prefix(task_idx),

allow_slow_inference=self._advanced_arguments.allow_slow_inference,

)

)

else:

assert not self._is_multitask

assert len(self._multitask) == 1

self._models = [

tf_op.ModelV2(

model_path=path,

verbose=False,

file_prefix=file_prefix,

allow_slow_inference=self._advanced_arguments.allow_slow_inference,

)

]

# Instantiate the model's graph

input_model_signature = input_model_signature_fn(inspector)

@tf.function

def f(x):