o "j$?@sddlZddlZddlZddlmZddlmZddlmZm Z ddl m Z ddl m Z ddlmZdd lmZmZmZmZdd lmZd d ZdddZdddZGdddeZGdddeZGdddeZGdddeZddZGdddejZ dS) N)nn)PyLayer) global_gatherglobal_scatter)check_nccl_version_for_p2p)in_dynamic_mode)recompute_hybrid)BaseGate GShardGate NaiveGate SwitchGate) count_by_gatecCs<|jdgkrt||d}|Stjd|jdg|jd}|S)Nrr dtype)shapepaddleZ index_selectemptyr)inpposinp_bufrq/var/www/html/Deteccion_Ine/venv/lib/python3.10/site-packages/paddle/incubate/distributed/models/moe/moe_layer.py_local_scatter&s rTcCsv|jdgkr,|j}tj|dd}tjtj||jdgdd||dd}tj||d}|Stj||jdg|jd}|S)NrZfloat32r)rrT) overwrite)rrrcastZscatterZzeros)rrout_batch_size maybe_overlapZ origin_dtyperrrr _local_gather.s rc Cs|dur |s dStr;|durtjjn|}t|j}|d|j9<t ||j }|j ||}| |S|durAdn|j}|durOtjjjn|j}tj|d|d|d|S)Nruse_calc_streamring_idnranks)Z is_memberrr distributedZ collectiveZ_get_default_grouplistrr"rrZ process_group all_gatherwaitidZ_get_global_groupZ _legacy_C_opsZ c_allgather)Ztensorgroupr Z tensor_shapeouttaskr!r"rrr _all_gather@s6  r+c@,eZdZdZe dddZeddZdS) MoEScatterz Scatter input samples from [batch x sequences] to contiguous alone experts. If `world_size` is greater than 1, the samples will first be locally scattered, and then exchanged across workers. Nc CsRt||}|dkrt||||d} n|} |jd||f|_|||f} |j| | S)Nr r(r)rrrmoe_argssave_for_backward) ctxrrlocal_expert_countglobal_expert_countfwd_batch_size world_sizer(Zlocal_input_bufZglobal_input_buf variablesrrrforwardjs   zMoEScatter.forwardc CsP|\}}}|j\}}}|dkrt||||d}n|}t|||} | dddfSNr r.) saved_tensorr/rr) r1Zgradrr2r3Zinp_batch_sizer5r(Z local_grad_inZgrad_inrrrbackwards   zMoEScatter.backwardN__name__ __module__ __qualname____doc__ staticmethodr7r:rrrrr-cs r-c@r,) MoEGatherz Gather output samples from contiguous alone experts back to [batch x sequences]. Works symmetrically with MoEScatter. Nc CsX|dkr t||||d}n|}t|||dd} |jd||f|_|||f} |j| | S)Nr r.F)rr)rrrr/r0) r1Zglobal_output_bufrr2r3local_batch_sizer5r(Zlocal_output_bufoutputr6rrrr7s   zMoEGather.forwardc CsN|\}}}|j\}}}t||}|dkrt||||d} n|} | dddfSr8)r9r/rr) r1grad_outrr2r3r4r5r(Z grad_out_bufZglobal_grad_out_bufrrrr:s   zMoEGather.backwardr;r<rrrrrBs rBc@(eZdZdZeddZeddZdS) AllGatherzP A wrapper for the All-Gather function to support auto-differentiation. cCs8g}tjj|||dtj|dd}||jdf|_|SNr.rZaxis)rr#r%concatrargs)r1rrankr5r(Z tensor_listrDrrrr7s zAllGather.forwardcCs.|j\}}tj|dg||g|d|gdS)Nrr ZaxesZstartsZends)rKrslice)r1rErLZdim0rrrr:s zAllGather.backwardNr<rrrrrGs  rGc@rF)SlicezK A wrapper for the Slice function to support auto-differentiation. c CsN|jd}||}||}t|||}tj|dg|g|gd}||f|_|S)NrrM)rminrrNrK) r1rrLr5r(BrCZ batch_startZ batch_endrrrr7s  z Slice.forwardcCs|j\}}t||dS)Nr.)rKr+)r1rEr5r(rrrr:s  zSlice.backwardNr<rrrrrOs   rOc Csrt||||d\}}}t|||gjdd}t|}Wdn1s-wY|||||fSrH)rrZno_gradreshape_sumintitem) gate num_expertr5 moe_grouprr2r3fwd_expert_countr4rrrprepare_forwards" rZcs4eZdZdZ     dfdd ZddZZS) MoELayera MoE Layer Args: d_model (int): Model dimention. experts (nn.LayerList): Expert networks list. gate (dict|NaiveGate|SwitchGate|NaiveGate): - If gate is a dict: gate is a gate network config, containing 2 keys: `type` (str) value can be: "naive", "gshard", "switch" or None, default is "gshard". `top_k` (int) Default value is 2. else gate is an instance of NaiveGate|SwitchGate|NaiveGate: moe_group: moe group for experts communication. mp_group: mp group for mp communication. recompute_interval (int, optional): Whether to use recompute, default 0, means to disable recompute. recompute_ctx (dict, optional): The context for recompute, if recompute_interval > 1, recompute_ctx must be given. Examples: .. code-block:: python >>> # doctest: +SKIP('Until Distributed move successfully, just skip it') >>> from paddle.nn import layer, LayerList >>> from paddle.distributed.moe import MoElayer >>> from paddle.distributed.collective import Group >>> from paddle.distributed import fleet >>> moe_group = Group(fleet.worker_index(), ... 0, ... list(range(fleet.worker_num()))) >>> mp_group = None >>> num_experts=8 >>> dim_feedforward=512 >>> d_model=8 >>> top_k=2 >>> class ExpertLayer(Layer): ... def __init__(self, d_model, d_hidden, name=None,rank=0, windex = 0, num_expert=1): ... super().__init__() ... self.htoh4 = nn.Linear(d_model, d_hidden) ... self.h4toh = nn.Linear(d_hidden, d_model) ... def forward(self, x): ... x = self.htoh4(x) ... x = self.h4toh(x) ... return x >>> gate_config = { ... "type": "gshard", ... "top_k": top_k, ... } >>> experts_list = LayerList() >>> for expi in range(num_experts): ... exp_layer = ExpertLayer(d_model, dim_feedforward // top_k, windex=expi, num_expert=num_experts) ... experts_list.append(exp_layer) >>> moeLayer = MoELayer(d_model = d_model, ... experts=experts_list, ... gate=gate_config, ... moe_group=moe_group, ... mp_group=mp_group, ... recompute_interval=0) Nrcst||_|duri}t|ttfsJd||_d|_|jdur)|jj|_t ||_ ||_ |dus7J||_ |jdkrJt dddkrJt||_||_t|tr|dd|_|dd}|d ksj|durxt|jt ||j|jd }nK|dkrt|jt ||j|j|jd }n7|d krt|jt ||j|j|jd }n#td t|t|tr|j|_nt|trtdt|td||_dS)Nz9gate config' type must be dict or an instance of BaseGater ZPADDLE_DISTRI_BACKENDZxccltop_ktypeZgshardZnaive)rWr5topk)rWr5r_r(switchzWe only support naive gate, gshard gate and switch gate, but you choose {} gate.zUnimplemented gate type: z/gate's type must be either dict or moe.BaseGate)super__init__ recompute_ctx isinstancedictr r(r5r"lenrWrecompute_intervalexpertsosgetenvrmp_groupd_modelgetr\r r r AssertionErrorformatstr TypeErrorr^rV)selfrlrhrVrXrkrgrc __class__rrrbKsv            zMoELayer.__init__c Cst|jdks J|j}|d|dg}d}d}|jdur&|jj}|jj}|dkr3t||||j}||\}}t ||j |j |j \}}} } } d} t|jdkrW|jd} |jdgkrb|| } n|} | |j kskJt|| || | |j |j }|j}dd}|jdks|jddkr||| |j}n t|j||| |j}|jd}t|jdkr||jd9}t|||| ||j |j }|d|j |g}||jdd|j g}t||d|g}|dkrt||||j}t||}|S)Nrr]rr cSs|jddkr |Sg}d}t|tjsJt|t|ksJt|D]\}}|dkr,q#|||||||||}q#tj|ddS)NrrI) rrdnpZndarrayrf enumerateappendrrJ)xrYrhyZ last_indexidxZ expert_countrrr experts_fwds z%MoELayer.forward..experts_fwd)rfrrRrkrLr"rOapplyrVrZrWr5r(r\r-rlrgnumpyrhrrcrBZreshaperZbmmrG)rrrZ origin_shapeZmp_rankZmp_sizevaluerVrr2r3rYr4r_Ztemp_posryrlr|rrrrr7s        zMoELayer.forward)NNNrN)r=r>r?r@rbr7 __classcell__rrrsrr[sGOr[)T)NT)!rir~rvrrZpaddle.autogradrZ"paddle.distributed.utils.moe_utilsrrZ#paddle.distributed.utils.nccl_utilsrZpaddle.frameworkrZ!paddle.incubate.distributed.fleetrrVr r r r utilsrrrr+r-rBrGrOrZZLayerr[rrrrs(        #24