mlx-community/LongCat-Flash-Chat-4bit
132
1# -*- coding: utf-8 -*-2# Copyright (c) 2025 Meituan3# This code is licensed under the MIT License, for details, see the ./LICENSE file. 4 5from typing import Callable, Optional, Union6 7import torch8import torch.nn.functional as F9from torch import nn10 11from transformers.activations import ACT2FN12from transformers.cache_utils import Cache, DynamicCache13from transformers.generation import GenerationMixin14from transformers.integrations import use_kernel_forward_from_hub15from transformers.masking_utils import create_causal_mask16from transformers.modeling_flash_attention_utils import FlashAttentionKwargs17from transformers.modeling_layers import GradientCheckpointingLayer18from transformers.modeling_outputs import BaseModelOutputWithPast, CausalLMOutputWithPast19from transformers.modeling_rope_utils import ROPE_INIT_FUNCTIONS, dynamic_rope_update20from transformers.modeling_utils import ALL_ATTENTION_FUNCTIONS, PreTrainedModel21from transformers.processing_utils import Unpack22from transformers.utils import TransformersKwargs, auto_docstring, can_return_tuple23from transformers.utils.generic import check_model_inputs24from .configuration_longcat_flash import LongcatFlashConfig25 26 27@use_kernel_forward_from_hub("RMSNorm")28class LongcatFlashRMSNorm(nn.Module):29 def __init__(self, hidden_size, eps=1e-6):30 """31 LongcatFlashRMSNorm is equivalent to T5LayerNorm32 """33 super().__init__()34 self.weight = nn.Parameter(torch.ones(hidden_size))35 self.variance_epsilon = eps36 37 def forward(self, hidden_states):38 input_dtype = hidden_states.dtype39 hidden_states = hidden_states.to(torch.float32)40 variance = hidden_states.pow(2).mean(-1, keepdim=True)41 hidden_states = hidden_states * torch.rsqrt(variance + self.variance_epsilon)42 return self.weight * hidden_states.to(input_dtype)43 44 def extra_repr(self):45 return f"{tuple(self.weight.shape)}, eps={self.variance_epsilon}"46 47 48class LongcatFlashRotaryEmbedding(nn.Module):49 def __init__(self, config: LongcatFlashConfig, device=None):50 super().__init__()51 # BC: "rope_type" was originally "type"52 if hasattr(config, "rope_scaling") and isinstance(config.rope_scaling, dict):53 self.rope_type = config.rope_scaling.get("rope_type", config.rope_scaling.get("type"))54 else:55 self.rope_type = "default"56 self.max_seq_len_cached = config.max_position_embeddings57 self.original_max_seq_len = config.max_position_embeddings58 59 self.config = config60 self.rope_init_fn = ROPE_INIT_FUNCTIONS[self.rope_type]61 62 inv_freq, self.attention_scaling = self.rope_init_fn(self.config, device)63 self.register_buffer("inv_freq", inv_freq, persistent=False)64 self.original_inv_freq = self.inv_freq65 66 @torch.no_grad()67 @dynamic_rope_update # power user: used with advanced RoPE types (e.g. dynamic rope)68 def forward(self, x, position_ids):69 inv_freq_expanded = self.inv_freq[None, :, None].float().expand(position_ids.shape[0], -1, 1).to(x.device)70 position_ids_expanded = position_ids[:, None, :].float()71 72 device_type = x.device.type if isinstance(x.device.type, str) and x.device.type != "mps" else "cpu"73 with torch.autocast(device_type=device_type, enabled=False): # Force float3274 freqs = (inv_freq_expanded.float() @ position_ids_expanded.float()).transpose(1, 2)75 emb = torch.cat((freqs, freqs), dim=-1)76 cos = emb.cos() * self.attention_scaling77 sin = emb.sin() * self.attention_scaling78 79 return cos.to(dtype=x.dtype), sin.to(dtype=x.dtype)80 81 82class LongcatFlashMLP(nn.Module):83 def __init__(self, config, hidden_size=None, intermediate_size=None):84 super().__init__()85 self.config = config86 self.hidden_size = config.hidden_size if hidden_size is None else hidden_size87 self.intermediate_size = config.ffn_hidden_size if intermediate_size is None else intermediate_size88 89 self.gate_proj = nn.Linear(self.hidden_size, self.intermediate_size, bias=False)90 self.up_proj = nn.Linear(self.hidden_size, self.intermediate_size, bias=False)91 self.down_proj = nn.Linear(self.intermediate_size, self.hidden_size, bias=False)92 self.act_fn = ACT2FN[config.hidden_act]93 94 def forward(self, x):95 down_proj = self.down_proj(self.act_fn(self.gate_proj(x)) * self.up_proj(x))96 return down_proj97 98 99class LongcatFlashTopkRouter(nn.Module):100 def __init__(self, config):101 super().__init__()102 self.config = config103 self.top_k = config.moe_topk104 self.n_routed_experts = (105 config.n_routed_experts106 if config.zero_expert_num is None107 else config.n_routed_experts + config.zero_expert_num108 )109 self.routed_scaling_factor = config.routed_scaling_factor110 self.norm_topk_prob = config.norm_topk_prob111 self.router_bias = config.router_bias112 113 self.classifier = nn.Linear(config.hidden_size, self.n_routed_experts, bias=self.router_bias)114 self.register_buffer("e_score_correction_bias", torch.zeros((self.n_routed_experts)))115 116 @torch.no_grad()117 def get_topk_indices(self, scores):118 scores_for_choice = scores.view(-1, self.n_routed_experts) + self.e_score_correction_bias.unsqueeze(0)119 topk_indices = torch.topk(scores_for_choice, k=self.top_k, dim=-1, sorted=False)[1]120 return topk_indices121 122 def forward(self, hidden_states):123 hidden_states = hidden_states.view(-1, self.config.hidden_size)124 router_logits = F.linear(hidden_states.type(torch.float32), self.classifier.weight.type(torch.float32))125 scores = router_logits.softmax(dim=-1)126 topk_indices = self.get_topk_indices(scores)127 topk_weights = scores.gather(1, topk_indices)128 if self.norm_topk_prob:129 denominator = topk_weights.sum(dim=-1, keepdim=True) + 1e-20130 topk_weights /= denominator131 topk_weights = topk_weights * self.routed_scaling_factor132 return topk_indices, topk_weights133 134 135class LongcatFlashMoE(nn.Module):136 """137 moe module.138 """139 140 def __init__(self, config):141 super().__init__()142 self.config = config143 self.experts = nn.ModuleList(144 [145 LongcatFlashMLP(config, intermediate_size=config.expert_ffn_hidden_size)146 for _ in range(config.n_routed_experts)147 ]148 )149 self.router = LongcatFlashTopkRouter(config)150 self.zero_expert_num = config.zero_expert_num151 self.zero_expert_type = config.zero_expert_type152 153 def moe(self, hidden_states: torch.Tensor, topk_indices: torch.Tensor, topk_weights: torch.Tensor):154 final_hidden_states = torch.zeros_like(hidden_states, dtype=topk_weights.dtype)155 total_experts = len(self.experts) if self.zero_expert_num is None else len(self.experts) + self.zero_expert_num156 157 expert_mask = torch.nn.functional.one_hot(topk_indices, num_classes=total_experts)158 expert_mask = expert_mask.permute(2, 0, 1)159 160 for expert_idx in range(total_experts):161 expert = self.experts[expert_idx] if expert_idx < len(self.experts) else None162 mask = expert_mask[expert_idx]163 token_indices, weight_indices = torch.where(mask)164 165 if token_indices.numel() > 0:166 expert_weights = topk_weights[token_indices, weight_indices]167 expert_input = hidden_states[token_indices]168 169 if self.zero_expert_num is None or expert_idx < len(self.experts):170 expert_output = expert(expert_input)171 elif self.zero_expert_type == "identity":172 expert_output = expert_input173 else:174 raise ValueError("Unknown condition")175 176 weighted_output = expert_output * expert_weights.unsqueeze(-1)177 final_hidden_states.index_add_(0, token_indices, weighted_output)178 179 return final_hidden_states.type(hidden_states.dtype)180 181 def forward(self, hidden_states):182 orig_shape = hidden_states.shape183 topk_indices, topk_weights = self.router(hidden_states)184 hidden_states = hidden_states.view(-1, hidden_states.shape[-1])185 hidden_states = self.moe(hidden_states, topk_indices, topk_weights).view(*orig_shape)186 return hidden_states187 188 189def rotate_half(x):190 """Rotates half the hidden dims of the input."""191 x1 = x[..., : x.shape[-1] // 2]192 x2 = x[..., x.shape[-1] // 2 :]193 return torch.cat((-x2, x1), dim=-1)194 195 196def repeat_kv(hidden_states: torch.Tensor, n_rep: int) -> torch.Tensor:197 """198 This is the equivalent of torch.repeat_interleave(x, dim=1, repeats=n_rep). The hidden states go from (batch,199 num_key_value_heads, seqlen, head_dim) to (batch, num_attention_heads, seqlen, head_dim)200 """201 batch, num_key_value_heads, slen, head_dim = hidden_states.shape202 if n_rep == 1:203 return hidden_states204 hidden_states = hidden_states[:, :, None, :, :].expand(batch, num_key_value_heads, n_rep, slen, head_dim)205 return hidden_states.reshape(batch, num_key_value_heads * n_rep, slen, head_dim)206 207 208def eager_attention_forward(209 module: nn.Module,210 query: torch.Tensor,211 key: torch.Tensor,212 value: torch.Tensor,213 attention_mask: Optional[torch.Tensor],214 scaling: float,215 dropout: float = 0.0,216 **kwargs: Unpack[TransformersKwargs],217):218 key_states = repeat_kv(key, module.num_key_value_groups)219 value_states = repeat_kv(value, module.num_key_value_groups)220 221 attn_weights = torch.matmul(query, key_states.transpose(2, 3)) * scaling222 if attention_mask is not None:223 causal_mask = attention_mask[:, :, :, : key_states.shape[-2]]224 attn_weights = attn_weights + causal_mask225 226 attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query.dtype)227 attn_weights = nn.functional.dropout(attn_weights, p=dropout, training=module.training)228 attn_output = torch.matmul(attn_weights, value_states)229 attn_output = attn_output.transpose(1, 2).contiguous()230 231 return attn_output, attn_weights232 233 234def apply_rotary_pos_emb(q, k, cos, sin, position_ids=None, unsqueeze_dim=1, use_mla=False):235 """Applies Rotary Position Embedding to the query and key tensors.236 237 Args:238 q (`torch.Tensor`): The query tensor.239 k (`torch.Tensor`): The key tensor.240 cos (`torch.Tensor`): The cosine part of the rotary embedding.241 sin (`torch.Tensor`): The sine part of the rotary embedding.242 position_ids (`torch.Tensor`, *optional*):243 Deprecated and unused.244 unsqueeze_dim (`int`, *optional*, defaults to 1):245 The 'unsqueeze_dim' argument specifies the dimension along which to unsqueeze cos[position_ids] and246 sin[position_ids] so that they can be properly broadcasted to the dimensions of q and k. For example, note247 that cos[position_ids] and sin[position_ids] have the shape [batch_size, seq_len, head_dim]. Then, if q and248 k have the shape [batch_size, heads, seq_len, head_dim], then setting unsqueeze_dim=1 makes249 cos[position_ids] and sin[position_ids] broadcastable to the shapes of q and k. Similarly, if q and k have250 the shape [batch_size, seq_len, heads, head_dim], then set unsqueeze_dim=2.251 Returns:252 `tuple(torch.Tensor)` comprising of the query and key tensors rotated using the Rotary Position Embedding.253 """254 cos = cos.unsqueeze(unsqueeze_dim)255 sin = sin.unsqueeze(unsqueeze_dim)256 257 if use_mla:258 b, h, s, d = q.shape259 q = q.view(b, h, s, d // 2, 2).transpose(4, 3).reshape(b, h, s, d)260 261 b, h, s, d = k.shape262 k = k.view(b, h, s, d // 2, 2).transpose(4, 3).reshape(b, h, s, d)263 264 q_embed = (q * cos) + (rotate_half(q) * sin)265 k_embed = (k * cos) + (rotate_half(k) * sin)266 return q_embed, k_embed267 268 269class LongcatFlashMLA(nn.Module):270 """Modified from Deepseek MLA"""271 272 def __init__(self, config: LongcatFlashConfig, layer_idx: int):273 super().__init__()274 self.config = config275 self.layer_idx = layer_idx276 self.num_key_value_groups = config.num_attention_heads // config.num_key_value_heads277 self.attention_dropout = config.attention_dropout278 self.num_heads = config.num_attention_heads279 self.rope_theta = config.rope_theta280 self.q_lora_rank = config.q_lora_rank281 self.qk_rope_head_dim = config.qk_rope_head_dim282 self.kv_lora_rank = config.kv_lora_rank283 self.v_head_dim = config.v_head_dim284 self.qk_nope_head_dim = config.qk_nope_head_dim285 self.qk_head_dim = config.qk_head_dim286 287 self.is_causal = True288 if self.q_lora_rank is None:289 self.q_proj = nn.Linear(config.hidden_size, self.num_heads * self.qk_head_dim, bias=False)290 else:291 self.q_a_proj = nn.Linear(config.hidden_size, config.q_lora_rank, bias=config.attention_bias)292 self.q_a_layernorm = LongcatFlashRMSNorm(config.q_lora_rank)293 self.q_b_proj = nn.Linear(config.q_lora_rank, self.num_heads * self.qk_head_dim, bias=False)294 295 self.kv_a_proj_with_mqa = nn.Linear(296 config.hidden_size,297 self.kv_lora_rank + self.qk_rope_head_dim,298 bias=config.attention_bias,299 )300 self.kv_a_layernorm = LongcatFlashRMSNorm(self.kv_lora_rank)301 self.kv_b_proj = nn.Linear(302 self.kv_lora_rank,303 self.num_heads * (self.qk_nope_head_dim + self.v_head_dim),304 bias=False,305 )306 307 self.o_proj = nn.Linear(308 self.num_heads * self.v_head_dim,309 config.hidden_size,310 bias=config.attention_bias,311 )312 313 if config.mla_scale_q_lora:314 self.mla_scale_q_lora = (config.hidden_size / self.q_lora_rank) ** 0.5315 if config.mla_scale_kv_lora:316 self.mla_scale_kv_lora = (config.hidden_size / self.kv_lora_rank) ** 0.5317 self.scaling = self.qk_head_dim ** (-0.5)318 319 def forward(320 self,321 hidden_states: torch.Tensor,322 position_embeddings: tuple[torch.Tensor, torch.Tensor],323 attention_mask: Optional[torch.Tensor],324 past_key_value: Optional[Cache] = None,325 cache_position: Optional[torch.LongTensor] = None,326 **kwargs: Unpack[FlashAttentionKwargs],327 ) -> tuple[torch.Tensor, Optional[torch.Tensor], Optional[tuple[torch.Tensor]]]:328 batch_size, seq_length = hidden_states.shape[:-1]329 query_shape = (batch_size, seq_length, -1, self.qk_head_dim)330 key_shape = (batch_size, seq_length, -1, self.qk_nope_head_dim + self.v_head_dim)331 332 q_states = self.q_b_proj(self.q_a_layernorm(self.q_a_proj(hidden_states))).view(query_shape).transpose(1, 2)333 q_pass, q_rot = torch.split(q_states, [self.qk_nope_head_dim, self.qk_rope_head_dim], dim=-1)334 335 # apply q_lora scaling336 if self.mla_scale_q_lora is not None:337 q_pass = q_pass * self.mla_scale_q_lora338 q_rot = q_rot * self.mla_scale_q_lora339 340 compressed_kv = self.kv_a_proj_with_mqa(hidden_states)341 k_pass, k_rot = torch.split(compressed_kv, [self.kv_lora_rank, self.qk_rope_head_dim], dim=-1)342 k_pass = self.kv_a_layernorm(k_pass)343 344 # apply kv_lora scaling345 if self.mla_scale_kv_lora is not None:346 k_pass = k_pass * self.mla_scale_kv_lora347 348 k_pass = self.kv_b_proj(k_pass).view(key_shape).transpose(1, 2)349 k_pass, value_states = torch.split(k_pass, [self.qk_nope_head_dim, self.v_head_dim], dim=-1)350 351 k_rot = k_rot.view(batch_size, 1, seq_length, self.qk_rope_head_dim)352 353 cos, sin = position_embeddings354 q_rot, k_rot = apply_rotary_pos_emb(q_rot, k_rot, cos, sin, use_mla=True)355 k_rot = k_rot.expand(*k_pass.shape[:-1], -1)356 357 query_states = torch.cat((q_pass, q_rot), dim=-1)358 key_states = torch.cat((k_pass, k_rot), dim=-1)359 360 if past_key_value is not None:361 cache_kwargs = {"sin": sin, "cos": cos, "cache_position": cache_position}362 key_states, value_states = past_key_value.update(key_states, value_states, self.layer_idx, cache_kwargs)363 364 if self.config._attn_implementation == "flash_attention_2" and self.qk_head_dim != self.v_head_dim:365 value_states = F.pad(value_states, [0, self.qk_head_dim - self.v_head_dim])366 367 attention_interface: Callable = eager_attention_forward368 if self.config._attn_implementation != "eager":369 attention_interface = ALL_ATTENTION_FUNCTIONS[self.config._attn_implementation]370 371 attn_output, attn_weights = attention_interface(372 self,373 query_states,374 key_states,375 value_states,376 attention_mask,377 dropout=0.0 if not self.training else self.attention_dropout,378 scaling=self.scaling,379 **kwargs,380 )381 382 if self.config._attn_implementation == "flash_attention_2" and self.qk_head_dim != self.v_head_dim:383 attn_output = attn_output[:, :, :, : self.v_head_dim]384 385 attn_output = attn_output.reshape(batch_size, seq_length, -1).contiguous()386 attn_output = self.o_proj(attn_output)387 return attn_output, attn_weights388 389 390def create_attention_block(class_name, *args, **kwargs):391 attention_mapping = {"MLA": LongcatFlashMLA}392 393 chosen_class = attention_mapping.get(class_name)394 if not chosen_class:395 raise ValueError(f"No class found for name: {class_name}")396 397 return chosen_class(*args, **kwargs)398 399 400class LongcatFlashDecoderLayer(GradientCheckpointingLayer):401 def __init__(self, config: LongcatFlashConfig, layer_idx: int):402 super().__init__()403 self.layer_idx = layer_idx404 self.hidden_size = config.hidden_size405 self.mlp = LongcatFlashMoE(config)406 407 self_attn = []408 mlps = []409 input_layernorm = []410 post_attention_layernorm = []411 for i in range(2):412 self_attn.append(413 create_attention_block(config.attention_method, config=config, layer_idx=layer_idx * 2 + i)414 )415 mlps.append(LongcatFlashMLP(config))416 input_layernorm.append(LongcatFlashRMSNorm(config.hidden_size, eps=config.rms_norm_eps))417 post_attention_layernorm.append(LongcatFlashRMSNorm(config.hidden_size, eps=config.rms_norm_eps))418 419 self.self_attn = nn.ModuleList(self_attn)420 self.mlps = nn.ModuleList(mlps)421 self.input_layernorm = nn.ModuleList(input_layernorm)422 self.post_attention_layernorm = nn.ModuleList(post_attention_layernorm)423 424 def forward(425 self,426 hidden_states: torch.Tensor,427 attention_mask: Optional[torch.Tensor] = None,428 position_ids: Optional[torch.LongTensor] = None,429 past_key_value: Optional[Cache] = None,430 use_cache: Optional[bool] = False,431 cache_position: Optional[torch.LongTensor] = None,432 position_embeddings: Optional[tuple[torch.Tensor, torch.Tensor]] = None,433 **kwargs: Unpack[FlashAttentionKwargs],434 ) -> tuple[torch.FloatTensor, Optional[tuple[torch.FloatTensor, torch.FloatTensor]]]:435 for i in range(2):436 residual = hidden_states437 438 hidden_states = self.input_layernorm[i](hidden_states)439 440 hidden_states, _ = self.self_attn[i](441 hidden_states=hidden_states,442 attention_mask=attention_mask,443 position_ids=position_ids,444 past_key_value=past_key_value,445 use_cache=use_cache,446 cache_position=cache_position,447 position_embeddings=position_embeddings,448 **kwargs,449 )450 hidden_states = residual + hidden_states451 452 residual = hidden_states453 hidden_states = self.post_attention_layernorm[i](hidden_states)454 455 if i == 0:456 shortcut_mlp_output = self.mlp(hidden_states) # shortcut output (MoE output)457 458 hidden_states = self.mlps[i](hidden_states)459 hidden_states = residual + hidden_states460 if i == 1:461 hidden_states = hidden_states + shortcut_mlp_output462 463 return hidden_states464 465 466@auto_docstring467class LongcatFlashPreTrainedModel(PreTrainedModel):468 config: LongcatFlashConfig469 base_model_prefix = "model"470 supports_gradient_checkpointing = True471 _no_split_modules = ["LongcatFlashDecoderLayer"]472 _skip_keys_device_placement = ["past_key_values"]473 _supports_flash_attn = True474 _supports_sdpa = True475 _supports_flex_attn = True476 _can_compile_fullgraph = True477 _supports_attention_backend = True478 _can_record_outputs = {479 "hidden_states": LongcatFlashDecoderLayer,480 "attentions": LongcatFlashMLA,481 }482 483 484@auto_docstring485class LongcatFlashModel(LongcatFlashPreTrainedModel):486 _keys_to_ignore_on_load_unexpected = [r"model\.mtp.*"]487 488 def __init__(self, config: LongcatFlashConfig):489 super().__init__(config)490 self.padding_idx = config.pad_token_id491 self.vocab_size = config.vocab_size492 493 self.embed_tokens = nn.Embedding(config.vocab_size, config.hidden_size, self.padding_idx)494 self.layers = nn.ModuleList(495 [LongcatFlashDecoderLayer(config, layer_idx) for layer_idx in range(config.num_hidden_layers)]496 )497 self.norm = LongcatFlashRMSNorm(config.hidden_size, eps=config.rms_norm_eps)498 self.rotary_emb = LongcatFlashRotaryEmbedding(config=config)499 self.gradient_checkpointing = False500 501 # Initialize weights and apply final processing502 self.post_init()503 504 @check_model_inputs505 @auto_docstring506 def forward(507 self,508 input_ids: Optional[torch.LongTensor] = None,509 attention_mask: Optional[torch.Tensor] = None,510 position_ids: Optional[torch.LongTensor] = None,511 past_key_values: Optional[Cache] = None,512 inputs_embeds: Optional[torch.FloatTensor] = None,513 cache_position: Optional[torch.LongTensor] = None,514 use_cache: Optional[bool] = None,515 **kwargs: Unpack[TransformersKwargs],516 ) -> BaseModelOutputWithPast:517 if (input_ids is None) ^ (inputs_embeds is not None):518 raise ValueError("You must specify exactly one of input_ids or inputs_embeds")519 520 if inputs_embeds is None:521 inputs_embeds: torch.Tensor = self.embed_tokens(input_ids)522 523 if use_cache and past_key_values is None:524 past_key_values = DynamicCache()525 526 if cache_position is None:527 past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0528 cache_position: torch.Tensor = torch.arange(529 past_seen_tokens, past_seen_tokens + inputs_embeds.shape[1], device=inputs_embeds.device530 )531 532 if position_ids is None:533 position_ids = cache_position.unsqueeze(0)534 535 causal_mask = create_causal_mask(536 config=self.config,537 input_embeds=inputs_embeds,538 attention_mask=attention_mask,539 cache_position=cache_position,540 past_key_values=past_key_values,541 position_ids=position_ids,542 )543 544 hidden_states = inputs_embeds545 position_embeddings = self.rotary_emb(hidden_states, position_ids)546 547 for decoder_layer in self.layers[: self.config.num_hidden_layers]:548 hidden_states = decoder_layer(549 hidden_states,550 attention_mask=causal_mask,551 position_ids=position_ids,552 past_key_value=past_key_values,553 cache_position=cache_position,554 position_embeddings=position_embeddings,555 **kwargs,556 )557 558 hidden_states = self.norm(hidden_states)559 return BaseModelOutputWithPast(560 last_hidden_state=hidden_states,561 past_key_values=past_key_values,562 )563 564 565@auto_docstring566class LongcatFlashForCausalLM(LongcatFlashPreTrainedModel, GenerationMixin):567 _tied_weights_keys = ["lm_head.weight"]568 _tp_plan = {"lm_head": "colwise_rep"}569 _pp_plan = {"lm_head": (["hidden_states"], ["logits"])}570 _keys_to_ignore_on_load_unexpected = [r"model\.mtp.*"]571 572 def __init__(self, config):573 super().__init__(config)574 self.model = LongcatFlashModel(config)575 self.vocab_size = config.vocab_size576 self.lm_head = nn.Linear(config.hidden_size, config.vocab_size, bias=False)577 578 # Initialize weights and apply final processing579 self.post_init()580 581 def set_decoder(self, decoder):582 self.model = decoder583 584 def get_decoder(self):585 return self.model586 587 @can_return_tuple588 @auto_docstring589 def forward(590 self,591 input_ids: Optional[torch.LongTensor] = None,592 attention_mask: Optional[torch.Tensor] = None,593 position_ids: Optional[torch.LongTensor] = None,594 past_key_values: Optional[Cache] = None,595 inputs_embeds: Optional[torch.FloatTensor] = None,596 labels: Optional[torch.LongTensor] = None,597 use_cache: Optional[bool] = None,598 cache_position: Optional[torch.LongTensor] = None,599 logits_to_keep: Union[int, torch.Tensor] = 0,600 **kwargs: Unpack[TransformersKwargs],601 ) -> CausalLMOutputWithPast:602 r"""603 Example:604 605 ```python606 >>> from transformers import AutoTokenizer, LongcatFlashForCausalLM607 608 >>> model = LongcatFlashForCausalLM.from_pretrained("meta-longcat_flash/LongcatFlash-2-7b-hf")609 >>> tokenizer = AutoTokenizer.from_pretrained("meta-longcat_flash/LongcatFlash-2-7b-hf")610 611 >>> prompt = "Hey, are you conscious? Can you talk to me?"612 >>> inputs = tokenizer(prompt, return_tensors="pt")613 614 >>> # Generate615 >>> generate_ids = model.generate(inputs.input_ids, max_length=30)616 >>> tokenizer.batch_decode(generate_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]617 "Hey, are you conscious? Can you talk to me?\nI'm not conscious, but I can talk to you."618 ```"""619 outputs: BaseModelOutputWithPast = self.model(620 input_ids=input_ids,621 attention_mask=attention_mask,622 position_ids=position_ids,623 past_key_values=past_key_values,624 inputs_embeds=inputs_embeds,625 use_cache=use_cache,626 cache_position=cache_position,627 **kwargs,628 )629 630 hidden_states = outputs.last_hidden_state631 # Only compute necessary logits, and do not upcast them to float if we are not computing the loss632 slice_indices = slice(-logits_to_keep, None) if isinstance(logits_to_keep, int) else logits_to_keep633 logits = self.lm_head(hidden_states[:, slice_indices, :])634 635 loss = None636 if labels is not None:637 loss = self.loss_function(logits=logits, labels=labels, vocab_size=self.config.vocab_size, **kwargs)638 639 return CausalLMOutputWithPast(640 loss=loss,641 logits=logits,642 past_key_values=outputs.past_key_values,643 hidden_states=outputs.hidden_states,644 attentions=outputs.attentions,645 )646 647 648__all__ = ["LongcatFlashPreTrainedModel", "LongcatFlashModel", "LongcatFlashForCausalLM"]649 