code: def _run(self) -> None:
      # Special code path only used for DeepSpeed Autotuning.
      if self._dsat_mode:
          ops = self.context._core.searcher.operations()
          op = next(ops)
          for _ in range(op.length):
              with dsat.dsat_reporting_context(
                  core_context=self.context._core, op=op
              ):
                  _ = self._train_for_step(
                      step_id=self.steps_completed + 1,
                      num_batches=1,
                      total_batches_processed=self.steps_completed,
                  )
  
      assert self.workloads is not None
      for w, response_func in self.workloads:
          try:
              if w.kind == workload.Workload.Kind.RUN_STEP:
                  action = "training"
                  metrics = self._train_for_step(
                      w.step_id,
                      w.num_batches,
                      w.total_batches_processed,
                  )
                  response = {
                      "metrics": metrics,
                      "stop_requested": self.context.get_stop_requested(),
                  }  # type: workload.Response
                  metrics = self.context.distributed.broadcast(metrics)
                  for callback in self.callbacks.values():
                      callback.on_training_workload_end(
                          avg_metrics=metrics["avg_metrics"],
                          batch_metrics=metrics["batch_metrics"],
                      )
              elif (
                  w.kind
                  == workload.Workload.Kind.COMPUTE_VALIDATION_METRICS
              ):
                  action = "validation"
                  response = {
                      "metrics": self._compute_validation_metrics(),
                      "stop_requested": self.context.get_stop_requested(),
                  }
              elif w.kind == workload.Workload.Kind.CHECKPOINT_MODEL:
                  action = "checkpointing"
                  metadata = {
                      "steps_completed": self.steps_completed,
                      "framework": f"torch-{torch.__version__}",
                      "format": "pickle",
                  }
                  with self.context._core.checkpoint.store_path(
                      metadata, shard=True
                  ) as (
                      path,
                      storage_id,
                  ):
                      self._save(path)
                  response = {"uuid": storage_id}
                  for callback in self.callbacks.values():
                      callback.on_checkpoint_upload_end(uuid=storage_id)
              else:
                  raise AssertionError(
                      "Unexpected workload: {}".format(w.kind)
                  )
  
          except det.InvalidHP as e:
              logger.info(
                  f"Invalid hyperparameter exception during {action}: {e}"
              )
              response = workload.InvalidHP()
          response_func(response)
          self.context._maybe_reset_tbd_writer()
          self.upload_tb_files()
  
imports: import abc
  import contextlib
  import logging
  import os
  import pathlib
  import pickle
  import random
  import time
  import warnings
  from typing import Any, Callable, Dict, Iterator, List, Optional, Type, Union, cast
  import deepspeed
  import numpy
  import torch
  from deepspeed.runtime import dataloader
  import determined
  from determined import layers, pytorch, util, workload
  from determined.pytorch import deepspeed
  from determined.pytorch import dsat
label_desc: Use image node_id 6 for calling the DeepSpeedTrialController obj's underlying member method code with example usage: obj._run() without return types