o }oiI @s2ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z ddl mZmZmZmZmZmZddlZddlZddlmZddlmZddlmZeeZdZ dZ!d Z"d d d Z#e$d Z%d Z&dZ'ddZ(GdddeZ)Gddde)Z*GdddZ+de(dfde,fddZ-ddZ.ddZ/   dHde,ded e0fd!d"Z1d#d$Z2d%e3d&ed'e4d(e5d)e3d*e4f d+d,Z6d%e3d-ee3fd.d/Z7dId%e3d&ed0eeed-e3fd1d2Z8d3d4Z9d5d6Z:d7d8Z;d9d:ZZ>d?e,de,d-e,fd@dAZ?dBdCZ@dDdEZAd-ee4eBffdFdGZCdS)JN) lru_cachepartial)AnyCallableListOptionalTypeUnion) TokenizerSpec)Dataset)AppStateiSystem) TEXT_TO_VALUE VALUE_TO_TEXTz\{%-?\s+generation\s+-?%\}z0.2idxcCstj|tjdd}t||kd}|j}|}t|dks)|ddt|kr2|t|dg}t|dkrW|d|ddkrW|dt|dkrW|d|ddksBtj||d}|j ~|S) z Build index of delimiter positions between samples in memmap. Can be provided externally. Returns a 1D array of ints. rdtypemoderr) npmemmapuint8wherertolistlenpopasarray_mmapclose)fn newline_intmdatamidx midx_dtyper+W/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/llm/gpt/data/utils.pybuild_index_from_memdata6s     r-cseZdZdZddddeddfdeedeedeed eed ee d d ee eeege fd ee deeffdd Z ddZ ddZddZddZddZddeefddZZS)_TextMemMapDatasetzO Allow per-line lazy access to multiple text files using numpy memmap. rNT dataset_pathsr' header_linesworkers tokenizerr build_index_fnsort_dataset_pathsindex_mapping_dirc stg_t|tr|g}t|dkrtd|__|_ |_ |_ |_ |r4t j _ tdtjoBtj} | rN| rXtjdkrXt||j |d| rbtsbtj| rttjdkrtt||j |d| r~ts~tjtdt} fddj D} td t| d tjt| d td tfd d| D} | _| _jd_ dS)a Args: dataset_paths: list of JSONL file paths. newline_int: ASCII code to use to interpret newlines in file. header_lines: number of header lines in JSON files. workers: number of workers to use for creating index files. tokenizer: tokenizer to use to convert text to tokens. build_index_fn: a callable build_index_fn(fn, newline_int) -> midx [np.array] that returns the index of newlines in a file fn must be pickleable (to be used in multiprocessing.Pool.map). sort_dataset_paths: whether to sort datasets by paths. index_mapping_dir: directory to save the index mapping to. If None, will write to the same folder as the dataset. r-files_list must contain at leat one file namezBuilding data filesr)r2r4r6zLoading data filescsg|]}|qSr+) load_file).0r&)r6selfr+r, sz/_TextMemMapDataset.__init__..z Time loading  mem-mapped files: secondszComputing global indicescsg|] \}}t|qSr+)r!)r9_r))r1r+r,r;srN)!super__init__mdata_midx_list isinstancestrr! ValueError _newline_int _header_lines _files_list_workerr3_sort_dataset_pathssortedloggerinfotorch distributed is_availableis_initializedget_rankbuild_index_fileslightning_prepare_databarrierr local_ranktimedatetime timedeltarcumsum midx_bins_size) r:r0r'r1r2r3r4r5r6is_distributed start_timerBr[ __class__)r1r6r:r,rAYsb            z_TextMemMapDataset.__init__cCs(|jr|jD] \}}|jqdSdSN)rBr$r%)r:r(r)r+r+r,__del__s  z_TextMemMapDataset.__del__cC|jSra)r\r:r+r+r,__len__z_TextMemMapDataset.__len__c Csl|t|ks |dkrtd|dt|dtj||jdd}|dkr+|j|dnd}|||j}|j|\}}|dkrFd}|d}n ||dd}||}z ||||} Wn'ty} zt d| t d |d |d |d || d } ~ wwz| | } W| Sty} zt d| t d| d|d |d |d | | d } ~ ww)z4 Return a string from binary memmap rzIndex z if out of dataset range with z samplesF)rightrz)Error while fetching sample from memmap: z file_id: z , file_idx: z, i: z, j: NzQError while building data from text, possible issue with sample expected format: zsample: z , file_id: ) r! IndexErrorrdigitizer[rGrB_fetch_sample_from_memmap ExceptionrLerror_build_data_from_text) r:rfile_idbase_idxfile_idxr(r)ijsampleedatar+r+r, __getitem__s8 " (z_TextMemMapDataset.__getitem__cCs|||d}|S)z Fetchs the text sample. Can be overriden by child-classes to support loading of partial samples and alternative decode methods. zutf-8)tobytesdecode)r:r(rqrrtextr+r+r,rjsz,_TextMemMapDataset._fetch_sample_from_memmapcCs"|jdur |j|}|S|}|S)zMAllows child-classes to modify the parsing of raw text, prior to tokenizationN)r3 text_to_ids)r:ryrur+r+r,rms  z(_TextMemMapDataset._build_data_from_textc Cs$td|t||}tj|tjdd}t|rtj|dddd}t||j kr4t d|j dt |d d  }t |}Wd n1sKwYd |vri|d }|j |kritd |j d||dd} t| krt dtdtd| d|||fStd|d|d)z Loads a text file as np.int8. Returns: mdata - memorymap of np.int8 midx - indices pointing to the end-of-line (or end of file) position size - number of lines in file zLoading rr.npyT allow_pickle mmap_modezMissing header, expected z header lines.inforbNr'z$Mismatch in newline_int, expected = z but loaded versionz0.0z+Version mismatch: Please delete existing '.z' files. Expected version = z, but file version = z. File path = zMemory Map for z- is not found, missing one or more of files: z .{.npy,.info})rLrM _index_fnrrr_index_file_existsloadr!rG RuntimeErroropenpicklerFwarningget__idx_version____idx_suffix__rE) r:r&r6idx_fnr(r)fp idx_info_dictr' idx_versionr+r+r,r8s:     z_TextMemMapDataset.load_filera)__name__ __module__ __qualname____doc__r-rrDrintrrboolrArbrervrjrmr8 __classcell__r+r+r_r,r.Ts@  i'  r.csteZdZdZ      ddeedeedeed eed eed d ee d eeffdd Z ddZ Z S)_JSONLMemMapDatasetz4 Memory-mapped iteration over a JSONL file. r/rNTr0r'r1r2r3r r5r6c stj|||||||ddS)a2 Args: dataset_paths: list of JSONL file paths. newline_int: ASCII code to use to interpret newlines in file. header_lines: number of header lines in JSON files. workers: number of workers to use for creating index files. tokenizer: tokenizer to use to convert text to tokens. sort_dataset_paths: whether to sort datasets by paths. index_mapping_dir: directory to save the index mapping to. If None, will write to the same folder as the dataset. )r0r'r1r2r3r5r6N)r@rA)r:r0r'r1r2r3r5r6r_r+r,rA9s z_JSONLMemMapDataset.__init__c CsNzt|}W|Sty&}ztd|td||d}~ww)z8Return a dictionary of data based on a single JSON line.z Exception: z datapoint: N)jsonloadsrkrLrl)r:ryrecordrtr+r+r,rmXs z)_JSONLMemMapDataset._build_data_from_text)r/rNNTN) rrrrrrDrrrrrArmrr+r+r_r,r4s0 rc@seZdZdZ     d deded ed ed ed ed efddZddZdedefddZdefddZ ddZ ddZ dede j fddZdS)!_OnlineSampleMappingax This class replaces NeMo's get_samples_mapping function which pre-computes. It is used to create a sample mapping for certain number of samples, including pseudo-random shuffling. The sampler allows to down, or upsample a given dataset. Shuffling leads to pseudo-random shuffling, where blocks are shuffled, and each block is internally shuffled. @BrrTF dataset_size num_samples block_size cache_maxsizeseedshuffletruncate_to_block_boundaryc Cs\||_||_|dur |n|j|_||_||_||_||_t|j|j|_t |j|j|_|j|j}t t |j|j|_ |jr`|r`|j|jkrP|j|8_|j d8_ |j|8_d}t |j } t |j |j} |rx|| d<d|_nd|_|rt jj|jd} | t |j } | | } | | } | |_| |_t | |_t|dd|j|_dS) a Args: dataset_size (int): Size of the dataset. num_samples (int): Number of samples the dataset should contain. block_size (int): Size of each sample block. This is used to shuffle the samples. None will be replaced with dataset size. cache_maxsize (int): Maximum size of the blocks cache for the get_sample_block function. seed (int): Seed for the random number generator used for shuffling. shuffle (bool): Whether to shuffle the samples. truncate_to_block_boundary (bool): Whether to truncate the last block to the block boundary. NrrrTFr)maxsizetyped)rrrrrrrmaxrequired_samplesminrrceil num_blocksarangefull use_digitizerandom RandomState permutationblock_idx_listblock_size_listrZ block_binsrget_sample_block) r:rrrrrrrlast_block_sizerr local_rngrr+r+r,rAms@     z_OnlineSampleMapping.__init__cCs>d|jd|jd|jd|jd|jd|jd|jdS) Nz!OnlineSampleMapping(dataset_size=z, num_samples=z , block_size=z, cache_maxsize=z, seed=z , shuffle=z, truncate_to_block_boundary=))rrrrrrrrdr+r+r,__str__sz_OnlineSampleMapping.__str__rreturnc st|trP|}|j|j|j}}}t|dur|nd}|jkr%j}t|dur-|nj}|jkr9j}|dur?|nd}fddt|||D}|S|jkrYtd|dkrj|j7}|dkrjtdj rut |j }n|j }|}|j |} || } | ddfS)Nrrcsg|]}|qSr+r+)r9rrdr+r,r;z4_OnlineSampleMapping.__getitem__..zIndex out of range)rCslicestartstopstep handle_indexrrangerhrrrirrr) r:rslcrrr sample_slice block_idx sample_block local_idx sample_idxr+rdr,rvs2        z _OnlineSampleMapping.__getitem__cCrcra)rrdr+r+r,rerfz_OnlineSampleMapping.__len__cCs&|j|j|j|j|j|j|j|jffS)z>> import numpy as np >>> import torch >>> from torch.utils.data import TensorDataset >>> from nemo_chem.data.fasta_dataset import handle_index >>> dataset = TensorDataset(torch.tensor(-np.arange(5))) >>> handle_index(dataset, 1) 1 >>> handle_index(dataset, -2) 3 rrzIndex out of range: )r!rh)datasetrr+r+r,r8s  rcCsddl}tdd|DS)a" This function checks whether it is invoked in lightning's hook "prepare_data", which is run only on rank 0. TextMemMapDataset contains a torch.distributed.barrier operation, so when run inside the single-process hook prepare_data, the barrier operation would hang forever. rNcSs$g|]}|jdkod|jdvqS) prepare_dataprepare_packed_sequence_datar)function code_context)r9framer+r+r,r;bsz*lightning_prepare_data..)inspectanystack)rr+r+r,rTYs rTTsamples_mappingsanity_check_dist_workersc  Cs$ddlm} |s|stdttjjd}|s"ttjjd}| dur2tj | tj |} n|} | d |7} |ttjjdkrM| d |7} |ttjjdkr_| d |7} | d |7} | d |7} | d |7} | d 7} | durt jdkrtj | st|d ddurt|dddurt|td | |jjtjksJ|jjtjksJt jdk}t}td |zddlm}Wn tytdw||j|j|||||||rdnd } tdtj| | ddtd | td t|| rPt jt j !dg}t jj"|| j#dddt jj"|| $d|d%t j&t jj&| 'dksPJ| durtd | t}tj(| ddd} td t|td | j)dt*|d rt*|drt+|| S) z`Get a list that maps a sample index to a starting sentence index, end sentence index, and lengthr)parallel_statez4Need to specify either max_num_samples or num_epochsrNz _{}_indexmapz_{}epz_{}mnsz_{}mslz _{:0.2f}sspz_{}sr{doc_idxsizeszP > WARNING: could not find index map file {}, building the indices on rank 0 ...z, > building samples index mapping for {} ...) helpers_cppzhCould not compile megatron dataset C++ helper functions and therefore cannot import helpers python file.rz% > done building samples index mapingTr}z > saved the index mapping in {}zB > elasped time to build and save samples mapping (seconds): {:4f})with_context_parallel)groupz" > loading indexed mapping from {}rr|z* loaded indexed file in {:3.3f} secondsz total number of samples: {}), megatron.corerrEriinfoint32rint64rpathjoinbasenameformatrNrOrRisfilegetattr#_make_indexed_dataset_compatibilityprintrrrrWrLrMmegatron.core.datasetsr ImportError build_mappingsaverUcuda LongTensor all_reduceget_data_parallel_group!get_pipeline_model_parallel_groupitemget_world_sizeget_tensor_model_parallel_grouprshapehasattr"_deallocate_indexed_dataset_memory)indexed_dataset data_prefix num_epochsmax_num_samplesmax_seq_lengthshort_seq_probrname binary_headr6rrrindexmap_filenameverboser^rcountsr+r+r,_get_samples_mappingis (       rcCs\t|dddust|dddurtdtjt|dtjd|_tjt|tjd|_ |S)zMMake any dataset compatible with IndexedDataset for Megatron samples mapping.rNrz0Dataset already has doc_idx or sizes attributes.rr) rAttributeErrorrrr!rronesrr)rr+r+r,rs rsourcer3name_end_token_idslabel_start_idsspecial_tokensnum_turn_start_tokensc Cst||\}}}} ||} t| } ||} t| } g}g}tt| d| t| s?t d| d| d| |dD]'}|t |d}|t }|t|d}| t|| t|qCdd|dD}| d}|D] }||vsJd q{t | } t| d| <t | } t| ||| ||| ||||| | tk}|d ksJd t| tkd d }| d|}| |d}t| |||dS)a& Given a conversation list. This transform: 1. Add signal '### ' at the beginning each sentence, with end signal ' '; 2. Concatenate conversations together; 3. Tokenize the concatenated conversation; 4. Make a deepcopy as the target. Mask human words with IGNORE_INDEX. Nz`First few tokens of the conversation are not the same as the header tokens. target[:header_len]=z header_tokens= conversationsvaluecSsg|]}|dqS)fromr+)r9sentencer+r+r,r;rz_preprocess..,z!mask role not in the conversationrz mask is emptyrr input_ids loss_mask context_ids answer_ids)'_get_header_conversation_type_mask_rolerzcopydeepcopyr!rNequaltensorrLr PREFIX_STRappendsplitr IGNORE_INDEX _mask_targetsrrrnonzerodict)rr3rrrrheader conversation data_type mask_roler target header_tokens header_lenidstokenized_lenssid1id2tokenized_sentencespeakers split_maskmasklast_ignore_index_posr"r#r+r+r, _preprocesss`              rArcCsjt|tr1|dr%dd|dD}|dr#|dd|dd|S|dr/|d}|S|}|S)aL Input source - HuggingFace AutoTokenizer messages format {"messages": [ {"role": "system","content":""}, {"role": "user","content":""}, {"role": "assistant","content":""} ]} source - can also be conversation format, these are converted to HF messages format Mask and type are ignored. Mask will apply to all non-assistant output tokens. {"conversations": [ {"from": "User","value":""}, {"from": "Assistant","value":"", "mask": "User", "system": "", "type": "TEXT_TO_VALUE"} ]} Output [ {"role": "system","content":""}, {"role": "user","content":""}, {"role": "assistant","content":""} ] rcSs"g|] }|d|ddqS)rrrolecontent)lower)r9convor+r+r,r;Ds"z/_convert_to_openai_messages..systemrrBmessages)rCr/rinsert)rchatr+r+r,_convert_to_openai_messages(s    rK tool_schemasc Cs$t|jds tdt|}d}t|tr|dp|}n|}t|jj du}|jj ||dd|d}|d}|r?|d}nd gt |}|j r[|d |j kr[||j g7}|d g7}d |vrnt ||ddd  d } nt |} |d| } || d} tt|t|t| t| d S) a Preprocess messages to apply chat template and tokenize. Returns a dictionary of tokens Input: source - HuggingFace AutoTokenizer messages format {"messages": [ {"role": "system","content":""}, {"role": "user","content":""}, {"role": "assistant","content":""} ]} source - can also be conversation format, these are converted to HF messages format Mask and type are ignored. Mask will apply to all non-assistant output tokens. {"conversations": [ {"from": "User","value":""}, {"from": "Assistant","value":"", "mask": "User", "system": "", "type": "TEXT_TO_VALUE"} ]} tokenizer - tokenizer to apply chat templates to tool_schemas - Optional tool_schemas to supply to apply_chat_template, these will be superseeded by tools supplied with the message Output { "input_ids": torch.LongTensor(), "mask": torch.BoolTensor(), "context_ids": torch.LongTensor(), "answer_ids": torch.LongTensor(), } * input_ids contain tokenized messages with chat template applied * mask corresponds to tokens of input_ids where 1 represents output tokens for the role `assistant` in both context and answer for multi-turn, and 0 to mask all other tokens, e.g. system, user, and tool calling. * context_ids contain tokenized messages with chat template applied for all messages except assistant's last * answer_ids contain tokenized messages with chat template applied for only the assistant's last generated output apply_chat_templatezQCannot apply chat template with tokenizer that is not a HuggingFace AutoTokenizerNtoolsT)rNtokenize return_dictreturn_assistant_tokens_maskr assistant_masksrrrr)rr3rErKrCr/rGENERATION_REGEXsearch chat_templaterMr!eos_idindexrNr BoolTensor) rr3rLrJrNtemplate_has_generation_kwdtokenized_chatr r?context_end_idxr"r#r+r+r,_chat_preprocessQsB #       r\c Cs| d} | d} t| } t|}|}|jd}tt|||D]\}\}}}|t}|t| || }t|t|}t| |}|dkr~|dkrt||ksSJt|||d}|dkrit ||||<q!|t|}||7}n |dkr~|t| }||krdS|||krt ||d|||ddst d|dkr|dks|durt ||||<nC||vr|dkr|dkrt ||| ||<n,||vr|dkrt ||| ||<n||vr|dkrt ||||<nt ||||<||7}q!dS) aThis function masks the tokens so the loss is computed only on the non-masked role's responses. For 'TEXT_TO_VALUE' type, the loss is computed on the value attributes. Args: target (Tensor): input ids tokenized_lens (List[int]): array of lengths of each turns speakers (List[str]): array of speakers of each turns header_len (int): the system prompt length s_ids (List[Tensor]): array of tokenized ids of each turns tokenizer (TokenizerSpec): tokenizer object mask_role (str): the speaker id to be masked from loss computation. gtype (str): either 'TEXT_TO_VALUE' or 'VALUE_TO_TEXT' name_end_token_ids (int): end of name token ids special_tokens (dict): special tokens used for the chat prompt. label_start_ids (list): list of label start token ids, num_turn_start_tokens (int): number of tokens of the turn_start str turn_start end_of_namerrNrrzAa sentence mismatches the corresponding piece in the conversation) rNr(r enumerateziprzr)r!$_identify_start_index_of_subsequencer,r'rLr)r4r8r=r6s_idsr3r3gtyperrrr TURN_TOKENEND_NAME_SIGNALcur_idxtgt_lenrq tokenized_lenspeakers_idr:r; skip_name_lenlocation newline_loc more_skip_lenr+r+r,r-sP          $  r-cCs|d}|d}d}d|vr"|d}|dur"|tvs"Jd|d|d}|dur8t|dkr8|dt|}|d d }|d t|||}t||d |||}||||fS) N end_of_turnr^type source type z not supportedrGr r?Usersystem_turn_startr)TYPE_INSTRUCTIONr SYSTEM_TOKEN_add_speaker_and_signal)rr END_SIGNALrer2r1r3r0r+r+r,r$s   r$cCs>|d}|d}|d}|d} d} |} t|D]\} } | d} |}|dur8| || || d|| d<nN|dkr[| || |d | vrOt| d ||nd| d|| d<n+|d kr~| || || d|d | vrxt| d ||nd| d<ntd |d | | d7} | |vr| t|d kr| |7} q| S)Nr]ro label_startr^rrrrlabelrrqzF not supported, only 'VALUE_TO_TEXT' and 'TEXT_TO_VALUE' are supportedr)r__response_value_formaterrEr!)r0rr3rcrrdrx LABEL_STARTre BEGIN_SIGNALr1rqr sentence_from role_tokenr+r+r,rwsn     rwcCs6t|tr |||S|durdStdt|d)NrzUnknown label type z, only str type is supported)rCrDrErp)rzry end_signalr+r+r,r{Ts  r{c CsLt|d|ddD]}t||||d|r#|SqdS)afind the location of the small tensor in the large tensor. e.g. small = [1,3], large = [2,3,1,3], returns 2 small = [3,2], large = [2,3,1,3], returns -1 Args: small (tensor): small tensor large (tensor): large tensor rrr)rsizerNr') subsequencesequencerqr+r+r,ra]s racCst||}t|r dStd||||}t|}t|jtjs-t d|jt |t d}td|dtj |d|ddtd |d t |t|d d dS) z&Helper function to build an index fileFzBuilding indexing for fn = z.midx must be an integer array, but got type = )r'rzSaving idx file = r{TrzSaving metadata file = rwb)rrrLrMrr# issubdtyperinteger TypeErrorr/rrrdumpr)r'r4r&r6rr)rur+r+r,rks    rr&cCs|r9|dr |dr|d}|dr|d}|dstj||dt}tjtj|dd|S|dt}|S)aReturn base file name of index files. This returns the base file name associated with specified index files. This base name is the base on top of which suffixes like .npy or .info are added. The parent directory is created if it does not already exist. fn may be specified in multiple ways: 1. file name: data.jsonl, 2. relative path to a file: relative/path/to/data.jsonl, 3. absolute path to a file: /absolute/path/to/data.jsonl. This function returns paths in the pattern of: 1. /path/to/input_mapping_dir/data.jsonl.idx 2. /path/to/input_mapping_dir/relative/path/to/data.jsonl.idx 3. /path/to/input_mapping_dir/absolute/path/to/data.jsonl.idx Args: fn: filename to get base name for. index_mapping_dir: directory to save the index mapping to. If None, will write to the same folder as the dataset. )/..rr.T)exist_ok) startswithlstriprrrrmakedirsdirname)r&r6rr+r+r,rs      rcCs(tj|drtj|drdSdS)z,Helper function to test if index file existsr{rTF)rrexists)rr+r+r,rs rcCsd|_d|_dS)z'Deallocate memory of an IndexedDataset.N)rr)rr+r+r,rs rc Cszddlm}Wnttfytdddlm}Ynwt|tr+||9}|St|t s2J|dks:|dur<|St |}t |t dkr|dkrP|}|St||}|dkrw|dkrwdt |}t d|d |d t |d |d ||kr|}|S|||}|S) z? Reconfigure trainer.limit_val_batches for pretraining r)get_num_microbatcheszCMegatron num_microbatches_calculator not found, using Apex version.gNinfg?zYou requested to check z of the val_dataloader but z * zX < 1. Please increase the `limit_val_batches` argument. Try at least `limit_val_batches=`) )megatron.core.num_microbatches_calculatorrrModuleNotFoundErrorloggingr(apex.transformer.pipeline_parallel.utilsrCrfloatr!rE) limit_batches dataloaderrdl_len_in_micro_batcheslimit_micro_batchesmin_percentager+r+r,_reconfigure_limit_batchessD      r)NNTra)Dr%rXrrmultiprocessingrrrrerW functoolsrrtypingrrrrrr numpyrrN1nemo.collections.common.tokenizers.tokenizer_specr nemo.core.classesr nemo.utilsr getLoggerrrLr)r,rvrucompilerSrrr-r.rrrDrSrrTrrrr/rlistrArKr\r-r$rwr{rarrrrrrr+r+r+r,s      a/6 #!   n  F$)U`9 '