第十三节 huggingface的trainner解读与Demo

    class SentimentDataset(torch.utils.data.Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels
    
    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item
    
    def __len__(self):
        return len(self.labels)
     # 创建数据集
    train_dataset = SentimentDataset(train_encodings, train_labels)
    val_dataset = SentimentDataset(val_encodings, val_labels)

    # 定义训练参数
    training_args = TrainingArguments(
    output_dir='./out_dirs',
    per_device_train_batch_size=2,
    per_device_eval_batch_size=2,
    num_train_epochs=3,
    logging_dir="./logs",
    evaluation_strategy="epoch",
    report_to="none"
    )

    # 定义Trainer对象
    trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    compute_metrics=lambda pred: {"accuracy": accuracy_score(pred.label_ids, pred.predictions.argmax(axis=1))}
    )

    # 开始训练
    trainer.train()
    
    # 评估模型
    eval_results = trainer.evaluate()
    # print(f"Accuracy: {eval_results['eval_accuracy']:.4f}")
    
    # 保存模型
    model.save_pretrained("./sentiment_model")

    # 加载模型
    model = BertForSequenceClassification.from_pretrained("./sentiment_model")
    
    # 预测一些示例文本
    example_texts = ["I love this!", "I hate it."]
    inputs = tokenizer(example_texts, padding=True, truncation=True, return_tensors="pt")
    outputs = model(**inputs)
    predicted_labels = torch.argmax(outputs.logits, dim=1).tolist()
    
    # 打印预测结果
    for text, label in zip(example_texts, predicted_labels):
    print(f"Text: {text} -- Predicted Label: {'positive' if label == 1 else 'negative'}")

    import os
    os.environ["CUDA_VISIBLE_DEVICES"] = "1"
    
    from torch.utils.data import Dataset, DataLoader
    
    import torch
    from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import accuracy_score
    import random
    
    # 随机种子
    seed = 42
    random.seed(seed)
    torch.manual_seed(seed)
    
    # 示例数据
    texts = ["I love Hugging Face!", "I hate this.", "This is fantastic!", "I dislike it."]
    labels = [1, 0, 1, 0]  # 1代表正面，0代表负面
    
    # 划分训练集和验证集
    train_texts, val_texts, train_labels, val_labels = train_test_split(texts, labels, test_size=0.2, random_state=seed)
    
    # 加载预训练的Bert tokenizer
    tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
    
    # 数据编码
    train_encodings = tokenizer(train_texts, truncation=True, padding=True)
    val_encodings = tokenizer(val_texts, truncation=True, padding=True)
    
    class SentimentDataset(torch.utils.data.Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels
    
    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item
    
    def __len__(self):
        return len(self.labels)
    
    # 创建数据集
    train_dataset = SentimentDataset(train_encodings, train_labels)
    val_dataset = SentimentDataset(val_encodings, val_labels)
    
    # 加载预训练的Bert模型
    model = BertForSequenceClassification.from_pretrained("bert-base-uncased", num_labels=2)
    
    # 定义训练参数
    training_args = TrainingArguments(
    output_dir='./out_dirs',
    per_device_train_batch_size=2,
    per_device_eval_batch_size=2,
    num_train_epochs=3,
    logging_dir="./logs",
    evaluation_strategy="epoch",
    report_to="none"
    )
    
    # 定义Trainer对象
    trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    compute_metrics=lambda pred: {"accuracy": accuracy_score(pred.label_ids, pred.predictions.argmax(axis=1))}
    )
    
    # 开始训练
    trainer.train()
    
    # 评估模型
    eval_results = trainer.evaluate()
    print(f"Accuracy: {eval_results['eval_accuracy']:.4f}")
    
    # 保存模型
    model.save_pretrained("./sentiment_model")
    
    # 加载模型
    model = BertForSequenceClassification.from_pretrained("./sentiment_model")
    
    # 预测一些示例文本
    example_texts = ["I love this!", "I hate it."]
    inputs = tokenizer(example_texts, padding=True, truncation=True, return_tensors="pt")
    outputs = model(**inputs)
    predicted_labels = torch.argmax(outputs.logits, dim=1).tolist()
    
    # 打印预测结果
    for text, label in zip(example_texts, predicted_labels):
    print(f"Text: {text} -- Predicted Label: {'positive' if label == 1 else 'negative'}")

    yield x：返回变量 x 的值。
    yield 10：返回整数 10。
    yield func()：返回函数 func() 的执行结果。
    yield from iterable：从可迭代对象中逐个返回元素。

    import itertools
    # 定义一个列表
    numbers_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]
    # 创建一个迭代器
    numbers_iterator = iter(numbers_list)
    # 记住第6次迭代的位置
    for a in itertools.islice(numbers_iterator, 6):
    print(a)
    # 需要重新开始迭代时
    print("Restarting iteration:")
    for num in itertools.chain([next(numbers_iterator)], numbers_iterator):
    print(num)

    import torch
    # 创建一个生成器对象
    generator = torch.Generator()
    # 设置生成器的种子
    generator.manual_seed(42)
    # 获取生成器的状态信息
    generator_state = generator.get_state()
    
    print(generator_state)

    import torch
    import pickle
    
    # 定义一个生成器函数
    def random_permutation_generator(n, generator_state, index):
    generator = torch.Generator()
    generator.set_state(generator_state)
    random_permutation = torch.randperm(n, generator=generator).tolist()
    
    for i in range(index, len(random_permutation)):
        yield random_permutation[i], i
    
    
    # 设置随机数生成器种子以确保结果可复现
    torch.manual_seed(42)
    
    n = 10  # 生成随机排列的范围为 0 到 n-1
    
    # 定义一个生成器
    generator = torch.Generator()
    generator.manual_seed(42)
    
    # 创建生成器对象
    gen = random_permutation_generator(n, generator.get_state(), 0)
    
    # 模拟迭代中断并保存生成器状态和已生成的索引到文件
    for i, (element, index) in enumerate(gen):
    print(element)
    if i == 4:
        with open('generator_state.pkl', 'wb') as f:
            state_index = (generator.get_state(), index)
            pickle.dump(state_index, f)
        break
    
    # 重新加载代码后恢复生成器状态和已生成的索引
    with open('generator_state.pkl', 'rb') as f:
    generator_state, index = pickle.load(f)
    
    gen = random_permutation_generator(n, generator_state, index)
    
    print("重新加载代码，从中断处继续生成随机排列的数字...")
    
    # 从中断处继续生成随机排列的数字
    for element, _ in gen:
    print(element)

    def train(
        self,
        resume_from_checkpoint: Optional[Union[str, bool]] = None,
        trial: Union["optuna.Trial", Dict[str, Any]] = None,
        ignore_keys_for_eval: Optional[List[str]] = None,
        **kwargs,
    ):
        """
        Main training entry point.
    
        Args:
            resume_from_checkpoint (`str` or `bool`, *optional*):
                If a `str`, local path to a saved checkpoint as saved by a previous instance of [`Trainer`]. If a
                `bool` and equals `True`, load the last checkpoint in *args.output_dir* as saved by a previous instance
                of [`Trainer`]. If present, training will resume from the model/optimizer/scheduler states loaded here.
            trial (`optuna.Trial` or `Dict[str, Any]`, *optional*):
                The trial run or the hyperparameter dictionary for hyperparameter search.
            ignore_keys_for_eval (`List[str]`, *optional*)
                A list of keys in the output of your model (if it is a dictionary) that should be ignored when
                gathering predictions for evaluation during the training.
            kwargs (`Dict[str, Any]`, *optional*):
                Additional keyword arguments used to hide deprecated arguments
        """
        if resume_from_checkpoint is False:
            resume_from_checkpoint = None
    
        # memory metrics - must set up as early as possible
        self._memory_tracker.start()
    
        args = self.args
    
        self.is_in_train = True
    
        # do_train is not a reliable argument, as it might not be set and .train() still called, so
        # the following is a workaround:
        if (args.fp16_full_eval or args.bf16_full_eval) and not args.do_train:
            self._move_model_to_device(self.model, args.device)
    
        if "model_path" in kwargs:
            resume_from_checkpoint = kwargs.pop("model_path")
            warnings.warn(
                "`model_path` is deprecated and will be removed in a future version. Use `resume_from_checkpoint` "
                "instead.",
                FutureWarning,
            )
        if len(kwargs) > 0:
            raise TypeError(f"train() received got unexpected keyword arguments: {', '.join(list(kwargs.keys()))}.")
        # This might change the seed so needs to run first.
        self._hp_search_setup(trial)
        self._train_batch_size = self.args.train_batch_size
    
        # Model re-init
        model_reloaded = False
        if self.model_init is not None:
            # Seed must be set before instantiating the model when using model_init.
            enable_full_determinism(self.args.seed) if self.args.full_determinism else set_seed(self.args.seed)
            self.model = self.call_model_init(trial)
            model_reloaded = True
            # Reinitializes optimizer and scheduler
            self.optimizer, self.lr_scheduler = None, None
    
        # Load potential model checkpoint,这个地方是resume
        if isinstance(resume_from_checkpoint, bool) and resume_from_checkpoint:
            resume_from_checkpoint = get_last_checkpoint(args.output_dir)
            if resume_from_checkpoint is None:
                raise ValueError(f"No valid checkpoint found in output directory ({args.output_dir})")
    
        if resume_from_checkpoint is not None and not is_sagemaker_mp_enabled() and not self.is_deepspeed_enabled:
            self._load_from_checkpoint(resume_from_checkpoint)
    
        # If model was re-initialized, put it on the right device and update self.model_wrapped
        if model_reloaded:
            if self.place_model_on_device:
                self._move_model_to_device(self.model, args.device)
            self.model_wrapped = self.model
    
        inner_training_loop = find_executable_batch_size(
            self._inner_training_loop, self._train_batch_size, args.auto_find_batch_size
        )  # 这个是模型类
        return inner_training_loop(
            args=args,
            resume_from_checkpoint=resume_from_checkpoint,
            trial=trial,
            ignore_keys_for_eval=ignore_keys_for_eval,
        )

     if resume_from_checkpoint is False:
     resume_from_checkpoint = None

    args = self.args
    self.is_in_train = True

    # do_train is not a reliable argument, as it might not be set and .train() still called, so the following is a workaround:
    if (args.fp16_full_eval or args.bf16_full_eval) and not args.do_train:
    self._move_model_to_device(self.model, args.device)

     def _move_model_to_device(self, model, device):
     model = model.to(device)
     # Moving a model to an XLA device disconnects the tied weights, so we have to retie them.
     if self.args.parallel_mode == ParallelMode.TPU and hasattr(model, "tie_weights"):
         model.tie_weights()

     if "model_path" in kwargs:
     resume_from_checkpoint = kwargs.pop("model_path")
     warnings.warn(
         "`model_path` is deprecated and will be removed in a future version. Use `resume_from_checkpoint` "
         "instead.",
         FutureWarning,
     )

    self._hp_search_setup(trial)

    self._train_batch_size = self.args.train_batch_size

     # Model re-init
     model_reloaded = False
     if self.model_init is not None:
     # Seed must be set before instantiating the model when using model_init.
     enable_full_determinism(self.args.seed) if self.args.full_determinism else set_seed(self.args.seed)
     self.model = self.call_model_init(trial)
     model_reloaded = True
     # Reinitializes optimizer and scheduler
     self.optimizer, self.lr_scheduler = None, None

    # Load potential model checkpoint,这个地方是resume
    if isinstance(resume_from_checkpoint, bool) and resume_from_checkpoint:
    resume_from_checkpoint = get_last_checkpoint(args.output_dir)
    if resume_from_checkpoint is None:
        raise ValueError(f"No valid checkpoint found in output directory ({args.output_dir})")
    
    if resume_from_checkpoint is not None and not is_sagemaker_mp_enabled() and not self.is_deepspeed_enabled:
    self._load_from_checkpoint(resume_from_checkpoint)

        # If model was re-initialized, put it on the right device and update self.model_wrapped
        if model_reloaded:
            if self.place_model_on_device:
                self._move_model_to_device(self.model, args.device)
            self.model_wrapped = self.model

     inner_training_loop = find_executable_batch_size(
     self._inner_training_loop, self._train_batch_size, args.auto_find_batch_size
     )  # 这个是模型类

        return inner_training_loop(
            args=args,
            resume_from_checkpoint=resume_from_checkpoint,
            trial=trial,
            ignore_keys_for_eval=ignore_keys_for_eval,
        )

    在Hugging Face的Trainer类中，self.accelerator.free_memory()函数通常用于释放加速器（Accelerator）上的内存。加速器通常指的是GPU或TPU等硬件加速设备，用于加速深度学习模型的训练过程。通过调用这个函数，Trainer可以释放加速器上的内存，以便在训练过程中及时释放不再需要的内存资源，从而优化内存使用并提高训练效率。

        # Data loader and number of training steps
        train_dataloader = self.get_train_dataloader()

        dataloader_params = {
            "batch_size": self._train_batch_size,
            "collate_fn": data_collator,
            "num_workers": self.args.dataloader_num_workers,
            "pin_memory": self.args.dataloader_pin_memory,
        }  # 这个就是dataloader相关参数

    dataloader_params["sampler"] = self._get_train_sampler()
    dataloader_params["drop_last"] = self.args.dataloader_drop_last
    dataloader_params["worker_init_fn"] = seed_worker

    return self.accelerator.prepare(DataLoader(train_dataset, **dataloader_params))  # 这里是

        # Setting up training control variables:
        # number of training epochs: num_train_epochs
        # number of training steps per epoch: num_update_steps_per_epoch
        # total number of training steps to execute: max_steps
        total_train_batch_size = self._train_batch_size * args.gradient_accumulation_steps * args.world_size
    
        len_dataloader = None
        if has_length(train_dataloader): 
            len_dataloader = len(train_dataloader)
            num_update_steps_per_epoch = len_dataloader // args.gradient_accumulation_steps
            num_update_steps_per_epoch = max(num_update_steps_per_epoch, 1)
            num_examples = self.num_examples(train_dataloader)  # 获得样本数量
            if args.max_steps > 0: # -1
                max_steps = args.max_steps
                num_train_epochs = args.max_steps // num_update_steps_per_epoch + int(
                    args.max_steps % num_update_steps_per_epoch > 0
                )
                # May be slightly incorrect if the last batch in the training dataloader has a smaller size but it's
                # the best we can do.
                num_train_samples = args.max_steps * total_train_batch_size
            else: 
                max_steps = math.ceil(args.num_train_epochs * num_update_steps_per_epoch)
                num_train_epochs = math.ceil(args.num_train_epochs)
                num_train_samples = self.num_examples(train_dataloader) * args.num_train_epochs
        elif args.max_steps > 0:  # Rely on max_steps when dataloader does not have a working size
            max_steps = args.max_steps
            # Setting a very large number of epochs so we go as many times as necessary over the iterator.
            # 设置非常大的训练周期数，以便我们可以尽可能多次地遍历迭代器
            num_train_epochs = sys.maxsize
            num_update_steps_per_epoch = max_steps
            num_examples = total_train_batch_size * args.max_steps
            num_train_samples = args.max_steps * total_train_batch_size
        else:
            raise ValueError(
                "args.max_steps must be set to a positive value if dataloader does not have a length, was"
                f" {args.max_steps}"
            )

    def has_length(dataset):
    """
    Checks if the dataset implements __len__() and it doesn't raise an error
    """
    try:
        return len(dataset) is not None
    except TypeError:
        # TypeError: len() of unsized object
        return False

    def num_examples(self, dataloader: DataLoader) -> int:
        """
        Helper to get number of samples in a [`~torch.utils.data.DataLoader`] by accessing its dataset. When
        dataloader.dataset does not exist or has no length, estimates as best it can
        """
        try:
            dataset = dataloader.dataset
            # Special case for IterableDatasetShard, we need to dig deeper
            if isinstance(dataset, IterableDatasetShard):
                return len(dataloader.dataset.dataset)
            return len(dataloader.dataset)
        except (NameError, AttributeError, TypeError):  # no dataset or length, estimate by length of dataloader
            return len(dataloader) * self.args.per_device_train_batch_size

    max_steps = math.ceil(args.num_train_epochs * num_update_steps_per_epoch)
    num_train_epochs = math.ceil(args.num_train_epochs)
    num_train_samples = self.num_examples(train_dataloader) * args.num_train_epochs

        # Compute absolute values for logging, eval, and save if given as ratio
        if args.logging_steps and args.logging_steps < 1:
            args.logging_steps = math.ceil(max_steps * args.logging_steps)
        if args.eval_steps and args.eval_steps < 1:
            args.eval_steps = math.ceil(max_steps * args.eval_steps)
        if args.save_steps and args.save_steps < 1:
            args.save_steps = math.ceil(max_steps * args.save_steps)
    
        if DebugOption.UNDERFLOW_OVERFLOW in self.args.debug:
            if self.args.n_gpu > 1:
                # nn.DataParallel(model) replicates the model, creating new variables and module
                # references registered here no longer work on other gpus, breaking the module
                raise ValueError(
                    "Currently --debug underflow_overflow is not supported under DP. Please use DDP"
                    " (torch.distributed.launch)."
                )
            else:
                debug_overflow = DebugUnderflowOverflow(self.model)  # noqa

    delay_optimizer_creation = (
            self.sharded_ddp is not None
            and self.sharded_ddp != ShardedDDPOption.SIMPLE
            or is_sagemaker_mp_enabled()
            or self.fsdp is not None
        )  # False
    
        # We need to reset the scheduler, as its parameters may be different on subsequent calls
        if self._created_lr_scheduler:  # False
            self.lr_scheduler = None
            self._created_lr_scheduler = False
    
        if self.is_deepspeed_enabled: # 这个是deepspeed使用用到的，暂时为False
            self.optimizer, self.lr_scheduler = deepspeed_init(self, num_training_steps=max_steps)
    
        if not delay_optimizer_creation: 
            self.create_optimizer_and_scheduler(num_training_steps=max_steps)

        delay_optimizer_creation = (
            self.sharded_ddp is not None
            and self.sharded_ddp != ShardedDDPOption.SIMPLE
            or is_sagemaker_mp_enabled()
            or self.fsdp is not None
        )  # False

    # We need to reset the scheduler, as its parameters may be different on subsequent calls
    if self._created_lr_scheduler:  # False
    self.lr_scheduler = None
    self._created_lr_scheduler = False
    
    if self.is_deepspeed_enabled: # 这个是deepspeed使用用到的，暂时为False
    self.optimizer, self.lr_scheduler = deepspeed_init(self, num_training_steps=max_steps)
    # 本实列采用方式
    if not delay_optimizer_creation: 
    self.create_optimizer_and_scheduler(num_training_steps=max_steps)

    def create_optimizer_and_scheduler(self, num_training_steps: int):
        """
        Setup the optimizer and the learning rate scheduler.
    
        We provide a reasonable default that works well. If you want to use something else, you can pass a tuple in the
        Trainer's init through `optimizers`, or subclass and override this method (or `create_optimizer` and/or
        `create_scheduler`) in a subclass.
     		
        """
        self.create_optimizer()
        if IS_SAGEMAKER_MP_POST_1_10 and smp.state.cfg.fp16:
            # If smp >= 1.10 and fp16 is enabled, we unwrap the optimizer
            optimizer = self.optimizer.optimizer
        else:
            optimizer = self.optimizer
        self.create_scheduler(num_training_steps=num_training_steps, optimizer=optimizer)

        optimizer_kwargs = {"lr": args.learning_rate}  # 这个地方获得学习率，很重要
    
        adam_kwargs = {
            "betas": (args.adam_beta1, args.adam_beta2),
            "eps": args.adam_epsilon,
        } # 获得betas与eps

    self.create_scheduler(num_training_steps=num_training_steps, optimizer=optimizer)

    def create_scheduler(self, num_training_steps: int, optimizer: torch.optim.Optimizer = None):
        """
        Setup the scheduler. The optimizer of the trainer must have been set up either before this method is called or
        passed as an argument.
    
        Args:
            num_training_steps (int): The number of training steps to do.
        """
        if self.lr_scheduler is None:
            self.lr_scheduler = get_scheduler(
                self.args.lr_scheduler_type,
                optimizer=self.optimizer if optimizer is None else optimizer,
                num_warmup_steps=self.args.get_warmup_steps(num_training_steps),
                num_training_steps=num_training_steps,
            )
            self._created_lr_scheduler = True
        return self.lr_scheduler

        self.state = TrainerState()
        self.state.is_hyper_param_search = trial is not None

    @dataclass
    class TrainerState:
    """
      <提示>
    
    在这个类中，一个步骤被理解为一个更新步骤。当使用梯度累积时，一个更新步骤可能需要多次前向和后向传递：如果您使用 `gradient_accumulation_steps=n`，那么一个更新步骤需要经过 *n* 个批次。
    
    </提示>
    
    参数:
    epoch (`float`, *可选*):
        仅在训练期间设置，表示训练所处的时期（小数部分表示当前时期完成的百分比）。
    global_step (`int`, *可选*，默认为 0):
        在训练期间，表示已完成的更新步数。
    max_steps (`int`, *可选*，默认为 0):
        当前训练中要执行的更新步数。
    total_flos (`float`, *可选*，默认为 0):
        自训练开始以来模型执行的总浮点操作数（以浮点数存储以避免溢出）。
    log_history (`List[Dict[str, float]]`, *可选*):
        自训练开始以来已完成的日志列表。
    best_metric (`float`, *可选*):
        在跟踪最佳模型时，迄今为止遇到的最佳指标值。
    best_model_checkpoint (`str`, *可选*):
        在跟踪最佳模型时，迄今为止遇到的最佳模型检查点的名称值。
    is_local_process_zero (`bool`, *可选*，默认为 `True`):
        此进程是否为本地（例如，在多台机器上以分布方式进行训练时在一台机器上）主要进程。
    is_world_process_zero (`bool`, *可选*，默认为 `True`):
        此进程是否为全局主要进程（在多台机器上以分布方式进行训练时，对于一个进程，这只会是 `True`）。
    is_hyper_param_search (`bool`, *可选*，默认为 `False`):
        我们是否在使用 Trainer.hyperparameter_search 进行超参数搜索过程中。这将影响数据在 TensorBoard 中记录的方式。
    
    """
    
    epoch: Optional[float] = None
    global_step: int = 0
    max_steps: int = 0
    num_train_epochs: int = 0
    total_flos: float = 0
    log_history: List[Dict[str, float]] = None
    best_metric: Optional[float] = None
    best_model_checkpoint: Optional[str] = None
    is_local_process_zero: bool = True
    is_world_process_zero: bool = True
    is_hyper_param_search: bool = False
    trial_name: str = None
    trial_params: Dict[str, Union[str, float, int, bool]] = None
    
    def __post_init__(self):
        if self.log_history is None:
            self.log_history = []
    
    def save_to_json(self, json_path: str):
        """Save the content of this instance in JSON format inside `json_path`."""
        json_string = json.dumps(dataclasses.asdict(self), indent=2, sort_keys=True) + "\n"
        with open(json_path, "w", encoding="utf-8") as f:
            f.write(json_string)
    
    @classmethod
    def load_from_json(cls, json_path: str):
        """Create an instance from the content of `json_path`."""
        with open(json_path, "r", encoding="utf-8") as f:
            text = f.read()
        return cls(**json.loads(text))

        # Activate gradient checkpointing if needed
        if args.gradient_checkpointing: 
            self.model.gradient_checkpointing_enable()

    model = self._wrap_model(self.model_wrapped)

    def _wrap_model(self, model, training=True, dataloader=None):
        if self.args.use_ipex:
            dtype = torch.bfloat16 if self.use_cpu_amp else torch.float32
            model = self.ipex_optimize_model(model, training, dtype=dtype)
    
        if is_sagemaker_mp_enabled():
            # Wrapping the base model twice in a DistributedModel will raise an error.
            if isinstance(self.model_wrapped, smp.model.DistributedModel):
                return self.model_wrapped
            return smp.DistributedModel(model, backward_passes_per_step=self.args.gradient_accumulation_steps)
         ...

        if is_sagemaker_mp_enabled() and resume_from_checkpoint is not None:
            self._load_from_checkpoint(resume_from_checkpoint, model)
    
        # as the model is wrapped, don't use `accelerator.prepare`
        # this is for unhandled cases such as
        # Fairscale Sharded DDP, FSDP-XLA, SageMaker MP/DP, DataParallel, IPEX
        use_accelerator_prepare = True if model is self.model else False
    
        if delay_optimizer_creation:
            self.create_optimizer_and_scheduler(num_training_steps=max_steps)
    
        # prepare using `accelerator` prepare
        if use_accelerator_prepare: 
            self.model.train()
            if hasattr(self.lr_scheduler, "step"):
                if self.use_apex:
                    model = self.accelerator.prepare(self.model)
                else: 
                    model, self.optimizer = self.accelerator.prepare(self.model, self.optimizer)
            else:
                # to handle cases wherein we pass "DummyScheduler" such as when it is specified in DeepSpeed config.
                model, self.optimizer, self.lr_scheduler = self.accelerator.prepare(
                    self.model, self.optimizer, self.lr_scheduler
                )
    
        if self.is_fsdp_enabled:
            self.model = model
    
        # for the rest of this function `model` is the outside model, whether it was wrapped or not
        if model is not self.model: 
            self.model_wrapped = model
    
        # backward compatibility
        if self.is_deepspeed_enabled: 
            self.deepspeed = self.model_wrapped
    
        # deepspeed ckpt loading
        if resume_from_checkpoint is not None and self.is_deepspeed_enabled:
            deepspeed_load_checkpoint(self.model_wrapped, resume_from_checkpoint)
    
        # Check if saved optimizer or scheduler states exist
        self._load_optimizer_and_scheduler(resume_from_checkpoint)

    if is_sagemaker_mp_enabled() and resume_from_checkpoint is not None:
            self._load_from_checkpoint(resume_from_checkpoint, model)

    # deepspeed ckpt loading
     if resume_from_checkpoint is not None and self.is_deepspeed_enabled:
     deepspeed_load_checkpoint(self.model_wrapped, resume_from_checkpoint)

    # prepare using `accelerator` prepare
        if use_accelerator_prepare: 
            self.model.train()
            if hasattr(self.lr_scheduler, "step"):
                if self.use_apex:
                    model = self.accelerator.prepare(self.model)
                else: 
                    model, self.optimizer = self.accelerator.prepare(self.model, self.optimizer)
            else:
                # to handle cases wherein we pass "DummyScheduler" such as when it is specified in DeepSpeed config.
                model, self.optimizer, self.lr_scheduler = self.accelerator.prepare(
                    self.model, self.optimizer, self.lr_scheduler
                )

       # Check if saved optimizer or scheduler states exist
        self._load_optimizer_and_scheduler(resume_from_checkpoint)

     self.state.epoch = 0
     start_time = time.time()
     epochs_trained = 0
     steps_trained_in_current_epoch = 0
     steps_trained_progress_bar = None
    
     # Check if continuing training from a checkpoint
     if resume_from_checkpoint is not None and os.path.isfile(
     os.path.join(resume_from_checkpoint, TRAINER_STATE_NAME)
     ):
     self.state = TrainerState.load_from_json(os.path.join(resume_from_checkpoint, TRAINER_STATE_NAME))
     epochs_trained = self.state.global_step // num_update_steps_per_epoch
     if not args.ignore_data_skip:
         steps_trained_in_current_epoch = self.state.global_step % (num_update_steps_per_epoch)
         steps_trained_in_current_epoch *= args.gradient_accumulation_steps
     else:
         steps_trained_in_current_epoch = 0
    
     logger.info("  Continuing training from checkpoint, will skip to saved global_step")
     logger.info(f"  Continuing training from epoch {epochs_trained}")
     logger.info(f"  Continuing training from global step {self.state.global_step}")
     if not args.ignore_data_skip:
         logger.info(
             f"  Will skip the first {epochs_trained} epochs then the first"
             f" {steps_trained_in_current_epoch} batches in the first epoch."
         )

        # Update the references
        self.callback_handler.model = self.model
        self.callback_handler.optimizer = self.optimizer
        self.callback_handler.lr_scheduler = self.lr_scheduler
        self.callback_handler.train_dataloader = train_dataloader
        if self.hp_name is not None and self._trial is not None:
            # use self._trial because the SigOpt/Optuna hpo only call `_hp_search_setup(trial)` instead of passing trial
            # parameter to Train when using DDP.
            self.state.trial_name = self.hp_name(self._trial)
        if trial is not None:
            assignments = trial.assignments if self.hp_search_backend == HPSearchBackend.SIGOPT else trial
            self.state.trial_params = hp_params(assignments)
        else: # finetune-lora
            self.state.trial_params = None
        # This should be the same if the state has been saved but in case the training arguments changed, it's safer
        # to set this after the load.
        self.state.max_steps = max_steps
        self.state.num_train_epochs = num_train_epochs
        self.state.is_local_process_zero = self.is_local_process_zero()
        self.state.is_world_process_zero = self.is_world_process_zero()
    
        # tr_loss is a tensor to avoid synchronization of TPUs through .item()
        tr_loss = torch.tensor(0.0).to(args.device)
        # _total_loss_scalar is updated everytime .item() has to be called on tr_loss and stores the sum of all losses
        self._total_loss_scalar = 0.0
        self._globalstep_last_logged = self.state.global_step
        model.zero_grad()
    
        self.control = self.callback_handler.on_train_begin(args, self.state, self.control)
    
        # Skip the first epochs_trained epochs to get the random state of the dataloader at the right point.
        if not args.ignore_data_skip:
            for epoch in range(epochs_trained):
                for _ in train_dataloader:
                    break

    total_batched_samples = 0
    for epoch in range(epochs_trained, num_train_epochs):  # 训练总共epoch数，我们的传参
       epoch_iterator = train_dataloader
    
       # Reset the past mems state at the beginning of each epoch if necessary.
       if args.past_index >= 0:
       self._past = None
    
       steps_in_epoch = (
       len(epoch_iterator)
       if len_dataloader is not None
       else args.max_steps * args.gradient_accumulation_steps
       )
      ...
       for step, inputs in enumerate(epoch_iterator):
       total_batched_samples += 1
       if rng_to_sync:
           self._load_rng_state(resume_from_checkpoint)
           rng_to_sync = False

        if args.past_index and hasattr(self, "_past"):
            # Clean the state at the end of training
            delattr(self, "_past")
        logger.info("\n\nTraining completed. Do not forget to share your model on huggingface.co/models =)\n\n")
        if args.load_best_model_at_end and self.state.best_model_checkpoint is not None:
            # Wait for everyone to get here so we are sur the model has been saved by process 0.
            if is_torch_tpu_available():
                xm.rendezvous("load_best_model_at_end")
            elif args.parallel_mode == ParallelMode.DISTRIBUTED:
                dist.barrier()
            elif is_sagemaker_mp_enabled():
                smp.barrier()
    
            self._load_best_model()
    
        # add remaining tr_loss
        self._total_loss_scalar += tr_loss.item()
        train_loss = self._total_loss_scalar / self.state.global_step
    
        metrics = speed_metrics("train", start_time, num_samples=num_train_samples, num_steps=self.state.max_steps)
        self.store_flos()
        metrics["total_flos"] = self.state.total_flos
        metrics["train_loss"] = train_loss
    
        self.is_in_train = False
    
        self._memory_tracker.stop_and_update_metrics(metrics)
    
        self.log(metrics)

        run_dir = self._get_output_dir(trial) # 获得路径
        checkpoints_sorted = self._sorted_checkpoints(use_mtime=False, output_dir=run_dir)

     def _sorted_checkpoints(
        self, output_dir=None, checkpoint_prefix=PREFIX_CHECKPOINT_DIR, use_mtime=False
    ) -> List[str]:
        ordering_and_checkpoint_path = []
    
        glob_checkpoints = [str(x) for x in Path(output_dir).glob(f"{checkpoint_prefix}-*") if os.path.isdir(x)]
    
        for path in glob_checkpoints:
            if use_mtime:
                ordering_and_checkpoint_path.append((os.path.getmtime(path), path))
            else:
                regex_match = re.match(f".*{checkpoint_prefix}-([0-9]+)", path)
                if regex_match is not None and regex_match.groups() is not None:
                    ordering_and_checkpoint_path.append((int(regex_match.groups()[0]), path))
    
        checkpoints_sorted = sorted(ordering_and_checkpoint_path)
        checkpoints_sorted = [checkpoint[1] for checkpoint in checkpoints_sorted]
        # Make sure we don't delete the best model.
        if self.state.best_model_checkpoint is not None:
            best_model_index = checkpoints_sorted.index(str(Path(self.state.best_model_checkpoint)))
            for i in range(best_model_index, len(checkpoints_sorted) - 2):
                checkpoints_sorted[i], checkpoints_sorted[i + 1] = checkpoints_sorted[i + 1], checkpoints_sorted[i]
        return checkpoints_sorted

        # Delete the last checkpoint when save_total_limit=1 if it's different from the best checkpoint and process allowed to save.
        if self.args.should_save and self.state.best_model_checkpoint is not None and self.args.save_total_limit == 1:
            for checkpoint in checkpoints_sorted:
                if checkpoint != self.state.best_model_checkpoint:
                    logger.info(f"Deleting older checkpoint [{checkpoint}] due to args.save_total_limit")
                    shutil.rmtree(checkpoint)

    self.control = self.callback_handler.on_train_end(args, self.state, self.control)

    return TrainOutput(self.state.global_step, train_loss, metrics)

    class TrainOutput(NamedTuple):
    global_step: int
    training_loss: float
    metrics: Dict[str, float]