class_docstr: Phoenix: A smart search AutoML algorithm.
class_docstr_tok: ["Phoenix",":","A","smart","search","AutoML","algorithm","."]
docstr: Builds a keras model based on hparams.
docstr_tok: ["Builds","a","keras","model","based","on","hparams","."]
code: def keras_model_builder(
      self,
      hparams,
      run_config=None,
      is_training=None,
      input_layer_fn=None,
      compile_model=True,
  ):
      """Builds a keras model based on hparams."""
      if compile_model:
          # Validate keras mode config is correct.
          if self._phoenix_spec.HasField("ensemble_spec"):
              logging.error("Cannot run ensembling in keras mode.")
              self._phoenix_spec.ClearField("ensemble_spec")
          if self._phoenix_spec.HasField("distillation_spec"):
              logging.error("Cannot run distillation in keras mode.")
              self._phoenix_spec.ClearField("distillation_spec")
  
      run_config_ = run_config
      if run_config is None:
          run_config_ = tf.estimator.RunConfig()
  
      if isinstance(hparams, hp.HParams):
          params = hparams
      elif isinstance(
          hparams, keras_tuner.engine.hyperparameters.HyperParameters
      ):
          parameter_values = aggregate_initial_architecture(
              hparams.values
          )
          params = hp.HParams(**parameter_values)
      else:
          parameter_values = aggregate_initial_architecture(
              hparams.values()
          )
          params = hp.HParams(**parameter_values)
  
      logging.info(run_config_)
      logging.info(run_config_.model_dir)
      my_id = architecture_utils.DirectoryHandler.get_trial_id(
          run_config_.model_dir, self._phoenix_spec
      )
  
      # Get all information we have so far.
      trials = []
      # TODO(b/172564129): Only the chief needs the trials. Test to see if
      # workers need them
      if not self._phoenix_spec.HasField("replay"):
          trials = self._metadata.get_completed_trials()
      else:
          params = _merge_hparams(
              params,
              hp.HParams.from_proto(
                  self._phoenix_spec.replay.towers[my_id - 1].hparams
              ),
          )
          params.set_hparam(
              "initial_architecture",
              self._phoenix_spec.replay.towers[my_id - 1].architecture[
                  :
              ],
          )
  
      # Update our database - clean up and sync ops.
      if run_config_.is_chief:
          self._metadata.before_generating_trial_model(
              my_id, run_config_.model_dir
          )
  
      # Determine whether to do ensemble search or distillation on this trial.
      trial_mode = trial_utils.get_trial_mode(
          self._phoenix_spec.ensemble_spec,
          self._phoenix_spec.distillation_spec,
          my_id,
      )
  
      generators = self._controller.get_generators(my_id, trials)
      towers = {}
      for name, generator in generators.items():
          logging.info(generators)
          tower = generator.instance.generate(
              input_layer_fn=input_layer_fn,
              trial_mode=trial_mode,
              logits_dimension=self._logits_dimension,
              hparams=params,
              run_config=run_config_,
              is_training=is_training,
              trials=generator.relevant_trials,
          )
          towers[name] = tower
  
      if (
          input_layer_fn is not None
          and self._phoenix_spec.is_input_shared
      ):
          shared_input_tensor, shared_lengths = input_layer_fn(
              is_training=is_training, scope_name="Phoenix/SharedInput"
          )
          for generator_towers in towers.values():
              for tower in generator_towers:
                  tower.add_feature_columns_input_layer(
                      shared_input_tensor, shared_lengths
                  )
      elif input_layer_fn is not None:
          logging.info(towers)
          for generator_towers in towers.values():
              for tower in generator_towers:
                  if not tower.has_input_tensor():
                      input_tensor, lengths = input_layer_fn(
                          is_training=is_training,
                          scope_name="{}/input".format(tower.name),
                      )
                      tower.add_feature_columns_input_layer(
                          input_tensor, lengths
                      )
  
      if compile_model:
          # Keras mode: Search only
          return self.keras_compile(towers, params)
  
      return towers, trials
  
imports: import functools
  import inspect
  from absl import logging
  import keras_tuner
  from model_search import controller
  from model_search import hparam
  from model_search import loss_fns
  from model_search import metric_fns
  from model_search import task_manager
  from model_search.architecture import architecture_utils
  from model_search.generators import base_tower_generator
  from model_search.generators import trial_utils
  from model_search.meta import transfer_learning
  from model_search.metadata import ml_metadata_db
  from model_search.proto import phoenix_spec_pb2
  from model_search.proto import transfer_learning_spec_pb2
  import numpy
  import tensorflow.compat.v2
label_desc: Use image node_id 4 for calling the Phoenix obj's underlying member method code with example usage: obj.keras_model_builder(hparams, run_config, is_training, input_layer_fn, compile_model) and returns: towers,  trials, self