|
|
|
|
|
|
| """
|
| This module implements a Gemma-style transformer architecture with:
|
| - RMSNorm (Root Mean Square Layer Normalization)
|
| - RoPE (Rotary Position Embeddings)
|
| - GeGLU (Gated Linear Unit with GELU activation)
|
| - GQA (Grouped Query Attention)
|
|
|
| Architecture Specifications:
|
| - hidden_size: 768
|
| - num_layers: 12
|
| - intermediate_size: 3072
|
| - num_attention_heads: 12
|
| - num_key_value_heads: 4 (GQA ratio of 3:1)
|
| - vocab_size: 50257 (GPT-2 tokenizer)
|
| """
|
|
|
| import math
|
| import logging
|
| from typing import Optional, Tuple, Dict, Any, Union, List
|
|
|
| import torch
|
| import torch.nn as nn
|
| import torch.nn.functional as F
|
| from torch import Tensor
|
|
|
| from transformers import PretrainedConfig, PreTrainedModel
|
| from transformers.modeling_outputs import CausalLMOutputWithPast
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
|
|
| class GemmaConfig(PretrainedConfig):
|
| """
|
| Configuration class for the Gemma-style model.
|
|
|
| Inherits from HuggingFace PretrainedConfig for compatibility with
|
| AutoConfig and the transformers ecosystem.
|
| """
|
| model_type = "gemma_custom"
|
|
|
| def __init__(
|
| self,
|
| hidden_size: int = 768,
|
| num_layers: int = 12,
|
| intermediate_size: int = 3072,
|
| num_attention_heads: int = 12,
|
| num_key_value_heads: int = 4,
|
| max_position_embeddings: int = 1024,
|
| vocab_size: int = 50257,
|
| rope_theta: float = 10000.0,
|
| rms_norm_eps: float = 1e-6,
|
| hidden_act: str = "gelu_pytorch_tanh",
|
| attention_dropout: float = 0.0,
|
| hidden_dropout: float = 0.0,
|
| tie_word_embeddings: bool = True,
|
| initializer_range: float = 0.02,
|
| **kwargs
|
| ):
|
| self.hidden_size = hidden_size
|
| self.num_layers = num_layers
|
| self.intermediate_size = intermediate_size
|
| self.num_attention_heads = num_attention_heads
|
| self.num_key_value_heads = num_key_value_heads
|
| self.max_position_embeddings = max_position_embeddings
|
| self.vocab_size = vocab_size
|
| self.rope_theta = rope_theta
|
| self.rms_norm_eps = rms_norm_eps
|
| self.hidden_act = hidden_act
|
| self.attention_dropout = attention_dropout
|
| self.hidden_dropout = hidden_dropout
|
| self.initializer_range = initializer_range
|
|
|
|
|
| self.head_dim = self.hidden_size // self.num_attention_heads
|
| self.num_key_value_groups = self.num_attention_heads // self.num_key_value_heads
|
|
|
| super().__init__(tie_word_embeddings=tie_word_embeddings, **kwargs)
|
|
|
|
|
| class RMSNorm(nn.Module):
|
| """
|
| Root Mean Square Layer Normalization.
|
|
|
| RMSNorm(x) = x * rsqrt(mean(x^2) + eps) * weight
|
| """
|
|
|
| def __init__(self, hidden_size: int, eps: float = 1e-6):
|
| super().__init__()
|
| self.weight = nn.Parameter(torch.ones(hidden_size))
|
| self.variance_epsilon = eps
|
|
|
| def forward(self, hidden_states: Tensor) -> Tensor:
|
| input_dtype = hidden_states.dtype
|
| hidden_states = hidden_states.to(torch.float32)
|
| variance = hidden_states.pow(2).mean(-1, keepdim=True)
|
| hidden_states = hidden_states * torch.rsqrt(variance + self.variance_epsilon)
|
| return self.weight * hidden_states.to(input_dtype)
|
|
|
|
|
| class RotaryEmbedding(nn.Module):
|
| """Rotary Position Embedding (RoPE)."""
|
|
|
| def __init__(self, dim: int, max_position_embeddings: int = 1024, theta: float = 10000.0):
|
| super().__init__()
|
| self.dim = dim
|
| self.max_position_embeddings = max_position_embeddings
|
| self.theta = theta
|
|
|
| inv_freq = 1.0 / (self.theta ** (torch.arange(0, dim, 2, dtype=torch.float32) / dim))
|
| self.register_buffer("inv_freq", inv_freq, persistent=False)
|
| self._set_cos_sin_cache(max_position_embeddings)
|
|
|
| def _set_cos_sin_cache(self, seq_len: int):
|
| t = torch.arange(seq_len, dtype=torch.float32)
|
| freqs = torch.einsum("i,j->ij", t, self.inv_freq)
|
| emb = torch.cat((freqs, freqs), dim=-1)
|
| self.register_buffer("cos_cached", emb.cos(), persistent=False)
|
| self.register_buffer("sin_cached", emb.sin(), persistent=False)
|
|
|
| def forward(self, x: Tensor, position_ids: Tensor) -> Tuple[Tensor, Tensor]:
|
| seq_len = position_ids.max() + 1
|
| if seq_len > self.cos_cached.shape[0]:
|
| self._set_cos_sin_cache(seq_len)
|
| self.cos_cached = self.cos_cached.to(x.device)
|
| self.sin_cached = self.sin_cached.to(x.device)
|
|
|
| cos = self.cos_cached[position_ids].unsqueeze(2)
|
| sin = self.sin_cached[position_ids].unsqueeze(2)
|
| return cos.to(x.dtype), sin.to(x.dtype)
|
|
|
|
|
| def rotate_half(x: Tensor) -> Tensor:
|
| x1 = x[..., : x.shape[-1] // 2]
|
| x2 = x[..., x.shape[-1] // 2 :]
|
| return torch.cat((-x2, x1), dim=-1)
|
|
|
|
|
| def apply_rotary_pos_emb(q: Tensor, k: Tensor, cos: Tensor, sin: Tensor) -> Tuple[Tensor, Tensor]:
|
| q_embed = (q * cos) + (rotate_half(q) * sin)
|
| k_embed = (k * cos) + (rotate_half(k) * sin)
|
| return q_embed, k_embed
|
|
|
|
|
| class GemmaMLP(nn.Module):
|
| """Gemma-style MLP with GeGLU activation."""
|
|
|
| def __init__(self, config: GemmaConfig):
|
| super().__init__()
|
| self.gate_proj = nn.Linear(config.hidden_size, config.intermediate_size, bias=False)
|
| self.up_proj = nn.Linear(config.hidden_size, config.intermediate_size, bias=False)
|
| self.down_proj = nn.Linear(config.intermediate_size, config.hidden_size, bias=False)
|
| self.act_fn = nn.GELU(approximate="tanh")
|
|
|
| def forward(self, x: Tensor) -> Tensor:
|
| gate = self.act_fn(self.gate_proj(x))
|
| up = self.up_proj(x)
|
| return self.down_proj(gate * up)
|
|
|
|
|
| class GemmaAttention(nn.Module):
|
| """Grouped Query Attention (GQA) module."""
|
|
|
| def __init__(self, config: GemmaConfig, layer_idx: int):
|
| super().__init__()
|
| self.config = config
|
| self.layer_idx = layer_idx
|
| self.hidden_size = config.hidden_size
|
| self.num_heads = config.num_attention_heads
|
| self.head_dim = config.head_dim
|
| self.num_key_value_heads = config.num_key_value_heads
|
| self.num_key_value_groups = config.num_key_value_groups
|
| self.attention_dropout = config.attention_dropout
|
|
|
| self.q_proj = nn.Linear(self.hidden_size, self.num_heads * self.head_dim, bias=False)
|
| self.k_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=False)
|
| self.v_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=False)
|
| self.o_proj = nn.Linear(self.num_heads * self.head_dim, self.hidden_size, bias=False)
|
|
|
| self.rotary_emb = RotaryEmbedding(
|
| self.head_dim,
|
| max_position_embeddings=config.max_position_embeddings,
|
| theta=config.rope_theta
|
| )
|
|
|
| def forward(
|
| self,
|
| hidden_states: Tensor,
|
| attention_mask: Optional[Tensor] = None,
|
| position_ids: Optional[Tensor] = None,
|
| ) -> Tensor:
|
| batch_size, seq_len, _ = hidden_states.shape
|
|
|
| query_states = self.q_proj(hidden_states)
|
| key_states = self.k_proj(hidden_states)
|
| value_states = self.v_proj(hidden_states)
|
|
|
| query_states = query_states.view(batch_size, seq_len, self.num_heads, self.head_dim)
|
| key_states = key_states.view(batch_size, seq_len, self.num_key_value_heads, self.head_dim)
|
| value_states = value_states.view(batch_size, seq_len, self.num_key_value_heads, self.head_dim)
|
|
|
| cos, sin = self.rotary_emb(query_states, position_ids)
|
| query_states, key_states = apply_rotary_pos_emb(query_states, key_states, cos, sin)
|
|
|
| query_states = query_states.transpose(1, 2)
|
| key_states = key_states.transpose(1, 2)
|
| value_states = value_states.transpose(1, 2)
|
|
|
| if self.num_key_value_groups > 1:
|
| key_states = key_states.repeat_interleave(self.num_key_value_groups, dim=1)
|
| value_states = value_states.repeat_interleave(self.num_key_value_groups, dim=1)
|
|
|
| scale = 1.0 / math.sqrt(self.head_dim)
|
| attn_weights = torch.matmul(query_states, key_states.transpose(-2, -1)) * scale
|
|
|
| if attention_mask is not None:
|
| attn_weights = attn_weights + attention_mask
|
|
|
| attn_weights = F.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query_states.dtype)
|
| attn_weights = F.dropout(attn_weights, p=self.attention_dropout, training=self.training)
|
|
|
| attn_output = torch.matmul(attn_weights, value_states)
|
| attn_output = attn_output.transpose(1, 2).contiguous()
|
| attn_output = attn_output.reshape(batch_size, seq_len, self.hidden_size)
|
|
|
| return self.o_proj(attn_output)
|
|
|
|
|
| class GemmaDecoderLayer(nn.Module):
|
| """Single transformer decoder layer with pre-norm architecture."""
|
|
|
| def __init__(self, config: GemmaConfig, layer_idx: int):
|
| super().__init__()
|
| self.hidden_size = config.hidden_size
|
| self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
|
| self.post_attention_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
|
| self.self_attn = GemmaAttention(config, layer_idx)
|
| self.mlp = GemmaMLP(config)
|
|
|
| def forward(
|
| self,
|
| hidden_states: Tensor,
|
| attention_mask: Optional[Tensor] = None,
|
| position_ids: Optional[Tensor] = None,
|
| ) -> Tensor:
|
| residual = hidden_states
|
| hidden_states = self.input_layernorm(hidden_states)
|
| hidden_states = self.self_attn(hidden_states, attention_mask, position_ids)
|
| hidden_states = residual + hidden_states
|
|
|
| residual = hidden_states
|
| hidden_states = self.post_attention_layernorm(hidden_states)
|
| hidden_states = self.mlp(hidden_states)
|
| hidden_states = residual + hidden_states
|
|
|
| return hidden_states
|
|
|
|
|
| class GemmaModel(nn.Module):
|
| """Core Gemma transformer model (without LM head)."""
|
|
|
| def __init__(self, config: GemmaConfig):
|
| super().__init__()
|
| self.config = config
|
| self.embed_tokens = nn.Embedding(config.vocab_size, config.hidden_size)
|
| self.layers = nn.ModuleList([
|
| GemmaDecoderLayer(config, layer_idx)
|
| for layer_idx in range(config.num_layers)
|
| ])
|
| self.norm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
|
| self.gradient_checkpointing = False
|
|
|
| def forward(
|
| self,
|
| input_ids: Tensor,
|
| attention_mask: Optional[Tensor] = None,
|
| position_ids: Optional[Tensor] = None,
|
| ) -> Tensor:
|
| batch_size, seq_len = input_ids.shape
|
| hidden_states = self.embed_tokens(input_ids)
|
|
|
| if position_ids is None:
|
| position_ids = torch.arange(seq_len, device=input_ids.device).unsqueeze(0).expand(batch_size, -1)
|
|
|
| causal_mask = self._create_causal_mask(seq_len, hidden_states.device, hidden_states.dtype)
|
|
|
| for layer in self.layers:
|
| if self.gradient_checkpointing and self.training:
|
| hidden_states = torch.utils.checkpoint.checkpoint(
|
| layer, hidden_states, causal_mask, position_ids, use_reentrant=False
|
| )
|
| else:
|
| hidden_states = layer(hidden_states, causal_mask, position_ids)
|
|
|
| return self.norm(hidden_states)
|
|
|
| def _create_causal_mask(self, seq_len: int, device: torch.device, dtype: torch.dtype) -> Tensor:
|
| mask = torch.full((seq_len, seq_len), float("-inf"), device=device, dtype=dtype)
|
| mask = torch.triu(mask, diagonal=1)
|
| return mask.unsqueeze(0).unsqueeze(0)
|
|
|
|
|
| class GemmaForCausalLM(PreTrainedModel):
|
| """
|
| Gemma model with language modeling head for causal text generation.
|
|
|
| Inherits from HuggingFace PreTrainedModel for full compatibility with
|
| AutoModelForCausalLM and the transformers ecosystem.
|
| """
|
| config_class = GemmaConfig
|
| supports_gradient_checkpointing = True
|
| _no_split_modules = ["GemmaDecoderLayer"]
|
| _supports_param_buffer_assignment = False
|
|
|
| def __init__(self, config: GemmaConfig):
|
| super().__init__(config)
|
| self.config = config
|
|
|
| self.model = GemmaModel(config)
|
| self.lm_head = nn.Linear(config.hidden_size, config.vocab_size, bias=False)
|
|
|
| if config.tie_word_embeddings:
|
| self.lm_head.weight = self.model.embed_tokens.weight
|
|
|
| self.post_init()
|
|
|
| @classmethod
|
| def from_pretrained(cls, pretrained_model_name_or_path, *args, **kwargs):
|
| """
|
| Custom from_pretrained that properly loads weights for this custom model.
|
| This overrides the default behavior to ensure weights are loaded correctly.
|
| """
|
| import os
|
| from huggingface_hub import hf_hub_download
|
|
|
|
|
| trust_remote_code = kwargs.pop("trust_remote_code", True)
|
| torch_dtype = kwargs.pop("torch_dtype", None)
|
| device_map = kwargs.pop("device_map", None)
|
|
|
|
|
| config = cls.config_class.from_pretrained(
|
| pretrained_model_name_or_path,
|
| trust_remote_code=trust_remote_code,
|
| **kwargs
|
| )
|
|
|
|
|
| model = cls(config)
|
|
|
|
|
| if os.path.isdir(pretrained_model_name_or_path):
|
| weight_file = os.path.join(pretrained_model_name_or_path, "pytorch_model.bin")
|
| else:
|
|
|
| weight_file = hf_hub_download(
|
| repo_id=pretrained_model_name_or_path,
|
| filename="pytorch_model.bin"
|
| )
|
|
|
|
|
| state_dict = torch.load(weight_file, map_location="cpu")
|
| model.load_state_dict(state_dict, strict=False)
|
|
|
|
|
| if torch_dtype is not None:
|
| model = model.to(torch_dtype)
|
| if device_map == "auto":
|
| if torch.cuda.is_available():
|
| model = model.to("cuda")
|
| elif device_map is not None:
|
| model = model.to(device_map)
|
|
|
| return model
|
|
|
| def get_input_embeddings(self):
|
| return self.model.embed_tokens
|
|
|
| def set_input_embeddings(self, value):
|
| self.model.embed_tokens = value
|
|
|
| def get_output_embeddings(self):
|
| return self.lm_head
|
|
|
| def set_output_embeddings(self, new_embeddings):
|
| self.lm_head = new_embeddings
|
|
|
| def _init_weights(self, module: nn.Module):
|
| std = self.config.initializer_range
|
| if isinstance(module, nn.Linear):
|
| module.weight.data.normal_(mean=0.0, std=std)
|
| if module.bias is not None:
|
| module.bias.data.zero_()
|
| elif isinstance(module, nn.Embedding):
|
| module.weight.data.normal_(mean=0.0, std=std)
|
|
|
| def num_parameters(self, only_trainable: bool = False) -> int:
|
| return sum(p.numel() for p in self.parameters() if not only_trainable or p.requires_grad)
|
|
|
| def forward(
|
| self,
|
| input_ids: Tensor,
|
| attention_mask: Optional[Tensor] = None,
|
| position_ids: Optional[Tensor] = None,
|
| labels: Optional[Tensor] = None,
|
| return_dict: Optional[bool] = None,
|
| **kwargs
|
| ) -> Union[Tuple, CausalLMOutputWithPast]:
|
| return_dict = return_dict if return_dict is not None else self.config.use_return_dict
|
|
|
| hidden_states = self.model(input_ids, attention_mask, position_ids)
|
| logits = self.lm_head(hidden_states)
|
|
|
| loss = None
|
| if labels is not None:
|
| shift_logits = logits[..., :-1, :].contiguous()
|
| shift_labels = labels[..., 1:].contiguous()
|
| loss_fct = nn.CrossEntropyLoss()
|
| loss = loss_fct(
|
| shift_logits.view(-1, self.config.vocab_size),
|
| shift_labels.view(-1)
|
| )
|
|
|
| if not return_dict:
|
| output = (logits,)
|
| return ((loss,) + output) if loss is not None else output
|
|
|
| return CausalLMOutputWithPast(
|
| loss=loss,
|
| logits=logits,
|
| past_key_values=None,
|
| hidden_states=None,
|
| attentions=None,
|
| )
|
|
|
| def prepare_inputs_for_generation(
|
| self, input_ids, past_key_values=None, attention_mask=None, **kwargs
|
| ):
|
| if past_key_values is not None:
|
| input_ids = input_ids[:, -1:]
|
|
|
| position_ids = kwargs.get("position_ids", None)
|
| if attention_mask is not None and position_ids is None:
|
| position_ids = attention_mask.long().cumsum(-1) - 1
|
| position_ids.masked_fill_(attention_mask == 0, 1)
|
| if past_key_values:
|
| position_ids = position_ids[:, -1].unsqueeze(-1)
|
|
|
| return {
|
| "input_ids": input_ids,
|
| "attention_mask": attention_mask,
|
| "position_ids": position_ids,
|
| "past_key_values": past_key_values,
|
| }
|
|
|
| def generate(
|
| self,
|
| input_ids: Tensor,
|
| max_new_tokens: int = 50,
|
| temperature: float = 1.0,
|
| top_k: int = 50,
|
| top_p: float = 0.95,
|
| do_sample: bool = True,
|
| eos_token_id: Optional[int] = None,
|
| **kwargs
|
| ) -> Tensor:
|
| """Custom generate method for simple autoregressive generation."""
|
| self.eval()
|
|
|
| for _ in range(max_new_tokens):
|
| with torch.no_grad():
|
| outputs = self.forward(input_ids)
|
| next_token_logits = outputs.logits[:, -1, :] / temperature
|
|
|
| if top_k > 0:
|
| indices_to_remove = next_token_logits < torch.topk(next_token_logits, top_k)[0][..., -1, None]
|
| next_token_logits[indices_to_remove] = float("-inf")
|
|
|
| if top_p < 1.0:
|
| sorted_logits, sorted_indices = torch.sort(next_token_logits, descending=True)
|
| cumulative_probs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)
|
| sorted_indices_to_remove = cumulative_probs > top_p
|
| sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
|
| sorted_indices_to_remove[..., 0] = 0
|
| indices_to_remove = sorted_indices_to_remove.scatter(1, sorted_indices, sorted_indices_to_remove)
|
| next_token_logits[indices_to_remove] = float("-inf")
|
|
|
| if do_sample:
|
| probs = F.softmax(next_token_logits, dim=-1)
|
| next_token = torch.multinomial(probs, num_samples=1)
|
| else:
|
| next_token = torch.argmax(next_token_logits, dim=-1, keepdim=True)
|
|
|
| input_ids = torch.cat([input_ids, next_token], dim=-1)
|
|
|
| if eos_token_id is not None and (next_token == eos_token_id).all():
|
| break
|
|
|
| return input_ids
|
|
|
|
|
| def create_model_from_config(config_dict: Optional[Dict[str, Any]] = None) -> GemmaForCausalLM:
|
| """Factory function to create a model from a configuration dictionary."""
|
| if config_dict is None:
|
| config = GemmaConfig()
|
| else:
|
| config = GemmaConfig(**{k: v for k, v in config_dict.items() if hasattr(GemmaConfig, k) or k in ['hidden_size', 'num_layers', 'intermediate_size', 'num_attention_heads', 'num_key_value_heads', 'max_position_embeddings', 'vocab_size', 'rope_theta', 'rms_norm_eps', 'hidden_act', 'attention_dropout', 'hidden_dropout', 'tie_word_embeddings', 'initializer_range']})
|
|
|
| model = GemmaForCausalLM(config)
|
| logger.info(f"Created model with {model.num_parameters():,} parameters")
|
| return model
|
|
|
