import json import math import os from dataclasses import dataclass from pathlib import Path from typing import Optional, Tuple import torch from torch import nn from flash_attn import flash_attn_func from fairseq.model_parallel.megatron.mpu import ( ColumnParallelLinear, RowParallelLinear, copy_to_model_parallel_region, VocabParallelEmbedding ) from fairscale.nn import checkpoint_wrapper from .rms_norm import RMSNorm from .kernel.rotary import apply_rotary_emb from .model_parallel_init import init_method, vocab_init_method def precompute_freqs_cis(dim: int, end: int, theta: float) -> torch.Tensor: freqs = 1.0 / (theta ** (torch.arange(0, dim, 2)[: (dim // 2)].float() / dim)) t = torch.arange(end, device=freqs.device) # type: ignore freqs = torch.outer(t, freqs).float() # type: ignore return freqs @dataclass class ModelArgs: dim: int n_layers: int head_dim: int hidden_dim: int n_heads: int n_kv_heads: int norm_eps: float vocab_size: int max_batch_size: int = 0 max_seq_len: int = -1 model_parallel_size: int = 1 load_checkpoint: bool = False rope_theta: float = 10000.0 sliding_window: Optional[int] = None class Attention(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.args = args self.dim = args.dim self.head_dim = args.head_dim self.hidden_dim = args.n_heads * args.head_dim self.key_value_dim = args.n_kv_heads * args.head_dim self.n_heads = args.n_heads // args.model_parallel_size self.n_kv_heads = args.n_kv_heads // args.model_parallel_size self.activate_sliding_window = args.sliding_window is not None self.cache_len = args.sliding_window - 1 if self.activate_sliding_window else args.max_seq_len self.repeats = self.n_heads // self.n_kv_heads self.scale = self.args.head_dim**-0.5 self.wq = ColumnParallelLinear(self.dim, self.hidden_dim, bias=False, gather_output=False, init_method=init_method) self.wk = ColumnParallelLinear(self.dim, self.key_value_dim, bias=False, gather_output=False, init_method=init_method) self.wv = ColumnParallelLinear(self.dim, self.key_value_dim, bias=False, gather_output=False, init_method=init_method) self.wo = RowParallelLinear(self.hidden_dim, self.dim, bias=False, input_is_parallel=True, init_method=init_method) def forward( self, x: torch.Tensor, rel_pos: Tuple[torch.Tensor, torch.Tensor], start_pos: int, incremental_state = None, ) -> torch.Tensor: bsz, seqlen, _ = x.shape xq, xk, xv = self.wq(x), self.wk(x), self.wv(x) xq = xq.view(bsz, seqlen, self.n_heads, self.head_dim) xk = xk.view(bsz, seqlen, self.n_kv_heads, self.head_dim) xv = xv.view(bsz, seqlen, self.n_kv_heads, self.head_dim) xq = apply_rotary_emb(xq, *rel_pos) xk = apply_rotary_emb(xk, *rel_pos) if incremental_state is not None: if "cache_k" not in incremental_state: incremental_state["cache_k"] = torch.zeros( ( self.args.max_batch_size, self.cache_len, self.n_kv_heads, self.head_dim, ) ).to(xk) incremental_state["cache_v"] = torch.zeros( ( self.args.max_batch_size, self.cache_len, self.n_kv_heads, self.head_dim, ) ).to(xv) key = torch.cat([incremental_state["cache_k"][:, :start_pos], xk], dim=1) value = torch.cat([incremental_state["cache_v"][:, :start_pos], xv], dim=1) if key.shape[1] > self.cache_len: incremental_state["cache_k"][:bsz] = key[:, -self.cache_len:] incremental_state["cache_v"][:bsz] = value[:, -self.cache_len:] else: incremental_state["cache_k"][:bsz, start_pos : start_pos + seqlen] = xk incremental_state["cache_v"][:bsz, start_pos : start_pos + seqlen] = xv else: key, value = xk, xv output = flash_attn_func(xq, key, value, causal=True, window_size=(self.args.sliding_window - 1, 0) if self.activate_sliding_window else (-1, -1)) return self.wo(output.view(bsz, seqlen, self.n_heads * self.head_dim)) class FeedForward(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.w1 = ColumnParallelLinear(args.dim, args.hidden_dim, bias=False, gather_output=False, init_method=init_method) self.w2 = RowParallelLinear(args.hidden_dim, args.dim, bias=False, input_is_parallel=True, init_method=init_method) self.w3 = ColumnParallelLinear(args.dim, args.hidden_dim, bias=False, gather_output=False, init_method=init_method) def forward(self, x) -> torch.Tensor: return self.w2(nn.functional.silu(self.w1(x)) * self.w3(x)) class TransformerBlock(nn.Module): def __init__(self, args: ModelArgs): super().__init__() self.n_heads = args.n_heads self.dim = args.dim self.attention = Attention(args) self.attention_norm = RMSNorm(args.dim, eps=args.norm_eps) self.ffn_norm = RMSNorm(args.dim, eps=args.norm_eps) self.args = args self.feed_forward: nn.Module self.feed_forward = FeedForward(args=args) def forward( self, x: torch.Tensor, rel_pos: Tuple[torch.Tensor, torch.Tensor], start_pos: int, incremental_state = None ) -> torch.Tensor: r = self.attention.forward(self.attention_norm(x), rel_pos, start_pos, incremental_state) h = x + r r = self.feed_forward.forward(self.ffn_norm(h)) out = h + r return out class Transformer(nn.Module): def __init__( self, args: ModelArgs, mp_rank: int = 0, checkpoint_activations: bool = False ): super().__init__() self.args = args self.vocab_size = args.vocab_size self.n_layers = args.n_layers self._precomputed_freqs_cis: Optional[torch.Tensor] = None self._window_precomputed_freqs_cis: Optional[torch.Tensor] = None self._global_precomputed_freqs_cis: Optional[torch.Tensor] = None assert self.vocab_size > 0 self.mp_rank = mp_rank self.checkpoint_activations = checkpoint_activations self.tok_embeddings = VocabParallelEmbedding( args.vocab_size, args.dim, -1, init_method=vocab_init_method ) self.norm = RMSNorm(args.dim, eps=args.norm_eps) self.output = nn.Linear(args.dim, args.vocab_size // args.model_parallel_size, bias=False) # Initialize all layers but slice off those not of this rank. layers = [TransformerBlock(args=args) for idx in range(args.n_layers)] if checkpoint_activations: layers = [checkpoint_wrapper(layer) for layer in layers] self.layers = nn.ModuleList(layers) self.n_local_layers = len(self.layers) @property def dtype(self) -> torch.dtype: return next(self.parameters()).dtype @property def device(self) -> torch.device: return next(self.parameters()).device def build_rel_pos(self, x, start_pos): if self._precomputed_freqs_cis is None: theta = self.args.rope_theta self._precomputed_freqs_cis = precompute_freqs_cis( self.args.head_dim, self.args.max_seq_len, theta ) if self._precomputed_freqs_cis.device != self.device: self._precomputed_freqs_cis = self._precomputed_freqs_cis.to( device=self.device ) cos = torch.cos(self._precomputed_freqs_cis[start_pos:start_pos+x.size(1)]) sin = torch.sin(self._precomputed_freqs_cis[start_pos:start_pos+x.size(1)]) rel_pos = (cos.to(x.dtype), sin.to(x.dtype)) return rel_pos def forward_partial( self, input_ids: torch.Tensor, start_pos: Optional[int] = 0, incremental_state = None, ) -> torch.Tensor: h = self.tok_embeddings(input_ids) rel_pos = self.build_rel_pos(h, start_pos) for local_layer_id, layer in enumerate(self.layers): if incremental_state is not None: if local_layer_id not in incremental_state: incremental_state[local_layer_id] = {} h = layer(h, rel_pos, start_pos, incremental_state=incremental_state[local_layer_id] if incremental_state is not None else None) return self.norm(h) def forward( self, input_ids: torch.Tensor, start_pos: Optional[int] = 0, incremental_state = None, ) -> torch.Tensor: h = self.forward_partial(input_ids, start_pos, incremental_state) if self.args.model_parallel_size > 1: h = copy_to_model_parallel_region(h) outs = self.output(h) return outs.float(), None def load_state_dict(self, state_dict, strict=False, assign=False): state_to_load = {} for k, v in state_dict.items(): if k.startswith("tok_embeddings") or k.startswith("output"): state_to_load[k] = v.view(self.args.model_parallel_size, self.vocab_size // self.args.model_parallel_size, self.args.dim)[self.mp_rank] elif "wq" in k or "wk" in k or "wv" in k or "w1" in k or "w3" in k: state_to_load[k] = v.view(self.args.model_parallel_size, -1, v.shape[1])[self.mp_rank] elif "wo" in k or "w2" in k: state_to_load[k] = v.view(v.shape[0], self.args.model_parallel_size, -1)[:, self.mp_rank] else: state_to_load[k] = v super().load_state_dict(state_to_load, strict=False, assign=assign) print("Loaded state dict from checkpoint.")