o }oi}!@sddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z mZmZmZmZmZddlZddlZddlmZmZddlmZddlmZdd lmZmZdd l m!Z!dd l"m#Z#e rtddlmZdd l$m%Z%d e&defddZ'             d6deddd e(d!e)d"e)d#e(d$ee&d%e&d&e(d'e&d(e(d)ed*d+e&d,e&d-e&dd.f d/d0Z*Gd1d.d.e!Z+Gd2d3d3eZ,Gd4d5d5e,Z-dS)7N) lru_cache)Path)sample) TYPE_CHECKINGAnyDictListLiteralMappingOptionalUnion) DatasetDict load_dataset) TokenizerSpec)CustomRetrievalDataModule)_get_samples_mapping_JSONLMemMapDataset)Dataset)NEMO_DATASETS_CACHE)PackedSequenceSpecsnamereturncCstt|}|jddd|S)zHRetrieve the root path for the dataset. Create the folder if not exists.T)parentsexist_ok)rrmkdir)routputrZ/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/llm/gpt/data/reranker.pyget_dataset_root&s rFTrighttrainfirstquerypos_docneg_docpath tokenizerr seq_lengthadd_bosadd_eosseedindex_mapping_dirtruncation_methodmemmap_workers data_typenum_hard_negativesnegative_sample_strategyrandomr% question_keypos_keyneg_keyReRankerDatasetcKs2tdt|||||||||| | | | | |d|S)z.Create ReRankerDataset for reranking training.) file_pathr*max_seq_lengthr,r-r1r.r/r0r2r3r4r7r8r9Nr)r:str)r)r*r+r,r-r.r/r0r1r2r3r4r7r8r9kwargsrrrcreate_reranker_dataset.s&r?c'@seZdZdZ            d7dededededededededededeededee eefdedede dd ed!ed"ef&d#d$Z d%d&Z d'd(Z d)d*Zd+d,Zd-d.Zd/d0Zd1d2Zed3d4Zd5d6ZdS)8r:a} A dataset class for training reranking models that handles query-document pairs with positive and negative examples. This dataset processes JSONL files containing query-document triplets (query, positive document, negative documents) and prepares them for reranking model training. It supports various tokenization options, sequence length constraints, and negative sampling strategies. The dataset expects each example to contain: - A query/question - One or more positive documents (relevant to the query) - Multiple negative documents (irrelevant to the query) During processing, it: 1. Formats each query-document pair with appropriate separators 2. Applies tokenization with optional BOS/EOS tokens 3. Handles sequence length constraints through truncation 4. Samples negative examples based on the specified strategy 5. Prepares attention masks and position IDs for model input The collate function combines multiple examples into batches, handling padding and attention masks appropriately for the reranking task. Args: file_path (str): Path to a JSONL dataset with (query,pos_doc,neg_doc) triplets. tokenizer (TokenizerSpec): Tokenizer for processing text. max_seq_length (int, optional): Maximum sequence length for each example. Defaults to 1024. min_seq_length (int, optional): Minimum sequence length for each example. Defaults to 1. add_bos (bool, optional): Whether to add beginning of sequence token. Defaults to True. add_eos (bool, optional): Whether to add end of sequence token. Defaults to True. max_num_samples (int, optional): Maximum number of samples to load. Defaults to None. seed (int, optional): Random seed for data shuffling. Defaults to 1234. index_mapping_dir (str, optional): Directory to save index mapping. Defaults to None. virtual_tokens (int, optional): Number of virtual tokens to add. Defaults to 0. memmap_workers (Optional[int], optional): Number of workers for memmap loading. Defaults to None. truncation_method (str, optional): Truncation method ('left' or 'right'). Defaults to 'right'. special_tokens (Optional[Mapping[str, str]], optional): Special tokens for formatting. Defaults to None. data_type (str, optional): Type of data ('train', 'query', or 'doc'). Defaults to 'train'. num_hard_negatives (int, optional): Number of negative examples to use. Defaults to 4. negative_sample_strategy (Literal["random", "first"], optional): Strategy for sampling negatives. Defaults to 'first'. question_key (str, optional): Key for question in input data. Defaults to 'question'. pos_key (str, optional): Key for positive document in input data. Defaults to 'pos_doc'. neg_key (str, optional): Key for negative documents in input data. Defaults to 'neg_doc'. TNr rr!r#r$r%questionr'r(r;r*r<min_seq_lengthr,r-max_num_samplesr.r/virtual_tokensr1r0special_tokensr2r3r4r5r7r8r9cCs||_||_||_||_||_||_||_||_| |_| |_ | |_ |jj r)|jj n|jj |_ ||_| dks=| dks=Jd|dksI|dksIJd| durWdd d d d d |_n| |_||_||_||_||_||_|d rtd|t|d }t|}Wdn1swYt|d dd}|D] }|t|d qWdn1swY|d d}t|gdd| | d|_d|_ |!t"d|jd|jd|jd|jd|jd|j d|jd|jddS)a~ file_path: Path to a JSONL dataset with (query,pos_doc,neg_doc) triplets in jsonl format. tokenizer: Tokenizer for the dataset. Instance of a class that inherits TokenizerSpec. max_seq_length (int): maximum sequence length for each dataset examples. Examples will either be truncated to fit this length or dropped if they cannot be truncated. min_seq_length (int): min length of each data example in the dataset. Data examples will be dropped if they do not meet the min length requirements. add_bos (bool): Whether to add a beginning of sentence token to each data example add_eos (bool): Whether to add an end of sentence token to each data example seed: Random seed for data shuffling. max_num_samples: Maximum number of samples to load. This can be > dataset length if you want to oversample data. If None, all samples will be loaded. index_mapping_dir: Directory to save the index mapping to. If None, will write to the same folder as the dataset. truncation_method: Truncation from which position. Options: ['left', 'right'] special_tokens: special tokens for the chat prompts, a dictionary of {token_type: token}. Default: { 'system_turn_start': '', 'turn_start': '', 'label_start': '', 'end_of_turn': ' ', 'end_of_name": ' ' } negative_sample_strategy: Strategy for negative samples. Options: ['random', 'first'] leftr!z2truncation_method must be either "left" or "right"r6r%z;negative_sample_strategy must be either "random" or "first"Nz z z  )system_turn_start turn_start label_start end_of_turn end_of_namez.jsonz$Converting JSON file to JSONL file: r.jsonlwr) dataset_pathsr* header_linesr/workersz#Creating ReRankerDataset with seed=z , add_bos=z , add_eos=z, max_seq_length=z, min_seq_length=z, pad_token_id=z, negative_sample_strategy=z, num_hard_negatives=.)#r*r;r<rCr,r-rDr.r/rEr0pad_ideos_id pad_token_idr4rFr2r3r7r8r9endswithloggingwarningopenjsonloadreplacewritedumpsrindexed_datasetsamples_mapping_build_samples_mappingwarn)selfr;r*r<rCr,r-rDr.r/rEr1r0rFr2r3r4r7r8r9fdataitemrrr__init__s/      zReRankerDataset.__init__c CsR|jdur$t|j|jd|j|jdd|j|jddd|jd |_dSd|_dS)Nr"r/F) ra data_prefix num_epochsrDr<short_seq_probr.r binary_headr/) rDrrar;r<r.splitr/rbrerrrrcs  z&ReRankerDataset._build_samples_mappingcCs,|jdur t|jS|jdusJt|jSN)rDlenrarbrqrrr__len__s   zReRankerDataset.__len__c Cst|tjr |}|jdur*|t|jksJ|j|\}}}t|tjr*|}|dur8|t|jks7Jnd}|dkrGt||}d}nd}z |j|}|rUd|d<Wntyp}zt d|d|j |d}~ww| |S)NrkrTF__AUTOGENERATED__zError while loading example z from dataset ) isinstancenpint64rhrbrsuint32ra ExceptionrYerrorr;_process_example)reidx_ auto_gen_idxexampleerrr __getitem__ s2      zReRankerDataset.__getitem__csdd|D}|j|j}|j}t|tr!|dn|}t|jks.Jdjdkr;t |jd}njdkrH|dj}nt d jt|jks[Jd d d j |}fd d|D}j rj jgj |}fdd|D}jrj jg|}fdd|D}|djd}fdd|D}jr|j jg}fdd|D}|||d}|S)z Create an example by concatenating text and answer. Truncation is carried out when needed, but it is performed only on the prompt side. BOS, EOS, and SEP, are added if specified. cSsi|]\}}||qSrr).0kvrrr /sz4ReRankerDataset._process_example..rz$Error: not enough negative documentsr6)rr%Nz"Invalid negative sample strategy: z3Error in sampling required number of hard negativescSsd|d|S)Nz question:z passage:r)qprrr format_textAsz5ReRankerDataset._process_example..format_textcsg|] }j|qSr)r* text_to_ids)rexrrBrerr Ez4ReRankerDataset._process_example..cs g|] }jjgj|qSr)r*rVrErnrqrrrKs csg|] }jjg|qSr)r*bos_idrrqrrrOrAcsg|] }|djdqSNrA)r<rrqrrrSrcsg|] }|jjgqSr)r*rVrrqrrrWr)positive negativesmetadata)itemsr7r8r9rvlistrsr3r4r ValueErrorr*rrErVr,rr<r-)rerrr'r(rrprocessed_examplerrrr|)s>     z ReRankerDataset._process_examplecCst|tjr dd|DS|S)NcSsg|]}|qSr)tolist)rrhrrrrbsz7ReRankerDataset._maybe_cast_to_list..)rvrwndarray)rexrrr_maybe_cast_to_list`s z#ReRankerDataset._maybe_cast_to_listcCs||d||Srr)rermrrr_ceil_to_nearestesz ReRankerDataset._ceil_to_nearestcsJ||}|j|jdkrfdd|D}|Sfdd|D}|S)NrGcs"g|] }gt||qSrrsrr max_lengthrUrrrl"z1ReRankerDataset._collate_item..cs"g|] }|gt|qSrrrrrrrnr)rrWr0)rerhrrrr _collate_itemhs  zReRankerDataset._collate_itemcCs8t|}|jdkrd|||d<|Sd|d|<|S)ztCreate `attention_mask`. Args: input_ids: A 1D tensor that holds the indices of tokens. rGrAN)torchzerosr0)rer item_lengthattention_maskrrr_create_attention_mask2qs   z'ReRankerDataset._create_attention_mask2c s g}g}g}d|D]@}||d||d|t|d|dD]}|||t|q'tt|dgdd|dDRq tjdjks\Jfdd |D}t|}fd d |D}t|}tj |d }|t ||||d } | S) z0 Collate query passage together rkrrrcss|]}t|VqdSrrr)rndrrr sz-ReRankerDataset.collate_fn..csg|]}|qSr)r)rrsrrerrrz.ReRankerDataset.collate_fn..csg|]}ttqSr)rrange)rr~rrrrrr) input_idstoken_type_idsrr position_ids) appendrsmaxminr<rrstack LongTensorr zeros_like) rebatchrrlengthsrhrrrprocessed_batchrrr collate_fns4  *  zReRankerDataset.collate_fn)r@rATTNr NrNr!Nr#r$r%rBr'r()__name__ __module__ __qualname____doc__r=rintboolr r r rircrtrr|rrrrno_gradrrrrrrr:Vs2      o7  c2seZdZdZ                d,deeeefdeedeedeedeedeede dedde de deee de de de de de d e d!e d"ed#d$ed%ed&ed'ee ee ff.fd(d) Z ed*d+ZZS)-CustomReRankerDataModulea7 A data module for managing reranking datasets that handles data loading, preprocessing, and batching. This module extends CustomRetrievalDataModule to provide specialized functionality for reranking tasks. It manages the creation and organization of training, validation, and test datasets for reranking models, with support for automatic dataset splitting and various data loading configurations. The module can work with either: 1. A single data file that will be automatically split into train/val/test sets 2. Separate files for training, validation, and testing Key features: - Automatic dataset splitting with configurable ratios - Support for both JSON and JSONL file formats - Configurable batch sizes and data loading parameters - Efficient data loading with memory mapping - Support for packed sequence specifications - Customizable data keys for query and document fields Args: data_root (Union[str, List[str]]): Path(s) to the training data file(s) in JSON/JSONL format. val_root (Optional[str]): Path to validation data file. If None, will split from data_root. test_root (Optional[str]): Path to test data file. If None, will split from data_root. val_ratio (Optional[float]): Ratio of data to use for validation when splitting. Defaults to 0.04. test_ratio (Optional[float]): Ratio of data to use for testing when splitting. Defaults to 0.01. dataset_identifier (Optional[str]): Unique identifier for the dataset. If None, generated from data_root. seq_length (int): Maximum sequence length for model input. Defaults to 2048. tokenizer (Optional[TokenizerSpec]): Tokenizer for text processing. Defaults to None. micro_batch_size (int): Batch size for each training step. Defaults to 4. global_batch_size (int): Total batch size across all GPUs. Defaults to 8. rampup_batch_size (Optional[List[int]]): Batch sizes for training rampup. Defaults to None. force_redownload (bool): Whether to force redownload of dataset. Defaults to False. delete_raw (bool): Whether to delete raw data after processing. Defaults to True. seed (int): Random seed for reproducibility. Defaults to 1234. memmap_workers (int): Number of workers for memory-mapped file loading. Defaults to 1. num_workers (int): Number of workers for data loading. Defaults to 8. pin_memory (bool): Whether to pin memory for faster GPU transfer. Defaults to True. persistent_workers (bool): Whether to keep workers alive between epochs. Defaults to False. packed_sequence_specs (Optional[PackedSequenceSpecs]): Specifications for packed sequences. Defaults to None. query_key (str): Key for query field in data. Defaults to "question". pos_doc_key (str): Key for positive document field in data. Defaults to "pos_doc". neg_doc_key (str): Key for negative document field in data. Defaults to "neg_doc". dataset_kwargs (Optional[Dict[str, Any]]): Additional arguments for dataset creation. Defaults to None. N{Gz?{Gz?rr$FTr rArBr'r( data_rootval_root test_root val_ratio test_ratiodataset_identifierr+r*rmicro_batch_sizeglobal_batch_sizerampup_batch_sizeforce_redownload delete_rawr.r1 num_workers pin_memorypersistent_workerspacked_sequence_specsr query_key pos_doc_key neg_doc_keydataset_kwargscs|durtt|}||_tjdid|d|d|d|d|d|d|d |d | d | d | d | d| d|d|d|d|d|d|d|d|d|d|dS)a Custom DataModule for Finetuning reranking Dataset. Args: data_root (Union[str, List[str]]): The JSON/JSONL data file(s) used for training/validation/test. if val_root/test_root is not present, data_root will be split to training and val/test based on val_ratio/test_ratio. val_root (Optional[str]): The JSON/JSONL data file used for validation. If not provided, validation set will be split from data_root. test_root (Optional[str]): The JSON/JSONL data file used for test. If not provided, test set will be split from data_root. val_ratio (Optional[float]): The ratio of validation set when splitting from data_root. test_ratio (Optional[float]): The ratio of test set when splitting from data_root. dataset_identifier (str): Dataset identifier when saving the dataset to NEMO_HOME. seq_length (int, optional): The maximum sequence length for the input and output text. Defaults to 2048. tokenizer (Optional[TokenizerSpec], optional): The tokenizer to use for preprocessing the text. If not provided, a Megatron GPT2 BPE tokenizer will be used. micro_batch_size (int, optional): The micro batch size for training. Defaults to 4. global_batch_size (int, optional): The global batch size for training. Defaults to 8. rampup_batch_size (Optional[List[int]], optional): A list of batch sizes for ramping up during training. Defaults to None. seed (int, optional): The random seed for data shuffling. Defaults to 1234. memmap_workers (int, optional): The number of worker processes for loading data using TextMemMapDataset. Defaults to 1. num_workers (int, optional): The number of worker processes for data loading. Defaults to 8. pin_memory (bool, optional): Whether to pin memory during data loading for faster GPU training. Defaults to True. persistent_workers (bool, optional): Whether to keep data loading workers persistent across epochs. Defaults to False. dataset_kwargs (Optional[Dict[str, Any]], optional): Keyword arguments to pass into the GPTSFTDataset class Nrrrrrrr+r*rrrrrr.r1rrrrrrrrr)hashlibmd5r=encode hexdigestrsuperri)rerrrrrrr+r*rrrrrr.r1rrrrrrrr __class__rrrisd8      z!CustomReRankerDataModule.__init__cKs"t|f|j|j|j|jd|S)N)r*r+r1r.r?r*r+r1r.rer)r>rrr_create_dataset2sz(CustomReRankerDataModule._create_dataset)NNrrNrNr$rNFTr rArTFNrBr'r(N)rrrrr r=rr floatrrrrrirr __classcell__rrrrrs/     Vrc seZdZdZdddddddddd d ddifd ed ed eddededeeededededededededee ee fffdd Z d,ddZ dd Z d-d#ed$ed%efd&d'Zd(d)Zed*d+ZZS).SpecterReRankerDataModuleaA data module for fine-tuning on the Specter dataset. This class inherits from the `CustomReRankerDataModule` class and is specifically designed for fine-tuning models on the SPECTER Datasets. It handles data download, preprocessing, splitting, and preparing the data in a format suitable for training, validation, and testing. Args: force_redownload (bool, optional): Whether to force re-download the dataset even if it exists locally. Defaults to False. delete_raw (bool, optional): Whether to delete the raw downloaded dataset after preprocessing. Defaults to True. See FineTuningDataModule for the other args Nir$rFTr rAr dataset_rootr+r*rrrrrrr.r1rrrrcsp||_||_td|_| |_|tjtddtddtdd|||||| | | | | ddi|ddS)Nspecterztraining.jsonlzvalidation.jsonlz test.jsonlr3rA)rrrr+r*rrrr.r1rrrr)rrrrr. prepare_datarri)rerr+r*rrrrrr.r1rrrrrrrriMs0     z"SpecterReRankerDataModule.__init__rcCs*|jr|jr|}||dSdS)z Prepare dataset for fine-tuning.N) train_pathexistsr_download_data_preprocess_and_split_data)redsetrrrrxsz&SpecterReRankerDataModule.prepare_datacCs:td|jjdtddt|j|jrddSddS)Nz Downloading z...zsentence-transformers/spectertripletr) cache_dir download_mode)rYinforrrr=rrrqrrrrsz(SpecterReRankerDataModule._download_data皙?333333?r train_ratiorc Csdtd|jjdd||}i}|d}|j|||jd}|dj||||jd}|d|d<|d|d<|d|d<|D]G\} }|j| d } | j d d d "} |D]} | t | d | d| dgddq\Wdn1s~wYt| d| qF|j r|jD]} | rt| qd t| jvr| qdSdS)aPreprocesses and splits the downloaded dataset into training, validation, and test sets. Args: dset (DatasetDict): The downloaded dataset object. split_val_from_train (bool, optional): Whether to split the validation set from the training set. If False, the validation set is split from the test set. Defaults to True. val_proportion (float, optional): The proportion of the training or test set to be used for the validation split. Defaults to 0.05. zPreprocessing z! to jsonl format and splitting...rAr#) test_sizer.testtraining validationrOrPzutf-8)encodinganchorrnegative)rBr'r(rHNz split saved to )rYrrrgettrain_test_splitr.rrr[r_r\r`riterdiris_dirshutilrmtreer=runlink)rerrrr save_splitsdataset split_datasetsplit_dataset2 split_name output_filerforrrrrs@      z4SpecterReRankerDataModule._preprocess_and_split_datacCsdS)z?No need to reconfigure trainer.limit_val_batches for finetuningNrrqrrrreconfigure_limit_batchessz3SpecterReRankerDataModule.reconfigure_limit_batchescKs$t|f|j|j|j|jdd|S)NrB)r*r+r1r.r7rrrrrrsz)SpecterReRankerDataModule._create_dataset)rN)rr)rrrrr=rr rrrrrirrr rrrrrrrrrrr>sf       + )r) rFTr Nr!r"r#r$r%r&r'r().rr\rYr functoolsrpathlibrr6rtypingrrrrr r r r numpyrwrdatasetsr r"nemo.collections.common.tokenizersr'nemo.collections.llm.gpt.data.retrievalr#nemo.collections.llm.gpt.data.utilsrrnemo.core.classesrnemo.lightning.baser-nemo.collections.llm.gpt.data.packed_sequencerr=rrrr?r:rrrrrrs    (             ([