code: def test_model_checkpoint_callback_2(
      model_and_optimizers: TrainerParameters,
      driver,
      device,
      only_state_dict,
  ):
      try:
          path = Path.cwd().joinpath("test_model_checkpoint")
          path.mkdir(exist_ok=True, parents=True)
  
          from fastNLP.core.callbacks.callback_event import Event
  
          @Trainer.on(Event.on_train_epoch_end())
          def raise_exception(trainer):
              if (
                  trainer.driver.get_local_rank() == 0
                  and trainer.cur_epoch_idx == 4
              ):
                  raise NotImplementedError
  
          callbacks = [
              CheckpointCallback(
                  folder=path,
                  every_n_epochs=None,
                  every_n_batches=None,
                  last=False,
                  on_exceptions=NotImplementedError,
                  topk=None,
                  monitor=None,
                  only_state_dict=only_state_dict,
                  save_object="model",
              ),
          ]
  
          with pytest.raises(NotImplementedError):
              trainer = Trainer(
                  model=model_and_optimizers.model,
                  driver=driver,
                  device=device,
                  optimizers=model_and_optimizers.optimizers,
                  train_dataloader=model_and_optimizers.train_dataloader,
                  evaluate_dataloaders=model_and_optimizers.evaluate_dataloaders,
                  input_mapping=model_and_optimizers.input_mapping,
                  output_mapping=model_and_optimizers.output_mapping,
                  metrics=model_and_optimizers.metrics,
                  n_epochs=10,
                  callbacks=callbacks,
                  output_from_new_proc="all",
              )
  
              trainer.run()
  
          if dist.is_initialized():
              dist.destroy_process_group()
              if FASTNLP_DISTRIBUTED_CHECK in os.environ:
                  os.environ.pop(FASTNLP_DISTRIBUTED_CHECK)
  
          # 检查生成保存模型文件的数量是不是正确的；
          all_saved_model_paths = {
              w.name: w
              for w in path.joinpath(
                  os.environ[FASTNLP_LAUNCH_TIME]
              ).iterdir()
          }
  
          if not isinstance(device, list):
              assert (
                  "model-epoch_4-batch_100-exception_NotImplementedError"
                  in all_saved_model_paths
              )
              exception_model_path = all_saved_model_paths[
                  "model-epoch_4-batch_100-exception_NotImplementedError"
              ]
          # ddp 下的文件名不同，因为同样的数据，ddp 用了更少的步数跑完；
          else:
              assert (
                  "model-epoch_4-batch_52-exception_NotImplementedError"
                  in all_saved_model_paths
              )
              exception_model_path = all_saved_model_paths[
                  "model-epoch_4-batch_52-exception_NotImplementedError"
              ]
  
          assert len(all_saved_model_paths) == 1
          all_state_dicts = [exception_model_path]
  
          for folder in all_state_dicts:
              trainer = Trainer(
                  model=model_and_optimizers.model,
                  driver="torch",
                  device=0,
                  optimizers=model_and_optimizers.optimizers,
                  train_dataloader=model_and_optimizers.train_dataloader,
                  evaluate_dataloaders=model_and_optimizers.evaluate_dataloaders,
                  input_mapping=model_and_optimizers.input_mapping,
                  output_mapping=model_and_optimizers.output_mapping,
                  metrics=model_and_optimizers.metrics,
                  n_epochs=2,
                  output_from_new_proc="all",
              )
  
              trainer.load_model(
                  folder, only_state_dict=only_state_dict
              )
              trainer.run()
              evaluator = Evaluator(
                  model=model_and_optimizers.model,
                  driver="torch",
                  device=0,
                  dataloaders=model_and_optimizers.evaluate_dataloaders,
                  input_mapping=model_and_optimizers.input_mapping,
                  output_mapping=model_and_optimizers.output_mapping,
                  metrics=model_and_optimizers.metrics,
              )
              evaluator.load_model(
                  folder, only_state_dict=only_state_dict
              )
              evaluator.run()
              trainer.driver.barrier()
  
      finally:
          rank_zero_rm(path)
          # pass
  
      if dist.is_initialized():
          dist.destroy_process_group()
  
imports: import os
  import pytest
  from typing import Any
  from dataclasses import dataclass
  from pathlib import Path
  import re
  import time
  from fastNLP.core.callbacks.checkpoint_callback import CheckpointCallback
  from fastNLP.core.controllers.trainer import Trainer
  from fastNLP import Evaluator
  from fastNLP.envs import FASTNLP_LAUNCH_TIME, FASTNLP_DISTRIBUTED_CHECK
  from tests.helpers.utils import magic_argv_env_context
  from fastNLP.envs.distributed import rank_zero_rm
  from tests.helpers.models.torch_model import TorchNormalModel_Classification_1
  from tests.helpers.datasets.torch_data import TorchArgMaxDataset
  from tests.helpers.utils import Capturing
  from fastNLP.core.log import logger
  from fastNLP.envs.imports import _NEED_IMPORT_TORCH
label_desc: Use image node_id 3 for calling a global function with example usage: test_model_checkpoint_callback_2(model_and_optimizers, driver, device, only_state_dict) without return types