o 3wi @sddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl Z ddlmZdd lmZmZmZmZdd lmZdd lmZmZdd lmZmZmZdd lmZee Z!ddZ"ddZ#dBddZ$dBddZ%ddZ&dCddZ'dCddZ(dDddZ)dCdd Z*dEd"e+d#e+d$e,fd%d&Z- dFd"e+d'e+d$e,d(e,fd)d*Z.d+e j/j0d,e j1fd-d.Z2d+e j/j0d/e3fd0d1Z4d2e j5j6d3e3fd4d5Z7d+e j/j0fd6d7Z8d+e j/j0d8e j/j0fd9d:Z9d+e j/j0d8e e j/j0ge,ffd;d<Z:d=d>Z;d?e3d8e3fd@dAZr4r5r9optim_state_dict_configrr@r,r pathjoin process_indexloggerinfotorchrLOCAL_STATE_DICTSHARDED_STATE_DICTFileSystemWriter)r? acceleratorr) output_dir model_indexr*dist_cprBFSDPr3is_multi_processctxr+r( weights_nameoutput_model_fileckpt_dirrrrsave_fsdp_modelesj       "    $rhcCsddlmm}ddlm}ddlm}ddlm} ||j | j kr2|j dk} | |j _ | |j _|jdkrB| ||j |j |jnt} t|} | |j | j krt||uru|jdkru|jsu|jsl|jdkrltd WddS|dkr~tdntd|d} tj|| }td ||j p|j}|rtj |d d }ni}td |n|j | j!kr|dkrtd |jdn td|d |jd} tj|| }td |tj |d d }td |nD|j | j"kr3t|vrtj|td|n|}td |dt#||| di}|j ||$||d|d}td |t%|||| d}Wd|S1sGwY|S)Nr)DefaultLoadPlannerrCr2rzzSet the `sync_module_states` flag to `True` so that model states are synced across processes when initializing FSDP objectrHrIzLoading model from T weights_onlyzModel loaded from rJr)rG)r(storage_readerrM)&rNrOrPrQrir:rDr3wait_for_everyoner;r<rSr>r4r5r9rTrr@typerWis_fsdp2sync_module_states ValueErrorr rrUrVrXrYis_main_processrZloadr[r\r,FileSystemReaderr0)r?r^r) input_dirr`r*rarirbr3rcrdr+reinput_model_file load_modelr(rg load_resultrrrload_fsdp_modelsz       "    ..rycCsddlmm}ddlm}ddlm}ddlm} tj |dd|j dkr2| ||j |j |j nt} t|} | |j dkrOdd lm} | ||| d } n|||} |j | jkr|jdkr|dkritd ntd |d }tj||}td |t| |td|n9tj|td |}tj |ddtd ||jd| i|||dtd|WddSWddS1swYdS)NrrArCr2TrErr)get_optimizer_state_dictr#rHrIzSaving Optimizer state to zOptimizer state saved in optimizerrK)rNrOrPrQrBr:rDr3rrRr9r;r>rTrr@r'rzoptim_state_dictr<rWr rUrVrXrYrZrr])r?r^r{r)r_optimizer_indexrarBrbr3rdr+rz optim_stateoptim_state_nameoutput_optimizer_filergrrrsave_fsdp_optimizersL           "rcCsddlmm}ddlm}ddlm} ||jdkr)|||j|j |j nt } t |} | |j| j krtd} |jdksD|j jss|dkrMtdntd|d} tj|| }td|tj|dd } td |n8t|vrtj|td|n|}td |d |i} |j| |||d | d } td||jdkr|j||| d}||nddlm}|||| | dWddSWddS1swYdS)NrrCr2rrHrIzLoading Optimizer state from TrjzOptimizer state loaded from zLoading Optimizer from r{) checkpoint_idrlzOptimizer loaded from )r)optimr|)set_optimizer_state_dictr#)rNrOrPr:rDr3rmr9r;r>rTrr@r<rWr5r rrUrVrXrYrZrsr(rtoptim_state_dict_to_loadr/r'r)r?r^r{r)rur}r*rarbr3rdr+r~optimizer_nameinput_optimizer_filerg flattened_osdrrrrload_fsdp_optimizersV          "rTcheckpoint_dir save_pathsafe_serializationcCsddlmm}ddlmmm}i}t|}|jdd|j|||| dd|r5|t n|t }t | dkrI|t|d}t|||d|S)z Passthrough to `torch.distributed.checkpoint.format_utils.dcp_to_torch_save` Will save under `save_path` as either `model.safetensors` or `pytorch_model.bin`. rNTrE)rlrMno_distr)r)rNrOrP)torch.distributed.checkpoint.format_utils format_utilsrmkdir_load_state_dictrt_EmptyStateDictLoadPlannerr r lenkeyslistr)rrrradist_cp_format_utilsr(rrr)_distributed_checkpoint_to_merged_weightsJs  r output_pathremove_checkpoint_dirc Cs2t|}ddlm}tddstd|sh|d}|d}d|d }|rD|rD|d 7}|d |d |d 7}|d7}t||rV|d7}|d|d7}t||rd|d7}|d|d7}t||}|jrtd|t |||} td| |rtd|t || dS)a? Merge the weights from sharded FSDP model checkpoints into a single combined checkpoint. Should be used if `SHARDED_STATE_DICT` was used for the model. Weights will be saved to `{output_path}/model.safetensors` if `safe_serialization` else `pytorch_model.bin`. Note: this is a CPU-bound process. Args: checkpoint_dir (`str`): The directory containing the FSDP checkpoints (can be either the model or optimizer). output_path (`str`): The path to save the merged checkpoint. safe_serialization (`bool`, *optional*, defaults to `True`): Whether to save the merged weights with safetensors (recommended). remove_checkpoint_dir (`bool`, *optional*, defaults to `False`): Whether to remove the checkpoint directory after merging. r) PartialStatez>=z2.3.0z/`merge_fsdp_weights` requires PyTorch >= 2.3.0`pytorch_model_fsdp_0 optimizer_0zTried to load from z) but couldn't find a valid metadata file.zE However, potential model and optimizer checkpoint directories exist.zPlease pass in either z/pytorch_model_fsdp_0 or z /optimizer_0zinstead.z8 However, a potential model checkpoint directory exists.zPlease try passing in z/pytorch_model_fsdp_0 instead.z< However, a potential optimizer checkpoint directory exists.z/optimizer_0 instead.zMerging FSDP weights from z.Successfully merged FSDP weights and saved to z"Removing old checkpoint directory N) raccelerate.staterrrqexistsrrrXrYrshutilrmtreerm) rrrrrmodel_path_existsoptimizer_path_existserrstaterrrrmerge_fsdp_weightsfs<        rr)devicec st|dd}|s Si|D]%}|d}d|dd|d}}||}t||}dt|<qdtjjffdd }|S)N_tied_weights_keys.modulecstt}|jddD]\}}t|vr|t||q |}|D]\}}|D]}|}|dur>t|||<q,t|||q,q&|S)NF)recurse)rrnamed_parametersidappenditemsr=setattr)r params_to_tienparamid_key _param_names param_name _tied_params param_init_fnrrparam_init_fn_tied_params  z7ensure_weights_retied..param_init_fn_tied_param)r=splitrV get_submodulerrZnnModule) rr)r _tied_namesnamermodrrrrrensure_weights_retieds    rfull_sdcCs@ddlm}ddlm}|}i}dd}dd}|jr\t||D]4\\} } } | j } | | j } |j | d| d|| | | j} ||| | \}}|| ||} | || <q&n;|D]6\} } | j } tj| | j | jd }|j |d| d||| | j} ||| |\}}|| ||} | || <q`|j|d d |S) a Loads the full state dict (could be only on rank 0) into the sharded model. This is done by broadcasting the parameters from rank 0 to all other ranks. This function modifies the model in-place. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to load the state dict into, expected to be on meta device or a VRAM spike can occur full_sd (`dict`): The full state dict to load, can only be on rank 0 rN)distribute_tensorc Ssz||}Wnty"|dd\}}||}t||}Ynwttd}d}|o1|jtjk} |jj r;| s;|j}|duoB| |fS)Nrr float8_e4m3fn) get_parameter_or_bufferAttributeErrorrsplitrr=hasattrrZdtyperis_floating_point is_contiguous) r)r empty_param old_parambase_param_namelocal_param_name submoduleis_torch_e4m3fn_available casting_dtypeis_param_float8_e4m3fnrrr_infer_parameter_dtypes    z:fsdp2_load_full_state_dict.._infer_parameter_dtypecSs$|dur |j|d}|r|}|S)N)r)to contiguous)tensor to_contiguousrrrr_cast_and_contiguouss  z8fsdp2_load_full_state_dict.._cast_and_contiguous)srcgroup)rrT)assign)torch.distributedrOtorch.distributed.tensorrr(rrziprvalues device_meshdetachr device_type broadcast get_group placementsrZemptysizerr/)r^r)rdistrmeta_sharded_sd sharded_sdrrr full_param sharded_paramrsharded_tensorrr full_tensorrrrfsdp2_load_full_state_dictsD     rr{mappingcsDz|jD]}fdd|dD|d<qWdSty!tdw)a Switches the parameters of the optimizer to new ones (sharded parameters in usual case). This function modifies the optimizer in-place. Args: optimizer (`torch.optim.Optimizer`): Optimizer instance which contains the original model parameters mapping (`dict`): Mapping from the original parameter (specified by `data_ptr`) to the sharded parameter Raises: KeyError: If a parameter in the optimizer couldn't be switched to its sharded version. This should never happen and indicates a bug. If we kept the original params instead of raising, the training wouldn't be numerically correct and weights wouldn't get updated. csg|]}|jqSr)data_ptr.0prrr )sz5fsdp2_switch_optimizer_parameters..paramszA parameter in the optimizer couldn't be switched to its sharded version. This breaks the training. Please raise an issue on GitHub.N) param_groupsKeyError)r{r param_grouprrr!fsdp2_switch_optimizer_parameterss  rc Csddlm}t|jj|}t|ddddD]3\}}t|ddkr-|dd\}}nd}|}|r8| |n|}||rJ||d d }| ||q|S) a8 Applies the activation checkpointing to the model. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to apply the activation checkpointing to Returns: `torch.nn.Module`: The model with the activation checkpointing applied r)checkpoint_wrapperT) return_fqnsNrrrF)preserve_rng_state) ;torch.distributed.algorithms._checkpoint.checkpoint_wrapperrfsdp2_prepare_auto_wrap_policyrr?rrrrrregister_module) r^r)rauto_wrap_policy_func layer_namelayer parent_name child_name parent_modulerrrfsdp2_apply_ac2s   r returncs(ddlm}m}m}t||pt|ot|j|}|r|S|jj}| || }|j |j |j p3|d}d} |D]\} } | jjdkrJd} nq<|jrv| svt|dddtfdd |D} |td }t|d rv|t||} | d urt|d d D]}| |rt||s||fi|qt||s||fi||jrt||||jr| s| D](\}}||j}d|vr|dd\}}| |}n|}|}|j!||ddqt|d r|t"|dd }|j#dkr|d us|tj$kr|tj$}|j%rt&'d|S)a"Prepares the model for FSDP2 in-place. Also returns the model to avoid misuse of the original model. Args: accelerator (`Accelerator`): The accelerator instance model (`torch.nn.Module`): The model to prepare Returns: `torch.nn.Module`: Prepared model r) FSDPModuleMixedPrecisionPolicy fully_shard)reshard_after_forwardoffload_policy mp_policyF Params4bitT)rfqnscsi|] \}}|vr||qSrrrkvnon_persistent_buffer_fqnsrr z'fsdp2_prepare_model..meta tie_weightsNrrr) persistentrnoz~FSDP upcast of low precision parameters to fp32 (since mixed_precision != 'no') may affect the precision of model checkpoints.)(torch.distributed.fsdpr r r isinstancer _orig_modrr?set_auto_wrap_policyr(rr7mixed_precision_policyr __class____name__cpu_ram_efficient_loadingrcopydeepcopy named_buffersrrZrrrrrrrrrregister_bufferr=mixed_precisionfloat32rrwarningswarn)r^r)r r r is_type_fsdp fsdp2_plugin original_sd fsdp2_kwargsmodel_has_params4bitrroriginal_non_persistent_buffersrrfqn buffer_tensor parent_fqnlocal_buffer_namer  model_dtyperrrfsdp2_prepare_modelSsp               " r:c sddlm}m}j}t|tjr|j}||ur^t|dd}|dur$g}t |}j dur0j }t |D]}t ||}|durHt d|d|q5dtjjdtffd d } | S||urqdtjjdtffd d } | SdS) a!Prepares the auto wrap policy based on its type, done to mimic the behaviour of FSDP1 auto wrap policy. Args: fsdp2_plugin (`FullyShardedDataParallelPlugin`): Instance of `FullyShardedDataParallelPlugin` containing the configuration options auto_wrap_policy_type (`str`): Either `transformer` or `size` model (`torch.nn.Module`): The model to wrap Returns: `Callable[[torch.nn.Module], bool]`: The auto wrap policy function to be applied to the model r)size_based_auto_wrap_policytransformer_auto_wrap_policy_no_split_modulesNz+Could not find the transformer layer class z in the model.rr csjdurdSt|tS)NF)transformer_cls_names_to_wrapr tuple)rr0transformer_cls_to_wraprrpolicys z.fsdp2_prepare_auto_wrap_policy..policycs tdd|D}|jkS)Ncss|]}|VqdS)N)numelrrrr szAfsdp2_prepare_auto_wrap_policy..policy..)sum parametersmin_num_params)rmodule_num_params)r0rrrBs )torch.distributed.fsdp.wrapr;r<auto_wrap_policyr  functoolspartialfuncr=rr>setr rqaddrZrrbool) r0r)r;r<fnno_split_modulesr> layer_classtransformer_clsrBrr@rrs.      rcKsddlm}|di|S)a Returns a `GradScaler` for FSDP2, as the current implementation of `get_grad_scaler` doesn't accept other args. We need this as current `get_grad_scaler` accepts only `distributed_type` as arg, which doesn't differentiate between FSDP1 and FSDP2 r) GradScalerNr)torch.amp.grad_scalerrU)kwargsrUrrrget_fsdp2_grad_scalers rX named_paramscCs:dd|D}dd|D}dd|D}|S)a6Removes parameter name modifiers in order to map them back to their original names. See huggingface/accelerate#3554 for more context. Args: named_params (`dict`): The named parameters dictionary to canonicalize. Returns: `dict`: The canonicalized named parameters dictionary cSi|] \}}|dd|qS)z._checkpoint_wrapped_modulereplacerrrrr rz,fsdp2_canonicalize_names..cSs,i|]\}}|dr|ddn||qS)z _orig_mod.r[) startswithr]rrrrrs cSrZ)z ._orig_modr[r\rrrrrr)r)rYrrrfsdp2_canonicalize_namess  r_)FN)rF)r)T)TF)=r'rKrrr- collectionsr contextlibrpathlibrtypingrrZloggingr constantsr r r r dataclassesr modelingrrotherrrrversionsrr%rXrrr,r0r@rhryrrstrrPrrrrrrdictrr Optimizerrr r:rrXr_rrrrsZ            : H 03 7.M!$o5