code: def train_workflow(
      model_manager: ModelManager,
      train_dataset: Optional[Dataset],
      eval_dataset: Optional[Dataset],
      *,
      num_epochs: int,
      use_gpu: bool,
      named_model_ids: ModuleNameToEntityId,
      child_workflow_id: int,
      setup_data: Optional[Dict[str, bytes]] = None,
      normalization_data_map: Optional[
          Dict[str, NormalizationData]
      ] = None,
      reward_options: Optional[RewardOptions] = None,
      reader_options: Optional[ReaderOptions] = None,
      resource_options: Optional[ResourceOptions] = None,
      warmstart_path: Optional[str] = None,
  ) -> RLTrainingOutput:
      writer = SummaryWriter()
      logger.info(
          "TensorBoard logging location is: {}".format(writer.log_dir)
      )
  
      if setup_data is not None:
          data_module = model_manager.get_data_module(
              setup_data=setup_data,
              reward_options=reward_options,
              reader_options=reader_options,
              resource_options=resource_options,
          )
          assert data_module is not None
          data_module.setup()
      else:
          data_module = None
  
      if normalization_data_map is None:
          assert data_module is not None
          normalization_data_map = (
              data_module.get_normalization_data_map()
          )
  
      warmstart_input_path = warmstart_path or None
      trainer_module = model_manager.build_trainer(
          use_gpu=use_gpu,
          reward_options=reward_options,
          normalization_data_map=normalization_data_map,
      )
  
      if not reader_options:
          reader_options = ReaderOptions()
  
      if not resource_options:
          resource_options = ResourceOptions()
  
      with summary_writer_context(writer):
          train_output, lightning_trainer = model_manager.train(
              trainer_module,
              train_dataset,
              eval_dataset,
              None,
              data_module,
              num_epochs,
              reader_options,
              resource_options,
              checkpoint_path=warmstart_input_path,
          )
  
      output_paths = {}
      for (
          module_name,
          serving_module,
      ) in model_manager.build_serving_modules(
          trainer_module, normalization_data_map
      ).items():
          torchscript_output_path = f"{model_manager.__class__.__name__}_{module_name}_{round(time.time())}.torchscript"
          torch.jit.save(serving_module, torchscript_output_path)
          logger.info(
              f"Saved {module_name} to {torchscript_output_path}"
          )
          output_paths[module_name] = torchscript_output_path
      return dataclasses.replace(
          train_output, output_paths=output_paths
      )
  
imports: import dataclasses
  import logging
  import time
  from typing import Dict, Optional
  import torch
  from reagent.core.parameters import NormalizationData
  from reagent.core.tensorboardX import summary_writer_context
  from reagent.data.manual_data_module import get_sample_range
  from reagent.data.oss_data_fetcher import OssDataFetcher
  from reagent.model_managers.model_manager import ModelManager
  from reagent.model_managers.union import ModelManager__Union
  from reagent.publishers.union import ModelPublisher__Union
  from reagent.validators.union import ModelValidator__Union
  from reagent.workflow.env import get_new_named_entity_ids, get_workflow_id
  from reagent.workflow.types import Dataset, ModuleNameToEntityId, ReaderOptions, RecurringPeriod, ResourceOptions, RewardOptions, RLTrainingOutput, TableSpec
  from torch.utils.tensorboard import SummaryWriter
label_desc: Use image node_id 3 for calling a global function with example usage: train_workflow(model_manager, train_dataset, eval_dataset) and returns: dataclasses