o ίio&@sjddlZddlZddlZddlZddlmZddlmZmZddl m Z ddl m Z m Z mZddlm ZddlmZddlmZmZdd lmZmZmZmZmZddlZdd lmZm Z m!Z!e"Z# d Z$Gd d d eZ%GdddeZ&GdddeZ'GdddeZ(GdddeZ)de*de+de+de+de+f ddZ,de*de+de+de+de+f ddZ- daded e.d!e.d"e*d#ee*f d$d%Z/d&ee*efd'ee*efd(e*d)ee*e0fd*ee*eff d+d,Z1d-d.Z2ded/e(fd0d1Z3d2ed3e+d4e+d5e+d6e+d7e)f d8d9Z4e$fd:e*d;e+dZ5d:e*d?e+d;e+deGdZd[d[Z?d\e?d?e+d;e+fd]d^Z@ dad\e?d7ee)fd_d`ZAdS)bNdeepcopy) dataclassfield)partial)EventProcessQueue)r)Path)EmptyFull)AnyDictIteratorOptional TypedDict) ChatFormat TokenizerArgsbuild_tokenizerz*.chunk.*.jsonlc@s:eZdZUdZeed<eed<eed<eed<eed<dS) JSONLStateaRepresents the current state of a JSON line reader. Attributes: content (Dict): The JSON content of the line. file_path (str): The path to the JSONL file. position (int): The file position after reading the line (in bytes). window (int): The window size used for iteration. offset (int): The offset used for iteration. current_iter (Optional[int]): Number of iterations over the jsonl file (for infinite iteration). file_pathposition block_sizeoffset current_iterN)__name__ __module__ __qualname____doc__str__annotations__intr"r"B/home/ubuntu/.local/lib/python3.10/site-packages/core/data/data.pyrBs   rc@sJeZdZUdZeed<eeefed<eeefed<eeefed<dS)MultiChoiceStateafRepresents the current state of a Multi choice iterator. Attributes: root_dir: path to dataset root directory sources Dict[str, float]: Dict from subdirectory to the weight used for sampling source_states: Dict[str, Any] Dict from source to iterator state rng_state: dict numpy bit generator state used to resume rng root_dirsourcessource_to_state rng_stateN) rrrrrr rfloatr r"r"r"r#r$Us  r$c@s:eZdZUeed<eed<eed<eed<eeed<dS)TokenizerStateit_statenameadd_bosadd_eospathN)rrrr r rboolrr"r"r"r#r*es r*c@s:eZdZUdZeed<eed<eed<eed<eed<dS)PackTokensStateaGRepresents the current state of a packing iterator. Attributes: start_token: int index to start reading from in the current sequence output_seq_len: int Length of sequences to output n_views: dict int Number of views to output. Each view is the same sequence but shifted by 1 from the previous start_tokenr+output_seq_lenn_viewsseq_lenN)rrrrr!r r r"r"r"r#r1ms  r1c@sBeZdZUdZeed<eed<eeefed<eed<eed<dS) PrefetchStateaRepresents the current state of a prefetching iterator. Attributes: prefetch_buffer: numpy array to store prefetched data seq_idx: int index of the current sequence to resume from rng_state: dict numpy bit generator state used to resume rng r+seq_idxr( prefetch_size batch_sizeN) rrrrr r r!rrr"r"r"r#r6}s  r6rrrrrc cs|dks ||kr td|dkr|dnd}t|||||d}t|dA}|||}r^|d7}|d||krPt|||||d}t||fV|}s1WddSWddS1siwYdS)azIterates over a JSON Lines file, yielding a line every `block_size` lines with an offset Example : If block_size = 3, offset = 1, iterator will yield lines 1 4 7 10 ... Example : If block_size = 2, offset = 0, iterator will yield lines 0 2 4 6 ... Args: file_path (str): Path to the JSONL file. position (int): The file position (in bytes) from which to start reading. block_size (int): The number of lines to skip between yields offset (int): The initial number of lines skipped Yields: JSONLState: Represents the state of each line read according to window and offset. rz&JSONL iterator offset value is invalidrrrrrrN) RuntimeErrorropenseekreadlinetelljsonloads) rrrrr current_linestatefileliner"r"r# read_jsonls8   "rHccsFz t|||||}|D] \}}||fVq |d7}d}q|w)zBMakes the block jsonl iterator infinite and updates n_iter counterTr:r)rHclose)rrrrritcontent jsonl_stater"r"r# loop_on_jsonls   rMiteratorr-r.tokenizer_typetokenizer_pathc cs$t||d}|dkrt|}|D]~\}}d|vrc|dks!Jdg} d} d|vr-|d} |dD]+} | ddkr;dnd } | d } | dkrT| rT| | | d| d d q1| | | d q1|| }n d|vsod|vsoJdd|vrudnd}||} |j| ||d}|t|||||dfVqdS)a] Tokenizes text from an iterator of content-state pairs using a specified tokenizer. Parameters: - iterator: An iterable of (content, state) pairs where content is a dict with a 'text' or 'content' key. - tokenizer: Tokenizer object with an `encode` method to convert text to tokens, supporting `add_bos` and `add_eos`. - add_bos (bool): Flag to add a beginning-of-sequence token. - add_eos (bool): Flag to add an end-of-sequence token. Yields: - (tokens, state) pairs, where `tokens` is a list of tokenized text, and `state` is the original state from the iterator. )r,r/llama3 conversationsz7conversations should be tokenized with llama3 tokenizerNimagefromhuman assistantvalue num_chunks)rolerKtyperX)rYrKtextrKz1JSON line must contain either text or content key)r-r.r+r-r.r,r/)rrappendencode_dialog_promptencoder*)rNr-r.rOrP tokenizer chat_formatrKrEdialogfirst_user_media conversationrYr[tokens content_keyr"r"r#tokenizesP       rgsource_to_iteratorr'r%r&r(ccst|}t|}t|}tj}||j_ t |t | } ||j || d} t || \} } i|| | i}t ||||jjd} | | fVq)a Iterates over multiple data sources, selecting sequences based on weighted random choice. Parameters: - source_to_iterator (Dict[str, Iterator]): Dict from source paths to their iterators. - source_to_state (Dict[str, State]): Initial state for each source, allowing state tracking. - root_dir str: Root dir of data sources - sources Dict[str, float]: Dict from subdirectory to the weight used for sampling - rng_state (dict): State of the random number generator for reproducibility. Yields: - Tuple of (seq, multi_choice_state) where `seq` is the next sequence from the chosen source, and `multi_choice_state` includes the current state of all sources and the RNG. This function ensures that sequences are chosen from the provided sources based on the specified weights, maintaining state information for each source and the RNG to allow for reproducible iteration. T)pr%r&r'r()lenlistkeysvaluesnprandom default_rng bit_generatorrEarraysumchoicenextr$)rhr'r%r&r( n_sourcespossible_sourcesweightsrng norm_weights source_choiceseqrEmulti_choice_stater"r"r# choose_sources&    rcCs\d}|dkr|d8}|||d7}|dkst||}||d<|d|=d|vr,|d=|S)aI Calculates the state to resume iteration after the buffer is cleared. This function determines the starting point for resuming iteration by rewinding `n_views` from the `end_token`. It handles cases where the rewind goes beyond the current sequence, adjusting the starting sequence and token index accordingly. rr:r5r2Nr)r2statesseq_to_resume_from resume_stater"r"r#get_empty_buffer_state?s  rempty_buffer_statec csg}g}|d}|d}|d}|d}||d}t|D]\} \} } |} d} | s|t| ks=Jd|dt| |t|}t|t| |}||} || || | }|t|||||d t||ksqJd t||krt|}|jdksJd tj j j ||d d }||d}t ||}||d}t||dksJ||fV|t| krd }d} | }| r+qdS)a Iterates over tokens, packing them into chunks. This function aggregates tokens into a buffer and yields fixed-size chunks with dimensions `(output_seq_len, n_views)`, where each column represents shifted sequences of tokens. It ensures continuity in token sequences across chunks, preventing boundary effects and maintaining consistency regardless of `n_views`. Parameters: - iterator: An iterator that yields pairs of (tokens, state), where tokens is a 1D sequence of tokens and state contains all necessary information to resume iterator from current position. - it_state: State of the iterator currently. - start_token (int): The index of the first token to start reading from for the first sequence. - output_seq_len (int): The length of the output sequences to be generated. - n_views (int): The number of shifted views to include in each output chunk. Yields: - numpy.ndarray: An array of shape `(output_seq_len, n_views)` containing the packed tokens. - PackTokensState: The state required to resume packing tokens from where the last returned chunk. The function handles the complexity of determining the correct state for resuming iteration after the buffer is cleared, ensuring seamless continuation of token sequences. r3r4r2r+r:FzStart token index z bigger than sequence )r2r5r+r3r4zBuffer overflowz#Iterator should return 1D sequencesraxisNT) enumeraterkminextendr]r1rorsndimlib stride_trickssliding_window_viewr)rNrbufferrr3r4r2previous_state buffer_sizeirerE end_tokensample_is_read free_spacer5out rewinded_idxr"r"r# pack_tokens[sd         r data_loaderr9r8r5r4rEccsbdtj||||ftd}tj}|d|j_|d}|dkr&||ks*Jd|d} |d} t||D] } t|\|| <} q8|j |ddt||D] } t|\|| <} qP|} ||d krj| } |jj} t | |d || ||d }||||d | |fVt|D]} t|\|||| <}q||d kr|} |j |dd|d |}q^) a Prepare batch in advance and shuffle them to reduce correlation inside batches (for ex when very long document is encountered). This function aggregates batches into a buffer and yields fixed-size batch size and seqlen with dimensions `(batch_size, seqlen, n_views)`, It uses a prefetch buffer to store batches in advance and shuffles them, the prefetch buffer is similar to `reservoir sampling`, but by block to preserve a smooth, easy and deterministic reloading. To ensure more uniform sequence sampling -> prefetch_size * batch_size * seq_len >> max_document_seqlength. Parameters: - iterator: An iterator that yields pairs of (sequence, state), where is a random sequence sampled from a corpus (as done by pack_tokens for example). - batch_size: The desired batch size. - prefetch_size: The number of batches to prefetch in advance. - seq_len (int): The length of the output sequences to be generated. - n_views (int): The number of shifted views to include in each output chunk. Yields: - numpy.ndarray: An array of shape `(batch_size, seq_len, n_views)` containing the packed tokens. - PrefetchState: The state required to resume prefetched batch. Contains also the internal of iterator. r)dtyper(r7rzAPrefetch state seq_idx should be in 0 <= seq_idx < prefetch_size.r+rTr:r+r7r(r9r8) roonesr!rprqrrrErangervshuffler6copy)rr9r8r5r4rEprefetch_bufferrzr7 _rng_state _it_stater next_it_state_idx pack_stater"r"r#&batch_and_shuffle_prefetched_sequencessJ    "   r dataset_path world_size file_patterncCsddt||D}t|}|dkrtd|d|||kr.||}|d|}n ||dks8Jd|dksCJd||S)NcSsg|]}t|qSr")r).0rir"r"r# sz,find_and_sanitize_chunks..rzNo valid chunks of pattern z in z3World size should be a multiple of number of chunkszNo valid chunks in )r globrkloggerfatal)rrrdataset_chunksn_chunks n_discardr"r"r#find_and_sanitize_chunkssrrankc CsTt|||}|t|}g}|D]}t|D]}|t|d||ddqq||S)aW Distributes the chunk files in a dataset path to each worker. If world_size is smaller than the number of chunks, the extra chunks are discarded. Otherwise, world_size is assumed to be a multiple of number of chunks. In that case there are world_size//nb_chunks workers on each chunk file, reading with different offsets. rr;)rrkrr]r) rrrrrn_ranks_per_chunkrank_to_jsonl_iterator_params chunk_pathrr"r"r#distribute_data_to_ranks   rseedc Csvt}|D] }td|d|d|ttj|||||}|||<qtj |||fj j } t |||| d} | S)NzDistributing data to rank z for dataset z from root rj) dictrinforosr/joinrorprqrrrEr$) r%r&rrrrdata_path_to_jsonl_staterrLmulti_rng_stater~r"r"r#init_choice_state"s* rtokenizer_name image_size patch_size max_num_tilesvision_input_typecCsdt|||||| d}t|| | | | d}td|||dd}tj|d||fjj}t|d|||dS)N)r%r&rrrrr\r)r2r+r3r4r5r:r) rr*r1rorprqrrrEr6)r%r&r9r8r5r4rrrr-r.rrPrrrrrr~tokenizer_staterprefetch_rng_stater"r"r# init_stateAsD rcCsLt}|dD]}|d|}t|d|d|d|d|d||<q|S)Nr&r'rrrrr)rrM) multi_state path_to_itersourcerLr"r"r# setup_sourcesys   rccs|d}|d}|d}t|}t||d|d|d|dd}t||d|d|d |d }t||}t||d |d |d |d|d}|V|D]}|qP|dS)Nr+r'r%r&r()rhr'r%r&r(r-r.r,r/r3r4r9r8)rr5r4r9r8rE)rrrgrrrnrI)rErrrrdata_itrJr"r"r#build_dataloadersF   rqueue stop_eventc Cs|7}|D]#}|s#z |j|ddWn tyYnw|r |r)n qWddSWddS1s=wYdS)z Producer function to fetch data from an iterable dataset and put it into a queue. Incorporates timeout management to avoid hanging on queue.put() when the queue is full. 皙?timeoutN)is_setputr )rriterator_builderrNitemr"r"r# feed_buffers&  "rproducerccsH|jdur z |jdd}|VWn tyYnw|jdustd)z Consumer function to process items from the queue. Handles cases where the queue might be empty by implementing timeouts on queue.get(). NrrzDData loader quit unexpectedly, real error has been raised previously)exitcodegetr r=)rrrr"r"r#consume_buffers     rrc cs t|d}t}tt|||fd}td|t||}z:|VW|| |j dd|j durGtd|j d| n td |j d |j td dS|| |j dd|j dur}td|j d| n td |j d |j td w) zb Context manager to setup and manage asynchronous iteration with producer-consumer model. )maxsize)targetargszAsync dataloader startedg?rNzKilling async data process z ...zAsync data process z exited with code zAsync dataloader cleaned up)r rrrrrstartrsetrIrrpidkill)rrrrrconsumerr"r"r#async_iterators:          rc@seZdZUdZeeed<eedZ e ee fed<dZ e ed<dZe ed<dZe ed <d Ze ed <d Zeed <d Zeed<d Zeed<dZe ed<dZee ed<dZee ed<dZee ed<dZeeed<eedZeed<dS)DataArgsNr%)default_factoryr&r9ir5r4*rTr-r. load_async@r8rrrrr`)rrrr%rrr rrr&rr)r9r!r5r4rr-r0r.rr8rrrrrr`r"r"r"r#rs         rrcCstdid|jd|jd|jd|jd|jd|jd|jd|d |d |jj d |jj d |j d |j d|j d|jd|jd|jS)Nr%r&r5r9r8r4rrrrrPr-r.rrrrr")rr%r&r5r9r8r4rr`r,r/r-r.rrrr)rrrr"r"r#init_dataloader_state_from_argssF     rcCs"tt|}|jrt|j|S|SN)rrrrr8)rrE data_builderr"r"r#build_dataloader_from_args"s  rr)B contextlibrBloggingrrr dataclassesrr functoolsrmultiprocessingrrr multiprocessing.synchronize EventClasspathlibr rr r typingr rrrrnumpyrocore.tokenizerrrr getLoggerrTRAIN_DATA_FILE_PATTERNrr$r*r1r6rr!rHrMr0rgr)rrrrrrrrrcontextmanagerrrrrrrrr"r"r"r#sn    ) 3  >     . L K    ,       8 *