o }oiKl @sddlmZddlZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z mZdd lmZmZmZmZmZmZdd lmZdd lmZdd lmZdd lmZddlmZddl m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-m.Z.ddl/m0Z0m1Z1m2Z2m3Z3ddl4m5Z5GdddeeZ6dej dej dej fddZ7dej dej d ed!e8de9e:f d"d#Z;dS)$)AnyN)LightningModule) DictConfig) PeftModel)Tensor) fully_shard) ReplicateShard)ColwiseParallelPrepareModuleInputRowwiseParallelSequenceParallel loss_parallelparallelize_module) DynamicCache)resample) AutoTokenizer) get_pad_id) HFHubMixin)maybe_install_lora)ASRBLEU)BLEU)configure_optimizers is_frozen)fp32_precision)load_pretrained_hfsetup_audio_codecsetup_speech_encoder) AudioSignal LabelsType LengthsType NeuralType)loggingc seZdZdeddffdd ZeddZedefdd Zedefd d Z edefd d Z eddZ edefddZ edefddZ edefddZ d@dedeeeffddZdefddZdedefddZdAdd ZdAd!d"ZdBdAd$d%Zdedefd&d'ZdAd(d)ZdAd*d+Zd,ed-efd.d/Zdejfd0d1Ze 2dCd3ejd4ejd5edeeejffd6d7Z fd8d9Z!d:d;Z"edefdd?Z$Z%S)DDuplexS2SModelcfgreturnNcs(t|tsJdt|dtt|_tj j j _ j j j _tjjdd_tjjjjd}|j_|j_jj_j`tttjfddtjD_tj jj!j"j#j_$%dtj&j'j(j)gj*d d _+d _,dS) NzYou must pass the config to DuplexS2SModel as a Python dict to support hyperparameter serialization in PTL checkpoints (we got: 'type(cfg)=z').T)use_fast)pretrained_weightscs g|] }tjjjjqS)torchnn Embeddingspeech_vocab_size embed_tokens embedding_dim).0_selfr(f/home/ubuntu/.local/lib/python3.10/site-packages/nemo/collections/speechlm2/models/duplex_s2s_model.py Osz+DuplexS2SModel.__init__.._control_codesdeviceF)- isinstancedicttypesuper__init__save_hyperparametersrr$r audio_codecvector_quantizercodebook_size_per_group_codebook_size num_groups_num_codebooksrpretrained_llm tokenizerrr'trainmodelllmlm_headr-rrr)r* ModuleListrangeembed_audio_tokensLinearconfig hidden_sizer, audio_headregister_buffertensor speech_bos_id speech_eos_idspeech_delay_idr7 _use_fsdp_use_tp)r2r$rH __class__r1r3r<0s>        zDuplexS2SModel.__init__cC |jdS)zVReturn the size of the audio codec codebook including extra speech BOS and EOS tokens.rAr1r(r(r3r,_ z DuplexS2SModel.speech_vocab_sizecCs|jS)zBIndicates start of utterance generation (not start of inference!).r\r1r(r(r3rSdszDuplexS2SModel.speech_bos_idcCrZ)z&Indicates end of utterance generation.r\r1r(r(r3rTir]zDuplexS2SModel.speech_eos_idcCrZ)z4Indicates start of inference (the very first frame).r\r1r(r(r3rUnr]zDuplexS2SModel.speech_delay_idcC|jjS)z&Return the size of the text tokenizer.)rE vocab_sizer1r(r(r3text_vocab_sizesszDuplexS2SModel.text_vocab_sizecCr`N)rEbos_idr1r(r(r3 text_bos_idxzDuplexS2SModel.text_bos_idcCr`rc)rEeos_idr1r(r(r3 text_eos_id|rfzDuplexS2SModel.text_eos_idcCs t|jS)a Text pad ID is used as a 'blank' for frames when the model is not speaking and for frames where the model is speaking but has already predicted the entire text channel's content. Example: flow: |---user---||-------assistant--------||-user-| text channel: 0000000000 1xxxxxxx0000000000000002 000000 Where 0 indicates PAD ID, 1 indicates BOS ID, 2 indacates EOS ID, and x indicates tokens corresponding to actual text )rrEr1r(r(r3 text_pad_ids zDuplexS2SModel.text_pad_id input_embedsc Csv|j|||dudd}|jdd\}}||d}||d|||j|j}||d}|dur9|d|d<|S) a\ Implements a fully offline forward pass through the entire model. The flow is the following: |-> |audio_head| -> |audio codes| |source speech + prev target text| -> |llm| -| |-> |lm_head| -> |token ids | NT) inputs_embedspast_key_values use_cache return_dictr_last_hidden_state) text_logits audio_logitsrlcache)rHshaperIrPviewrCr,) r2rjrroutBTrprqansr(r(r3forwards   zDuplexS2SModel.forwardbatchc Cs|j|d|dd\}}|d}|jd|jd}dkrencode transposeclamp_r"warningwherererSrhrTfullrUrW device_meshsizer-rKrCadd_rLr$get)r2rzsource_encodedsource_encoded_lensrdiff target_codestarget_codes_lenstlslbtt input_ids tp_world_size remainder text_inputsr audio_inputsrrjcbidxr(r(r3prepare_inputss          (zDuplexS2SModel.prepare_inputs batch_idxc Csd|jj|jj|jfD] }t|r|q ||}||d}|d}t;t j j j |d dd|d dddd|}t j j j |d  dd |d  dd dd||j}Wdn1sgwY|jj||jj|} |djdd \} } | t |jdur|jjdjdd nd||| | |t j|| | d } |j| dd| S)Nrjrrprr^rsum) reductionrqr_rlr)loss learning_rate text_loss audio_loss batch_sizesequence_length num_frames padding_ratioT)on_step)r preprocessorencoderrHrevalrrrr)r* functional cross_entropyflattenrCr$text_loss_weightaudio_loss_weightrs as_tensor_trainertrainer optimizers param_groupsrfloat32log_dict) r2rzrminputsforward_outputsrrrrrvrwrxr(r(r3 training_step#sJ   (   zDuplexS2SModel.training_stepcCs t|dSrc)rr1r(r(r3on_train_epoch_startJ z#DuplexS2SModel.on_train_epoch_startcCs*|t|jj|_t|_dSrc)rrr$ scoring_asrresetasr_bleurbleur1r(r(r3on_validation_epoch_startMsz(DuplexS2SModel.on_validation_epoch_startvalcCs|j}|D]\}}|j|d|||jdddq |j}|D]\}}|j|d|||jdddq)dS)Nr0T)on_epoch sync_dist)rcomputeitemslogrr7r)r2prefixrkrrr(r(r3on_validation_epoch_endRs & &z&DuplexS2SModel.on_validation_epoch_endc Cs|D]O\}}|dur q||d|d}t$|jj||dt|ddd|dddtjdWdn1sAwY|j j||d|d d qdS) Nr{r| target_textsri"Vi>r)namerefs pred_audiopred_audio_lenstext)rrhyps) roffline_inferencerrupdaterrr)rr)r2rzrr dataset_batchresultsr(r(r3validation_stepZs"zDuplexS2SModel.validation_stepcCs|Src)rr1r(r(r3on_test_epoch_startnz"DuplexS2SModel.on_test_epoch_startcCs |jddS)Ntest)r)rr1r(r(r3on_test_epoch_endqrz DuplexS2SModel.on_test_epoch_endargskwargscOs|j|i|Src)rr2rrr(r(r3 test_steptszDuplexS2SModel.test_stepcCshtjd|j|jd}tjd|jf|j|jd}||}t|jD]}| |j ||d|fq!|S)aK Return the partial embedding corresponding to the start frame of the model. It consists of the sum of text embedding of pad ID, and sum of audio token embeddings corresponding to an all-zero frame. This is consistent with how the model is trained. The returned shape is (1, embedding_dim). )r^)rr7r^.) r)rrir7rArUr-rKrCrrL)r2text_bos audio_bosrjrr(r(r3_get_bos_embeddingws  z!DuplexS2SModel._get_bos_embeddingTr~input_signal_lens decode_audioc Cs`|j||d\}}|j\}}}|jrRtj|g|jd} tjj| tjjj dt | } | |krQ|dd|d|ddf} | d| |d} tj || gdd}n|} ||jdd9}t} tj|| |j|jtjd }tj|| |jtjd }|ddd f|7<||ddddf| d }|d jd dddd f|ddd f<|djd dddd f|ddd f<td|jdD]u}|dd|f||dd|df7<t|jD]}|dd|f|j||dd|d|f7<q||dd||df|dd }|d jd dddd f|dd|f<|djd dddd f|dd|f<q|jrW| |krW|ddd|f}|ddd|f}t|||j|jd|||d}|rt||j}t+t|j j!|"dd|d\}}Wdn 1swYWdn 1swY||d<||d<|S)a Autoregressive prediction. Args: input_signal: a batch of waveforms with shape (B, T) with source sampling rate. input_signal_lens: example lengths as number of samples of shape (B,). decode_audio: bool, whether to decode audio codes to waveform. Returns: A dict with keys: * "text": generated text, de-tokenized to strings, properly skipping text_pad_id; list of length B. * "audio": generated waveform of shape (B, T3) (`decode_audio=True`). * "audio_len" output lengths as number of waveform samples of shape (B,) (when `decode_audio=True`). * "tokens_text": generated text tokens of shape (B, T2). * "tokens_len" output lengths as number of tokens of shape (B,). * "tokens_audio": generated audio codes of shape (B, T2, K) where `K=num_codebooks`. r}r6)opNr^rrr)r7rr)rrrprrqrr)rEpad_id)r tokens_text tokens_audio tokens_lenr_)tokensrrr)#rrsrVr)rRr7 distributed all_reduceReduceOpMAXintitemrepeatrr$rremptyrCrrargmaxrKr-rL tokens_to_strrErireplace_control_speech_codesr5rrr>decoder)r2r~rrrjlengthsrvT_localHT_tensorrw last_framepadrr gen_audiogen_textrxtrgen_audio_codespredicted_audiopredicted_audio_lensr(r(r3rsd  ((.6$(*   z DuplexS2SModel.offline_inferencecs>ttj|i|WddS1swYdSrc)rr;backwardrrXr(r3rs"zDuplexS2SModel.backwardcCst|Src)rr1r(r(r3rrz#DuplexS2SModel.configure_optimizersc Csjtdtdtdddtdtdddtdtdddtdtddd tdtd |jjd gd S) z Return a typing schema for optimal batch size calibration for various sequence lengths using OOMptimizer. r{)rvrwinput)rr: seq_lengthr|)rvrrroutput)rr:rra)clsr)r9r!rr rrErar1r(r(r3oomptimizer_schemas z!DuplexS2SModel.oomptimizer_schemac CsZ|j}|dur dS|j}t|tr|jj}|d}dkrd|_tt ft dfddt d}t ||||j D]_}t ttttt ddt tt dft fdtttt ddd }|j}d D]*}t||}||d krtd |d |d |dt||||qit |||q;|j|jfD]} t | |tt dt dddq|d} dkr+| jdksJd|_d| i} t|j D]\} } t| fi| |j | <qt|jfi| |_t|jD]} t|j| fi| |j| <qt|jfi| |_t|jfi| |_t|jfi| |_t|jfi| |_dSdS)Nrr^T) input_layoutsdesired_input_layoutsuse_local_output)zlayers.0norm)output_layouts)rr ) input_layernormzself_attn.q_projzself_attn.k_projzself_attn.v_projzself_attn.o_projpost_attention_layernormmlpz mlp.gate_projz mlp.up_projz mlp.down_proj) num_headsnum_key_value_headsrOrz attn_layer.=z$ is not divisible by tp_mesh.size()=z:: set a different tensor parallelism size to avoid errors.rF)rr#r! data_parallelmesh) rrHr8r base_modelrGrrWr rr r rlayersr r self_attngetattrr"rsetattrrIrPndimrV enumeraterr-rKrCrLr)r2rrHtp_meshplantransformer_block attn_layerattrrrdp_mesh fsdp_configidxlayerr(r(r3configure_models       zDuplexS2SModel.configure_modelrc)r%N)r)T)&__name__ __module__ __qualname__r9r<propertyr,rrSrTrUrbrerhrirstrryrrrrrrrrrrr)rrboolrrrrr< __classcell__r(r(rXr3r#/sd/    s '      Xr# speech_codes control_codesr%cCs&tt|||ddddf|S)z Replaces control codes (speech BOS, EOS, etc) in `speech_codes` with the first frame which is assumed to consist of 'valid' codes representing silence. Nr^)r)risin)rDrEr(r(r3r cs&r rr rErcCsLg}t||D]\}}|d|}|||k}|||q |Src)zipcpuappend ids_to_text)rr rErrxhyp_idshyp_lenr(r(r3r ks   r )<typingrr) lightningr omegaconfrpeftrrtorch.distributed.fsdprtorch.distributed.tensorrr !torch.distributed.tensor.parallelr r r r rr transformersr-nemo.collections.audio.parts.utils.resamplingr"nemo.collections.common.tokenizersr%nemo.collections.speechlm2.data.utilsr'nemo.collections.speechlm2.parts.hf_hubr%nemo.collections.speechlm2.parts.lorar1nemo.collections.speechlm2.parts.metrics.asr_bleur-nemo.collections.speechlm2.parts.metrics.bleur,nemo.collections.speechlm2.parts.optim_setuprr*nemo.collections.speechlm2.parts.precisionr+nemo.collections.speechlm2.parts.pretrainedrrrnemo.core.neural_typesrrr r! nemo.utilsr"r#r rlistrAr r(r(r(r3s:                 8*