code: def _create_ranking_model(
      tf_transform_output, hparams
  ) -> tf.keras.Model:
      """Creates a Keras ranking model."""
      (
          context_feature_specs,
          example_feature_specs,
          _,
      ) = features.get_features()
      (
          context_keras_inputs,
          example_keras_inputs,
      ) = struct2tensor_parsing_utils.create_keras_inputs(
          context_feature_specs,
          example_feature_specs,
          features.LIST_SIZE_FEATURE_NAME,
      )
      (
          context_features,
          example_features,
          mask,
      ) = _preprocess_keras_inputs(
          context_keras_inputs,
          example_keras_inputs,
          tf_transform_output,
          hparams,
      )
  
      # Since argspec inspection is expensive, for keras layer,
      # layer_obj._call_spec.arg_names is a property that uses cached argspec for
      # call. We use this to determine whether the layer expects `inputs` as first
      # argument.
      # TODO(b/185176464): update tfr dependency to remove this branch.
      flatten_list = tfr.keras.layers.FlattenList()
  
      # TODO(kathywu): remove the except branch once changes to the call function
      # args in the Keras Layer have been released.
      try:
          first_arg_name = flatten_list._call_spec.arg_names[
              0
          ]  # pylint: disable=protected-access
      except AttributeError:
          first_arg_name = flatten_list._call_fn_args[
              0
          ]  # pylint: disable=protected-access
      if first_arg_name == "inputs":
          (
              flattened_context_features,
              flattened_example_features,
          ) = flatten_list(
              inputs=(context_features, example_features, mask)
          )
      else:
          (
              flattened_context_features,
              flattened_example_features,
          ) = flatten_list(context_features, example_features, mask)
  
      # Concatenate flattened context and example features along `list_size` dim.
      context_input = [
          tf.keras.layers.Flatten()(flattened_context_features[name])
          for name in sorted(flattened_context_features)
      ]
      example_input = [
          tf.keras.layers.Flatten()(flattened_example_features[name])
          for name in sorted(flattened_example_features)
      ]
      input_layer = tf.concat(context_input + example_input, 1)
      dnn = tf.keras.Sequential()
      if hparams["use_batch_norm"]:
          dnn.add(
              tf.keras.layers.BatchNormalization(
                  momentum=hparams["batch_norm_moment"]
              )
          )
      for layer_size in hparams["hidden_layer_dims"]:
          dnn.add(tf.keras.layers.Dense(units=layer_size))
          if hparams["use_batch_norm"]:
              dnn.add(
                  tf.keras.layers.BatchNormalization(
                      momentum=hparams["batch_norm_moment"]
                  )
              )
          dnn.add(tf.keras.layers.Activation(activation=tf.nn.relu))
          dnn.add(tf.keras.layers.Dropout(rate=hparams["dropout_rate"]))
  
      dnn.add(tf.keras.layers.Dense(units=1))
  
      # Since argspec inspection is expensive, for keras layer,
      # layer_obj._call_spec.arg_names is a property that uses cached argspec for
      # call. We use this to determine whether the layer expects `inputs` as first
      # argument.
      restore_list = tfr.keras.layers.RestoreList()
  
      # TODO(kathywu): remove the except branch once changes to the call function
      # args in the Keras Layer have been released.
      try:
          first_arg_name = flatten_list._call_spec.arg_names[
              0
          ]  # pylint: disable=protected-access
      except AttributeError:
          first_arg_name = flatten_list._call_fn_args[
              0
          ]  # pylint: disable=protected-access
      if first_arg_name == "inputs":
          logits = restore_list(inputs=(dnn(input_layer), mask))
      else:
          logits = restore_list(dnn(input_layer), mask)
  
      model = tf.keras.Model(
          inputs={**context_keras_inputs, **example_keras_inputs},
          outputs=logits,
          name="dnn_ranking_model",
      )
      model.compile(
          optimizer=tf.keras.optimizers.Adagrad(
              learning_rate=hparams["learning_rate"]
          ),
          loss=tfr.keras.losses.get(hparams["loss"]),
          metrics=tfr.keras.metrics.default_keras_metrics(),
      )
      return model
  
file_path: tfx/tfx/examples/ranking/ranking_utils.py
imports: import tensorflow
  import tensorflow_ranking
  import tensorflow_transform
  from tfx.examples.ranking import features
  from tfx.examples.ranking import struct2tensor_parsing_utils
  from tfx_bsl.public import tfxio
label_desc: Use image node_id 6 for calling a global function with example usage: _create_ranking_model(tf_transform_output, hparams) and returns: model