j/ ddlmZmZddlmZmZmZddlmZ ddl m Z m Z m Z mZddlmZded efd Zdd ed eefd Zd ZdZ ddedee jjdedee jfdZddZ ddZdS))reducewraps)AnyCallableOptionalN) tree_flattentree_map tree_reducetree_unflatten)Modulemodelfncvfd}tj|tfd}|S)aTransform the passed function ``fn`` to a function that computes the gradients of ``fn`` wrt the model's trainable parameters and also its value. Args: model (mlx.nn.Module): The model whose trainable parameters to compute gradients for fn (Callable): The scalar function to compute gradients for Returns: A callable that returns the value of ``fn`` and the gradients wrt the trainable parameters of ``model`` c>||i|SNupdate)paramsargskwargsrrs V/lsinfo/ai/hellotax_ai/base_platform/venv/lib/python3.11/site-packages/mlx/nn/utils.pyinner_fnz value_and_grad..inner_fns* Vr4"6"""cNg|Ri|\}}||fSrtrainable_parameters)rrvaluegradr value_grad_fns rwrapped_value_grad_fnz-value_and_grad..wrapped_value_grad_fn!s<#mE$>$>$@$@R4RRR6RR td{r)mxvalue_and_gradr)rrrr"r!s`` @rr$r$ sk######%h//M 2YYY ! rmodulec~fd}tj|tfd}|S)aPTransform the passed callable to one that performs gradient checkpointing with respect to the trainable parameters of the module (and the callable's inputs). Args: module (mlx.nn.Module): The module for whose parameters we will be performing gradient checkpointing. fn (Callable, optional): The function to checkpoint. If not provided it defaults to the provided module. Returns: A callable that saves the inputs and outputs during the forward pass and recomputes all intermediate states during the backward pass. Nc>||i|Srr)rrrrr%s rrzcheckpoint..inner_fn=s* fr4"6"""rc@g|Ri|Srr)rrcheckpointed_fnr%s rwrapped_checkpointed_fnz+checkpoint..wrapped_checkpointed_fnCs/v::<<NtNNNvNNNr)r# checkpointr)r%rrr*r)s`` @rr+r+)s~ z######mH--O 2YYOOOOOYO #"rcnd|D}d|D}d|D}d|D}||||fS)Ncg|]\}}|Sr.).0k_s r z!_extract_info..Ks   $!QA   rc"g|] \}}|j Sr.)shaper/r1gs rr2z!_extract_info..L ' ' '$!Qag ' ' 'rc"g|] \}}|j Sr.)sizer5s rr2z!_extract_info..Ms % % %1QV % % %rc"g|] \}}|j Sr.)dtyper5s rr2z!_extract_info..Nr7rr.)flatkeysshapessizesdtypess r _extract_inforAJs]  $   D ' '$ ' ' 'F % % % % %E ' '$ ' ' 'F  &&rcg}g}d}tt|D]D}||||||zz }||kr||g}d}E|r||g}|S)Nr)rangelenappend)r=r?itemsizecommunication_size grad_groups grad_groupgrad_group_sizeis r_group_by_sizerLRsKJO 3t99    !58h.. 0 0 0   z * * *JO:&&& r gradientsgroupall_reduce_sizecommunication_streamc> ptjdkr|S|dkrt fd|St | t dkr|St \  t fd Dst|dSt  dj|}g}|D]}tfd|dg}tj fd|D tj z tj |dd | fd t!|Dt#|S) aAverage the gradients across the distributed processes in the passed group. This helper enables concatenating several gradients of small arrays to one big all reduce call for better networking performance. Args: gradients (Any): The Python tree containing the gradients (it should have the same structure across processes) group (Optional[mlx.core.distributed.Group]): The group of processes to average the gradients. If set to ``None`` the global group is used. Default: ``None``. all_reduce_size (int): Group arrays until their size in bytes exceeds this number. Perform one communication step per group of arrays. If less or equal to 0 array grouping is disabled. Default: ``32MiB``. communication_stream (Optional[mlx.core.Stream]): The stream to use for the communication. If unspecified the default communication stream is used which can vary by back-end. Default: ``None``. r rcNtj|z S)NrOstream)r# distributedall_sum)xNrQrOs rz#average_gradients..s1bn,,+-  rc30K|]}|dkVdS)rNr.)r/dtr@s r z$average_gradients..s+44r2?444444rc.||d|zgzS)Nr.)rXyr?s rrZz#average_gradients..s!quuQx/?.@*@rcRg|]#}|dd$Sr r_reshape)r/rK flat_gradss rr2z%average_gradients..s0BBB!Aq!))"--BBBr)rUrOr_c3pK|]0\}}|||fV1dSrrc)r/rKjbig_gradr=r>s rr]z$average_gradients..sZ""Aqa(1+--fQi889""""""r)r#rVinitr9r r rDrAallaverage_gradientsrLr concatenaterWsplitextend enumerater )rNrOrPrQrHnew_flat_gradsrIindicesrYrhr@rer=r>r?s ` ` @@@@@@@rrkrkcs0  *R^((**E AAvv!          "),, z??a   '4J&?&?#feV4444V44444 :$Yq99 9$T5&)./RR %  J@@@@*qcRRG~BBBBzBBBH&&%9'  x'!B$-88H  ! !""""""%j11"""     n---rctd|d}tj||}tj|}tj||dzz dt fd|}||fS)NcT||zSr)squaresum)accr6s rrZz"_clip_grads_fsdp..ssQXXZZ^^5E5E/ErgrOgư>g?c|zSrr.)r6 normalizers rrZz"_clip_grads_fsdp..s Q^r)r r#rVrWsqrtminimumr ) grads_slicemax_normrO local_norm_sqglobal_norm_sq grad_normrys @r_clip_grads_fsdprs E E{TWXXM^++M+GGN''IH D(893??J3333[AAK  !!rc !"#$|ptj}|||ndz}|dkrC|+t ||\}} |||| fS|||St |"t |#t"\} } $} | dj} t| $| |}|!| }i}i}t|D]\}}tj !"fd|Dd}tj ||||z ||<|+tj ||||||<tj !#fd|Dd}||||<d} |t |||\}} |||}g}t|D]\}}tj||||}!$fd |D}g}d}|D]}||z }||tj||dd d}t|D]A\}}|| |||| |fBt%|} || | fS| S) a Perform a distributed optimizer step by sharding gradients and optimizer states across ranks. This helper function performs the following steps: 1. Reduce-scatter the gradients across ranks so each rank gets a shard of the averaged gradients. 2. Optionally clip the sharded gradients by global norm. 3. Apply the optimizer update on the local parameter slice using the sharded gradients. 4. All-gather the updated parameter slices from all ranks to reconstruct the full parameters tree. This is similar to PyTorch's FSDP with `reshard_after_forward=False`. Args: gradients (Any): The Python tree containing the full gradients (it should have the same structure as ``parameters``). Each gradient's first dimension must be divisible by ``fsdp_group.size()``. parameters (Any): The Python tree containing the full parameters (it should have the same structure across processes). Each parameter's first dimension must be divisible by ``fsdp_group.size()``. optimizer: Optimizer with an ``apply_gradients`` method. fsdp_group (Optional[mlx.core.distributed.Group]): The group of processes for FSDP sharding. If ``None``, the global group is used. dp_group (Optional[mlx.core.distributed.Group]): The group of processes for data-parallel gradient averaging. Required when ``fsdp_group`` is smaller than the world (e.g. FSDP intra-node, DDP inter-node). Default: ``None``. communication_size (int): Group arrays until their size in bytes exceeds this number. Perform one communication step per group of arrays. If less or equal to 0 array grouping is disabled. Default: ``32MiB``. communication_stream (Optional[mlx.core.Stream]): The stream to use for the communication. If unspecified the default communication stream is used which can vary by back-end. Default: ``None``. max_norm (Optional[float]): If provided, clip gradients to this maximum global norm before applying the optimizer update. Default: ``None``. Returns: If ``max_norm`` is ``None``, returns the updated full-parameter tree. Otherwise returns ``(parameters, grad_norm)``, where ``grad_norm`` is the global gradient norm before clipping. Example: >>> optimizer = optim.SGD(learning_rate=0.01) >>> # Without gradient clipping >>> updated_params = fsdp_apply_gradients(grads, params, optimizer) >>> model.update(updated_params) >>> >>> # With gradient clipping >>> updated_params, grad_norm = fsdp_apply_gradients( ... grads, params, optimizer, max_norm=1.0 ... ) >>> model.update(updated_params) Nr rcTg|]$}|dd%Srbrc)r/rKSres rr2z(fsdp_apply_gradients..s2 @ @ @Z]1  % %a , , @ @ @r)axisrTcTg|]$}|dd%Srbrc)r/rKr flat_paramss rr2z(fsdp_apply_gradients..s2 A A A![^A  & &q" - - A A Arrwc&g|] }|zSr.r.)r/rKrr?s rr2z(fsdp_apply_gradients..3s!888uQx1}888rr_)r#rVrir9rapply_gradientsr rArLrankrorl sum_scatterrW all_gatherrErmrdr )%rN parameters optimizer fsdp_groupdp_grouprGrQr}rYrr=r>r@rFgroups fsdp_rank grad_slices param_slices group_idx arr_grouprh big_paramupdated_param_slicesnew_flat big_gathered split_sizes split_indicesrvsparts idx_in_grouprKresultrrerr?s% @@@@rfsdp_apply_gradientsrs|4r~2244J0DX]]___!LAAvv  #3Ix#H#H Iy,,Y CCYN N((J???i((Jz**K"/ ";";D&%ay~H D%3E F FFA!!IKL )& 1 177 9> @ @ @ @ @i @ @ @q    N & & 3G '    I  %'^%;%;I&h?S&<&&K "N A A A A Ay A A A   #,I"6 YI!1 " " "  Y %44[,OOH )& 1 1OO 9~00  +'1  98888i888   & &A 1HC   % % % %}SbS'9BBB(33 O OOL! OOT!WeL&9&A&A&)&L&LM N N N N OH % %Fy  Mrr)NrMN)NNrMNN) functoolsrrtypingrrrmlx.corecorer#utilsr r r r layers.baserr$r+rArLrVGroupintStreamrkrrr.rrrs$#######**********GGGGGGGGGGGG!&!h!!!!:##v#8H#5####B'''&-1'04 J.J.J. BN( )J.J.#29- J.J.J.J.Z"""" # GGGGGGr