modeling_kambo.py

# coding=utf-8
"""Kambo-v1: a hybrid short-convolution / grouped-query-attention MoE.

The backbone is 24 layers. Six of them (3, 7, 11, 15, 19, 23) are grouped-query
attention with RoPE and QK-norm; the other eighteen are double-gated causal
short convolutions. Every layer's feed-forward is a mixture of experts: 16
routed experts at top-2 plus one shared expert that sees every token.

Two consequences shape this file:

  * Incremental decoding needs two different caches. The attention layers need
    the usual keys and values. The convolution layers need no keys or values at
    all -- only the last ``conv_kernel - 1`` columns of their pre-convolution
    signal, a few kilobytes that stay constant no matter how long the context
    grows. ``KamboCache`` holds both, and the model tells `generate` to leave
    cache construction alone (``_supports_default_dynamic_cache`` is False).

  * The convolution carries no positional encoding, so it cannot tell a padding
    token from a real one by position. Left-padded batches therefore zero the
    pre-convolution signal at padded positions, which is exactly what the
    causal left-pad does at the start of a sequence. Without that, the first
    two real tokens of a padded row convolve against the padding and a batch of
    two prompts does not reproduce the same two prompts run one at a time.
"""

from typing import List, Optional, Tuple, Union

import torch
import torch.nn as nn
import torch.nn.functional as F
import transformers
from transformers.modeling_outputs import BaseModelOutputWithPast, CausalLMOutputWithPast
from transformers.modeling_utils import PreTrainedModel
from transformers.generation import GenerationMixin

from .configuration_kambo import KamboConfig


# ---------------------------------------------------------------------------
# Cache
# ---------------------------------------------------------------------------

class KamboCache:
    """Per-layer state for incremental decoding.

    Deliberately not a subclass of ``transformers.Cache``: that contract assumes
    every layer stores keys and values, and eighteen of these layers store a
    convolution window instead. The model opts out of the default cache
    machinery and builds this itself in ``prepare_inputs_for_generation``.
    """

    def __init__(self):
        self.key_cache: dict = {}
        self.value_cache: dict = {}
        self.conv_states: dict = {}
        self._seen = 0

    def get_seq_length(self, layer_idx: int = 0) -> int:
        return self._seen

    # `generate` calls this on some paths to size a new cache.
    def get_max_cache_shape(self):
        return None

    def get_mask_sizes(self, cache_position, layer_idx: int = 0):
        return self._seen + cache_position.shape[0], self._seen

    def update_attention(self, key, value, layer_idx: int):
        if layer_idx in self.key_cache:
            key = torch.cat([self.key_cache[layer_idx], key], dim=2)
            value = torch.cat([self.value_cache[layer_idx], value], dim=2)
        self.key_cache[layer_idx] = key
        self.value_cache[layer_idx] = value
        return key, value

    def reorder(self, beam_idx: torch.LongTensor):
        for d in (self.key_cache, self.value_cache, self.conv_states):
            for i, t in d.items():
                d[i] = t.index_select(0, beam_idx.to(t.device))

    # Beam search calls this name on the cache object.
    def reorder_cache(self, beam_idx):
        self.reorder(beam_idx)

    def batch_select_indices(self, indices):
        self.reorder(indices)

    def crop(self, max_length: int):
        """Assisted decoding rolls the cache back when a draft is rejected.

        The attention layers can be sliced, but a convolution state is a sliding
        window that cannot be reconstructed from a shorter prefix without
        re-running the layer. Rather than return a silently wrong state, refuse:
        the caller sees an error instead of degraded output.
        """
        raise NotImplementedError(
            "Kambo caches a convolution window that cannot be cropped. "
            "Speculative/assisted decoding is not supported; use plain generate()."
        )

    def __len__(self):
        return self._seen


# ---------------------------------------------------------------------------
# Primitives
# ---------------------------------------------------------------------------

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

    def forward(self, x):
        dt = x.dtype
        x = x.float()
        x = x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps)
        return (x * self.weight.float()).to(dt)

    def extra_repr(self):
        return f"{tuple(self.weight.shape)}, eps={self.eps}"


def _rope_cache(seq: int, head_dim: int, theta: float, device, dtype):
    inv = 1.0 / (theta ** (torch.arange(0, head_dim, 2, device=device).float() / head_dim))
    t = torch.arange(seq, device=device).float()
    f = torch.outer(t, inv)
    return torch.cos(f).to(dtype), torch.sin(f).to(dtype)


def _apply_rope(x, cos, sin):
    """Split-half rotary embedding.

    ``cos``/``sin`` are ``head_dim // 2`` wide and are NOT duplicated to the full
    head width. The rotation pairs channel ``i`` with channel ``i + head_dim/2``.
    This is not the interleaved convention used by most Llama-family code; the
    weights were trained under this one, and swapping the two produces fluent
    output that is subtly and permanently wrong.
    """
    x1, x2 = x.chunk(2, dim=-1)
    return torch.cat([x1 * cos - x2 * sin, x2 * cos + x1 * sin], dim=-1)


class KamboShortConv(nn.Module):
    """Double-gated causal depthwise convolution.

    ``in_proj`` produces three streams; the convolution runs on ``b * v`` and its
    output is gated again by ``c``. No positional encoding of any kind.
    """

    def __init__(self, config: KamboConfig):
        super().__init__()
        d, k = config.hidden_size, config.conv_kernel
        self.k = k
        self.in_proj = nn.Linear(d, 3 * d, bias=False)
        self.conv = nn.Conv1d(d, d, k, groups=d, bias=False)
        self.out_proj = nn.Linear(d, d, bias=False)

    def forward(self, x, cache: Optional[KamboCache] = None, layer_idx: int = 0,
                token_mask: Optional[torch.Tensor] = None):
        b, c, v = self.in_proj(x).chunk(3, dim=-1)
        g = (b * v).transpose(1, 2)                      # [B, D, T]

        # Padding contributes zero, matching the zeros the causal left-pad
        # supplies at the start of a sequence.
        if token_mask is not None:
            g = g * token_mask[:, None, :].to(g.dtype)

        if cache is None or layer_idx not in cache.conv_states:
            past = g.new_zeros(g.shape[0], g.shape[1], self.k - 1)
        else:
            past = cache.conv_states[layer_idx]

        full = torch.cat([past, g], dim=-1)              # [B, D, (k-1) + T]
        if cache is not None:
            # Keep exactly k-1 columns regardless of T (T may be 1, or shorter
            # than k-1 on a very short prompt).
            cache.conv_states[layer_idx] = full[..., -(self.k - 1):].detach().clone()

        y = self.conv(full).transpose(1, 2)              # [B, T, D]
        return self.out_proj(c * y)


class KamboAttention(nn.Module):
    def __init__(self, config: KamboConfig, layer_idx: int):
        super().__init__()
        d, hd = config.hidden_size, config.head_dim
        self.layer_idx = layer_idx
        self.nq = config.num_attention_heads
        self.nkv = config.num_key_value_heads
        self.hd = hd
        self.rep = self.nq // self.nkv
        self.q_proj = nn.Linear(d, self.nq * hd, bias=False)
        self.k_proj = nn.Linear(d, self.nkv * hd, bias=False)
        self.v_proj = nn.Linear(d, self.nkv * hd, bias=False)
        self.o_proj = nn.Linear(self.nq * hd, d, bias=False)
        self.q_norm = KamboRMSNorm(hd, config.rms_norm_eps)
        self.k_norm = KamboRMSNorm(hd, config.rms_norm_eps)

    def forward(self, x, cos, sin, attn_bias=None, cache=None, use_causal=False):
        B, T, _ = x.shape
        q = self.q_proj(x).view(B, T, self.nq, self.hd).transpose(1, 2)
        k = self.k_proj(x).view(B, T, self.nkv, self.hd).transpose(1, 2)
        v = self.v_proj(x).view(B, T, self.nkv, self.hd).transpose(1, 2)

        # QK-norm first, rotary second. The reverse order also runs.
        q, k = self.q_norm(q), self.k_norm(k)
        q, k = _apply_rope(q, cos, sin), _apply_rope(k, cos, sin)

        if cache is not None:
            k, v = cache.update_attention(k, v, self.layer_idx)

        k = k.repeat_interleave(self.rep, dim=1)
        v = v.repeat_interleave(self.rep, dim=1)

        o = F.scaled_dot_product_attention(
            q, k, v, attn_mask=attn_bias, is_causal=use_causal
        )
        return self.o_proj(o.transpose(1, 2).reshape(B, T, -1))


class KamboMoE(nn.Module):
    """16 routed experts at top-2, plus one shared expert on every token.

    Inference is exactly dropless: tokens are sorted by expert and each expert
    runs one GEMM over its own rows. Training used a capacity-based batched
    path for speed, which can drop an assignment when an expert is
    oversubscribed; at inference there is no throughput reason to accept that
    approximation, and the loop is the path the capacity version approximates.
    """

    def __init__(self, config: KamboConfig):
        super().__init__()
        d, dff, E = config.hidden_size, config.d_ff, config.n_experts
        self.E, self.k, self.d, self.dff = E, config.top_k, d, dff
        self.router = nn.Linear(d, E, bias=False)
        self.w1 = nn.Parameter(torch.empty(E, d, dff))
        self.w3 = nn.Parameter(torch.empty(E, d, dff))
        self.w2 = nn.Parameter(torch.empty(E, dff, d))
        self.sw1 = nn.Linear(d, dff, bias=False)
        self.sw3 = nn.Linear(d, dff, bias=False)
        self.sw2 = nn.Linear(dff, d, bias=False)

    def forward(self, x):
        B, T, D = x.shape
        xf = x.reshape(-1, D)

        # The router runs in fp32 and must be written out explicitly: a plain
        # module call would be demoted to bf16 under autocast, and this is the
        # one place in the model where that changes which experts are selected.
        dev_type = xf.device.type
        with torch.autocast(device_type=dev_type, enabled=False):
            logits = F.linear(xf.float(), self.router.weight.float())
            probs = logits.softmax(-1)
            topv, topi = probs.topk(self.k, dim=-1)
            topv = topv / topv.sum(-1, keepdim=True)

        out = self.sw2(F.silu(self.sw1(xf)) * self.sw3(xf))

        flat_e = topi.reshape(-1)
        flat_w = topv.reshape(-1).to(x.dtype)
        order = torch.argsort(flat_e)
        tok = torch.div(order, self.k, rounding_mode="floor")
        counts = torch.bincount(flat_e, minlength=self.E).tolist()

        xs = xf[tok]
        ws = flat_w[order].unsqueeze(-1)
        ys = torch.empty_like(xs)
        s = 0
        for e in range(self.E):
            n = counts[e]
            if n == 0:
                continue
            xe = xs[s:s + n]
            h = F.silu(xe @ self.w1[e]) * (xe @ self.w3[e])
            ys[s:s + n] = h @ self.w2[e]
            s += n

        out = out.index_add(0, tok, (ys * ws).to(out.dtype))
        return out.view(B, T, D)


class KamboDecoderLayer(nn.Module):
    def __init__(self, config: KamboConfig, layer_idx: int):
        super().__init__()
        self.layer_idx = layer_idx
        self.is_attn = layer_idx in config.gqa_layers
        self.input_layernorm = KamboRMSNorm(config.hidden_size, config.rms_norm_eps)
        if self.is_attn:
            self.self_attn = KamboAttention(config, layer_idx)
        else:
            self.conv = KamboShortConv(config)
        self.post_attention_layernorm = KamboRMSNorm(config.hidden_size, config.rms_norm_eps)
        self.moe = KamboMoE(config)

    def forward(self, x, cos=None, sin=None, attn_bias=None, cache=None,
                use_causal=False, token_mask=None):
        h = self.input_layernorm(x)
        if self.is_attn:
            h = self.self_attn(h, cos, sin, attn_bias=attn_bias, cache=cache,
                               use_causal=use_causal)
        else:
            h = self.conv(h, cache=cache, layer_idx=self.layer_idx,
                          token_mask=token_mask)
        x = x + h
        x = x + self.moe(self.post_attention_layernorm(x))
        return x


# ---------------------------------------------------------------------------
# Model
# ---------------------------------------------------------------------------

class KamboPreTrainedModel(PreTrainedModel):
    config_class = KamboConfig
    base_model_prefix = "model"
    supports_gradient_checkpointing = True
    _no_split_modules = ["KamboDecoderLayer"]
    _skip_keys_device_placement = "past_key_values"
    _supports_sdpa = True

    def _init_weights(self, module):
        std = 0.02
        if isinstance(module, (nn.Linear, nn.Conv1d)):
            module.weight.data.normal_(mean=0.0, std=std)
            if getattr(module, "bias", None) is not None:
                module.bias.data.zero_()
        elif isinstance(module, nn.Embedding):
            module.weight.data.normal_(mean=0.0, std=std)
        elif isinstance(module, KamboRMSNorm):
            module.weight.data.fill_(1.0)
        elif isinstance(module, KamboMoE):
            for p in (module.w1, module.w2, module.w3):
                p.data.normal_(mean=0.0, std=std)


def _build_attn_bias(attention_mask, q_len, kv_len, past_len, device, dtype):
    """Additive [B, 1, q_len, kv_len] mask: causal AND not-padding."""
    q_pos = torch.arange(q_len, device=device) + past_len
    k_pos = torch.arange(kv_len, device=device)
    allowed = (k_pos[None, :] <= q_pos[:, None])[None, None, :, :]

    if attention_mask is not None:
        pad = attention_mask[:, None, None, :].bool()
        allowed = allowed & pad

    # A row that is entirely masked would softmax over all -inf and produce
    # NaN, which then propagates through the whole sequence. Fully padded rows
    # exist in real batches; let such a row attend to itself and discard the
    # result downstream rather than poisoning the batch.
    allowed = allowed | (~allowed.any(dim=-1, keepdim=True))

    bias = torch.zeros(allowed.shape, device=device, dtype=dtype)
    return bias.masked_fill(~allowed, torch.finfo(dtype).min)


class KamboModel(KamboPreTrainedModel):
    def __init__(self, config: KamboConfig):
        super().__init__(config)
        self.embed_tokens = nn.Embedding(config.vocab_size, config.hidden_size)
        self.layers = nn.ModuleList(
            [KamboDecoderLayer(config, i) for i in range(config.num_hidden_layers)]
        )
        self.norm = KamboRMSNorm(config.hidden_size, config.rms_norm_eps)
        self.gradient_checkpointing = False
        self._rope = None
        self.post_init()

    def get_input_embeddings(self):
        return self.embed_tokens

    def set_input_embeddings(self, value):
        self.embed_tokens = value

    def _rope_for(self, position_ids, dtype, device):
        need = int(position_ids.max().item()) + 1
        if self._rope is None or self._rope[0].shape[0] < need or self._rope[0].device != device:
            size = max(need, self.config.max_position_embeddings)
            self._rope = _rope_cache(size, self.config.head_dim,
                                     self.config.rope_theta, device, torch.float32)
        cos, sin = self._rope
        # [B, T, hd/2] -> [B, 1, T, hd/2] so each row uses its own positions,
        # which is what makes left-padded batches agree with unpadded singles.
        return (cos[position_ids].unsqueeze(1).to(dtype),
                sin[position_ids].unsqueeze(1).to(dtype))

    def forward(
        self,
        input_ids: Optional[torch.LongTensor] = None,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        past_key_values: Optional[KamboCache] = None,
        inputs_embeds: Optional[torch.FloatTensor] = None,
        use_cache: Optional[bool] = None,
        output_hidden_states: Optional[bool] = None,
        return_dict: Optional[bool] = None,
        **kwargs,
    ):
        use_cache = use_cache if use_cache is not None else self.config.use_cache
        return_dict = return_dict if return_dict is not None else True
        output_hidden_states = bool(output_hidden_states)

        if (input_ids is None) == (inputs_embeds is None):
            raise ValueError("Pass exactly one of input_ids or inputs_embeds.")
        if inputs_embeds is None:
            inputs_embeds = self.embed_tokens(input_ids)

        x = inputs_embeds
        B, T, _ = x.shape
        device = x.device

        if self.gradient_checkpointing and self.training:
            use_cache = False
        if use_cache and past_key_values is None:
            past_key_values = KamboCache()
        past_len = past_key_values.get_seq_length() if past_key_values is not None else 0
        kv_len = past_len + T

        if position_ids is None:
            if attention_mask is not None:
                # cumsum over the full mask handles left padding: the first real
                # token gets position 0 no matter how much padding precedes it.
                pos_full = (attention_mask.long().cumsum(-1) - 1).clamp(min=0)
                position_ids = pos_full[:, -T:]
            else:
                position_ids = torch.arange(past_len, kv_len, device=device).unsqueeze(0).expand(B, T)

        cos, sin = self._rope_for(position_ids, x.dtype, device)

        # The fast path -- a single unpadded sequence -- is exactly what the
        # training code ran, so parity is checked against it directly.
        use_causal = attention_mask is None and past_len == 0 and T > 1
        attn_bias = None
        if not use_causal and not (attention_mask is None and T == 1 and past_len == 0):
            attn_bias = _build_attn_bias(attention_mask, T, kv_len, past_len, device, x.dtype)

        token_mask = attention_mask[:, -T:] if attention_mask is not None else None

        all_hidden = [] if output_hidden_states else None
        for layer in self.layers:
            if all_hidden is not None:
                all_hidden.append(x)
            if self.gradient_checkpointing and self.training:
                x = self._gradient_checkpointing_func(
                    layer.__call__, x, cos, sin, attn_bias, past_key_values,
                    use_causal, token_mask,
                )
            else:
                x = layer(x, cos, sin, attn_bias=attn_bias, cache=past_key_values,
                          use_causal=use_causal, token_mask=token_mask)

        x = self.norm(x)
        if all_hidden is not None:
            all_hidden.append(x)

        if past_key_values is not None:
            past_key_values._seen = kv_len

        if not return_dict:
            return tuple(v for v in (x, past_key_values, all_hidden) if v is not None)
        return BaseModelOutputWithPast(
            last_hidden_state=x,
            past_key_values=past_key_values if use_cache else None,
            hidden_states=tuple(all_hidden) if all_hidden is not None else None,
        )


# transformers 5 expects a {tied: source} mapping here; 4.x expects a flat list
# and raises on a dict. Both spellings mean the same thing -- lm_head shares the
# embedding matrix -- so pick by version rather than pinning users to one.
_TIED = ({"lm_head.weight": "model.embed_tokens.weight"}
         if int(transformers.__version__.split(".")[0]) >= 5
         else ["lm_head.weight"])


class KamboForCausalLM(KamboPreTrainedModel, GenerationMixin):
    _tied_weights_keys = _TIED

    def __init__(self, config: KamboConfig):
        super().__init__(config)
        self.model = KamboModel(config)
        self.vocab_size = config.vocab_size
        self.lm_head = nn.Linear(config.hidden_size, config.vocab_size, bias=False)
        self.post_init()

    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):
        self.lm_head = new

    def get_decoder(self):
        return self.model

    # Tell `generate` not to build a Cache for us: eighteen of these layers
    # hold a convolution window, not keys and values. Honoured identically by
    # transformers 4.x and 5.x, both of which take this as the signal that the
    # model supplies its own cache in prepare_inputs_for_generation.
    def _supports_default_dynamic_cache(self) -> bool:
        return False

    def forward(
        self,
        input_ids: Optional[torch.LongTensor] = None,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.LongTensor] = None,
        past_key_values: Optional[KamboCache] = None,
        inputs_embeds: Optional[torch.FloatTensor] = None,
        labels: Optional[torch.LongTensor] = None,
        use_cache: Optional[bool] = None,
        output_hidden_states: Optional[bool] = None,
        return_dict: Optional[bool] = None,
        logits_to_keep: Union[int, torch.Tensor] = 0,
        **kwargs,
    ):
        return_dict = return_dict if return_dict is not None else True
        # transformers renamed this argument; accept the older spelling too.
        if "num_logits_to_keep" in kwargs:
            logits_to_keep = kwargs.pop("num_logits_to_keep")

        out = self.model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            position_ids=position_ids,
            past_key_values=past_key_values,
            inputs_embeds=inputs_embeds,
            use_cache=use_cache,
            output_hidden_states=output_hidden_states,
            return_dict=True,
        )

        h = out.last_hidden_state
        if isinstance(logits_to_keep, int):
            if logits_to_keep > 0:
                h = h[:, -logits_to_keep:, :]
        else:
            h = h[:, logits_to_keep, :]
        logits = self.lm_head(h).float()

        loss = None
        if labels is not None:
            loss = self.loss_function(
                logits=logits, labels=labels, vocab_size=self.config.vocab_size, **kwargs
            )

        if not return_dict:
            return tuple(v for v in (loss, logits, out.past_key_values) if v is not None)
        return CausalLMOutputWithPast(
            loss=loss,
            logits=logits,
            past_key_values=out.past_key_values,
            hidden_states=out.hidden_states,
        )

    def prepare_inputs_for_generation(
        self,
        input_ids,
        past_key_values=None,
        attention_mask=None,
        inputs_embeds=None,
        cache_position=None,
        use_cache=True,
        **kwargs,
    ):
        if use_cache and past_key_values is None:
            past_key_values = KamboCache()

        past_len = past_key_values.get_seq_length() if past_key_values is not None else 0
        if past_len > 0:
            input_ids = input_ids[:, past_len:]

        position_ids = kwargs.get("position_ids")
        if position_ids is None and attention_mask is not None:
            position_ids = (attention_mask.long().cumsum(-1) - 1).clamp(min=0)
        if position_ids is not None:
            position_ids = position_ids[:, -input_ids.shape[1]:]

        model_inputs = {
            "input_ids": input_ids,
            "past_key_values": past_key_values,
            "attention_mask": attention_mask,
            "position_ids": position_ids,
            "use_cache": use_cache,
        }
        # Only the last position's logits are ever sampled; computing the full
        # [B, T, 151936] head over a long prompt is pure waste.
        if past_len == 0 and input_ids.shape[1] > 1:
            model_inputs["logits_to_keep"] = 1
        return model_inputs

    def _reorder_cache(self, past_key_values, beam_idx):
        if past_key_values is not None:
            past_key_values.reorder(beam_idx)
        return past_key_values


__all__ = ["KamboConfig", "KamboModel", "KamboForCausalLM", "KamboPreTrainedModel", "KamboCache"]