Pi3ddlZddlmZddlmZmZmZmZddlZddl m Z ddl m Z m Z mZddlmZmZmZddlmZddlmZejd ZeGd d ZGd d ZdS)N) dataclass)AnyDictOptionalUnion) DictConfig)_init_optim_stateset_model_state_dictset_state_dict)configtrainingutils)DistributedCheckpointer)OptimizerInBackwardWrapperDEBUGceZdZUdZeed<eed<eed<eed<dZeee e fed<dee e ffd Z dS) TrainingProgressz- This is training progress metadata. seed epochs_run total_epochsmax_steps_per_epochNdataloader_state_dictreturnc tj|jtj|jtj|jtj|jtj |j iS)N) r SEED_KEYr EPOCHS_KEYrTOTAL_EPOCHS_KEYr MAX_STEPS_KEYrDATALOADER_KEYrselfs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/torchtune/training/checkpointing/_checkpoint_client.py state_dictzTrainingProgress.state_dict&sA  ty    %t'8  "D$<  #T%?   ) __name__ __module__ __qualname____doc__int__annotations__rrrstrrobjectr#r$r"rrs IIIOOO6:8DcN3::: Df-      r$rc eZdZdZdeddfdZdZdZdej j d e ej j efd ed eddf d Zdej j d e ej j efd ed eddf d Zdej j d e ej j efd ed eddf dZdeeeffdZdej j d e ej j efdeeeffdZdS)CheckpointClientaK Stateful checkpointing client for TorchTune recipes. This class is responsible for saving and loading checkpoints using the user configured checkpointers or distributed checkpointer if asynchronous checkpointing is enabled. Args: cfg (DictConfig): Configuration object used to instantiate the recipe. cfgrNc||_d|_d|_|jdd|_|jdd|_|jdd|_tj|jj |_ tj \}|_ |j dk|_ dS)Nresume_from_checkpointFenable_async_checkpointingoptimizer_in_bwddevicer)_cfg _checkpointer_dcp_checkpointerget_resume_from_checkpoint_enable_async_checkpointing_optimizer_in_bwdr get_devicer6_deviceget_world_size_and_rank_rank _is_rank_zero)r!r0_s r"__init__zCheckpointClient.__init__:s ""&'+y}}5Mu'U'U$+/9== (%, , ("&/A5!I!I'ty/?@@@ 577 4:!Z1_r$c|js5|jrdn|j}tj|jj||_|jS)zF Builds and returns the user configured Checkpointer. F)should_load_recipe_state)r8r<r;r instantiater7 checkpointer)r!rFs r"_get_checkpointerz"CheckpointClient._get_checkpointerPs]! 321 % "(!3 &)A"""D !!r$c|js4|}t|j|j|_|jS)z Builds and returns the DistributedCheckpointer. DistributedCheckpointer is used for asynchronous checkpointing, if enabled. Uses the user configured checkpointer directory and outout directories. )checkpoint_dir output_dir)r9rIr_checkpoint_dir _output_dir)r!rHs r"_get_dcp_checkpointerz&CheckpointClient._get_dcp_checkpointer`sN % 1133L%<+;'3&&&D " %%r$model optimizertraining_progressepochc|jr-tdtj}i}||||tj<||tj <| }| ||d|jr5tdtj|z dddSdS)a0 Checkpoint the training state asynchronously as a distributed checkpoint. Saving asnchronously unblocks the training sooner to continue for the next epoch. The constructed checkpoint state dict contains the following information: - Model weights with key training.MODEL_KEY - Relevant recipe state, including optimizer, if training is not complete To correctly resume training from a distributed checkpoint, user needs to have both resume_from_checkpoint and enable_async_checkpointing flags set to True in the config. User does not need to provide any paths to checkpoint or recipe files. Latest intermediate and valid checkpoint will be loaded from the output directory and training progress will be restored automatically. z?Saving checkpoint asynchronously. Retrieving full state dict...T)rS save_asyncz$Saving asynchronous checkpoint took .2f secsN) rBloginfotime perf_counterupdater#r MODEL_KEYOPT_KEYrOsave_checkpoint)r!rPrQrRrScp_start ckpt_dict dcp_savers r"_save_checkpoint_asyncz'CheckpointClient._save_checkpoint_asyncps*   + HHV W W W(**H *5577888(-(8(8(:(: ($%&/&:&:&<&< ("#..00 !!  "      HH`t7H7J7JX7U````       r$c` dzjk }t|t }i jr-t dtj ii|r\tj |jj jr3t dtj z ddn|  rjr-t dtj}|roj s#tj||jj nY|jD]*\}} tj|| jj |<+n| jr3t dtj|z ddnd   fd } |r1jr | t"jd S| d S) a Checkpoint the training state synchronously. The constructed checkpoint state dict contains the following information: - Model weights with key training.MODEL_KEY - Relevant recipe state, including optimizer, if training is not complete To correctly resume training from this checkpoint, user needs to have both resume_from_checkpoint flag set to True and recipe file paths set in the config. zOSaving checkpoint. This may take some time. Retrieving full model state dict...r5z#Getting full model state dict took rVrWzGetting optimizer state dict...z"Getting optimizer state dict took NctjirHtjijr5t dtj z dddSdS)N)rSintermediate_checkpointzSaving checkpoint took rVrW) r\r r]r^r#rIr_rBrXrYrZr[)checkpoint_dictr`rSrgmodel_state_dictoptim_state_dictr!rRsr"_save_checkpoint_helperzGCheckpointClient._save_checkpoint_sync.._save_checkpoint_helpers  " "H$68H#I J J J' G&&(8:J'KLLL&&'8'C'C'E'EFFF  " " $ $ 4 4(? 5    ! Wd.?.A.AH.LWWWW  r$)rrI isinstancerrBrXrYrZr[r gather_cpu_state_dictr?r#r=get_full_optimizer_state_dict optim_mapitemstorch distributedbarrier)r!rPrQrRrSrHno_dist optim_startparamoptrkrhr`rgrirjs` `` @@@@@r"_save_checkpoint_syncz&CheckpointClient._save_checkpoint_syncs"#(!).?.L"L--//  /FGGG   + HHa   (**H  2 (="|     ! c$:K:M:MPX:Xcccc %//11  " $! 2:;;;"/11  :- '/'M!*#| ((($$'0&9&?&?&A&A s%B!3(:4<)!$-#7#7#9#9 ! e9J9L9L{9Zeeee $             *  &! *'')))   % % ' ' ' ' ' # # % % % % %r$c|dz|jk}|r!|jr|||||dS|||||dS)a Checkpoint the training state. The constructed checkpoint state dict contains the following information: - Model weights with key training.MODEL_KEY - Relevant recipe state, including optimizer state, if training is not complete If asynchronous checkpointing is enabled, the checkpoint will be saved asynchronously as a distributed checkpoint. Otherwise, the checkpoint will be saved synchronously with the checkpointer user has configured. reN)rr<rcrx)r!rPrQrRrSrgs r"r_z CheckpointClient.save_checkpointsm$#(!).?.L"L " St'G S  ' 'y:KU S S S S S  & &ui9JE R R R R Rr$cN|S)zp This method is used to load the base model from the checkpoint configured by the user. )rIload_checkpointr s r"load_base_checkpointz%CheckpointClient.load_base_checkpoint!s" %%''77999r$c|jrtj}|jst |i}|}|}|tj|tj |tj dtj dtj dtj di||}|jsdtj |vr4t!|||tj|tj nvt#||tjnTt#||tjtj |vr%||tj |jr3t&dtj|z dd|S)z This method is used to resume training from a distributed checkpoint state. Due to being distributed, this method is called on every rank. r)rirj)rPriz1DistributedCheckpointer loaded the checkpoint in rVz seconds.)rBrZr[r=r r#r\r r]r^rrrrrOr{r r load_state_dictrXrY)r!rPrQdcp_load_startrhrirjs r"load_distributed_checkpointz,CheckpointClient.load_distributed_checkpoint(s   1!.00N% ) i ( ( (*, ++--$//11"$4 "2!1#Q)1&   4466FFWW% M?22%4X5G%H%4X5E%F %%4X5G%H !!01C!D     ?22))/(:J*KLLL    HHwDDUDWDWZhDhwwww   r$)r%r&r'r(rrDrIrOrqnnModuleroptim Optimizerrrr)rcrxr_rr+rr|rr-r$r"r/r/0s- - ----,""" &&& *x*.0JJK*, *  *  ****Xj&xj&.0JJKj&, j&  j&  j&j&j&j&XSxS.0JJKS, S  S  SSSS2:d38n::::=x=.0JJK= c3h ======r$r/)rZ dataclassesrtypingrrrrrq omegaconfr'torch.distributed.checkpoint.state_dictr r r torchtuner r r.torchtune.training.checkpointing._checkpointerrtorchtune.training.memoryr get_loggerrXrr/r-r$r"rsY !!!!!!------------  .---------RRRRRR@@@@@@ew          *uuuuuuuuuur$