llava/model/ema.py

"""
Exponential Moving Average (EMA) for Model Weights

Maintains a shadow copy of model parameters with exponentially decayed averaging:
    θ_ema = decay * θ_ema + (1 - decay) * θ

Benefits:
- More stable inference (smoother weight updates)
- Better generalization (less prone to overfitting)
- Standard practice for diffusion models (IP2P, SD, DALL-E, etc.)

Usage:
    # Initialize
    ema = EMA(model.parameters(), decay=0.9999)
    
    # After optimizer.step()
    ema.update()
    
    # For validation/inference
    with ema.average_parameters():
        # Use EMA weights
        output = model(input)
"""

import torch
import torch.nn as nn
from contextlib import contextmanager
from typing import Iterable, Optional
from copy import deepcopy


class EMA:
    """
    Exponential Moving Average of model parameters.
    
    Args:
        parameters: Iterable of model parameters to track
        decay: EMA decay rate (default: 0.9999)
                Higher = slower update, more stable
                Common values: 0.999, 0.9999, 0.99999
        update_every: Update EMA every N steps (default: 1)
        min_decay: Minimum decay value (default: 0.0, no minimum)
        use_num_updates: Adjust decay based on num_updates (default: True)
    """
    def __init__(
        self,
        parameters: Iterable[nn.Parameter],
        decay: float = 0.9999,
        update_every: int = 1,
        min_decay: float = 0.0,
        use_num_updates: bool = True,
    ):
        self.decay = decay
        self.min_decay = min_decay
        self.update_every = update_every
        self.use_num_updates = use_num_updates
        
        # Store model parameters
        self.online_params = list(parameters)
        
        # Create shadow copy of parameters on the same device
        self.shadow_params = [p.clone().detach().to(p.device) for p in self.online_params]
        
        # Track number of updates
        self.num_updates = 0
        
        # Flag to track if we're currently using EMA weights
        self.using_ema = False
        
    def get_decay(self, num_updates: Optional[int] = None) -> float:
        """
        Compute current decay value.
        
        Optionally adjusts decay based on number of updates:
            decay = min(decay, (1 + num_updates) / (10 + num_updates))
        
        This starts with lower decay for faster initial updates,
        then gradually increases to target decay.
        """
        if num_updates is None:
            num_updates = self.num_updates
            
        decay = self.decay
        
        if self.use_num_updates:
            # Adjust decay based on num_updates (starts lower, gradually increases)
            decay = min(decay, (1 + num_updates) / (10 + num_updates))
        
        # Apply minimum decay
        decay = max(decay, self.min_decay)
        
        return decay
    
    @torch.no_grad()
    def update(self):
        """
        Update EMA parameters.
        
        Called after optimizer.step() during training.
        """
        self.num_updates += 1
        
        # Only update every N steps
        if self.num_updates % self.update_every != 0:
            return
        
        decay = self.get_decay()
        
        # Update shadow parameters
        for i, (shadow_param, online_param) in enumerate(zip(self.shadow_params, self.online_params)):
            if online_param.requires_grad:
                # Ensure shadow parameter is on the same device as online parameter
                if shadow_param.device != online_param.device:
                    self.shadow_params[i] = shadow_param.to(online_param.device)
                    shadow_param = self.shadow_params[i]
                
                # EMA update: θ_ema = decay * θ_ema + (1 - decay) * θ
                shadow_param.mul_(decay).add_(online_param.data, alpha=1 - decay)
    
    @torch.no_grad()
    def copy_to(self, parameters: Iterable[nn.Parameter]):
        """Copy EMA parameters to given parameters"""
        for shadow_param, param in zip(self.shadow_params, parameters):
            # Ensure shadow parameter is on the same device as target parameter
            if shadow_param.device != param.device:
                shadow_param = shadow_param.to(param.device)
            param.data.copy_(shadow_param.data)
    
    @torch.no_grad()
    def store(self):
        """Store current model parameters (before loading EMA)"""
        if self.using_ema:
            raise RuntimeError("Already using EMA parameters!")
        
        self.backup_params = [p.clone() for p in self.online_params]
    
    @torch.no_grad()
    def restore(self):
        """Restore original model parameters (after using EMA)"""
        if not self.using_ema:
            raise RuntimeError("Not currently using EMA parameters!")
        
        for param, backup_param in zip(self.online_params, self.backup_params):
            param.data.copy_(backup_param.data)
        
        del self.backup_params
    
    @torch.no_grad()
    def load_ema_into_model(self):
        """Load EMA parameters into model"""
        self.store()
        self.copy_to(self.online_params)
        self.using_ema = True
    
    @torch.no_grad()
    def restore_model_from_ema(self):
        """Restore model parameters from backup"""
        self.restore()
        self.using_ema = False
    
    @contextmanager
    def average_parameters(self):
        """
        Context manager for using EMA parameters.
        
        Usage:
            with ema.average_parameters():
                # Model now uses EMA weights
                output = model(input)
            # Model back to normal weights
        """
        self.load_ema_into_model()
        try:
            yield
        finally:
            self.restore_model_from_ema()
    
    def state_dict(self):
        """Get EMA state for checkpointing"""
        return {
            'shadow_params': self.shadow_params,
            'num_updates': self.num_updates,
            'decay': self.decay,
            'min_decay': self.min_decay,
            'update_every': self.update_every,
            'use_num_updates': self.use_num_updates,
        }
    
    def load_state_dict(self, state_dict):
        """Load EMA state from checkpoint"""
        self.shadow_params = state_dict['shadow_params']
        self.num_updates = state_dict['num_updates']
        self.decay = state_dict.get('decay', self.decay)
        self.min_decay = state_dict.get('min_decay', self.min_decay)
        self.update_every = state_dict.get('update_every', self.update_every)
        self.use_num_updates = state_dict.get('use_num_updates', self.use_num_updates)
    
    def __repr__(self):
        return (
            f"EMA(decay={self.decay}, num_updates={self.num_updates}, "
            f"current_decay={self.get_decay():.6f}, num_params={len(self.shadow_params)})"
        )


class EMAWarmupScheduler:
    """
    Scheduler for EMA decay with warmup.
    
    Gradually increases EMA decay from min to max over warmup steps.
    Useful for stable training start.
    
    Args:
        ema: EMA instance
        warmup_steps: Number of warmup steps
        min_decay: Starting decay value (default: 0.95)
        max_decay: Target decay value (default: 0.9999)
    """
    def __init__(
        self,
        ema: EMA,
        warmup_steps: int,
        min_decay: float = 0.95,
        max_decay: float = 0.9999,
    ):
        self.ema = ema
        self.warmup_steps = warmup_steps
        self.min_decay = min_decay
        self.max_decay = max_decay
        self.original_decay = ema.decay
    
    def step(self):
        """Update EMA decay based on num_updates"""
        if self.ema.num_updates < self.warmup_steps:
            # Linear warmup
            progress = self.ema.num_updates / self.warmup_steps
            self.ema.decay = self.min_decay + (self.max_decay - self.min_decay) * progress
        else:
            # Use original decay
            self.ema.decay = self.original_decay


if __name__ == "__main__":
    # Test EMA
    print("Testing EMA module...")
    
    # Create dummy model
    model = nn.Linear(10, 10)
    
    # Initialize EMA
    ema = EMA(model.parameters(), decay=0.9999)
    print(f"\n[1] Initialized: {ema}")
    
    # Simulate training
    print("\n[2] Simulating training...")
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    
    for step in range(100):
        # Forward pass
        x = torch.randn(4, 10)
        y = model(x)
        loss = y.mean()
        
        # Backward pass
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        # Update EMA
        ema.update()
        
        if step % 20 == 0:
            print(f"  Step {step}: decay={ema.get_decay():.6f}, num_updates={ema.num_updates}")
    
    # Test context manager
    print("\n[3] Testing context manager...")
    with torch.no_grad():
        x_test = torch.randn(4, 10)
        
        # Normal weights
        output_normal = model(x_test)
        
        # EMA weights
        with ema.average_parameters():
            output_ema = model(x_test)
        
        # Back to normal weights
        output_normal_2 = model(x_test)
        
        print(f"  Normal output mean: {output_normal.mean().item():.4f}")
        print(f"  EMA output mean: {output_ema.mean().item():.4f}")
        print(f"  Normal output (after EMA) mean: {output_normal_2.mean().item():.4f}")
        print(f"  ✓ Weights correctly restored: {torch.allclose(output_normal, output_normal_2)}")
    
    # Test state dict
    print("\n[4] Testing state dict...")
    state = ema.state_dict()
    print(f"  State dict keys: {list(state.keys())}")
    
    # Create new EMA and load state
    ema_new = EMA(model.parameters(), decay=0.999)  # Different decay
    ema_new.load_state_dict(state)
    print(f"  Loaded EMA: {ema_new}")
    print(f"  ✓ num_updates match: {ema_new.num_updates == ema.num_updates}")
    
    # Test warmup scheduler
    print("\n[5] Testing EMA warmup scheduler...")
    model2 = nn.Linear(10, 10)
    ema2 = EMA(model2.parameters(), decay=0.9999)
    scheduler = EMAWarmupScheduler(ema2, warmup_steps=50, min_decay=0.95, max_decay=0.9999)
    
    for step in range(100):
        # Simulate training
        ema2.update()
        scheduler.step()
        
        if step % 20 == 0:
            print(f"  Step {step}: decay={ema2.decay:.6f}")
    
    print("\n✓ All EMA tests passed!")