micro-glm-tinystories / models /micro_deepseek.py

Upload complete MicroGPT model (hf_model_micro_glm_ckpt_step_500)

507f954 verified 3 months ago

9.51 kB

"""
MicroDeepSeek: Tiny Mixture-of-Experts transformer in PyTorch.
Inspired by DeepSeek's MoE architecture with top-k routing.
Architecture:
- Standard transformer decoder with causal attention
- MoE FFN layers instead of standard MLP (4 experts, top-2 routing)
- Load balancing loss for expert utilization
- Tiny config to fit in 6GB VRAM
Reference: "DeepSeek-V2: A Strong, Economical, and Efficient Mixture-of-Experts Language Model"
"""

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


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

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


class CausalSelfAttention(nn.Module):
    def __init__(self, n_embd, n_head, block_size, dropout):
        super().__init__()
        assert n_embd % n_head == 0
        self.n_head = n_head
        self.head_dim = n_embd // n_head
        self.q = nn.Linear(n_embd, n_embd, bias=False)
        self.k = nn.Linear(n_embd, n_embd, bias=False)
        self.v = nn.Linear(n_embd, n_embd, bias=False)
        self.proj = nn.Linear(n_embd, n_embd, bias=False)
        self.attn_drop = nn.Dropout(dropout)
        self.resid_drop = nn.Dropout(dropout)
        self.register_buffer('mask', torch.tril(torch.ones(block_size, block_size)).view(1, 1, block_size, block_size))

    def forward(self, x):
        B, T, C = x.shape
        q = self.q(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
        k = self.k(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
        v = self.v(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
        att = (q @ k.transpose(-2, -1)) / math.sqrt(self.head_dim)
        att = att.masked_fill(self.mask[:, :, :T, :T] == 0, float('-inf'))
        att = F.softmax(att, dim=-1)
        att = self.attn_drop(att)
        y = att @ v
        y = y.transpose(1, 2).contiguous().view(B, T, C)
        return self.resid_drop(self.proj(y))


class MoEExpert(nn.Module):
    """A single expert FFN."""
    def __init__(self, n_embd, hidden_mult=4):
        super().__init__()
        self.fc1 = nn.Linear(n_embd, hidden_mult * n_embd, bias=False)
        self.fc2 = nn.Linear(hidden_mult * n_embd, n_embd, bias=False)

    def forward(self, x):
        x = F.gelu(self.fc1(x))
        x = self.fc2(x)
        return x


class TopKRouter(nn.Module):
    """
    Routes tokens to top-k experts.
    Returns: expert outputs, router z-loss (for load balancing)
    """
    def __init__(self, n_embd, num_experts, top_k=2):
        super().__init__()
        self.num_experts = num_experts
        self.top_k = top_k
        self.gate = nn.Linear(n_embd, num_experts, bias=False)

    def forward(self, x):
        """
        x: (B*T, C) - flattened tokens
        Returns:
          routing_weights: (B*T, top_k) - weights per selected expert
          expert_indices: (B*T, top_k) - selected expert ids
          router_loss: scalar - auxiliary load balancing loss
        """
        # Compute router logits
        gate_logits = self.gate(x)  # (B*T, num_experts)
        routing_weights = F.softmax(gate_logits, dim=-1)

        # Top-k selection
        top_k_weights, top_k_indices = torch.topk(routing_weights, self.top_k, dim=-1)

        # Normalize top-k weights
        top_k_weights = top_k_weights / (top_k_weights.sum(dim=-1, keepdim=True) + 1e-6)

        # Router z-loss for load balancing
        # Encourages uniform routing by penalizing large logits
        router_loss = torch.mean(gate_logits ** 2) * 0.01  # scale factor

        return top_k_weights, top_k_indices, router_loss


class MoEFeedForward(nn.Module):
    """
    Mixture of Experts FFN layer.
    Uses top-k routing with auxiliary load balancing loss.
    """
    def __init__(self, n_embd, num_experts=4, top_k=2, dropout=0.1):
        super().__init__()
        self.num_experts = num_experts
        self.top_k = top_k
        self.router = TopKRouter(n_embd, num_experts, top_k)
        self.experts = nn.ModuleList([MoEExpert(n_embd) for _ in range(num_experts)])
        self.drop = nn.Dropout(dropout)

    def forward(self, x):
        """
        x: (B, T, C)
        Returns: (B, T, C), router_loss
        """
        B, T, C = x.shape
        x_flat = x.reshape(-1, C)  # (B*T, C)

        # Route tokens
        routing_weights, expert_indices, router_loss = self.router(x_flat)

        # Initialize output buffer
        out = torch.zeros_like(x_flat)

        # Scatter tokens to selected experts
        for expert_id in range(self.num_experts):
            # Find tokens routed to this expert
            mask = (expert_indices == expert_id)
            if not mask.any():
                continue

            # Get the routing weights for this expert
            # mask shape: (B*T, top_k) - True where expert_id is selected
            token_mask = mask.any(dim=-1)  # (B*T,) - True if any top-k slot selected this expert
            token_indices = token_mask.nonzero(as_tuple=True)[0]  # indices of tokens that use this expert

            if len(token_indices) == 0:
                continue

            # Get the routing weight of this expert for each token
            # For each token, find which slot (0..top_k-1) has this expert
            token_weights = []
            for i in token_indices:
                # Find which slot(s) have this expert
                slots = (expert_indices[i] == expert_id).nonzero(as_tuple=True)[0]
                if len(slots) > 0:
                    # Take the weight from the first matching slot
                    token_weights.append(routing_weights[i, slots[0]])
                else:
                    token_weights.append(0.0)
            token_weights = torch.tensor(token_weights, device=x.device).unsqueeze(-1)

            # Run the expert
            expert_input = x_flat[token_indices]  # (N, C)
            expert_output = self.experts[expert_id](expert_input)  # (N, C)

            # Weighted sum into output
            out[token_indices] += expert_output * token_weights

        out = out.view(B, T, C)
        out = self.drop(out)

        # Add load balancing loss
        return out, router_loss


class MoEBlock(nn.Module):
    def __init__(self, n_embd, n_head, block_size, dropout, num_experts=4, top_k=2):
        super().__init__()
        self.ln1 = RMSNorm(n_embd)
        self.attn = CausalSelfAttention(n_embd, n_head, block_size, dropout)
        self.ln2 = RMSNorm(n_embd)
        self.moe = MoEFeedForward(n_embd, num_experts, top_k, dropout)

    def forward(self, x):
        x = x + self.attn(self.ln1(x))
        moe_out, router_loss = self.moe(self.ln2(x))
        x = x + moe_out
        return x, router_loss


class MicroDeepSeek(nn.Module):
    """
    Tiny MoE transformer inspired by DeepSeek.
    - Standard causal attention
    - MoE FFN with top-k routing (4 experts, top-2)
    - Auxiliary load balancing loss
    - Tiny config for 6GB VRAM
    """

    def __init__(self, vocab_size, block_size, n_layer=2, n_head=4, n_embd=128, dropout=0.1,
                 num_experts=4, top_k=2):
        super().__init__()
        self.block_size = block_size
        self.num_experts = num_experts
        self.top_k = top_k
        self.wte = nn.Embedding(vocab_size, n_embd)
        self.wpe = nn.Embedding(block_size, n_embd)
        self.blocks = nn.ModuleList([
            MoEBlock(n_embd, n_head, block_size, dropout, num_experts, top_k)
            for _ in range(n_layer)
        ])
        self.ln_f = RMSNorm(n_embd)
        self.lm_head = nn.Linear(n_embd, vocab_size, bias=False)
        self.lm_head.weight = self.wte.weight
        self.apply(self._init_weights)

    def _init_weights(self, module):
        if isinstance(module, nn.Linear):
            nn.init.normal_(module.weight, mean=0.0, std=0.02)
        elif isinstance(module, nn.Embedding):
            nn.init.normal_(module.weight, mean=0.0, std=0.02)

    def forward(self, idx, targets=None):
        B, T = idx.shape
        if T > self.block_size:
            raise ValueError(f'block size exceeded: {T} > {self.block_size}')
        pos = torch.arange(T, device=idx.device)
        x = self.wte(idx) + self.wpe(pos)

        total_router_loss = 0.0
        for block in self.blocks:
            x, router_loss = block(x)
            total_router_loss = total_router_loss + router_loss

        x = self.ln_f(x)
        logits = self.lm_head(x)
        loss = None
        if targets is not None:
            ce_loss = F.cross_entropy(logits.reshape(-1, logits.size(-1)), targets.reshape(-1))
            # Combine cross-entropy loss with auxiliary router loss
            loss = ce_loss + total_router_loss
        return logits, loss

    @torch.no_grad()
    def generate(self, idx, max_new_tokens, temperature=1.0, top_k=40):
        self.eval()
        for _ in range(max_new_tokens):
            idx_cond = idx[:, -self.block_size:]
            logits, _ = self(idx_cond)
            logits = logits[:, -1, :] / temperature
            if top_k is not None:
                v, _ = torch.topk(logits, min(top_k, logits.size(-1)))
                logits[logits < v[:, [-1]]] = -float('inf')
            probs = F.softmax(logits, dim=-1)
            idx_next = torch.multinomial(probs, num_samples=1)
            idx = torch.cat([idx, idx_next], dim=1)
        return idx

	"""
	MicroDeepSeek: Tiny Mixture-of-Experts transformer in PyTorch.
	Inspired by DeepSeek's MoE architecture with top-k routing.

	Architecture:
	- Standard transformer decoder with causal attention
	- MoE FFN layers instead of standard MLP (4 experts, top-2 routing)
	- Load balancing loss for expert utilization
	- Tiny config to fit in 6GB VRAM

	Reference: "DeepSeek-V2: A Strong, Economical, and Efficient Mixture-of-Experts Language Model"
	"""

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


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

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


	class CausalSelfAttention(nn.Module):
	def __init__(self, n_embd, n_head, block_size, dropout):
	super().__init__()
	assert n_embd % n_head == 0
	self.n_head = n_head
	self.head_dim = n_embd // n_head
	self.q = nn.Linear(n_embd, n_embd, bias=False)
	self.k = nn.Linear(n_embd, n_embd, bias=False)
	self.v = nn.Linear(n_embd, n_embd, bias=False)
	self.proj = nn.Linear(n_embd, n_embd, bias=False)
	self.attn_drop = nn.Dropout(dropout)
	self.resid_drop = nn.Dropout(dropout)
	self.register_buffer('mask', torch.tril(torch.ones(block_size, block_size)).view(1, 1, block_size, block_size))

	def forward(self, x):
	B, T, C = x.shape
	q = self.q(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
	k = self.k(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
	v = self.v(x).view(B, T, self.n_head, self.head_dim).transpose(1, 2)
	att = (q @ k.transpose(-2, -1)) / math.sqrt(self.head_dim)
	att = att.masked_fill(self.mask[:, :, :T, :T] == 0, float('-inf'))
	att = F.softmax(att, dim=-1)
	att = self.attn_drop(att)
	y = att @ v
	y = y.transpose(1, 2).contiguous().view(B, T, C)
	return self.resid_drop(self.proj(y))


	class MoEExpert(nn.Module):
	"""A single expert FFN."""
	def __init__(self, n_embd, hidden_mult=4):
	super().__init__()
	self.fc1 = nn.Linear(n_embd, hidden_mult * n_embd, bias=False)
	self.fc2 = nn.Linear(hidden_mult * n_embd, n_embd, bias=False)

	def forward(self, x):
	x = F.gelu(self.fc1(x))
	x = self.fc2(x)
	return x


	class TopKRouter(nn.Module):
	"""
	Routes tokens to top-k experts.
	Returns: expert outputs, router z-loss (for load balancing)
	"""
	def __init__(self, n_embd, num_experts, top_k=2):
	super().__init__()
	self.num_experts = num_experts
	self.top_k = top_k
	self.gate = nn.Linear(n_embd, num_experts, bias=False)

	def forward(self, x):
	"""
	x: (B*T, C) - flattened tokens
	Returns:
	routing_weights: (B*T, top_k) - weights per selected expert
	expert_indices: (B*T, top_k) - selected expert ids
	router_loss: scalar - auxiliary load balancing loss
	"""
	# Compute router logits
	gate_logits = self.gate(x) # (B*T, num_experts)
	routing_weights = F.softmax(gate_logits, dim=-1)

	# Top-k selection
	top_k_weights, top_k_indices = torch.topk(routing_weights, self.top_k, dim=-1)

	# Normalize top-k weights
	top_k_weights = top_k_weights / (top_k_weights.sum(dim=-1, keepdim=True) + 1e-6)

	# Router z-loss for load balancing
	# Encourages uniform routing by penalizing large logits
	router_loss = torch.mean(gate_logits ** 2) * 0.01 # scale factor

	return top_k_weights, top_k_indices, router_loss


	class MoEFeedForward(nn.Module):
	"""
	Mixture of Experts FFN layer.
	Uses top-k routing with auxiliary load balancing loss.
	"""
	def __init__(self, n_embd, num_experts=4, top_k=2, dropout=0.1):
	super().__init__()
	self.num_experts = num_experts
	self.top_k = top_k
	self.router = TopKRouter(n_embd, num_experts, top_k)
	self.experts = nn.ModuleList([MoEExpert(n_embd) for _ in range(num_experts)])
	self.drop = nn.Dropout(dropout)

	def forward(self, x):
	"""
	x: (B, T, C)
	Returns: (B, T, C), router_loss
	"""
	B, T, C = x.shape
	x_flat = x.reshape(-1, C) # (B*T, C)

	# Route tokens
	routing_weights, expert_indices, router_loss = self.router(x_flat)

	# Initialize output buffer
	out = torch.zeros_like(x_flat)

	# Scatter tokens to selected experts
	for expert_id in range(self.num_experts):
	# Find tokens routed to this expert
	mask = (expert_indices == expert_id)
	if not mask.any():
	continue

	# Get the routing weights for this expert
	# mask shape: (B*T, top_k) - True where expert_id is selected
	token_mask = mask.any(dim=-1) # (B*T,) - True if any top-k slot selected this expert
	token_indices = token_mask.nonzero(as_tuple=True)[0] # indices of tokens that use this expert

	if len(token_indices) == 0:
	continue

	# Get the routing weight of this expert for each token
	# For each token, find which slot (0..top_k-1) has this expert
	token_weights = []
	for i in token_indices:
	# Find which slot(s) have this expert
	slots = (expert_indices[i] == expert_id).nonzero(as_tuple=True)[0]
	if len(slots) > 0:
	# Take the weight from the first matching slot
	token_weights.append(routing_weights[i, slots[0]])
	else:
	token_weights.append(0.0)
	token_weights = torch.tensor(token_weights, device=x.device).unsqueeze(-1)

	# Run the expert
	expert_input = x_flat[token_indices] # (N, C)
	expert_output = self.experts[expert_id](expert_input) # (N, C)

	# Weighted sum into output
	out[token_indices] += expert_output * token_weights

	out = out.view(B, T, C)
	out = self.drop(out)

	# Add load balancing loss
	return out, router_loss


	class MoEBlock(nn.Module):
	def __init__(self, n_embd, n_head, block_size, dropout, num_experts=4, top_k=2):
	super().__init__()
	self.ln1 = RMSNorm(n_embd)
	self.attn = CausalSelfAttention(n_embd, n_head, block_size, dropout)
	self.ln2 = RMSNorm(n_embd)
	self.moe = MoEFeedForward(n_embd, num_experts, top_k, dropout)

	def forward(self, x):
	x = x + self.attn(self.ln1(x))
	moe_out, router_loss = self.moe(self.ln2(x))
	x = x + moe_out
	return x, router_loss


	class MicroDeepSeek(nn.Module):
	"""
	Tiny MoE transformer inspired by DeepSeek.
	- Standard causal attention
	- MoE FFN with top-k routing (4 experts, top-2)
	- Auxiliary load balancing loss
	- Tiny config for 6GB VRAM
	"""

	def __init__(self, vocab_size, block_size, n_layer=2, n_head=4, n_embd=128, dropout=0.1,
	num_experts=4, top_k=2):
	super().__init__()
	self.block_size = block_size
	self.num_experts = num_experts
	self.top_k = top_k
	self.wte = nn.Embedding(vocab_size, n_embd)
	self.wpe = nn.Embedding(block_size, n_embd)
	self.blocks = nn.ModuleList([
	MoEBlock(n_embd, n_head, block_size, dropout, num_experts, top_k)
	for _ in range(n_layer)
	])
	self.ln_f = RMSNorm(n_embd)
	self.lm_head = nn.Linear(n_embd, vocab_size, bias=False)
	self.lm_head.weight = self.wte.weight
	self.apply(self._init_weights)

	def _init_weights(self, module):
	if isinstance(module, nn.Linear):
	nn.init.normal_(module.weight, mean=0.0, std=0.02)
	elif isinstance(module, nn.Embedding):
	nn.init.normal_(module.weight, mean=0.0, std=0.02)

	def forward(self, idx, targets=None):
	B, T = idx.shape
	if T > self.block_size:
	raise ValueError(f'block size exceeded: {T} > {self.block_size}')
	pos = torch.arange(T, device=idx.device)
	x = self.wte(idx) + self.wpe(pos)

	total_router_loss = 0.0
	for block in self.blocks:
	x, router_loss = block(x)
	total_router_loss = total_router_loss + router_loss

	x = self.ln_f(x)
	logits = self.lm_head(x)
	loss = None
	if targets is not None:
	ce_loss = F.cross_entropy(logits.reshape(-1, logits.size(-1)), targets.reshape(-1))
	# Combine cross-entropy loss with auxiliary router loss
	loss = ce_loss + total_router_loss
	return logits, loss

	@torch.no_grad()
	def generate(self, idx, max_new_tokens, temperature=1.0, top_k=40):
	self.eval()
	for _ in range(max_new_tokens):
	idx_cond = idx[:, -self.block_size:]
	logits, _ = self(idx_cond)
	logits = logits[:, -1, :] / temperature
	if top_k is not None:
	v, _ = torch.topk(logits, min(top_k, logits.size(-1)))
	logits[logits < v[:, [-1]]] = -float('inf')
	probs = F.softmax(logits, dim=-1)
	idx_next = torch.multinomial(probs, num_samples=1)
	idx = torch.cat([idx, idx_next], dim=1)
	return idx