code: def run(cfg: Any) -> None:
      """Runs the routine.
  
      Args:
          cfg: config object with all the hyperparameters
      """
      os.makedirs(cfg.output_directory, exist_ok=True)
  
      # Force evaluation if user trains 0 epochs
      cfg.training.evaluate_before_training = (
          cfg.training.evaluate_before_training
          or cfg.training.epochs == 0
      )
  
      # Set the random seed for reproducibility
      # either random seed when user set it -1 or deterministic user chosen seed
      if cfg.environment.seed < 0:
          cfg.environment._seed = np.random.randint(1_000_000)
      else:
          cfg.environment._seed = cfg.environment.seed
  
      if (
          cfg.architecture.backbone_dtype in ["int8", "int4"]
          and cfg.environment.use_deepspeed
      ):
          raise ValueError(
              f"Deepspeed do not support backbone type {cfg.architecture.backbone_dtype}."
              + " Please set backbone type to float16 or bfloat16 for using deepspeed."
          )
  
      # Prepare environment
      if "WORLD_SIZE" in os.environ:
          cfg.environment._distributed = (
              int(os.environ["WORLD_SIZE"]) > 1
          )
      else:
          cfg.environment._distributed = False
  
      if cfg.environment._distributed:
          cfg.environment._local_rank = int(os.environ["LOCAL_RANK"])
          cfg.environment._device = (
              "cuda:%d" % cfg.environment._local_rank
          )
          if cfg.environment.use_deepspeed:
              deepspeed.init_distributed()
          else:
              torch.distributed.init_process_group(
                  backend="nccl", init_method="env://"
              )
          cfg.environment._cpu_comm = torch.distributed.new_group(
              backend="gloo"
          )
  
          cfg.environment._world_size = (
              torch.distributed.get_world_size()
          )
          cfg.environment._rank = torch.distributed.get_rank()
          torch.cuda.set_device(cfg.environment._rank)
          logger.info(
              f"Training in distributed mode with multiple processes, "
              f"1 GPU per process. Process {cfg.environment._rank}, "
              f"total: {cfg.environment._world_size} "
              f"local rank: {cfg.environment._local_rank}."
          )
  
          # Sync the random seed
          cfg.environment._seed = int(
              sync_across_processes(
                  np.array([cfg.environment._seed]),
                  cfg.environment._world_size,
                  group=cfg.environment._cpu_comm,
              )[0]
          )
      else:
          cfg.environment._local_rank = 0
          cfg.environment._device = (
              "cuda:0"
              if (
                  torch.cuda.is_available()
                  and len(cfg.environment.gpus) > 0
              )
              else "cpu"
          )
          if cfg.environment._device == "cpu":
              logger.warning("Training on CPU. This will be slow.")
  
      set_seed(cfg.environment._seed)
      if cfg.environment._local_rank == 0:
          logger.info(f"Problem Type: {cfg.problem_type}")
          logger.info(f"Global random seed: {cfg.environment._seed}")
  
      cfg = set_environment(cfg)
  
      # we need to get train dataframe and number of labels if not set or in training mode
      if cfg.environment._local_rank == 0:
          logger.info("Preparing the data...")
      train_df, val_df = get_data(cfg)
  
      if (
          len(val_df) > int(os.getenv("GPT_EVAL_MAX", 100))
          and "GPT" in cfg.prediction.metric
      ):
          logger.warning(
              f"More than {os.getenv('GPT_EVAL_MAX', 100)} validation records. "
              "Safeguarding against OpenAI API costs. Setting metric to BLEU. "
              "Change GPT_EVAL_MAX to run GPT validation."
          )
          cfg.prediction.metric = "BLEU"
  
      # prepare data
      if cfg.environment._local_rank == 0:
          logger.info("Preparing train and validation data")
      train_dataset = get_train_dataset(train_df=train_df, cfg=cfg)
      val_dataset = get_val_dataset(val_df=val_df, cfg=cfg)
      train_dataloader = get_train_dataloader(
          train_ds=train_dataset, cfg=cfg
      )
      val_dataloader = get_val_dataloader(val_ds=val_dataset, cfg=cfg)
  
      if cfg.environment._local_rank == 0:
          total_training_steps = (
              cfg.training.epochs
              * len(train_dataloader)
              * cfg.training.batch_size
              * cfg.environment._world_size
          )
  
          num_eval_epochs = get_number_of_validation_epochs(
              training_epochs=cfg.training.epochs,
              evaluation_epochs=cfg.training.evaluation_epochs,
          )
          val_batch_size = get_inference_batch_size(cfg)
  
          # if zero shot, validate once before training
          total_validation_steps = (
              len(val_dataloader)
              * (
                  num_eval_epochs
                  + int(cfg.training.evaluate_before_training)
              )
              * val_batch_size
              * cfg.environment._world_size
          )
  
      # Prepare model and optimizer
      if cfg.environment.use_deepspeed:
          ds_config = get_ds_config(cfg)
          # keep this object alive.
          dschf = HfDeepSpeedConfig(ds_config)  # noqa: F841
      with torch.device(cfg.environment._device):
          model = cfg.architecture.model_class(cfg)
          check_disk_space(
              model, cfg.output_directory, cfg.environment.use_deepspeed
          )
  
          # load model weights
          if cfg.architecture.pretrained_weights != "":
              # Do not load strictly if continue training from the previous experiment
              load_checkpoint(
                  cfg, model, strict=cfg.training.epochs == -1
              )
      model.to(cfg.environment._device)
  
      epoch_steps = len(train_dataloader)
      optimizer = get_optimizer(model=model, cfg=cfg)
      scheduler = get_scheduler(
          cfg=cfg, optimizer=optimizer, epoch_steps=epoch_steps
      )
  
      if getattr(cfg.architecture, "force_embedding_gradients"):
          for module in model.modules():
              if isinstance(module, torch.nn.Embedding):
                  for param in module.parameters():
                      param.requires_grad = True
                      param.data = param.data.float()
  
      if cfg.environment._distributed:
          (
              model,
              optimizer,
              train_dataloader,
              val_dataloader,
              scheduler,
          ) = wrap_model_distributed(
              model=model,
              optimizer=optimizer,
              lr_scheduler=scheduler,
              train_dataloader=train_dataloader,
              val_dataloader=val_dataloader,
              cfg=cfg,
          )
  
      if cfg.environment.compile_model:
          # deepspeed do not support torch.compile
          if cfg.environment.use_deepspeed:
              logger.warning(
                  "Deepspeed is active, but it doesn't support torch.compile."
                  "Skipping compilation for this experiment."
              )
          else:
              if cfg.environment._distributed:
                  model.module.backbone = torch.compile(
                      model.module.backbone
                  )
              else:
                  model.backbone = torch.compile(model.backbone)
  
      # Force settings when saving best checkpoint
      if cfg.training.save_best_checkpoint:
          cfg.training.train_validation_data = False
  
      # reset steps
      cfg.environment._curr_step = 0
      cfg.environment._curr_val_step = 0
  
      gc.collect()
  
      global_start_time = time.time()
      if cfg.environment._local_rank == 0:
          # re-save cfg
          save_config_yaml(f"{cfg.output_directory}/cfg.yaml", cfg)
  
          cfg.logging._logger = MainLogger(cfg)
  
          cfg.logging._logger.log(
              "internal",
              "total_training_steps",
              total_training_steps,
              step=0,
          )
  
          cfg.logging._logger.log(
              "internal",
              "total_validation_steps",
              total_validation_steps,
              step=0,
          )
  
          cfg.logging._logger.log(
              "internal",
              "global_start_time",
              global_start_time,
              step=cfg.environment._curr_step,
          )
          # re-save config
          save_config_yaml(f"{cfg.output_directory}/cfg.yaml", cfg)
  
      train_function = (
          run_train_rlhf
          if cfg.problem_type == "text_rlhf_language_modeling"
          else run_train
      )
      val_loss, val_metric = train_function(
          cfg=cfg,
          model=model,
          optimizer=optimizer,
          scheduler=scheduler,
          epoch_steps=epoch_steps,
          train_dataloader=train_dataloader,
          val_dataloader=val_dataloader,
          val_df=val_df,
      )
  
      # reset external logging
      if cfg.environment._local_rank == 0:
          cfg.logging._logger.reset_external()
  
      experiment_path = f"{cfg.output_directory}"
  
      if cfg.training.epochs == 0:
          checkpoint_path = cfg.output_directory
          if cfg.environment._local_rank == 0:
              logger.info(
                  f"Saving last model checkpoint to {checkpoint_path}"
              )
          save_checkpoint(model=model, path=checkpoint_path, cfg=cfg)
  
      if cfg.environment._local_rank == 0:
          save_config_yaml(f"{cfg.output_directory}/cfg.yaml", cfg)
          save_prediction_outputs(cfg.experiment_name, experiment_path)
  
          flag_path = os.path.join(cfg.output_directory, "flags.json")
          write_flag(flag_path, "status", "finished")
          time_took = time.time() - global_start_time
          if time_took > 86400:
              # if more than one day, show days
              # need to subtract 1 day from time_took since strftime shows day of year
              # which starts counting at 1
              time_took_formatted = time.strftime(
                  "%-jd %H:%M:%S", time.gmtime(float(time_took - 86400))
              )
          else:
              time_took_formatted = time.strftime(
                  "%H:%M:%S", time.gmtime(float(time_took))
              )
          write_flag(
              flag_path, "info", f"Runtime: {time_took_formatted}"
          )
  
imports: import os
  import argparse
  import gc
  import logging
  import sys
  import time
  from distutils import util
  from typing import Any, Callable, Dict, Tuple
  import deepspeed
  import numpy
  import pandas
  import torch
  from torch.cuda.amp import GradScaler, autocast
  from torch.utils.data import DataLoader
  from tqdm import tqdm
  from transformers.deepspeed import HfDeepSpeedConfig
  from llm_studio.src.datasets.text_utils import get_tokenizer
  from llm_studio.src.loggers import MainLogger
  from llm_studio.src.trl.trainer import PPOTrainer
  from llm_studio.src.utils.config_utils import load_config_py, load_config_yaml, save_config_yaml
  from llm_studio.src.utils.data_utils import get_data, get_inference_batch_size, get_train_dataloader, get_train_dataset, get_val_dataloader, get_val_dataset
  from llm_studio.src.utils.exceptions import LLMTrainingException
  from llm_studio.src.utils.export_utils import save_prediction_outputs
  from llm_studio.src.utils.gpu_utils import sync_across_processes
  from llm_studio.src.utils.logging_utils import TqdmToLogger, initialize_logging, log_plot, write_flag
  from llm_studio.src.utils.modeling_utils import activate_neftune, check_disk_space, get_ds_config, get_number_of_validation_epochs, get_optimizer, get_scheduler, load_checkpoint, reduce_metric, run_inference, save_checkpoint, save_predictions, unwrap_model, wrap_model_distributed
  from llm_studio.src.utils.utils import kill_ddp_processes, set_environment, set_seed
label_desc: Use image node_id 4 for calling a global function with example usage: run(cfg) without return types