# Base code imported from # https://github.com/state-spaces/mamba from functools import partial from typing import Optional from torch import nn, Tensor, cat from flash_attn.ops.triton.layer_norm import RMSNorm from mamba_ssm.modules.mamba2 import Mamba2 # TODO: this is just the bimamba wrapper, still need RCPS from .caduceus_configuration_caduceus import CaduceusConfig from .caduceus_modeling_caduceus import BiMambaWrapper as Caduceus from .modules_mha import CausalMHA from .modules_mlp import SwiGLU class Mamba2Wrapper(Mamba2): """ Mamba2 wrapper class that has the same inference interface as the CausalMHA class. """ def __init__(self, *args, flops_counter=None, **kwargs): super().__init__(*args, **kwargs) self.flops_counter = flops_counter def forward(self, *args, num_tokens, **kwargs): if self.flops_counter is not None: num_flops = ( 6 * int(num_tokens.sum().item()) * self.d_model * self.expand * self.d_state ) self.flops_counter.add_flops(num_flops) return super().forward(*args, **kwargs) def step(self, hidden_states, inference_params): # Don't use _get_states_from_cache because we want to assert that they exist conv_state, ssm_state = inference_params.key_value_memory_dict[ self.layer_idx ] # init class of Mamba2 accepts layer_idx result, conv_state, ssm_state = super().step( hidden_states, conv_state, ssm_state ) # Update the state cache in-place inference_params.key_value_memory_dict[self.layer_idx][0].copy_(conv_state) inference_params.key_value_memory_dict[self.layer_idx][1].copy_(ssm_state) return result class CaduceusWrapper(Caduceus): """ Mamba2 wrapper class that has the same inference interface as the CausalMHA class. """ def __init__(self, *args, flops_counter=None, **kwargs): cad_cfg = kwargs["config"] try: super().__init__(*args, **kwargs) self.d_model = cad_cfg.d_model self.expand = cad_cfg.layer_cfg.mamba_cfg.ssm_cfg["expand"] self.d_state = cad_cfg.layer_cfg.mamba_cfg.ssm_cfg["d_state"] except TypeError as e: # alternate processing depending on caducues version, TODO: standardize self.d_model = cad_cfg.d_model self.expand = cad_cfg.layer_cfg["mamba_cfg"]["ssm_cfg"]["expand"] self.d_state = cad_cfg.layer_cfg["mamba_cfg"]["ssm_cfg"]["d_state"] super().__init__( d_model=self.d_model, bidirectional=cad_cfg.bidirectional, bidirectional_strategy=cad_cfg.bidirectional_strategy, **cad_cfg.layer_cfg["mamba_cfg"]["ssm_cfg"], ) self.flops_counter = flops_counter def forward(self, *args, num_tokens, **kwargs): if self.flops_counter is not None: num_flops = ( 6 * int(num_tokens.sum().item()) * self.d_model * self.expand * self.d_state ) * 2 # Approximate as just 2 passes of caducues self.flops_counter.add_flops(num_flops) return super().forward(*args, **kwargs) def step(self, hidden_states, inference_params): # Don't use _get_states_from_cache because we want to assert that they exist conv_state, ssm_state = inference_params.key_value_memory_dict[ self.layer_idx ] # init class of Mamba2 accepts layer_idx result, conv_state, ssm_state = super().step( hidden_states, conv_state, ssm_state ) # Update the state cache in-place inference_params.key_value_memory_dict[self.layer_idx][0].copy_(conv_state) inference_params.key_value_memory_dict[self.layer_idx][1].copy_(ssm_state) return result def create_block( arch, d_model, d_intermediate=None, ssm_cfg=dict(), attn_cfg=dict(), norm_epsilon=1e-5, layer_idx=None, residual_in_fp32=True, device=None, dtype=None, flops_counter=None, ): factory_kwargs = {"device": device, "dtype": dtype} # Mixer if arch in ("t", "T"): mixer_cls = partial( CausalMHA, **attn_cfg, **factory_kwargs, layer_idx=layer_idx, flops_counter=flops_counter, ) elif arch in ("m", "M"): mixer_cls = partial( Mamba2Wrapper, **ssm_cfg, **factory_kwargs, layer_idx=layer_idx, flops_counter=flops_counter, ) elif arch in ("c", "C"): caduceus_config = { "bidirectional": True, "bidirectional_strategy": "add", "d_model": d_model, "layer_cfg": {"mamba_cfg": {"version": "v2", "ssm_cfg": ssm_cfg}}, } caduceus_config = CaduceusConfig(**caduceus_config) mixer_cls = partial( CaduceusWrapper, **factory_kwargs, config=caduceus_config, layer_idx=layer_idx, flops_counter=flops_counter, ) else: raise NotImplementedError # MLP if arch in ("T", "M", "C"): mlp_cls = partial( SwiGLU, d_intermediate=d_intermediate, **factory_kwargs, ) elif arch in ("t", "m", "c"): mlp_cls = nn.Identity else: raise NotImplementedError # Normalization norm_cls = partial(RMSNorm, eps=norm_epsilon, **factory_kwargs) if arch in ("c", "C"): block = RCPSBlock( d_model, mixer_cls, mlp_cls, norm_cls=norm_cls, residual_in_fp32=residual_in_fp32, flops_counter=flops_counter, ) else: block = Block( d_model, mixer_cls, mlp_cls, norm_cls=norm_cls, residual_in_fp32=residual_in_fp32, flops_counter=flops_counter, ) return block class Block(nn.Module): def __init__( self, d_model, mixer_cls=None, mlp_cls=None, norm_cls=None, residual_in_fp32=True, flops_counter=None, ): super().__init__() self.residual_in_fp32 = residual_in_fp32 self.norm1 = norm_cls(d_model) self.mixer = mixer_cls(d_model) self.flops_counter = flops_counter if mlp_cls is not nn.Identity: self.norm2 = norm_cls(d_model) self.mlp = mlp_cls(d_model) else: self.mlp = None assert RMSNorm is not None, "Triton is not installed" assert isinstance(self.norm1, RMSNorm), "Only RMSNorm is supported" def forward( self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None, mixer_kwargs=None, num_tokens=None, ): hidden_states, residual = self.norm1( hidden_states, residual=residual, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) if mixer_kwargs is None: mixer_kwargs = {} hidden_states = self.mixer( hidden_states, inference_params=inference_params, num_tokens=num_tokens, **mixer_kwargs, ) if self.mlp is not None: hidden_states, residual = self.norm2( hidden_states, residual=residual, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states = self.mlp(hidden_states) ########################################### # ------------- Update FLOPs --------------# ########################################### if self.flops_counter is not None: # Add FLOPs for mlps, defer mixer FLOPs to inner mixer layers norm_1_flops = 0.0 # ignore norm_2_flops = norm_1_flops # ignore if isinstance(self.mlp, SwiGLU): mlp_flops = ( 2 * int(num_tokens.sum().item()) * (3 * self.mlp.d_model * self.mlp.d_intermediate) ) # in_dim = d_model, out_dim = ffw_dim gate_flops = 5 * int(num_tokens.sum().item()) * self.mlp.d_model self.flops_counter.add_flops( mlp_flops + gate_flops + norm_1_flops + norm_2_flops ) # self.flops_counter.add_flops( # 2 # * int(num_tokens.sum().item()) # * sum([x.numel() for x in self.parameters()]) # ) return hidden_states, residual def allocate_inference_cache(self, batch_size, max_seqlen, dtype=None, **kwargs): return self.mixer.allocate_inference_cache( batch_size, max_seqlen, dtype=dtype, **kwargs ) def step(self, hidden_states, inference_params, residual=None): hidden_states, residual = self.norm1( hidden_states, residual=residual, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states = self.mixer.step(hidden_states, inference_params) if self.mlp is not None: hidden_states, residual = self.norm2( hidden_states, residual=residual, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states = self.mlp(hidden_states) return hidden_states, residual class RCPSBlock(Block): """ RCPS version of a block (taken from Caduceus) """ def __init__( self, d_model, mixer_cls=None, mlp_cls=None, norm_cls=None, residual_in_fp32=True, flops_counter=None, ): super().__init__( d_model, mixer_cls, mlp_cls, norm_cls, residual_in_fp32, flops_counter ) self.norm1 = norm_cls(d_model // 2) self.mixer = mixer_cls(d_model) if mlp_cls is not nn.Identity: self.norm2 = norm_cls(d_model // 2) self.mlp = mlp_cls(d_model) else: self.mlp = None assert RMSNorm is not None, "Triton is not installed" assert isinstance(self.norm1, RMSNorm), "Only RMSNorm is supported" def forward( self, hidden_states: Tensor, residual: Optional[Tensor] = None, inference_params=None, mixer_kwargs=None, num_tokens=None, ): ################################################## # ----------- Split Norm 1 into fwd and rc -------# ################################################## hidden_states_fwd, residual_fwd = self.norm1( hidden_states[..., hidden_states.shape[-1] // 2 :], residual=residual[..., hidden_states.shape[-1] // 2 :] if residual is not None else None, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states_rc, residual_rc = self.norm1( hidden_states[..., : hidden_states.shape[-1] // 2].flip(dims=[-2, -1]), residual=residual[..., : hidden_states.shape[-1] // 2].flip(dims=[-2, -1]) if residual is not None else None, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states = cat( [hidden_states_fwd, hidden_states_rc.flip(dims=[-2, -1])], dim=-1 ) residual = cat([residual_fwd, residual_rc.flip(dims=[-2, -1])], dim=-1) ################################################### # ------------ Run the Mixer ---------------------# ################################################### if mixer_kwargs is None: mixer_kwargs = {} hidden_states = self.mixer( hidden_states, inference_params=inference_params, num_tokens=num_tokens, **mixer_kwargs, ) ################################################## # ----------- Split Norm 2 into fwd and rc ------# ################################################## if self.mlp is not None: hidden_states_fwd, residual_fwd = self.norm2( hidden_states[..., hidden_states.shape[-1] // 2 :], residual=residual[..., hidden_states.shape[-1] // 2 :] if residual is not None else None, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states_rc, residual_rc = self.norm2( hidden_states[..., : hidden_states.shape[-1] // 2].flip(dims=[-2, -1]), residual=residual[..., : hidden_states.shape[-1] // 2].flip( dims=[-2, -1] ) if residual is not None else None, prenorm=True, residual_in_fp32=self.residual_in_fp32, ) hidden_states = cat( [hidden_states_fwd, hidden_states_rc.flip(dims=[-2, -1])], dim=-1 ) residual = cat([residual_fwd, residual_rc.flip(dims=[-2, -1])], dim=-1) hidden_states = self.mlp(hidden_states) ########################################### # ------------- Update FLOPs --------------# ########################################### if self.flops_counter is not None: # Add FLOPs for mlps, defer mixer FLOPs to inner mixer layers norm_1_flops = 0.0 # ignore norm_2_flops = norm_1_flops # ignore # TODO: Account for different types of MLPs? (Always SwiGLU for now, requires larger changes) if isinstance(self.mlp, SwiGLU): mlp_flops = ( 2 * int(num_tokens.sum().item()) * (3 * self.mlp.d_model * self.mlp.d_intermediate) ) # in_dim = d_model, out_dim = ffw_dim gate_flops = 5 * int(num_tokens.sum().item()) * self.mlp.d_model self.flops_counter.add_flops( mlp_flops + gate_flops + norm_1_flops + norm_2_flops ) # self.flops_counter.add_flops( # 2 # * int(num_tokens.sum().item()) # * sum([x.numel() for x in self.parameters()]) # ) return hidden_states, residual def step(self, hidden_states, inference_params, residual=None): raise NotImplementedError("RCPS version of step not implemented")