model/layers.py

"""基础算子：RMSNorm、解耦 RoPE、SwiGLU FFN。"""

import math
import torch
import torch.nn as nn
import torch.nn.functional as F


class RMSNorm(nn.Module):
    """比 LayerNorm 少一次均值统计，省算力且效果相当，是现代 LLM 的标配。

        y = x / sqrt(mean(x^2) + eps) * g
    """

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

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 统计量始终用 fp32 算，避免半精度下 rsqrt 抖动
        out = x.float() * torch.rsqrt(x.float().pow(2).mean(-1, keepdim=True) + self.eps)
        return (out * self.weight.float()).type_as(x)


# --------------------------------------------------------------------------
#  RoPE：旋转位置编码
#  MLA 里它是「解耦」的 —— 只作用在 Q/K 单独切出来的 qk_rope_head_dim 维上，
#  剩下的 qk_nope 部分不带位置信息，这样 K 才能被压进和位置无关的潜在向量里。
# --------------------------------------------------------------------------

def build_rope_cache(head_dim: int, max_seq_len: int, theta: float = 10000.0,
                     scaling: float = 1.0, device=None):
    """返回 (cos, sin)，形状均为 (max_seq_len, head_dim)。

    scaling > 1 时做 NTK-aware 频率插值：把 base 放大 scaling^(d/(d-2))，
    等价于低频维度插值、高频维度基本不动，用来做上下文长度外推。
    """
    if scaling > 1.0:
        theta = theta * (scaling ** (head_dim / (head_dim - 2)))
    inv_freq = 1.0 / (theta ** (torch.arange(0, head_dim, 2, dtype=torch.float32, device=device) / head_dim))
    t = torch.arange(max_seq_len, dtype=torch.float32, device=device)
    freqs = torch.outer(t, inv_freq)                 # (T, head_dim/2)
    emb = torch.cat([freqs, freqs], dim=-1)          # (T, head_dim)
    return emb.cos(), emb.sin()


def rotate_half(x: torch.Tensor) -> torch.Tensor:
    x1, x2 = x.chunk(2, dim=-1)
    return torch.cat([-x2, x1], dim=-1)


def apply_rope(x: torch.Tensor, cos: torch.Tensor, sin: torch.Tensor) -> torch.Tensor:
    """x: (B, H, T, D)，cos/sin: (T, D)。"""
    cos = cos[None, None, :, :].to(x.dtype)
    sin = sin[None, None, :, :].to(x.dtype)
    return x * cos + rotate_half(x) * sin


# --------------------------------------------------------------------------

class SwiGLU(nn.Module):
    """FFN(x) = W_down( silu(W_gate x) * W_up x )，比 ReLU-MLP 表达力更强。"""

    def __init__(self, dim: int, inter_dim: int):
        super().__init__()
        self.w_gate = nn.Linear(dim, inter_dim, bias=False)
        self.w_up = nn.Linear(dim, inter_dim, bias=False)
        self.w_down = nn.Linear(inter_dim, dim, bias=False)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.w_down(F.silu(self.w_gate(x)) * self.w_up(x))