transformers

transformers.Trainer

class Trainer:#这段代码根据训练数据集的类型和硬件环境，选择适当的采样器来处理数据集。def _get_train_sampler(self) -> Optional[torch.utils.data.sampler.Sampler]:if isinstance(self.train_dataset, torch.utils.data.IterableDataset):return Noneelif is_torch_tpu_available():return get_tpu_sampler(self.train_dataset)else:return (RandomSampler(self.train_dataset)if self.args.local_rank == -1else DistributedSampler(self.train_dataset))

class Trainer:#这段代码通过检查和设置采样器来创建一个适用于训练数据集的 DataLoader 对象，并返回它，以便在训练过程中使用。def get_train_dataloader(self) -> DataLoader:"""Returns the training :class:`~torch.utils.data.DataLoader`.Will use no sampler if :obj:`self.train_dataset` is a :obj:`torch.utils.data.IterableDataset`, a random sampler(adapted to distributed training if necessary) otherwise.Subclass and override this method if you want to inject some custom behavior."""if self.train_dataset is None:raise ValueError("Trainer: training requires a train_dataset.")train_sampler = self._get_train_sampler()return DataLoader(self.train_dataset,batch_size=self.args.train_batch_size,sampler=train_sampler,collate_fn=self.data_collator,drop_last=self.args.dataloader_drop_last,)

        # This might change the seed so needs to run first.self._hp_search_setup(trial)# Model re-initif self.model_init is not None:# Seed must be set before instantiating the model when using model_init.set_seed(self.args.seed)model = self.model_init()self.model = model.to(self.args.device)# Reinitializes optimizer and schedulerself.optimizer, self.lr_scheduler = None, None# Data loader and number of training stepstrain_dataloader = self.get_train_dataloader()if self.args.max_steps > 0:t_total = self.args.max_stepsnum_train_epochs = (self.args.max_steps // (len(train_dataloader) // self.args.gradient_accumulation_steps) + 1)else:t_total = int(len(train_dataloader) // self.args.gradient_accumulation_steps * self.args.num_train_epochs)num_train_epochs = self.args.num_train_epochsself.args.max_steps = t_totalself.create_optimizer_and_scheduler(num_training_steps=t_total)# Check if saved optimizer or scheduler states existif (model_path is not Noneand os.path.isfile(os.path.join(model_path, "optimizer.pt"))and os.path.isfile(os.path.join(model_path, "scheduler.pt"))):# Load in optimizer and scheduler statesself.optimizer.load_state_dict(torch.load(os.path.join(model_path, "optimizer.pt"), map_location=self.args.device))self.lr_scheduler.load_state_dict(torch.load(os.path.join(model_path, "scheduler.pt")))model = self.modelif self.args.fp16 and _use_apex:if not is_apex_available():raise ImportError("Please install apex from https://www.github.com/nvidia/apex to use fp16 training.")model, self.optimizer = amp.initialize(model, self.optimizer, opt_level=self.args.fp16_opt_level)# multi-gpu training (should be after apex fp16 initialization)if self.args.n_gpu > 1:model = torch.nn.DataParallel(model)# Distributed training (should be after apex fp16 initialization)if self.args.local_rank != -1:model = torch.nn.parallel.DistributedDataParallel(model,device_ids=[self.args.local_rank],output_device=self.args.local_rank,find_unused_parameters=True,)if self.tb_writer is not None:self.tb_writer.add_text("args", self.args.to_json_string())self.tb_writer.add_hparams(self.args.to_sanitized_dict(), metric_dict={})

train()

		#9. 训练循环# Train!if is_torch_tpu_available():total_train_batch_size = self.args.train_batch_size * xm.xrt_world_size()else:total_train_batch_size = (self.args.train_batch_size* self.args.gradient_accumulation_steps* (torch.distributed.get_world_size() if self.args.local_rank != -1 else 1))logger.info("***** Running training *****")logger.info("  Num examples = %d", self.num_examples(train_dataloader))# 训练示例数量logger.info("  Num Epochs = %d", num_train_epochs)#训练的总 Epoch 数logger.info("  Instantaneous batch size per device = %d", self.args.per_device_train_batch_size)#每个设备上的批量大小logger.info("  Total train batch size (w. parallel, distributed & accumulation) = %d", total_train_batch_size)#并行、分布式和梯度累积后的总训练批量大小logger.info("  Gradient Accumulation steps = %d", self.args.gradient_accumulation_steps)#梯度累积步数logger.info("  Total optimization steps = %d", t_total)#总优化步骤数# 这些变量用于跟踪训练过程中的全局步骤、当前 Epoch、已经训练的 Epoch 数和当前 Epoch 中已经训练的步骤数。self.global_step = 0self.epoch = 0epochs_trained = 0steps_trained_in_current_epoch = 0# 检查点恢复这部分代码用于从检查点恢复训练进度。如果提供了 model_path，则尝试从中解析出 global_step、epochs_trained 和 steps_trained_in_current_epoch，以便继续训练。如果解析失败，则从头开始训练。# Check if continuing training from a checkpointif model_path is not None:# set global_step to global_step of last saved checkpoint from model pathtry:self.global_step = int(model_path.split("-")[-1].split(os.path.sep)[0])epochs_trained = self.global_step // (len(train_dataloader) // self.args.gradient_accumulation_steps)steps_trained_in_current_epoch = self.global_step % (len(train_dataloader) // self.args.gradient_accumulation_steps)logger.info("  Continuing training from checkpoint, will skip to saved global_step")logger.info("  Continuing training from epoch %d", epochs_trained)logger.info("  Continuing training from global step %d", self.global_step)logger.info("  Will skip the first %d steps in the first epoch", steps_trained_in_current_epoch)except ValueError:self.global_step = 0logger.info("  Starting fine-tuning.")# 初始化损失和优化器 	•	tr_loss：初始化训练损失。logging_loss_scalar：用于日志记录的损失标量。model.zero_grad()：将模型的梯度归零。tr_loss = torch.tensor(0.0).to(self.args.device)logging_loss_scalar = 0.0model.zero_grad()# 训练循环disable_tqdm = self.args.disable_tqdm or not self.is_local_process_zero()#	•	disable_tqdm：控制进度条的显示。train_pbar = trange(epochs_trained, int(np.ceil(num_train_epochs)), desc="Epoch", disable=disable_tqdm)#train_pbar：用于显示 Epoch 进度的进度条。# ange(epochs_trained, int(np.ceil(num_train_epochs)))：从已经训练的 Epoch 开始，继续进行剩余的 Epoch 训练。如果使用分布式采样器，设置当前 Epoch。for epoch in range(epochs_trained, int(np.ceil(num_train_epochs))):if isinstance(train_dataloader, DataLoader) and isinstance(train_dataloader.sampler, DistributedSampler):train_dataloader.sampler.set_epoch(epoch)# 数据加载和迭代根据设备类型（TPU 或 GPU）设置数据加载器。	如果 past_index 大于等于 0，重置过去的内存状态。if is_torch_tpu_available():parallel_loader = pl.ParallelLoader(train_dataloader, [self.args.device]).per_device_loader(self.args.device)epoch_iterator = parallel_loaderelse:epoch_iterator = train_dataloader# Reset the past mems state at the beginning of each epoch if necessary.if self.args.past_index >= 0:self._past = None# epoch_pbar = tqdm(epoch_iterator, desc="Iteration", disable=disable_tqdm)# 迭代每个epoch ：inputs 通常包含如下键：	input_ids：输入文本的标记索引。	attention_mask：关注标记，用于指示模型应关注哪些标记。labels：目标标签。for step, inputs in enumerate(epoch_iterator):# Skip past any already trained steps if resuming trainingif steps_trained_in_current_epoch > 0:steps_trained_in_current_epoch -= 1epoch_pbar.update(1)continuetr_loss += self.training_step(model, inputs) #进入将输入喂入模型进行训练的阶段if (step + 1) % self.args.gradient_accumulation_steps == 0 or (# last step in epoch but step is always smaller than gradient_accumulation_stepslen(epoch_iterator) <= self.args.gradient_accumulation_stepsand (step + 1) == len(epoch_iterator)):if self.args.fp16 and _use_native_amp:self.scaler.unscale_(self.optimizer)torch.nn.utils.clip_grad_norm_(model.parameters(), self.args.max_grad_norm)elif self.args.fp16 and _use_apex:torch.nn.utils.clip_grad_norm_(amp.master_params(self.optimizer), self.args.max_grad_norm)else:torch.nn.utils.clip_grad_norm_(model.parameters(), self.args.max_grad_norm)if is_torch_tpu_available():xm.optimizer_step(self.optimizer)elif self.args.fp16 and _use_native_amp:self.scaler.step(self.optimizer)self.scaler.update()else:self.optimizer.step()self.lr_scheduler.step()model.zero_grad()self.global_step += 1self.epoch = epoch + (step + 1) / len(epoch_iterator)if (self.args.logging_steps > 0 and self.global_step % self.args.logging_steps == 0) or (self.global_step == 1 and self.args.logging_first_step):logs: Dict[str, float] = {}tr_loss_scalar = tr_loss.item()logs["loss"] = (tr_loss_scalar - logging_loss_scalar) / self.args.logging_steps# backward compatibility for pytorch schedulerslogs["learning_rate"] = (self.lr_scheduler.get_last_lr()[0]if version.parse(torch.__version__) >= version.parse("1.4")else self.lr_scheduler.get_lr()[0])logging_loss_scalar = tr_loss_scalarself.log(logs)if self.args.evaluate_during_training and self.global_step % self.args.eval_steps == 0:metrics = self.evaluate()self._report_to_hp_search(trial, epoch, metrics)if self.args.save_steps > 0 and self.global_step % self.args.save_steps == 0:# In all cases (even distributed/parallel), self.model is always a reference# to the model we want to save.if hasattr(model, "module"):assert (model.module is self.model), f"Module {model.module} should be a reference to self.model"else:assert model is self.model, f"Model {model} should be a reference to self.model"# Save model checkpointcheckpoint_folder = f"{PREFIX_CHECKPOINT_DIR}-{self.global_step}"if self.hp_search_backend is not None and trial is not None:run_id = (trial.numberif self.hp_search_backend == HPSearchBackend.OPTUNAelse tune.get_trial_id())checkpoint_folder += f"-run-{run_id}"output_dir = os.path.join(self.args.output_dir, checkpoint_folder)self.save_model(output_dir)if self.is_world_process_zero():self._rotate_checkpoints()if is_torch_tpu_available():xm.rendezvous("saving_optimizer_states")xm.save(self.optimizer.state_dict(), os.path.join(output_dir, "optimizer.pt"))xm.save(self.lr_scheduler.state_dict(), os.path.join(output_dir, "scheduler.pt"))elif self.is_world_process_zero():torch.save(self.optimizer.state_dict(), os.path.join(output_dir, "optimizer.pt"))torch.save(self.lr_scheduler.state_dict(), os.path.join(output_dir, "scheduler.pt"))epoch_pbar.update(1)if self.args.max_steps > 0 and self.global_step >= self.args.max_steps:breakepoch_pbar.close()train_pbar.update(1)if self.args.tpu_metrics_debug or self.args.debug:if is_torch_tpu_available():# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)xm.master_print(met.metrics_report())else:logger.warning("You enabled PyTorch/XLA debug metrics but you don't have a TPU ""configured. Check your training configuration if this is unexpected.")if self.args.max_steps > 0 and self.global_step >= self.args.max_steps:break

开始训练的示意图

tr_loss += self.training_step(model, inputs)

    def training_step(self, model: nn.Module, inputs: Dict[str, Union[torch.Tensor, Any]]) -> torch.Tensor:"""Perform a training step on a batch of inputs.Subclass and override to inject custom behavior.Args:model (:obj:`nn.Module`):The model to train.inputs (:obj:`Dict[str, Union[torch.Tensor, Any]]`):The inputs and targets of the model.The dictionary will be unpacked before being fed to the model. Most models expect the targets under theargument :obj:`labels`. Check your model's documentation for all accepted arguments.Return::obj:`torch.Tensor`: The tensor with training loss on this batch."""if hasattr(self, "_training_step"):warnings.warn("The `_training_step` method is deprecated and won't be called in a future version, define `training_step` in your subclass.",FutureWarning,)return self._training_step(model, inputs, self.optimizer)model.train()inputs = self._prepare_inputs(inputs)if self.args.fp16 and _use_native_amp:with autocast():outputs = model(**inputs)loss = outputs[0]else:outputs = model(**inputs) #获得输出# We don't use .loss here since the model may return tuples instead of ModelOutput.loss = outputs[0]

获得输出需要调用DAGN.py中的
class DAGN():
def forward():
# input_ids.size()=torch.Size([32,128])
flat_input_ids = input_ids.view(-1, input_ids.size(-1)) if input_ids is not None else None
flat_attention_mask = attention_mask.view(-1, attention_mask.size(-1)) if attention_mask is not None else None
flat_token_type_ids = token_type_ids.view(-1, token_type_ids.size(-1)) if token_type_ids is not None else None

   flat_passage_mask = passage_mask.view(-1, passage_mask.size(-1)) if passage_mask is not None else Noneflat_question_mask = question_mask.view(-1, question_mask.size(-1)) if question_mask is not None else Noneflat_argument_bpe_ids = argument_bpe_ids.view(-1, argument_bpe_ids.size(-1)) if argument_bpe_ids is not None else Noneflat_domain_bpe_ids = domain_bpe_ids.view(-1, domain_bpe_ids.size(-1)) if domain_bpe_ids is not None else None  flat_punct_bpe_ids = punct_bpe_ids.view(-1, punct_bpe_ids.size(-1)) if punct_bpe_ids is not None else None# 获取roberta模型最后一层的输出利用input和attention_mask;last_hidden_state=torch.Size([32,128,768])# BERT输入的所有token经过BERT编码后，会在每个位置输出一个大小为 hidden_size（在 BERT-base中是 768）的向量。last_hidden_state, p = self.roberta(flat_input_ids, attention_mask=flat_attention_mask, token_type_ids=None, return_dict = False)#last_hidden_state, p = self.bert(flat_input_ids, attention_mask=flat_attention_mask, token_type_ids=None, return_dict = False)sequence_output = last_hidden_state#torch.Size([32,128,768])pooled_output = p #torch.Size([32.768])

上述文件中指明了得到的内容：

@dataclass
class BaseModelOutputWithPooling(ModelOutput):"""Base class for model's outputs that also contains a pooling of the last hidden states.Args:last_hidden_state (:obj:`torch.FloatTensor` of shape :obj:`(batch_size, sequence_length, hidden_size)`):Sequence of hidden-states at the output of the last layer of the model.pooler_output (:obj:`torch.FloatTensor` of shape :obj:`(batch_size, hidden_size)`):Last layer hidden-state of the first token of the sequence (classification token)further processed by a Linear layer and a Tanh activation function. The Linearlayer weights are trained from the next sentence prediction (classification)objective during pretraining.hidden_states (:obj:`tuple(torch.FloatTensor)`, `optional`, returned when ``output_hidden_states=True`` is passed or when ``config.output_hidden_states=True``):Tuple of :obj:`torch.FloatTensor` (one for the output of the embeddings + one for the output of each layer)of shape :obj:`(batch_size, sequence_length, hidden_size)`.Hidden-states of the model at the output of each layer plus the initial embedding outputs.attentions (:obj:`tuple(torch.FloatTensor)`, `optional`, returned when ``output_attentions=True`` is passed or when ``config.output_attentions=True``):Tuple of :obj:`torch.FloatTensor` (one for each layer) of shape:obj:`(batch_size, num_heads, sequence_length, sequence_length)`.Attentions weights after the attention softmax, used to compute the weighted average in the self-attentionheads."""last_hidden_state: torch.FloatTensorpooler_output: torch.FloatTensor = Nonehidden_states: Optional[Tuple[torch.FloatTensor]] = Noneattentions: Optional[Tuple[torch.FloatTensor]] = None

图像解释：

dagn的forward中逐步计算损失函数

在dagn.py的第464行，在计算损失的时候，github中的原始代码调用错了算是计算的函数，改为
loss2 = self.get_con_lossL(positive_keys2, negative_keys2)