Llama 3 Source Code

Note

See github for more details.

Model

Overview of Llama3 model:

RMSNorm

\[y = \frac{x}{\sqrt{\mathbf{MeanSquare}[x] + \epsilon}}*\gamma\]

\(\mathbf{MeanSquare}[x]\) is calculated over \(C\) and is a vector of size \((N, L)\).

Note

See Normalization for comparisons of different normalization.

import torch
from torch import nn

class RMSNorm(torch.nn.Module):
    
    def __init__(self, dim: int, eps: float = 1e-6):
        super().__init__()
        self.eps = eps
        self.weight = nn.Parameter(torch.ones(dim))

    def _norm(self, x):
        return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps)

    def forward(self, x):
        # x shape: (N, L, C)
        # weight shape: (C,)
        output = self._norm(x.float()).type_as(x)
        return output * self.weight

Attention

RoPE

Given a position index \(m\in[0,c)\) and an embedding vector \(\mathbf{x} = [x_0,x_1,\dots,x_{d-1}]^{\top}\), where \(d\) is the dimension of the attention head, RoPE defines a vector-valued complex function \(\mathbf{f}(\mathbf{x}, m)\) as follows:

\[\mathbf{f}(\mathbf{x}, m) = [(x_0 + ix_{1})e^{im\theta_0},(x_2 + ix_{3})e^{im\theta_1},\dots,(x_{d-2} + ix_{d-1})e^{im\theta_{d/2-1}}]\]

where \(\theta_{j}=\theta^{-2j/d}\) (\(\theta\) is a hyper-parameter). Using RoPE, the self-attention score

\[\begin{split} \begin{split} \text{Re}\left \langle \mathbf{f}(\mathbf{q}, m), \mathbf{f}(\mathbf{k}, n) \right \rangle &= \text{Re}\left[\sum_{j=0}^{d/2-1}(q_{2j} + iq_{2j+1})(k_{2j} - ik_{2j+1})e^{i(m-n)\theta_{j}}\right]\\ &= a(m-n) \end{split} \end{split}\]

is only dependent on relative position \(m-n\) through trigonometric functions.

\[\text{RoPE}(\mathbf{x}, m) = \left[\text{Re}((x_0 + ix_{1})e^{im\theta_0}), \text{Im}((x_0 + ix_{1})e^{im\theta_0}), \text{Re}((x_2 + ix_{3})e^{im\theta_1}), \text{Im}((x_2 + ix_{3})e^{im\theta_1}), ...\right]\]

from typing import Tuple, Optional


def precompute_freqs_cis(dim: int, end: int, theta: float = 10000.0):
    # theta_j
    freqs = 1.0 / (theta ** (torch.arange(0, dim, 2)[: (dim // 2)].float() / dim))
    # 0,1,...,c-1
    t = torch.arange(end, device=freqs.device, dtype=torch.float32)
    # m * theta_j
    freqs = torch.outer(t, freqs)
    # exp ** (i*m*theta_j)
    freqs_cis = torch.polar(torch.ones_like(freqs), freqs)  # complex64
    return freqs_cis


def reshape_for_broadcast(freqs_cis: torch.Tensor, x: torch.Tensor):
    ndim = x.ndim
    assert 1 < ndim
    assert freqs_cis.shape == (x.shape[1], x.shape[-1])
    # (seqlen, head_dim // 2) -> (1, seqlen, 1, head_dim // 2)
    shape = [d if i == 1 or i == ndim - 1 else 1 for i, d in enumerate(x.shape)]
    return freqs_cis.view(*shape)


def apply_rotary_emb(
    xq: torch.Tensor,
    xk: torch.Tensor,
    freqs_cis: torch.Tensor,
) -> Tuple[torch.Tensor, torch.Tensor]:
    # view_as_complex: [[q_0, q_1], [q_2, q_3],...] -> [q_0 + i*q_1, q_2 + i*q_3, ...]
    # (bsz, seqlen, n_heads, head_dim) -> (bsz, seqlen, n_heads, head_dim // 2)
    xq_ = torch.view_as_complex(xq.float().reshape(*xq.shape[:-1], -1, 2))
    xk_ = torch.view_as_complex(xk.float().reshape(*xk.shape[:-1], -1, 2))
    freqs_cis = reshape_for_broadcast(freqs_cis, xq_)
    # (bsz, seqlen, n_heads, head_dim // 2) ->
    # (bsz, seqlen, n_heads, head_dim // 2, 2) ->
    # (bsz, seqlen, n_heads, head_dim)
    xq_out = torch.view_as_real(xq_ * freqs_cis).flatten(3)
    xk_out = torch.view_as_real(xk_ * freqs_cis).flatten(3)
    return xq_out.type_as(xq), xk_out.type_as(xk)

Attention

Note

We do not need to cache previous queries.
Attention is the only place that one position interact with other positions.

from dataclasses import dataclass

@dataclass
class ModelArgs:
    dim: int = 4096
    n_layers: int = 32
    n_heads: int = 32
    vocab_size: int = -1
    multiple_of: int = 256  # make SwiGLU hidden layer size multiple of large power of 2
    norm_eps: float = 1e-5
    rope_theta: float = 500000

    max_batch_size: int = 32
    max_seq_len: int = 2048

class Attention(nn.Module):
    def __init__(self, args: ModelArgs):
        super().__init__()
        self.n_heads = args.n_heads
        self.head_dim = args.dim // args.n_heads
        
        # Simple linear transformations for query, key, value
        self.wq = nn.Linear(args.dim, args.n_heads * self.head_dim, bias=False)
        self.wk = nn.Linear(args.dim, args.n_heads * self.head_dim, bias=False)
        self.wv = nn.Linear(args.dim, args.n_heads * self.head_dim, bias=False)
        # Output linear transformation
        self.wo = nn.Linear(args.n_heads * self.head_dim, args.dim, bias=False)
        
        # Cache
        self.cache_k = torch.zeros((args.max_batch_size, args.max_seq_len, self.n_heads, self.head_dim))
        self.cache_v = torch.zeros((args.max_batch_size, args.max_seq_len, self.n_heads, self.head_dim))
        
    def forward(
        self,
        x: torch.Tensor,
        start_pos: int,
        freqs_cis: torch.Tensor,
        mask: Optional[torch.Tensor],
    ):
        bsz, seqlen, _ = x.shape
        xq, xk, xv = self.wq(x), self.wk(x), self.wv(x)
        
        # Reshape for multi-head attention computation
        xq = xq.view(bsz, seqlen, self.n_heads, self.head_dim)
        xk = xk.view(bsz, seqlen, self.n_heads, self.head_dim)
        xv = xv.view(bsz, seqlen, self.n_heads, self.head_dim)
        
        # RoPE
        xq, xk = apply_rotary_emb(xq, xk, freqs_cis=freqs_cis)
        
        self.cache_k = self.cache_k.to(xq)
        self.cache_v = self.cache_v.to(xq)
        
        self.cache_k[:bsz, start_pos : start_pos + seqlen] = xk
        self.cache_v[:bsz, start_pos : start_pos + seqlen] = xv
        
        # Add previous keys and values
        keys = self.cache_k[:bsz, : start_pos + seqlen]
        values = self.cache_v[:bsz, : start_pos + seqlen]
        
        # Transpose for matrix multiplication
        xq = xq.transpose(1, 2)          # (bs, n_heads, seqlen, head_dim)
        keys = keys.transpose(1, 2)      # (bs, n_heads, cache_len + seqlen, head_dim)
        values = values.transpose(1, 2)  # (bs, n_heads, cache_len + seqlen, head_dim)
        
        # Scaled dot-product attention
        scores = torch.matmul(xq, keys.transpose(2, 3)) / math.sqrt(self.head_dim)
        if mask is not None:
            scores = scores + mask  # (bs, n_heads, seqlen, cache_len + seqlen)
        scores = F.softmax(scores.float(), dim=-1).type_as(xq)
        
        # Compute weighted average
        output = torch.matmul(scores, values)  # (bs, n_heads, seqlen, head_dim)
        output = output.transpose(1, 2).contiguous().view(bsz, seqlen, -1)  # (bs, seqlen, n_heads * head_dim)
        
        return self.wo(output)  # (bs, seqlen, dim)

FeedForward

The “position-wise feed-forward networks” (FFN) takes a vector \(x\) (the hidden representation at a particular position in the sequence) and passes it through two learned linear transformations, (represented by the matrices \(W_{1}\) and \(W_{2}\) and bias vectors \(b_{1}\) and \(b_{2}\)). A rectified-linear (ReLU) activation function applied between the two linear transformations.

\[\text{FFN}(x, W_{1}, W_{2}, b_{1}, b_{2}) = \max(0, xW_{1}+b_{1})W_{2} + b_{2}\]

If we use a version with no bias:

\[\text{FFN}_{\text{ReLU}}(x, W_{1}, W_{2}) = \max(0, xW_{1})W_{2}\]

Subsequent work has proposed replacing the ReLU with other nonlinear activation functions such as \(\text{Swish} = x\sigma(x)\) (also known as SiLU):

\[ \text{FFN}_{\text{Swish}}(x, W_{1}, W_{2}) = \text{Swish}(xW_{1})W_{2} \]

SwiGLU activation function

GLU is a neural network layer defined as the component-wise product of two linear transformations of the input, one of which is sigmoid-activated.

\[\text{GLU}(x, W, V, b, c) = \sigma(xW + b)\otimes(xV+c)\]

We propose additional variations on the Transformer FFN layer which use GLU or one of its variants in place of the first linear transformation and the activation function:

\[\begin{split} \begin{aligned} \text{FFN}_{\text{GLU}}(x, W, V, W_{2}) &= (\sigma(xW + b)\otimes(xV+c))W_{2}\\ \text{FFN}_{\text{ReGLU}}(x, W, V, W_{2}) &= (\max(0, xW + b)\otimes(xV+c))W_{2}\\ \text{FFN}_{\text{SwiGLU}}(x, W, V, W_{2}) &= (\text{Swish}_{1}(xW + b)\otimes(xV+c))W_{2}\\ \end{aligned} \end{split}\]

class FeedForward(nn.Module):
    def __init__(
        self,
        dim: int,
        hidden_dim: int,
        multiple_of: int
    ):
        super().__init__()
        # use a dimension of 2/3*4d instead of 4d as in PaLM
        hidden_dim = int(2 * hidden_dim / 3)
        # make SwiGLU hidden layer size multiple of large power of 2
        hidden_dim = multiple_of * ((hidden_dim + multiple_of - 1) // multiple_of)
        
        self.w1 = nn.Linear(dim, hidden_dim, bias=False)
        self.w2 = nn.Linear(hidden_dim, dim, bias=False)
        self.w3 = nn.Linear(dim, hidden_dim, bias=False)

    def forward(self, x):
        return self.w2(F.silu(self.w1(x)) * self.w3(x))

Transformer

class TransformerBlock(nn.Module):
    def __init__(self, layer_id: int, args: ModelArgs):
        super().__init__()
        self.n_heads = args.n_heads
        self.dim = args.dim
        self.head_dim = args.dim // args.n_heads
        self.attention = Attention(args)
        self.feed_forward = FeedForward(
            dim=args.dim,
            hidden_dim=4 * args.dim,
            multiple_of=args.multiple_of
        )
        self.layer_id = layer_id
        self.attention_norm = RMSNorm(args.dim, eps=args.norm_eps)
        self.ffn_norm = RMSNorm(args.dim, eps=args.norm_eps)

    def forward(
        self,
        x: torch.Tensor,
        start_pos: int,
        freqs_cis: torch.Tensor,
        mask: Optional[torch.Tensor],
    ):
        h = x + self.attention(self.attention_norm(x), start_pos, freqs_cis, mask)
        out = h + self.feed_forward(self.ffn_norm(h))
        return out

Mask illustration:

class Transformer(nn.Module):
    def __init__(self, params: ModelArgs):
        super().__init__()
        self.params = params
        self.vocab_size = params.vocab_size
        self.n_layers = params.n_layers
        
        self.tok_embeddings = nn.Embedding(params.vocab_size, params.dim)

        self.layers = torch.nn.ModuleList()
        for layer_id in range(params.n_layers):
            self.layers.append(TransformerBlock(layer_id, params))

        self.norm = RMSNorm(params.dim, eps=params.norm_eps)
        self.output = nn.Linear(params.dim, params.vocab_size, bias=False)

        # end = 2 * max_seq_len
        self.freqs_cis = precompute_freqs_cis(
            params.dim // params.n_heads,
            params.max_seq_len * 2,
            params.rope_theta,
        )

    @torch.inference_mode()
    def forward(self, tokens: torch.Tensor, start_pos: int):
        _bsz, seqlen = tokens.shape
        h = self.tok_embeddings(tokens)
        self.freqs_cis = self.freqs_cis.to(h.device)
        freqs_cis = self.freqs_cis[start_pos : start_pos + seqlen]

        mask = None
        if seqlen > 1:
            mask = torch.full((seqlen, seqlen), float("-inf"), device=tokens.device)

            mask = torch.triu(mask, diagonal=1)

            # When performing key-value caching, we compute the attention scores
            # only for the new sequence. Thus, the matrix of scores is of size
            # (seqlen, cache_len + seqlen), and the only masked entries are (i, j) for
            # j > cache_len + i, since row i corresponds to token cache_len + i.
            mask = torch.hstack(
                [torch.zeros((seqlen, start_pos), device=tokens.device), mask]
            ).type_as(h)

        for layer in self.layers:
            h = layer(h, start_pos, freqs_cis, mask)
        h = self.norm(h)
        output = self.output(h).float()
        return output