code: def _mosaic_train_loop_per_worker(config):
      """Per-worker training loop for Mosaic Composers."""
      trainer_init_per_worker = config.pop("_trainer_init_per_worker")
  
      # Replace Composer's Loggers with RayLogger
      ray_logger = RayLogger(keys=config.pop("log_keys", []))
  
      # initialize Composer trainer
      trainer: Trainer = trainer_init_per_worker(config)
  
      # Remove Composer's Loggers if there are any added in the trainer_init_per_worker
      # this removes the logging part of the loggers
      filtered_callbacks = list()
      for callback in trainer.state.callbacks:
          if not isinstance(callback, LoggerDestination):
              filtered_callbacks.append(callback)
      filtered_callbacks.append(ray_logger)
      trainer.state.callbacks = filtered_callbacks
  
      # this prevents data to be routed to all the Composer Loggers
      trainer.logger.destinations = (ray_logger,)
  
      # call the trainer
      trainer.fit()
  
file_path: ray/python/ray/train/mosaic/mosaic_trainer.py
imports: import inspect
  import warnings
  from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Type
  from composer.loggers.logger_destination import LoggerDestination
  from composer.trainer import Trainer
  from ray.train import Checkpoint, DataConfig, RunConfig, ScalingConfig
  from ray.train.mosaic._mosaic_utils import RayLogger
  from ray.train.torch import TorchConfig, TorchTrainer
  from ray.train.trainer import GenDataset
  from ray.util import PublicAPI
label_desc: Use image node_id 1 for calling a global function with example usage: _mosaic_train_loop_per_worker(config) without return types