o }oiUd@sddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z mZmZmZmZmZmZmZmZddlmZddlZddlZddlZddlZddlmZddl m!Z!m"Z"ddl#m$Z$dd l%m&Z&dd l'm(Z(dd lm)Z)m*Z*dd l+m,Z,dd l-m.Z.ddl/m0Z0ddl1m2Z3ddl4m5Z5ddl6m7Z7m8Z8m9Z9ddl:m;Z;mm?Z@ddlAmBZBddlCmDZDe ZE edZFeGddde!ZGGddde<ZHde eeFdeFfddZIdeejJejKffdd ZLd!eeee=ee=fdee=fd"d#ZMGd$d%d%e<ZNGd&d'd'ejOe8jPe8jQZRGd(d)d)e(ZSGd*d+d+eZTGd,d-d-e0ZUGd.d/d/ejVZWGd0d1d1ejXZYd2d3ZZededfd4d5Z[ej\]d6ej\j^ej\j_d7d8Z`d9d:Zaebd;krteadSdS)<N)contextmanager) dataclass)Path) AnyDictIterableIteratorListOptionalSequenceTuple TypedDictTypeVarUnion)TensorBoardLogger)ModelParallelConfigparallel_state)OptimizerConfig) ModelType)MegatronModule)Tensornn) DataLoader) transforms)MNIST) lightning)llm) NeMoLoggerioresume)DataTMegatronLossReduction ReductionT) callbacks)MegatronOptimizerModule)MegatronDataSamplerTc@s:eZdZUdZdZeed<dZeed<dej fddZ dS) ExampleConfigzExampleConfig is a dataclass that is used to configure the model. Timers from ModelParallelConfig are required for megatron forward compatibility. Fcalculate_per_token_lossfp8returncCst|S)zThis function is called by the strategy to construct the model. Note: Must pass self into Model since model requires having a config object. Returns: The model object. ) ExampleModelselfr.`/home/ubuntu/.local/lib/python3.10/site-packages/tests/collections/llm/test_mnist_model_nemo2.pyconfigure_modelCszExampleConfig.configure_modelN) __name__ __module__ __qualname____doc__r(bool__annotations__r)rModuler0r.r.r.r/r'9s   r'c@sDeZdZdZdededeeeffddZde edefdd Z d S) MSELossReductionz]A class used for calculating the loss, and for logging the reduced loss across micro batches.batch forward_outr*cCs@|d}|}|d}||dd}tj||}|d|ifS)a-Calculates the loss within a micro-batch. A micro-batch is a batch of data on a single GPU. Args: batch: A batch of data that gets passed to the original forward inside LitAutoEncoder. forward_out: the output of the forward method inside LitAutoEncoder. Returns: A tuple containing [, ReductionT] where the loss tensor will be used for backpropagation and the ReductionT will be passed to the reduce method (which currently only works for logging.). datax_hatravg)viewsizer functionalmse_loss)r-r9r:xoutputsr<xviewlossr.r.r/forwardQs   zMSELossReduction.forwardlosses_reduced_per_micro_batchcCstdd|D}|S)aZWorks across micro-batches. (data on single gpu). Note: This currently only works for logging and this loss will not be used for backpropagation. Args: losses_reduced_per_micro_batch: a list of the outputs of forward Returns: A tensor that is the mean of the losses. (used for logging). cSsg|]}|dqS)r>r.).0rFr.r.r/ qz+MSELossReduction.reduce..)torchstackmean)r-rH mse_lossesr.r.r/reducefs zMSELossReduction.reduceN) r1r2r3r4r rr r"rGr rPr.r.r.r/r8Nsr8seqr*cCs"|D] }|dur |Sqtd)z;Returns the first non-None value from the sequence or failsNznon-None value not found) ValueError)rQsr.r.r/ some_firstus rTc Cs(| rdkrtdr#d!r!"}|s$tdtjdr:dd}}||fStjjdrf}zt|}Wnty_}ztd|d}~ww|j|j fSt dr}d}t |}t |St dr} t | }t |S td ) Nrz!Looking up dtype on an empty listz!Looking up dtype on an empty dict)dtypedevicer.z6Cannot get dtype on a torch module with no parameters.)keysvalueszGot something we didnt expect)rRrLrrr7next parameters StopIterationrUrVdictrTget_dtype_devicelist TypeError) torch_objectr;rUrVmperYvallr.r.r/r^}s:$   r^batchescs0 rdkrdtjdrtjddS r9dkr9dtdr8fdddDS r]dkr]dtdr\tfddttdDS rdkrdtdr~fd d ttdDSd urd S rdkrtd  td )aITakes a sequence of batches and collates them into a single batch. This is distinct from the standard pytorch default_collator since it does not add the batch dimension, it's assumed the batch dimension is already present in the input, as would be the case when parallelizing across minibatches. IMPORTANT: The underlying data primitive _must_ be a torch Tensor. The input to this function is a recurisve type, there can be any amount of nesting between dictionaries, tuples, and lists, as long as the inner type is a n-d torch.Tensor. Examples: Outer container = Dict: [{'a': torch.tensor([1]), 'b': torch.tensor([2])}, {'a': torch.tensor([2]), 'b': torch.tensor([3])}] -> {'a': torch.tensor([1, 2]), 'b': torch.tensor([2, 3])} Outer container = List: [[torch.tensor([1]), torch.tensor([2])], [torch.tensor([2]), torch.tensor([3])]] -> [torch.tensor([1, 2]), torch.tensor([2, 3])] Outer container = Tuple: ([torch.tensor([1]), torch.tensor([2])], [torch.tensor([2]), torch.tensor([3])]) -> (torch.tensor([1, 2]), torch.tensor([2, 3])) Args: batches (Optional[Sequence[ReductionT]]): sequence of batches to collate into a single batch. Returns: A single batch of the same type as the elements of your input sequence. rWrr.)dimcs$i|]tfddDqS)cg|]}|qSr.r.rIr9keyr.r/rJrKz-batch_collator...batch_collatorrIrgrkr/ s$z"batch_collator..c3s&|]tfddDVqdS)crir.r.rjir.r/rJrKz,batch_collator...Nrmrorprrr/ s$z!batch_collator..cs"g|] tfddDqS)crir.r.rjrrr.r/rJrKz-batch_collator...rmrorprrr/rJs"z"batch_collator..Nz Cannot process an empty sequencez-Unsupported input structure in batch_collator) rLrcatr]tuplerangelenr_rRrpr.rpr/rns($"""" rnc@sFeZdZdZdededeejeffddZde edefddZ d S) PassthroughLossReductionaInternally in NeMo2.0 the forward step is always expected to return a loss reduction class, and forward is expected to return a loss. This class hijacks that mechanism to instead pass through the forward output unperturbed as the loss (to enable inference in the predict step), and then the reduce method is used to collate the batch of forward outputs into a single batch. This supports the model forward output being a tensor, dict, tuple, or list of tensors. The inner type _must always be a torch.Tensor_. r9r:r*cCs t|\}}tjd||d|fS)as_summary_ Args: batch (DataT): The batch of data that was passed through the model to generate output. forward_out (torch.Tensor): The output from your model's forward pass. Returns: Tuple[torch.Tensor, ReductionT]: A tuple containing the loss tensor (dummy in this case) and the forward output (unmodified). rW)rVrU)r^rLzeros)r-r9r:rUrVr.r.r/rGs z PassthroughLossReduction.forwardcCst|S)a:This overrides the standard reduce with a simplified version that just takes a list of your model's forward outputs and collates them togehter into a single output. Args: forward_out (List[ReductionT]): _description_ Returns: ReductionT: _description_ rm)r-r:r.r.r/rPs zPassthroughLossReduction.reduceN) r1r2r3r4r r rLrrGr rPr.r.r.r/rys  rycseZdZdZfddZddedeedefdd Z ddeefd d Z de fd d Z de fddZ de fddZde fddZdddZZS)LitAutoEncoderzhA very basic lightning module for testing the megatron strategy and the megatron-nemo2-bionemo contract.cs6t||_ttddddd|_|j|dS)zInitializes the model. Args: config: a Config object necessary to construct the actual nn.Module (the thing that has the parameters). g-C6?adamT)lr optimizeruse_distributed_optimizerconfigN)super__init__rr$roptimconnectr-r __class__r.r/rs  zLitAutoEncoder.__init__Nr9 batch_idxr*cCs|d}||S)arThis forward will be called by the megatron scheduler and it will be wrapped. !!! note The `training_step` defines the training loop and is independent of the `forward` method here. Args: batch: A dictionary of data. batch_idx: The index of the batch. Returns: The output of the model. r;)module)r-r9rrCr.r.r/rGs zLitAutoEncoder.forwardcCs |||S)aRThe training step is where the loss is calculated and the backpropagation is done. Background: - NeMo's Strategy overrides this method. - The strategies' training step will call the forward method of the model. - That forward method then calls the wrapped forward step of MegatronParallel which wraps the forward method of the model. - That wrapped forward step is then executed inside the Mcore scheduler, which calls the `_forward_step` method from the MegatronParallel class. - Which then calls the training_step function here. In this particular use case, we simply call the forward method of this class, the lightning module. Args: batch: A dictionary of data. requires `batch_idx` as default None. batch_idx: The index of the batch. r.)r-r9rr.r.r/ training_steps zLitAutoEncoder.training_stepcCtSNr8r,r.r.r/training_loss_reductionz&LitAutoEncoder.training_loss_reductioncCrrrr,r.r.r/validation_loss_reductionz(LitAutoEncoder.validation_loss_reductioncCrrrr,r.r.r/test_loss_reductionrz"LitAutoEncoder.test_loss_reductioncCrr)ryr,r.r.r/predict_loss_reductionrz%LitAutoEncoder.predict_loss_reductioncCs|j|_dSr)rr0rr,r.r.r/r0#szLitAutoEncoder.configure_modelr)r*N)r1r2r3r4rrr intrrGrr!rrrrr0 __classcell__r.r.rr/r{s r{csVeZdZdeddffdd ZdedeeeffddZd e eddfd d Z Z S) r+rr*Ncsdt|tj|_tdd|_t|_ tdd|_ tdd|_ t|_ tdd|_ dS)zConstructor of the model. Args: config: The config object is responsible for telling the strategy what model to create. @N)rrrencoder_or_decoder model_typerLinearlinear1ReLUrelulinear2linear3relu2linear4rrr.r/r(s   zExampleModel.__init__rCcCsX||dd}||}||}||}||}||}||}||dS)zForward pass of the model. Args: x: The input data. Returns: x_hat: The result of the last linear layer of the network. rr=)r<z)r?r@rrrrrr)r-rCrr<r.r.r/rG7s       zExampleModel.forward input_tensorcCsdS)a'This is needed because it is a megatron convention. Even if it is a no-op for single GPU testing. See megatron.model.transformer.set_input_tensor() Note: Currently this is a no-op just to get by an mcore function. Args: input_tensor: Input tensor. Nr.)r-rr.r.r/set_input_tensorIs zExampleModel.set_input_tensor) r1r2r3rrrrstrrGr rrr.r.rr/r+'sr+c@s&eZdZUeed<eed<eed<dS) MnistItemr;labelidxN)r1r2r3rr6rr.r.r.r/rVs  rcs&eZdZdedeffdd ZZS) MNISTCustomindexr*cst|\}}|||dS)aWraps the getitem method of the MNIST dataset such that we return a Dict instead of a Tuple or tensor. Args: index: The index we want to grab, an int. Returns: A dict containing the data ("x"), label ("y"), and index ("idx"). )r;rr)r __getitem__)r-rrCyrr.r/r]s  zMNISTCustom.__getitem__)r1r2r3rrrrr.r.rr/r\src@s@eZdZddZddZdddZddd Zd d Zd d ZdS)LossLoggingCallbackcCsg|_g|_dS)zqLog the loss at the end of each batch. For training do not reduce across the epoch but do so for validation/test.N) val_losses test_lossesr,r.r.r/rrs zLossLoggingCallback.__init__cCsNtjdkr#tr%t|tr|d}|}|jd|ddddddSdSdS)NrrF train_lossT)on_stepprog_barloggerrank_zero_only)rL distributedget_rankris_pipeline_last_stage isinstancer]log)r-trainer pl_modulerDr9rrFr.r.r/on_train_batch_endws  z&LossLoggingCallback.on_train_batch_endrcCDtjdkrtr t|tr|d}|}|j|dSdSdSNrrF) rLrrrrrr]rappendr-rrrDr9rdataloader_idxrFr.r.r/on_test_batch_ends  z%LossLoggingCallback.on_test_batch_endcCrr) rLrrrrrr]rrrr.r.r/on_validation_batch_ends  z+LossLoggingCallback.on_validation_batch_endcCbtjdkr+tr-t|jdkr/t|j}|j d|dddd|j dSdSdSdS)Nrval_lossTrrr) rLrrrrrxrrMrNrclear)r-rr avg_val_lossr.r.r/on_validation_epoch_endz+LossLoggingCallback.on_validation_epoch_endcCr)Nr test_lossTr) rLrrrrrxrrMrNrr)r-rr avg_test_lossr.r.r/on_test_epoch_endrz%LossLoggingCallback.on_test_epoch_endN)r) r1r2r3rrrrrrr.r.r.r/rqs   rcsheZdZddededdffdd Zd eddfd d Zdefd d ZdefddZ defddZ Z S)MNISTDataModule./ data_dir batch_sizer*NcsLt||_||_d|_d|_d|_d|_t|j|j|j|jd|_ dS)Nd)seq_lenmicro_batch_sizeglobal_batch_sizerampup_batch_size) rrrrrrmax_lenrr% data_sampler)r-rrrr.r/rs  zMNISTDataModule.__init__stagecCstt|jdtdd|_t|jdtdd|_t|jdtdd}tjjj |ddgt dd\|_ |_ dS) zeSets up the datasets Args: stage: can be one of train / test / predict. TF)download transformtrainii*) generatorN)rrrToTensor mnist_test mnist_predictrLutilsr; random_split Generator manual_seed mnist_train mnist_val)r-r mnist_fullr.r.r/setups zMNISTDataModule.setupcCt|j|jddSNr)r num_workers)rrrr,r.r.r/train_dataloaderz MNISTDataModule.train_dataloadercCrr)rrrr,r.r.r/val_dataloaderrzMNISTDataModule.val_dataloadercCrr)rrrr,r.r.r/test_dataloaderrzMNISTDataModule.test_dataloader)rr) r1r2r3rrrrrrrrrr.r.rr/rs  rcCs@dtjj_tjtrt tj rtj dSdS)zResets _GLOBAL_NUM_MICROBATCHES_CALCULATOR in megatron which is used in NeMo to initialized model parallel in nemo.collections.nlp.modules.common.megatron.megatron_init.initialize_model_parallel_for_nemo N) megatroncorenum_microbatches_calculator#_GLOBAL_NUM_MICROBATCHES_CALCULATORrLcuda empty_cacheris_initializeddestroy_model_parallelrdestroy_process_groupr.r.r.r/_reset_megatron_parallel_states   rccs$z tdVWtdStw)zIPuts you into a clean parallel state, and again tears it down at the end.N)rr.r.r.r/reset_megatron_parallel_states rGPUcCs0tjt}d|}tj|dtjtjddS)Nzpython T)shellstdoutstderr)ospathabspath__file__ subprocess check_callsysr)rcallr.r.r/Atest_train_mnist_litautoencoder_with_megatron_strategy_single_gpus  r cCsrt)}t|}|sJ|sJthd}tjddddddd}|}||}tt ||d}t t ||||d }t t d }t jddd ddd } t jd d| dddddttgd } t|d} tj|| | |tjdddd| Wdn1swYtt jddd ddtdddddd} t jd d| t |d} |jdd}t|d}t|sJd|dtt|j| jj }t!| j"|| #|d}t$|%d d!hksJd"||d!j&t'| j(dfksJ|d j&t'| j(d#fks J| Wdn1swYWddSWddS1s2wYdS)$znThis is the actual test that will get run in a subprocess so it does not contaminate the state of other tests.test_experimentTrrWz6{model_name}--{val_loss:.2f}-{step}-{consumed_samples}) save_lastmonitor save_top_kevery_n_train_stepsfilenamealways_save_context)save_dirname)log_dirr tensorboardckptrr)tensor_model_parallel_sizepipeline_model_parallel_sizeddpfind_unused_parametersenable_nemo_ckpt_iogpu) acceleratordevicesstrategylimit_val_batchesval_check_interval max_steps num_nodeslog_every_n_stepsr#)r)resume_if_existsresume_ignore_no_checkpoint)modelr;rrrNrF)rrr output_log)rrrrrr)r!r"r#default_root_dirz.ckptweightsz checkpoint z not found in ) dataloaders ckpt_pathrr<z>We expect forward output from predit_step, not the loss, got: r))tempfileTemporaryDirectoryrexistsis_dirr nl_callbacksModelCheckpointrrrr{r'nlMegatronStrategyTrainerrtrack_iorrrrr AutoResume _teardownr%last_model_pathreplacerlistdirparentr+rrnpredictrsetrXshaperxr) tmpdir_strtmpdirrcheckpoint_callbackroot_dirr tb_logger nemo_loggerr+r#r data_module pred_strategypredict_trainerr2unwrapped_trained_modelforward_outputr.r.r/@run_train_mnist_litautoencoder_with_megatron_strategy_single_gpus        8      <$rQ__main__)crrr r3 contextlibr dataclassesrpathlibrtypingrrrrr r r r r rrlightning.pytorchpytorchpl)megatron.core.num_microbatches_calculatorrpytestrLtorch.distributedlightning.pytorch.loggersr megatron.corerrmegatron.core.optimizerrmegatron.core.transformer.enumsr megatron.core.transformer.modulerrrtorch.utils.datar torchvisionrtorchvision.datasetsrnemorr9nemo.collectionsrnemo.lightningrrr nemo.lightning.megatron_parallelr r!r"nemo.lightning.pytorchr#r7nemo.lightning.pytorch.optimr$nemo.lightning.pytorch.pluginsr% TokenizerTyper&r'r8rTrUrVr^rnryLightningModuleIOMixinConnectorMixinr{r+rrCallbackrLightningDataModulerrrmark run_only_on integration pleasefixmer rQr1r.r.r.r/sp   4             '*)!G/,-   m