code: def main():
      # See all possible arguments in src/transformers/training_args.py
      # or by passing the --help flag to this script.
      # We now keep distinct sets of args, for a cleaner separation of concerns.
  
      parser = HfArgumentParser(
          (ModelArguments, DataTrainingArguments, TrainingArguments)
      )
      if len(sys.argv) == 2 and sys.argv[1].endswith(".json"):
          # If we pass only one argument to the script and it's the path to a json file,
          # let's parse it to get our arguments.
          model_args, data_args, training_args = parser.parse_json_file(
              json_file=os.path.abspath(sys.argv[1])
          )
      else:
          (
              model_args,
              data_args,
              training_args,
          ) = parser.parse_args_into_dataclasses()
  
      # Detecting last checkpoint.
      last_checkpoint = None
      if (
          os.path.isdir(training_args.output_dir)
          and training_args.do_train
          and not training_args.overwrite_output_dir
      ):
          last_checkpoint = get_last_checkpoint(
              training_args.output_dir
          )
          if (
              last_checkpoint is None
              and len(os.listdir(training_args.output_dir)) > 0
          ):
              raise ValueError(
                  f"Output directory ({training_args.output_dir}) already exists and is not empty. "
                  "Use --overwrite_output_dir to overcome."
              )
          elif last_checkpoint is not None:
              logger.info(
                  f"Checkpoint detected, resuming training at {last_checkpoint}. To avoid this behavior, change "
                  "the `--output_dir` or add `--overwrite_output_dir` to train from scratch."
              )
  
      # Setup logging
      logging.basicConfig(
          format="%(asctime)s - %(levelname)s - %(name)s -   %(message)s",
          datefmt="%m/%d/%Y %H:%M:%S",
          handlers=[logging.StreamHandler(sys.stdout)],
      )
      logger.setLevel(
          logging.INFO
          if is_main_process(training_args.local_rank)
          else logging.WARN
      )
  
      # Log on each process the small summary:
      logger.warning(
          f"Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}"
          + f"distributed training: {bool(training_args.local_rank != -1)}, 16-bits training: {training_args.fp16}"
      )
      # Set the verbosity to info of the Transformers logger (on main process only):
      if is_main_process(training_args.local_rank):
          transformers.utils.logging.set_verbosity_info()
          transformers.utils.logging.enable_default_handler()
          transformers.utils.logging.enable_explicit_format()
      logger.info(f"Training/evaluation parameters {training_args}")
  
      # Set seed before initializing model.
      set_seed(training_args.seed)
  
      # Get the datasets: you can either provide your own CSV/JSON training and evaluation files (see below)
      # or specify a GLUE benchmark task (the dataset will be downloaded automatically from the datasets Hub).
      #
      # For CSV/JSON files, this script will use as labels the column called 'label' and as pair of sentences the
      # sentences in columns called 'sentence1' and 'sentence2' if such column exists or the first two columns not named
      # label if at least two columns are provided.
      #
      # If the CSVs/JSONs contain only one non-label column, the script does single sentence classification on this
      # single column. You can easily tweak this behavior (see below)
      #
      # In distributed training, the load_dataset function guarantee that only one local process can concurrently
      # download the dataset.
      if data_args.task_name is not None:
          # Downloading and loading a dataset from the hub.
          datasets = load_dataset("glue", data_args.task_name)
      else:
          # Loading a dataset from your local files.
          # CSV/JSON training and evaluation files are needed.
          data_files = {
              "train": data_args.train_file,
              "validation": data_args.validation_file,
          }
  
          # Get the test dataset: you can provide your own CSV/JSON test file (see below)
          # when you use `do_predict` without specifying a GLUE benchmark task.
          if training_args.do_predict:
              if data_args.test_file is not None:
                  train_extension = data_args.train_file.split(".")[-1]
                  test_extension = data_args.test_file.split(".")[-1]
                  assert (
                      test_extension == train_extension
                  ), "`test_file` should have the same extension (csv or json) as `train_file`."
                  data_files["test"] = data_args.test_file
              else:
                  raise ValueError(
                      "Need either a GLUE task or a test file for `do_predict`."
                  )
  
          for key in data_files.keys():
              logger.info(
                  f"load a local file for {key}: {data_files[key]}"
              )
  
          if data_args.train_file.endswith(".csv"):
              # Loading a dataset from local csv files
              datasets = load_dataset("csv", data_files=data_files)
          else:
              # Loading a dataset from local json files
              datasets = load_dataset("json", data_files=data_files)
      # See more about loading any type of standard or custom dataset at
      # https://huggingface.co/docs/datasets/loading_datasets.html.
  
      # Labels
      if data_args.task_name is not None:
          is_regression = data_args.task_name == "stsb"
          if not is_regression:
              label_list = datasets["train"].features["label"].names
              num_labels = len(label_list)
          else:
              num_labels = 1
      else:
          # Trying to have good defaults here, don't hesitate to tweak to your needs.
          is_regression = datasets["train"].features["label"].dtype in [
              "float32",
              "float64",
          ]
          if is_regression:
              num_labels = 1
          else:
              # A useful fast method:
              # https://huggingface.co/docs/datasets/package_reference/main_classes.html#datasets.Dataset.unique
              label_list = datasets["train"].unique("label")
              label_list.sort()  # Let's sort it for determinism
              num_labels = len(label_list)
  
      # Load pretrained model and tokenizer
      #
      # In distributed training, the .from_pretrained methods guarantee that only one local process can concurrently
      # download model & vocab.
      config = AutoConfig.from_pretrained(
          model_args.config_name
          if model_args.config_name
          else model_args.model_name_or_path,
          num_labels=num_labels,
          finetuning_task=data_args.task_name,
          cache_dir=model_args.cache_dir,
          revision=model_args.model_revision,
          use_auth_token=True if model_args.use_auth_token else None,
      )
      tokenizer = AutoTokenizer.from_pretrained(
          model_args.tokenizer_name
          if model_args.tokenizer_name
          else model_args.model_name_or_path,
          cache_dir=model_args.cache_dir,
          use_fast=model_args.use_fast_tokenizer,
          revision=model_args.model_revision,
          use_auth_token=True if model_args.use_auth_token else None,
      )
      model = AutoModelForSequenceClassification.from_pretrained(
          model_args.model_name_or_path,
          from_tf=bool(".ckpt" in model_args.model_name_or_path),
          config=config,
          cache_dir=model_args.cache_dir,
          revision=model_args.model_revision,
          use_auth_token=True if model_args.use_auth_token else None,
      )
  
      # Preprocessing the datasets
      if data_args.task_name is not None:
          sentence1_key, sentence2_key = task_to_keys[
              data_args.task_name
          ]
      else:
          # Again, we try to have some nice defaults but don't hesitate to tweak to your use case.
          non_label_column_names = [
              name
              for name in datasets["train"].column_names
              if name != "label"
          ]
          if (
              "sentence1" in non_label_column_names
              and "sentence2" in non_label_column_names
          ):
              sentence1_key, sentence2_key = "sentence1", "sentence2"
          else:
              if len(non_label_column_names) >= 2:
                  sentence1_key, sentence2_key = non_label_column_names[
                      :2
                  ]
              else:
                  sentence1_key, sentence2_key = (
                      non_label_column_names[0],
                      None,
                  )
  
      # Padding strategy
      if data_args.pad_to_max_length:
          padding = "max_length"
      else:
          # We will pad later, dynamically at batch creation, to the max sequence length in each batch
          padding = False
  
      # Some models have set the order of the labels to use, so let's make sure we do use it.
      label_to_id = None
      if (
          model.config.label2id
          != PretrainedConfig(num_labels=num_labels).label2id
          and data_args.task_name is not None
          and not is_regression
      ):
          # Some have all caps in their config, some don't.
          label_name_to_id = {
              k.lower(): v for k, v in model.config.label2id.items()
          }
          if list(sorted(label_name_to_id.keys())) == list(
              sorted(label_list)
          ):
              label_to_id = {
                  i: int(label_name_to_id[label_list[i]])
                  for i in range(num_labels)
              }
          else:
              logger.warn(
                  "Your model seems to have been trained with labels, but they don't match the dataset: ",
                  f"model labels: {list(sorted(label_name_to_id.keys()))}, dataset labels: {list(sorted(label_list))}."
                  "\nIgnoring the model labels as a result.",
              )
      elif data_args.task_name is None and not is_regression:
          label_to_id = {v: i for i, v in enumerate(label_list)}
  
      if data_args.max_seq_length > tokenizer.model_max_length:
          logger.warn(
              f"The max_seq_length passed ({data_args.max_seq_length}) is larger than the maximum length for the"
              f"model ({tokenizer.model_max_length}). Using max_seq_length={tokenizer.model_max_length}."
          )
      max_seq_length = min(
          data_args.max_seq_length, tokenizer.model_max_length
      )
  
      def preprocess_function(examples):
          # Tokenize the texts
          args = (
              (examples[sentence1_key],)
              if sentence2_key is None
              else (examples[sentence1_key], examples[sentence2_key])
          )
          result = tokenizer(
              *args,
              padding=padding,
              max_length=max_seq_length,
              truncation=True,
          )
  
          # Map labels to IDs (not necessary for GLUE tasks)
          if label_to_id is not None and "label" in examples:
              result["label"] = [
                  (label_to_id[l] if l != -1 else -1)
                  for l in examples["label"]
              ]
          return result
  
      datasets = datasets.map(
          preprocess_function,
          batched=True,
          load_from_cache_file=not data_args.overwrite_cache,
      )
      if training_args.do_train:
          if "train" not in datasets:
              raise ValueError("--do_train requires a train dataset")
          train_dataset = datasets["train"]
          if data_args.max_train_samples is not None:
              train_dataset = train_dataset.select(
                  range(data_args.max_train_samples)
              )
  
      if training_args.do_eval:
          if (
              "validation" not in datasets
              and "validation_matched" not in datasets
          ):
              raise ValueError(
                  "--do_eval requires a validation dataset"
              )
          eval_dataset = datasets[
              "validation_matched"
              if data_args.task_name == "mnli"
              else "validation"
          ]
          if data_args.max_val_samples is not None:
              eval_dataset = eval_dataset.select(
                  range(data_args.max_val_samples)
              )
  
      if (
          training_args.do_predict
          or data_args.task_name is not None
          or data_args.test_file is not None
      ):
          if "test" not in datasets and "test_matched" not in datasets:
              raise ValueError("--do_predict requires a test dataset")
          test_dataset = datasets[
              "test_matched"
              if data_args.task_name == "mnli"
              else "test"
          ]
          if data_args.max_test_samples is not None:
              test_dataset = test_dataset.select(
                  range(data_args.max_test_samples)
              )
  
      # Log a few random samples from the training set:
      if training_args.do_train:
          for index in random.sample(range(len(train_dataset)), 3):
              logger.info(
                  f"Sample {index} of the training set: {train_dataset[index]}."
              )
  
      # Get the metric function
      if data_args.task_name is not None:
          metric = load_metric("glue", data_args.task_name)
      # TODO: When datasets metrics include regular accuracy, make an else here and remove special branch from
      # compute_metrics
  
      # You can define your custom compute_metrics function. It takes an `EvalPrediction` object (a namedtuple with a
      # predictions and label_ids field) and has to return a dictionary string to float.
      def compute_metrics(p: EvalPrediction):
          preds = (
              p.predictions[0]
              if isinstance(p.predictions, tuple)
              else p.predictions
          )
          preds = (
              np.squeeze(preds)
              if is_regression
              else np.argmax(preds, axis=1)
          )
          if data_args.task_name is not None:
              result = metric.compute(
                  predictions=preds, references=p.label_ids
              )
              if len(result) > 1:
                  result["combined_score"] = np.mean(
                      list(result.values())
                  ).item()
              return result
          elif is_regression:
              return {"mse": ((preds - p.label_ids) ** 2).mean().item()}
          else:
              return {
                  "accuracy": (preds == p.label_ids)
                  .astype(np.float32)
                  .mean()
                  .item()
              }
  
      # Data collator will default to DataCollatorWithPadding, so we change it if we already did the padding.
      if data_args.pad_to_max_length:
          data_collator = default_data_collator
      elif training_args.fp16:
          data_collator = DataCollatorWithPadding(
              tokenizer, pad_to_multiple_of=8
          )
      else:
          data_collator = None
  
      # Initialize our Trainer
      trainer = Trainer(
          model=model,
          args=training_args,
          train_dataset=train_dataset
          if training_args.do_train
          else None,
          eval_dataset=eval_dataset if training_args.do_eval else None,
          compute_metrics=compute_metrics,
          tokenizer=tokenizer,
          data_collator=data_collator,
      )
  
      # Training
      if training_args.do_train:
          checkpoint = None
          if last_checkpoint is not None:
              checkpoint = last_checkpoint
          elif os.path.isdir(model_args.model_name_or_path):
              # Check the config from that potential checkpoint has the right number of labels before using it as a
              # checkpoint.
              if (
                  AutoConfig.from_pretrained(
                      model_args.model_name_or_path
                  ).num_labels
                  == num_labels
              ):
                  checkpoint = model_args.model_name_or_path
  
          train_result = trainer.train(
              resume_from_checkpoint=checkpoint
          )
          metrics = train_result.metrics
          max_train_samples = (
              data_args.max_train_samples
              if data_args.max_train_samples is not None
              else len(train_dataset)
          )
          metrics["train_samples"] = min(
              max_train_samples, len(train_dataset)
          )
  
          trainer.save_model()  # Saves the tokenizer too for easy upload
  
          trainer.log_metrics("train", metrics)
          trainer.save_metrics("train", metrics)
          trainer.save_state()
  
      # Evaluation
      if training_args.do_eval:
          logger.info("*** Evaluate ***")
  
          # Loop to handle MNLI double evaluation (matched, mis-matched)
          tasks = [data_args.task_name]
          eval_datasets = [eval_dataset]
          if data_args.task_name == "mnli":
              tasks.append("mnli-mm")
              eval_datasets.append(datasets["validation_mismatched"])
  
          for eval_dataset, task in zip(eval_datasets, tasks):
              metrics = trainer.evaluate(eval_dataset=eval_dataset)
  
              max_val_samples = (
                  data_args.max_val_samples
                  if data_args.max_val_samples is not None
                  else len(eval_dataset)
              )
              metrics["eval_samples"] = min(
                  max_val_samples, len(eval_dataset)
              )
  
              trainer.log_metrics("eval", metrics)
              trainer.save_metrics("eval", metrics)
  
      if training_args.do_predict:
          logger.info("*** Test ***")
  
          # Loop to handle MNLI double evaluation (matched, mis-matched)
          tasks = [data_args.task_name]
          test_datasets = [test_dataset]
          if data_args.task_name == "mnli":
              tasks.append("mnli-mm")
              test_datasets.append(datasets["test_mismatched"])
  
          for test_dataset, task in zip(test_datasets, tasks):
              # Removing the `label` columns because it contains -1 and Trainer won't like that.
              test_dataset.remove_columns_("label")
              predictions = trainer.predict(
                  test_dataset=test_dataset
              ).predictions
              predictions = (
                  np.squeeze(predictions)
                  if is_regression
                  else np.argmax(predictions, axis=1)
              )
  
              output_test_file = os.path.join(
                  training_args.output_dir, f"test_results_{task}.txt"
              )
              if trainer.is_world_process_zero():
                  with open(output_test_file, "w") as writer:
                      logger.info(f"***** Test results {task} *****")
                      writer.write("index\tprediction\n")
                      for index, item in enumerate(predictions):
                          if is_regression:
                              writer.write(f"{index}\t{item:3.3f}\n")
                          else:
                              item = label_list[item]
                              writer.write(f"{index}\t{item}\n")
  
file_path: sagemaker-python-sdk/tests/data/huggingface/run_glue.py
imports: import logging
  import os
  import random
  import sys
  from dataclasses import dataclass, field
  from typing import Optional
  import numpy
  from datasets import load_dataset, load_metric
  import transformers
  from transformers import AutoConfig, AutoModelForSequenceClassification, AutoTokenizer, DataCollatorWithPadding, EvalPrediction, HfArgumentParser, PretrainedConfig, Trainer, TrainingArguments, default_data_collator, set_seed
  from transformers.trainer_utils import get_last_checkpoint, is_main_process
  from transformers.utils import check_min_version
label_desc: Use image node_id 1 for calling a global function with example usage: main() and returns: result, result, dict, dict