""" This is an experiment to: * entertain a thought to try and abide by HF's transformers API (to benefit from caching better) * conform to a single embedding (instead of a bunch of them) by folding/unfolding inputs * stop trying to make a mixed AR+NAR model work since it seems lobotomized if I keep trying to enforce both recurrent and parallel inferencing (despite a penalty cost) + I will not cave and go with codebook patterns, not yet. """ from ..config import cfg from ..data import fold_inputs, unfold_outputs import torch import torch.nn.functional as F from torch.nn.utils.rnn import pad_sequence from torch import Tensor from torch.nn import CrossEntropyLoss from torch.utils.checkpoint import checkpoint from torchmetrics.classification import BinaryAccuracy, MulticlassAccuracy, MulticlassPrecision import random import math import logging _logger = logging.getLogger(__name__) from einops import rearrange from tqdm import trange from .arch import * if cfg.model.arch_type not in AVAILABLE_ARCHES: raise ValueError(f"Requesting arch `{cfg.model.arch_type}` but not available") if cfg.model.arch_type in ["mamba","mamba2"]: LlmArchClass = MambaLMHeadModel elif cfg.model.arch_type == "llama": LlmArchClass = LlamaForCausalLM elif cfg.model.arch_type == "retnet": LlmArchClass = RetNetForCausalLM else: raise ValueError(f"Requesting arch `{cfg.model.arch_type}` but not available") class Model(LlmArchClass): def __init__( self, n_text_tokens = 256, n_audio_tokens = 1024, d_model=1024, n_layers=12, n_heads=16, p_dropout=0.1, config = cfg.model, ): self.hyper_config = config hf_attention = config.attention if config is not None else None gradient_checkpointing = config.gradient_checkpointing if config is not None else True # text_tokens + rvq levels + [audio tokens * codebooks] (prom) + [audio tokens * codebooks] (resp) + stop # vocab_size = n_text_tokens + cfg.model.max_levels + (n_audio_tokens * cfg.model.max_levels) + (n_audio_tokens * cfg.model.max_levels) + 1 if hf_attention == "auto": if AVAILABLE_ATTENTIONS: hf_attention = AVAILABLE_ATTENTIONS[0] else: hf_attention = "eager" if hf_attention == "xformers": hf_attention = "mem_efficient" text_start = 0 text_end = text_start + config.text_tokens lang_start = text_end lang_end = lang_start + config.langs rvq_start = lang_end rvq_end = rvq_start + config.resp_levels prom_start = rvq_end prom_end = prom_start + config.audio_tokens * config.resp_levels task_start = prom_end task_end = task_start + config.tasks tone_start = task_end tone_end = tone_start + config.tones resp_start = tone_end resp_end = resp_start + config.audio_tokens * config.resp_levels vocab_size = resp_end if config.arch_type == "llama": super().__init__(config=LlamaConfig( vocab_size=vocab_size, hidden_size=d_model, max_position_embeddings=cfg.dataset.frames_per_second * config.max_levels * 60, # max-length of 60 seconds intermediate_size=d_model*4, num_hidden_layers=n_layers, num_attention_heads=n_heads, attention_dropout=p_dropout, num_key_value_heads=n_heads, sliding_window=cfg.dataset.frames_per_second * config.max_levels * 12, hidden_act="gelu", is_encoder_decoder=False, is_decoder=True, attn_implementation=hf_attention, )) if gradient_checkpointing: self.gradient_checkpointing_enable(gradient_checkpointing_kwargs=dict( use_reentrant=False )) elif config.arch_type == "retnet": super().__init__(config=RetNetConfig( vocab_size=vocab_size, decoder_embed_dim=d_model, decoder_value_embed_dim =d_model * 2, decoder_retention_heads=n_heads, decoder_ffn_embed_dim=d_model * 4, decoder_layers=n_layers, dropout=p_dropout, checkpoint_activations=gradient_checkpointing, activation_fn="gelu", use_layernorm=False, use_biases=False, use_glu=True, #chunkwise_recurrent=self.causal and self.recurrent_chunk_size > 0, #recurrent_chunkwise_size=self.recurrent_chunk_size if self.causal else 0, #no_output_layer=True, #rotary_embedding_base=self.rotary_embedding_base, # 10000 decoder_normalize_before=True, )) elif config.arch_type in ["mamba","mamba2"]: super().__init__(config=MambaConfig( vocab_size=vocab_size, d_model=d_model, n_layer=n_layers*2, d_intermediate=0, # d_model*4, ssm_cfg={"layer": "Mamba2", "use_mem_eff_path": True} if config.arch_type == "mamba2" else {}, rms_norm=True, fused_add_norm=True, residual_in_fp32=False, )) self.backbone.gradient_checkpointing = gradient_checkpointing self.accuracy_metric = None if True else MulticlassAccuracy( vocab_size, top_k=10, average="micro", multidim_average="global", ignore_index=-100, ) def generate( self, *args, **kwargs ): if cfg.model.arch_type in ["mamba","mamba2"]: kwargs["cg"] = True if "attention_mask" in kwargs: kwargs.pop("attention_mask") if "do_sample" in kwargs: kwargs.pop("do_sample") if "min_length" in kwargs: kwargs.pop("min_length") """ if "position_ids" in kwargs: kwargs.pop("position_ids") if "max_new_tokens" in kwargs: kwargs.pop("max_new_tokens") if "max_length" not in kwargs: kwargs["max_length"] = 500 * (self.hyper_config.resp_levels if self.hyper_config.experimental.interleave else 1) if "num_last_tokens" not in kwargs: kwargs["num_last_tokens"] = self.hyper_config.experimental.causal_size """ input_ids = kwargs.pop("input_ids") attention_mask = kwargs.pop("attention_mask", None) position_ids = kwargs.pop("position_ids", None) stop_token = kwargs.pop("eos_token_id", 3) max_steps = kwargs.pop("max_new_tokens", 500) device = input_ids.device batch_size = input_ids.shape[0] sequence_list = [ inputs for inputs in input_ids ] position_list = [ positions for positions in position_ids ] start_positions = [ inputs.shape[0] for inputs in input_ids ] stopped = torch.zeros(batch_size, device=device).bool() config = self.hyper_config state = None disable_tqdm = False causal_size = config.experimental.causal_size # get next in sequence for n in trange(max_steps // max(1, causal_size), desc="AR", disable=disable_tqdm): output = super().forward( input_ids=torch.stack(sequence_list), #attention_mask=attention_mask, #past_key_values=state, #position_ids=torch.stack(position_list), #use_cache=False, #return_dict=False ) logits = output[0] # state = output[1] r = [ logit[-causal_size:].argmax(dim=1) for logit in logits ] # append tokens for i, ri in enumerate(r): if stop_token in ri: stopped[i] = True last_position_id = position_list[i][-1].item() + 1 sequence_list[i] = torch.cat([ sequence_list[i], ri.to(device) ], dim=0) #position_list[i] = torch.cat([ position_list[i], torch.tensor([ last_position_id + _ for _ in range( ri.shape[0] ) ], device=device, dtype=torch.int32) ]) # stop token found stopped |= r == stop_token if stopped.all().item(): break def _prune(l: Tensor, stop = stop_token): indices = (l == stop).nonzero() if len(indices) == 0: return l return l[: indices.min().item()] sequence_list = [ _prune(seq[start_positions[i]:], stop_token) for i, seq in enumerate(sequence_list) ] return torch.stack(sequence_list) """ return super().generate(*args, **kwargs) """ def forward( self, *args, **kwargs, ): config = self.hyper_config if "text_list" in kwargs: text_list = kwargs.pop("text_list", None) proms_list = kwargs.pop("proms_list", None) resps_list = kwargs.pop("resps_list", None) lang_list = kwargs.pop("lang_list", None) tone_list = kwargs.pop("tone_list", None) training = kwargs.pop("training", False) steps = kwargs.pop("steps", 500) batch_size = len(text_list) if training: quant_levels = None if config.experimental.interleave else [ random.randint( 0 if "ar" in config.capabilities else 1, config.max_levels - 1) for _ in range(batch_size) ] input_ids, attention_mask, position_ids = fold_inputs( text_list=text_list, prom_list=proms_list, resp_list=resps_list, targ_list=resps_list, quant_levels=quant_levels, ) target_ids, target_attention_mask, target_position_ids = fold_inputs( text_list=text_list, prom_list=proms_list, resp_list=resps_list, targ_list=resps_list, quant_levels=quant_levels, ignore_index=-100 ) return self.forward( input_ids=input_ids, labels=target_ids, position_ids=position_ids, quant_levels=quant_levels, ) if config.experimental.interleave: input_ids, attention_mask, position_ids = fold_inputs( text_list=text_list, prom_list=proms_list ) output = self.generate( input_ids=input_ids, position_ids=position_ids, attention_mask=attention_mask, eos_token_id=3, do_sample=True, max_new_tokens=steps*config.max_levels, ) return unfold_outputs( output )["resp_list"] resps_list = [ [] for _ in range(batch_size) ] for l in range(config.max_levels): quant_levels = [ l for _ in range(batch_size) ] input_ids, attention_mask, position_ids = fold_inputs(text_list=text_list, prom_list=proms_list, resp_list=resps_list, quant_levels=quant_levels) min_length = 1 for batch in input_ids: min_length = max( min_length, batch.shape[0] + 1 ) # to-do: figure out a way to do one forward pass but sample N tokens to replicate the NAR sample pass output = self.generate( input_ids=input_ids, attention_mask=attention_mask, position_ids=position_ids, eos_token_id=3, do_sample=True, max_new_tokens=steps, ) unfolded = unfold_outputs( output, quant_levels=quant_levels ) if l == 0: steps = 0 for batch, resp in enumerate(unfolded["resp_list"]): length = resp.shape[-1] # store length if l == 0: steps = max( steps, length ) # pad else: resp = resp[:steps] if length < steps: resp = torch.cat([ resp, torch.Tensor([ 0 for _ in range(steps-length) ]).to(resp) ]) resps_list[batch].append( resp ) for i, resp in enumerate( resps_list ): resps_list[i] = torch.stack( resp ).t() return resps_list if config.arch_type in ["mamba","mamba2"]: kwargs.pop("attention_mask", None) labels = kwargs.pop("labels", None) quant_levels = kwargs.pop("quant_levels", None) output = super().forward(*args, **kwargs) logits = output.logits # i HATE the correct way if labels is not None: if quant_levels is None: quant_levels = [0 for _ in range(labels.shape[0])] # predict the next token for AR, else predict in place loss = sum([ F.cross_entropy( logit[:-config.experimental.causal_size, :] if quant_level == 0 or "nar" not in config.capabilities else logit, label[config.experimental.causal_size:] if quant_level == 0 or "nar" not in config.capabilities else label, ignore_index=-100 ) for logit, label, quant_level in zip( logits, labels, quant_levels ) ]) self.loss = dict( nll = loss, ) if self.accuracy_metric is not None: self.stats = dict( acc = (sum([ self.accuracy_metric( logit, target ) for logit, target in zip( logits, labels ) ] ) / len( logits )).item() ) """ if config.loss_factors: sep = 3 # determine specific sections to focus on indices = [ [ idx for idx, token in enumerate( batch ) if token == sep ] for i, batch in enumerate( labels ) ] text_index = 0 resp_index = 1 # 1 includes everything non text, -3 includes pre_resp + resp (ignores prom, probably better to include prom here) labels_text = [ batch[:indices[i][text_index] + 1 ] for i, batch in enumerate( labels ) ] labels_resp = [ batch[indices[i][resp_index] + 1:] for i, batch in enumerate( labels ) ] logits_text = [ batch[:indices[i][text_index] + 1 ] for i, batch in enumerate( logits ) ] logits_resp = [ batch[indices[i][resp_index] + 1:] for i, batch in enumerate( logits ) ] loss_text = sum([ F.cross_entropy( logit[:-1, :], label[1:], ignore_index=-100 ) for logit, label in zip( logits_text, labels_text ) ]) / len(logits_text) * self.hyper_config.loss_factor("text") loss_resp = sum([ F.cross_entropy( logit[:-1, :], label[1:], ignore_index=-100 ) for logit, label in zip( logits_resp, labels_resp ) ]) / len(logits_resp) * self.hyper_config.loss_factor("resp") self.loss = dict( text = loss_text, resp = loss_resp, ) if self.accuracy_metric is not None: self.stats = dict( acc = dict( text = (sum([ self.accuracy_metric( logit, target ) for logit, target in zip( logits_text, labels_text ) ] ) / len( logits_text )).item(), resp = (sum([ self.accuracy_metric( logit, target ) for logit, target in zip( logits_resp, labels_resp ) ] ) / len( logits_resp )).item(), ) ) """ return output def example_usage(): cfg.trainer.backend = "local" cfg.hyperparameters.gradient_accumulation_steps = 1 if cfg.audio_backend == "dac": cfg.sample_rate = 44_100 from functools import partial from einops import repeat from tqdm import tqdm from ..emb.qnt import decode_to_file, unload_model from ..engines import Engine from ..utils import wrapper as ml import numpy as np import re device = "cuda" def tokenize(content): return torch.tensor( cfg.tokenizer.encode(content) ) def _load_quants(path) -> Tensor: qnt = np.load(path, allow_pickle=True)[()] return torch.from_numpy(qnt["codes"].astype(np.int16))[0, :cfg.model.max_levels, :].t().to(torch.int16) qnt = _load_quants(f"./data/qnt.{'dac' if cfg.audio_backend == 'dac' else 'enc'}") text_list = [ tokenize("ˈaɪ wɪl nˌɑːt ˈæsk ɐ sˈɛkənd tˈaɪm").to(device), #tokenize("ˈaɪ wɪl nˌɑːt ˈæsk ɐ sˈɛkənd tˈaɪm").to(device), ] prom_list = [ qnt[:cfg.dataset.frames_per_second, :].to(device), #qnt[:cfg.dataset.frames_per_second, :].to(device), ] resp_list = [ qnt[:, :].to(device), #qnt[cfg.dataset.frames_per_second:, :].to(device), #qnt[:cfg.dataset.frames_per_second, :].to(device), ] text_list = text_list[:1] prom_list = prom_list[:1] resp_list = resp_list[:1] kwargs = {} model = Model(**kwargs).to(device) steps = 50 # 100 if cfg.model.experimental.interleave else 300 optimizer = cfg.hyperparameters.optimizer.lower() if cfg.yaml_path is not None else "prodigy" scheduler = cfg.hyperparameters.scheduler.lower() if cfg.yaml_path is not None else "" learning_rate = cfg.hyperparameters.learning_rate if cfg.yaml_path is not None else None if cfg.optimizations.dadaptation: # do not combine the two if scheduler == "schedulefree": scheduler = "" learning_rate = 1.0 if optimizer == "prodigy": if learning_rate is None: learning_rate = 1.0 optimizer = ml.Prodigy elif optimizer == "adagrad": if learning_rate is None: learning_rate = 1.0e-2 optimizer = ml.Adagrad elif optimizer == "adamw": if learning_rate is None: learning_rate = 1.0e-4 optimizer = ml.AdamW elif optimizer == "sdg": if learning_rate is None: learning_rate = 1.0e-4 optimizer = ml.SGD else: raise ValueError(f"Unrecognized optimizer: {optimizer}") _logger.info(f"Optimizer: {optimizer}\tLearning rate: {learning_rate}") optimizer = optimizer(model.parameters(), lr=learning_rate) if scheduler == "schedulefree": if isinstance(optimizer, ml.AdamW): scheduler = ml.schedulefree.AdamWScheduleFree elif isinstance(optimizer, ml.SGD): scheduler = ml.schedulefree.SGDScheduleFree else: scheduler = None if scheduler is not None: _logger.info(f"Scheduler: {scheduler}") optimizer = scheduler( model.parameters(), lr = learning_rate ) if cfg.optimizations.replace and cfg.optimizations.linear: model = ml.replace_linear( model ) if cfg.optimizations.replace and cfg.optimizations.embedding: model = ml.replace_embedding( model ) engine = Engine(model=model, optimizer=optimizer) """ torch.save( { 'module': model.state_dict() }, f"./data/{cfg.model.arch_type}.pth" ) """ _logger.info(f"{LlmArchClass} parameter count: {sum(p.numel() for p in model.parameters() if p.requires_grad)}") @torch.inference_mode() def sample( name, steps=cfg.model.max_levels*cfg.dataset.frames_per_second*6 ): engine.eval() resp_list = model( text_list=text_list, proms_list=prom_list ) for i, batch in enumerate(resp_list): _ = decode_to_file(batch.to(device=device), f"data/{cfg.model.arch_type}.{cfg.audio_backend}.{i}.{name}.wav", device=device) unload_model() def train(): engine.train() t = trange(steps) for i in t: stats = {"step": i} stats |= engine.traverse(text_list=text_list, proms_list=prom_list, resps_list=resp_list, training=True) stats |= engine.gather_attribute("stats") stats |= {"grad_norm": engine.get_global_grad_norm()} tqdm.write(f"{stats}") """ torch.save( { 'module': model.state_dict() }, f"./data/{cfg.model.arch_type}.pth" ) """ #sample("init", 5) train() sample("final") if __name__ == "__main__": example_usage()