code: def train_val_test(
      model: Union[DDP, DMP],
      train_loader: data_utils.DataLoader,
      val_loader: data_utils.DataLoader,
      test_loader: data_utils.DataLoader,
      device: torch.device,
      optimizer: optim.Adam,
      lr_scheduler: optim.lr_scheduler.StepLR,
      num_epochs: int,
      metric_ks: List[int],
      export_root: str,
  ) -> None:
      """
      Train/validation/test loop. Ensure the dataloader will do the shuffling on each rank
      and will output the performance metrics like recalls and ndcgs
  
      Args:
          model (Union[DDP, DMP]): DMP or DDP model contains the Bert4Rec.
          train_loader (data_utils.DataLoader): DataLoader used for training.
          val_loader (data_utils.DataLoader): DataLoader used for validation.
          test_loader (data_utils.DataLoader): DataLoader used for testing.
          device (torch.device): the device to train/val/test
          optimizer (optim.Adam): Adam optimizer to train the model
          lr_scheduler (optim.lr_scheduler.StepLR): scheduler to control the learning rate
          num_epochs (int): the number of epochs to train
          metric_ks (List[int]): the metrics we want to validate
          export_root (str): the export root of the saved models
  
      Returns:
          None.
      """
      _validate(model, val_loader, device, -1, metric_ks)
      for epoch in range(num_epochs):
          # pyre-fixme[16] Undefined attribute [16]: has no attribute `set_epoch`
          train_loader.sampler.set_epoch(epoch)
          _train_one_epoch(
              model,
              train_loader,
              device,
              optimizer,
              lr_scheduler,
              epoch,
          )
          _validate(model, val_loader, device, epoch, metric_ks)
          if (epoch + 1) % 10 == 0:
              torch.save(
                  model.state_dict(),
                  export_root + f"epoch_{epoch + 1}_model.pth",
              )
              print(
                  f"epoch {epoch + 1} model has been saved to {export_root}"
              )
      _validate(model, test_loader, device, num_epochs, metric_ks, True)
  
imports: import argparse
  import os
  import sys
  from typing import Any, cast, Dict, List, Union
  import numpy
  import torch
  import torch.nn
  import torch.optim
  import torch.utils.data
  from fbgemm_gpu.split_embedding_configs import EmbOptimType
  from torch import distributed
  from torch.nn.parallel import DistributedDataParallel
  from torchrec.distributed.embedding import EmbeddingCollectionSharder
  from torchrec.distributed.model_parallel import DistributedModelParallel
  from torchrec.distributed.types import ModuleSharder
  from torchrec.optim.keyed import CombinedOptimizer, KeyedOptimizerWrapper
  from torchrec.optim.optimizers import in_backward_optimizer_filter
  from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
  from tqdm import tqdm
label_desc: Use image node_id 7 for calling a global function with example usage: train_val_test(model, train_loader, val_loader, test_loader, device, optimizer, lr_scheduler, num_epochs, metric_ks, export_root) without return types