minimax-m1-lightning-attention

name	minimax-m1-lightning-attention
title	MiniMax-M1: Scaling Test-Time Compute Efficiently with Lightning Attention
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2506.13585
keywords	["reasoning","test-time-compute","mixture-of-experts","lightning-attention","reinforcement-learning"]
description	Hybrid-attention MoE reasoning model supporting 1M token context and 80K token generation, combining lightning attention with CISPO RL algorithm for efficient scaling.

MiniMax-M1: Scaling Test-Time Compute Efficiently with Lightning Attention

Core Concept

MiniMax-M1 is an open-weight reasoning model combining a hybrid Mixture-of-Experts architecture with lightning attention mechanism. With 456B total parameters and 45.9B activated per token, it supports 1 million token context length and 80K token generation. A novel RL algorithm called CISPO (Clipped IS-weight Policy Optimization) achieves the efficiency gains of DAPO with 50% fewer training steps by clipping importance-sampling weights rather than token probabilities. The model excels through large-scale reinforcement learning on diverse reasoning tasks completed in three weeks using 512 H800 GPUs.

Architecture Overview

• Hybrid Attention Design: Combines transnormer blocks with lightning attention and periodic softmax attention for near-linear scaling
• Mixture-of-Experts: Sparse routing with 45.9B activated parameters enabling efficient scaling
• CISPO Algorithm: Clips importance sampling weights rather than token updates, matching DAPO with less training
• Multi-Stage Training: Continual pretraining → supervised fine-tuning → large-scale RL
• Diverse RL Data: Verifiable tasks (math, coding, reasoning) + general domain with model-based rewards

Implementation

Step 1: Implement Lightning Attention

Create efficient attention mechanism with near-linear complexity:

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

class LightningAttention(nn.Module):
    """
    Lightning attention: efficient attention with near-linear complexity.
    Combines linear recurrence with periodic exact attention.
    """
    def __init__(self, hidden_size, num_heads, use_rms_norm=True):
        super().__init__()
        self.hidden_size = hidden_size
        self.num_heads = num_heads
        self.head_dim = hidden_size // num_heads

        # Q, K, V projections
        self.q_proj = nn.Linear(hidden_size, hidden_size)
        self.k_proj = nn.Linear(hidden_size, hidden_size)
        self.v_proj = nn.Linear(hidden_size, hidden_size)
        self.o_proj = nn.Linear(hidden_size, hidden_size)

        # RMS normalization
        if use_rms_norm:
            self.norm_q = nn.Identity()
            self.norm_k = nn.Identity()
        else:
            self.norm_q = nn.LayerNorm(self.head_dim)
            self.norm_k = nn.LayerNorm(self.head_dim)

        # Recurrent weight for linear attention
        self.beta = nn.Parameter(torch.ones(num_heads) * 0.5)

    def forward(self, x, attention_mask=None, use_exact_attention=False):
        
        batch, seq_len, _ = x.shape

        
        q = .q_proj(x).reshape(batch, seq_len, .num_heads, .head_dim)
        k = .k_proj(x).reshape(batch, seq_len, .num_heads, .head_dim)
        v = .v_proj(x).reshape(batch, seq_len, .num_heads, .head_dim)

        
        q = .norm_q(q)
        k = .norm_k(k)

         use_exact_attention:
            
            scores = torch.matmul(q, k.transpose(-, -))
            scores = scores / (.head_dim ** )

             attention_mask   :
                scores = scores.masked_fill(~attention_mask, ())

            attn_weights = F.softmax(scores, dim=-)
            output = torch.matmul(attn_weights, v)

        :
            
            
            output = ._linear_attention(q, k, v, batch, seq_len)

        
        output = output.reshape(batch, seq_len, .hidden_size)
        output = .o_proj(output)

         output

     ():
        
        
        k_activated = F.elu(k) + 
        q_activated = F.elu(q) + 

        
        outputs = []

         t  (seq_len):
            q_t = q_activated[:, t]  
            k_t = k_activated[:, t]  
            v_t = v[:, t]  

             t == :
                
                numerator = torch.einsum(, k_t, v_t)
                denominator = k_t.(dim=, keepdim=)
            :
                
                beta_t = .beta.unsqueeze().unsqueeze(-)
                numerator = beta_t * numerator + ( - beta_t) * torch.einsum(
                    , k_t, v_t
                )
                denominator = beta_t * denominator + ( - beta_t) * k_t

            
            output_t = torch.einsum(, q_t, numerator) / (
                denominator.(dim=, keepdim=) + 
            )
            outputs.append(output_t)

        output = torch.stack(outputs, dim=)  

         output

Step 2: Implement Hybrid Attention Block

Combine lightning and softmax attention strategically:

class HybridAttentionBlock(nn.Module):
    """
    Block alternating between lightning and exact attention.
    Lightning for efficiency, softmax periodically for stability.
    """
    def __init__(self, hidden_size, num_heads, use_exact_every_n=4):
        super().__init__()
        self.hidden_size = hidden_size
        self.num_heads = num_heads
        self.use_exact_every_n = use_exact_every_n

        self.lightning_attn = LightningAttention(hidden_size, num_heads)

    def forward(self, x, step=0):
        """
        Args:
            x: [batch, seq_len, hidden_size]
            step: block step number (to determine exact vs lightning)
        """
        # Alternate between lightning and exact attention
        use_exact = (step % self.use_exact_every_n == 0)

        if use_exact:
            return self.lightning_attn(x, use_exact_attention=True)
        else:
            return self.lightning_attn(x, use_exact_attention=False)

Step 3: Implement CISPO RL Algorithm

Efficient policy optimization clipping IS-weights instead of log-probs:

class CISPOTrainer:
    """
    Clipped IS-weight Policy Optimization.
    More efficient than DAPO: clips importance samples rather than token updates.
    """
    def __init__(self, model, clip_ratio=0.2, target_kl=0.01):
        self.model = model
        self.clip_ratio = clip_ratio
        self.target_kl = target_kl

    def compute_importance_weights(self, log_probs_new, log_probs_old):
        """
        Compute importance sampling weights.

        Args:
            log_probs_new: [batch, seq_len] log probs from updated policy
            log_probs_old: [batch, seq_len] log probs from old policy

        Returns:
            is_weights: [batch, seq_len]
        """
        # Importance sampling ratio
        log_ratio = log_probs_new - log_probs_old
        is_weights = torch.exp(log_ratio)

        return is_weights

    def clip_importance_weights(self, is_weights, advantages):
        """
        Clip IS weights rather than policy updates.
        This is more sample-efficient than standard PPO clipping.

        Args:
            is_weights: [batch, seq_len] importance weights
            advantages: [batch, seq_len] advantage estimates

        Returns:
            clipped_loss: scalar loss
        """
        # Clip importance weights
        clipped_is_weights = torch.clamp(
            is_weights,
            1 - self.clip_ratio,
            1 + self.clip_ratio
        )

        # Compute loss with clipping
        # L = -min(is_weights * advantage, clipped_is_weights * advantage)
        unclipped_loss = -is_weights * advantages
        clipped_loss = -clipped_is_weights * advantages

        loss = torch.max(unclipped_loss, clipped_loss).mean()

         loss

     ():
        
        
        outputs = .model(batch_prompts, batch_responses)
        log_probs_new = outputs.log_probs  

        
        is_weights = .compute_importance_weights(
            log_probs_new, batch_log_probs_old
        )

        
        policy_loss = .clip_importance_weights(is_weights, batch_advantages)

        
        kl_div = (batch_log_probs_old - log_probs_new).mean()
        kl_loss = torch.clamp(kl_div - .target_kl, =)

        
        total_loss = policy_loss +  * kl_loss

         total_loss, {
            : policy_loss.item(),
            : kl_loss.item(),
            : is_weights.mean().item()
        }

Step 4: Implement Diverse Reward Models

Support multiple reward types for curriculum learning:

class MultiTaskRewardModel(nn.Module):
    """
    Reward model supporting multiple task categories:
    - Verifiable: math, coding (binary correctness)
    - General: quality scoring via model-based rewards
    """
    def __init__(self, task_type='mixed'):
        super().__init__()
        self.task_type = task_type

    def compute_reward(self, responses, task_labels, references=None):
        """
        Args:
            responses: [batch] generated responses
            task_labels: [batch] task category
            references: [batch] ground truth (for verifiable tasks)

        Returns:
            rewards: [batch] reward scores in [0, 1]
        """
        rewards = []

        for i, (response, task) in enumerate(zip(responses, task_labels)):
            if task in ['math', 'coding']:
                # Verifiable tasks: binary reward
                if references is not None:
                    is_correct = self._check_correctness(response, references[i])
                    reward = 1.0 if is_correct else 0.0
                else:
                    reward = 0.5
            else:
                # General tasks: use learned model-based reward
                reward = self._score_response(response)

            rewards.append(reward)

        return torch.tensor(rewards)

     ():
        
        response_clean = response.strip().lower()
        reference_clean = reference.strip().lower()

        
         re
        response_nums = re.findall(, response_clean)
        ref_nums = re.findall(, reference_clean)

         response_nums  ref_nums:
             (response_nums[-]) == (ref_nums[-])

         response_clean == reference_clean

     ():
        
        
        length_score = ((response) / , )
         length_score *  +   

Step 5: Training Loop with Curriculum Learning

Implement curriculum-based RL training:

def train_minimax_m1(model, train_dataloader, num_epochs=10,
                     device='cuda'):
    """
    Train MiniMax-M1 with CISPO and curriculum learning.
    """
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    cispo_trainer = CISPOTrainer(model)
    reward_model = MultiTaskRewardModel()

    # Curriculum: start with verifiable, then mix in general
    task_schedule = {
        0: {'verifiable_frac': 1.0, 'general_frac': 0.0},
        5: {'verifiable_frac': 0.7, 'general_frac': 0.3},
        10: {'verifiable_frac': 0.5, 'general_frac': 0.5}
    }

    for epoch in range(num_epochs):
        epoch_loss = 0
        epoch_metrics = {}

        # Get curriculum weights for this epoch
        curriculum = task_schedule.get(epoch, {'verifiable_frac': 0.5,
                                               'general_frac': 0.5})

        for batch_idx, batch in enumerate(train_dataloader):
            # Filter batch by curriculum
            verifiable_mask = batch['task'].isin(['math', 'coding'])
            verifiable_frac = curriculum['verifiable_frac']

            # Sample batch following curriculum
            if torch.rand() < verifiable_frac:
                filtered_batch = batch[verifiable_mask]
            :
                filtered_batch = batch[~verifiable_mask]

             (filtered_batch) == :
                

            
            rewards = reward_model.compute_reward(
                filtered_batch[],
                filtered_batch[],
                filtered_batch.get()
            )

            
            baseline = rewards.mean()
            advantages = rewards - baseline

            
            loss, metrics = cispo_trainer.training_step(
                filtered_batch[],
                filtered_batch[],
                advantages.to(device),
                filtered_batch[],
                rewards.to(device)
            )

            
            optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), )
            optimizer.step()

            epoch_loss += loss.item()

            
             k, v  metrics.items():
                 k   epoch_metrics:
                    epoch_metrics[k] = []
                epoch_metrics[k].append(v)

        
        avg_loss = epoch_loss / (train_dataloader)
        ()
         k, v  epoch_metrics.items():
            ()

Practical Guidance

• Lightning Attention Frequency: Use periodic exact attention every 4-8 blocks for stability
• Mixture-of-Experts Routing: Set to activate 45.9B/456B ≈ 10% parameters; adjust sparsity for speed
• CISPO vs DAPO: CISPO needs ~50% fewer steps; prefer for large-scale RL
• Curriculum Learning: Start verifiable (easy wins), gradually mix general tasks
• Reward Models: Use binary for verifiable tasks; train small reward model for general quality
• GPU Efficiency: Sparse activation reduces memory; monitor actual speedup
• Evaluation: Test on reasoning benchmarks (AIME, MATH, code), long-context tasks
• Implementation: Use vLLM or similar for efficient generation with large models

Reference

Paper: arXiv:2506.13585 Key metrics: 25% FLOPs at 100K generation vs. DeepSeek R1; 1M context + 80K generation CISPO advantage: 50% fewer training steps vs. DAPO Related work: Mixture-of-experts, test-time scaling, efficient transformers, policy optimization