o }oiQ @sJddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z ddl mZddlmZddlmZmZddlmZddlmZddlmZdd lmZdd lmZdd lm Z dd l!m"Z"e"d \Z#Z$ervddlm%Z%ddl&m'Z'ddZ(ddZ)ddZ*Gdddej+eZ,de,de-de-de e-e.fde e-e.ff ddZ/dS)N)Path) TYPE_CHECKINGAnyDictListOptionalTypeUnion)rank_zero_info)EVAL_DATALOADERSTRAIN_DATALOADERS) GPTDataset)MegatronDataset)data)WrappedDataLoader)IOMixin)MegatronDataSampler) safe_importtransformer_engineGPTDatasetConfig) TokenizerSpeccCs0|durdSzt|WdStyYdSw)z#Returns True if string is a number.NFT)float ValueError)sr^/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/llm/gpt/data/pre_training.pyis_number_tryexcept)s rcCsL|ddd}t|dkrdSttt|}t|r"t|s"Jd|dS)z( Check if the paths are zipped. NrFzGot malformatted zipped list)lenlistmapranyall)pathsevenis_numrrris_zipped_list4s r'cCsT|durtdt|tst|tr(t|r|ddd}|D]}t|qdSt|tr:|D]}t|q1dSt|tsHt|t sHtdt |}d}| ret |t j sctdt|ddS|r|t |t j sztdt|ddS|D])}t t||}|stdt|d t |t j stdt|dq~dS) z; Validate the accessibility of the dataset assets. NzExpected path to have a value.rz+Expected path to be of string or Path type.)z.binz.idxz Expected z to be readable.z to exist.)r isinstancetupler r'$validate_dataset_asset_accessibilitydictvaluesstrris_dirosaccessR_OKPermissionErrorexistsFileNotFoundError)r$ppathsufficessuffix file_pathrrrr+Bs@    r+c-seZdZdZdddddddddddddd dd dddefd eeBeeefBd e d e dde de de ee de de de de de de de de dede ede de e de e de e d e e d!df,fd"d# Zd$e d%e d&ee efd'ee effd(d)Zd@d+ed!dfd,d-Zd!efd.d/Zd!efd0d1Zd!efd2d3Zd!efd4d5ZedAd7d8Zd!eeeffd9d:Zd;eeefd!dfdd?ZZ S)BPreTrainingDataModulea} PyTorch Lightning-compatible data module for pre-training GPT-style models. Args: paths (Path | List | Dict[str, List]): Paths of the data distributions. Can be either a single path, a list of paths, or a dictionary. If a single path or a list of paths, the given paths will be used to generate the train, validation and test datasets. If providing a list of paths, the format can be either (1) a list of paths, e.g. ["path/to/dataset_1_prefix", "path/to/dataset_2_prefix"], or (2) a flattened, zipped list of weights and paths, e.g. ["30", "path/to/dataset_1_prefix", "70", "path/to/dataset_2_prefix"] If a dictionary is provided, it is expected to have the following form: { 'train': , 'validation': , 'test': } where each value is either a path or a list of paths as described above. In this case, each split will be generated using the given paths. Note that if limit_val_batches <= 1, we generate the entire validaton dataset, so weights should not be provided for the validation split. seq_length (int): Sequence length. tokenizer (Optional["TokenizerSpec"]): An instance of a TokenizerSpec object. micro_batch_size (int): Batch size per GPU. global_batch_size (int): Global batch size. rampup_batch_size (Optional[List[int]]): Rampup batch size, should be in format of [start_global_batch_size, batch_size_increment, ramup_samples]. num_workers (int): See ``torch.utils.data.DataLoader`` documentation. pin_memory (bool): See ``torch.utils.data.DataLoader`` documentation. persistent_workers (bool): See ``torch.utils.data.DataLoader`` documentation. reset_position_ids (bool): Option to reset the position IDs in the dataset at an interval. Not supported with fused and flash attention. create_attention_mask (bool): Option to enable the attention masks generation. Not supported with fused and flash attention. reset_attention_mask (bool): Option to reset the attention mask from the dataset. Not supported with fused and flash attention. eod_mask_loss (int): Option to enable the EOD mask loss. seed (int): Seed for generating the GPT dataset. split (str): A string of 3 comma-separated integers denoting how much of the distribution to allocate to train, validation, and test sets, respectively. Unused if ``paths`` is a dict. index_mapping_dir (Optional[str]): Path to a directory to write index mapping files. num_dataset_builder_threads (int): The number of threads to use for dataset building. num_train_samples (Optional[int]): The number of samples to use for training, defaults to total train steps times global batch size. num_val_samples (Optional[int]): The number of samples to use for validation, defaults to total validation steps times global batch size. num_test_samples (Optional[int]): The number of samples to use for testing, defaults to total test steps times global batch size. dataset_cls (Optional[Type[MegatronDataset]]): The dataset class to use for the data module. iNTFiz 900,50,50r(r$ seq_length tokenizerrmicro_batch_sizeglobal_batch_sizerampup_batch_size num_workers pin_memorypersistent_workersreset_position_idscreate_attention_maskreset_attention_mask eod_mask_lossseedsplitindex_mapping_dirnum_dataset_builder_threadsnum_train_samplesnum_val_samplesnum_test_samples dataset_clsreturncsptt|tttfs|g}ddlm}||_t |i}t|trE|dur1t d|d||d||d||dg|d<n||\}}t |d krSd}||g|d <||d <||_ ||_||_||_||_||_||_| |_| |_| p|t |_| |_| |_||_||_||_||_d|_||_||_ ||_!dd l"m#}|p|d d|_t$|j|j|j|d|_%dS)Nr)get_blend_from_listzsplit=zP will be ignored since datasets are being created from 3 separate distributions.train validationtestblend_per_splitr(blendrK)get_nmt_tokenizermegatronGPT2BPETokenizer)seq_lenr@rArB)&super__init__r)r r*r,megatron.core.datasets.utilsrSrQr+warningswarnr build_kwargsr>r@rAr?rCrDrErFHAVE_TErGrHrIrJrKrLrMinit_global_steprNrOrP3nemo.collections.nlp.modules.common.tokenizer_utilsrYr data_sampler)selfr$r>r?r@rArBrCrDrErFrGrHrIrJrKrLrMrNrOrPrQrSrbweightsrY __class__rrr^sb             zPreTrainingDataModule.__init__trainer_max_stepstrainer_val_check_intervaltrainer_limit_val_batchestrainer_limit_test_batchesc Cs`ddlm}|}|dksJd|dt||jj}|jdur9|j|ks.Jd|d|j}t||jj}||d|}t||jj} |} t| |jj} |jdurh|j| kseJd | d|j} |jdur}|j| kszJd | d|j} |d kr|d krt|t rd |j vsJdd} || | g} ||j | dd|j d \|_|_|_dS)z% Build the datasets. r)BlendedMegatronDatasetBuilderz max_steps z should be greater than 0Nz3num_train_samples must be greater than or equal to .r(z%num_val_samples must be greater than z&num_test_samples must be greater than gg?rXaWhen using a single data distribution, limit_val_batches <= 1.0 is not supported. If you'd like to run with a fractional value of limit_val_batches, please pass in separate datasets for the train, validation, and test datasets by providing a dictionary of paths, e.g.: paths={ 'train': [PATHS FOR TRAIN], 'validation': [PATHS FOR VALIDATION], 'test' :[PATHS FOR TEST], }cSsdS)NTrrrrr.sz-PreTrainingDataModule.build..)is_built_on_rankconfig)7megatron.core.datasets.blended_megatron_dataset_builderrointrfrArNrOrPr)rrbrQgpt_dataset_configbuild _train_ds_validation_ds_test_ds) rgrkrlrmrnro train_itersrN eval_itersrO test_itersrPtrain_valid_test_num_samplesrrrrwsN          zPreTrainingDataModule.buildstagecCsBt|dr |jdusJd|j|jj|jj|jj|jjddS)z( Setup the data module. trainerNz?Setup should be completed when trainer and config are attached.rkrlrmrn)hasattrrrw max_stepsval_check_intervallimit_val_batcheslimit_test_batches)rgrrrrsetup2s  zPreTrainingDataModule.setupcC|j|jddS)z+ Get the train dataloader. rTmode)_create_dataloaderrxrgrrrtrain_dataloaderRz&PreTrainingDataModule.train_dataloadercCr)z0 Get the validation dataloader. rUr)rryrrrrval_dataloaderXrz$PreTrainingDataModule.val_dataloadercCr)z* Get the test dataloader. rVr)rrzrrrrtest_dataloader^rz%PreTrainingDataModule.test_dataloaderc KsF|jj|_|j|j_td|||j|j|jt|dt j j d|}|S)N collate_fn)rdatasetrCrDrErr) r global_steprdrfrrCrDrEgetattrr dataloaderdefault_collate)rgrrkwargsrrrrrds   z(PreTrainingDataModule._create_dataloaderrc CsBddlm}|d|j|j|j|j|j|j|j|j |j d |j S)z4 Get the GPT dataset configuration. rr) random_seedsequence_lengthr? path_to_cacherFrGrHrIrMNr) "megatron.core.datasets.gpt_datasetrrJr>r?rLrFrGrHrIrMrb)rgrrrrrvrs  z(PreTrainingDataModule.gpt_dataset_configcCs|j|jj|j}d|iS)zCalled when saving a checkpoint, implement to generate and save datamodule state. Returns: A dictionary containing datamodule state. consumed_samples)rfcompute_consumed_samplesrrrd)rgrrrr state_dictsz PreTrainingDataModule.state_dictrc Csvzddlm}Wnttfytdddlm}Ynw|d}||j_||j_ ||ddd|j_ t dd S) zCalled when loading a checkpoint, implement to reload datamodule state given datamodule stat Args: state_dict: the datamodule state returned by ``state_dict``. r)update_num_microbatchesCMegatron num_microbatches_calculator not found, using Apex version.rF)rconsistency_checkr(z*** Loaded DataModule state dict successfully. IGNORE PTL's warning below about the dataloader not being resumable. This is warning is expected because we are handling dataloader resumption manually in NeMo. ***N) )megatron.core.num_microbatches_calculatorr ImportErrorModuleNotFoundErrorloggingwarning(apex.transformer.pipeline_parallel.utilsrfinit_consumed_samplesprev_consumed_samples if_first_stepr )rgrrrrrrload_state_dicts" z%PreTrainingDataModule.load_state_dictc Csddlm}||jj|j|j_||jj|j|j_zddlm}Wnt t fy8t dddl m}Ynw|jj|9_dS)zx Reconfigure trainer.limit_train_batches and trainer.limit_val_batches in terms of num of microbatches. r)_reconfigure_limit_batches)get_num_microbatchesrN)#nemo.collections.llm.gpt.data.utilsrrlimit_train_batchesrxrryrrrrrrrnum_sanity_val_steps)rgrrrrrreconfigure_limit_batchess   z/PreTrainingDataModule.reconfigure_limit_batches)r)rRr)!__name__ __module__ __qualname____doc__r rrrr.rurboolrrr^r rrwrr rr rrrrpropertyrvrrrr __classcell__rrrirr;ks5      S   A   r; datamodulerkrlrmrncCs\ddlm}|rJd|jdddddlm}|d||j||||ddS) a` Builds the index mapping cache for nemo.collections.llm.gpt.data.PreTrainingDataModule. Args: datamodule (PreTrainingDataModule): The pre-training data module to build. trainer_max_steps (int): The max_steps set in your trainer. trainer_val_check_interval (int): The interval at which to perform validation in your trainer. trainer_limit_val_batches (Union[int, float]): The number of validation batches to use in your trainer. trainer_limit_test_batches (Union[int, float]): The number of test batches to use in your trainer. Returns: None rNzHThis function cannot be called inside an existing torch.distributed job.r() world_sizerank)rz Building r)torch.distributed distributedis_initializedinit_process_group nemo.utilsrinforw)rrkrlrmrndistrrrrbuild_pretraining_datamodules   r)0rr0r`pathlibrtypingrrrrrrr lightning.pytorchpytorchpl%lightning.pytorch.utilities.rank_zeror !lightning.pytorch.utilities.typesr r rr 'megatron.core.datasets.megatron_datasetr torch.utilsrnemo.lightning.datarnemo.lightning.io.mixinrnemo.lightning.pytorch.pluginsrnemo.utils.import_utilsr_rcr1nemo.collections.common.tokenizers.tokenizer_specrrr'r+LightningDataModuler;rurrrrrrsH $             )^