Conceptual Framework Implementation (Python)

The code below illustrates the core mechanisms of DySL-VLA: the dynamic-static layer classification, trajectory continuity computation, prior-post skipping guidance, and the two-stage knowledge distillation training loop. This is educational pseudocode capturing the paper’s architectural ideas.

# ─────────────────────────────────────────────────────────────────────────────
# DySL-VLA: Dynamic-Static Layer-Skipping for VLA Model Inference
# Yang, Qi, Xie, Yu, Liu, Li · arXiv:2602.22896v2 [cs.RO] 2026
# Conceptual implementation of core components
# ─────────────────────────────────────────────────────────────────────────────

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import List, Tuple, Optional, Deque
from collections import deque
import math


# ─── Section 1: Layer Classification ─────────────────────────────────────────

def classify_vla_layers(
    model: nn.Module,
    calibration_data: torch.Tensor,
    static_ratio: float = 0.20
) -> Tuple[List[int], List[int]]:
    """
    Identify static (informative) vs dynamic layers using cosine similarity.
    
    A layer is 'informative' if its output activation differs significantly
    from its input (low cosine similarity between input and output).
    
    Args:
        model: VLA model (LLM backbone portion)
        calibration_data: Sample activations for measurement
        static_ratio: Fraction of layers to designate as static (paper: 20%)
        
    Returns:
        static_layers: Indices of always-executed layers
        dynamic_layers: Indices of potentially-skippable layers
    """
    similarities = []
    hooks = []
    layer_io = {}
    
    # Register forward hooks to capture input/output for each layer
    for i, layer in enumerate(model.layers):
        def make_hook(idx):
            def hook(module, input, output):
                inp = input[0].detach()
                out = output[0].detach() if isinstance(output, tuple) else output.detach()
                layer_io[idx] = (inp, out)
            return hook
        hooks.append(layer.register_forward_hook(make_hook(i)))
    
    # Run calibration forward pass
    with torch.no_grad():
        model(calibration_data)
    
    # Compute average cosine similarity per layer
    for i in range(len(model.layers)):
        inp, out = layer_io[i]
        inp_flat = inp.view(inp.shape[0], -1)
        out_flat = out.view(out.shape[0], -1)
        cos_sim = F.cosine_similarity(inp_flat, out_flat, dim=1).mean().item()
        similarities.append(cos_sim)
    
    # Remove hooks
    for h in hooks:
        h.remove()
    
    # Layers with lowest cosine similarity → most informative → static
    n_static = max(1, int(len(model.layers) * static_ratio))
    sorted_by_info = sorted(range(len(similarities)), key=lambda i: similarities[i])
    static_layers = sorted(sorted_by_info[:n_static])
    dynamic_layers = [i for i in range(len(model.layers)) if i not in static_layers]
    
    return static_layers, dynamic_layers


# ─── Section 2: Trajectory Continuity Signal ─────────────────────────────────

class TrajectoryContinuityMonitor:
    """
    Tracks trajectory continuity C_t = -(1/k) * sum_{j=t-k+1}^{t} ||A_j - A_{j-1}||_2
    
    A value close to 0 indicates smooth, uniform motion (safe to skip).
    A large negative value indicates high action variation (critical phase).
    """
    
    def __init__(self, k: int = 5, eta: float = 1e-3):
        """
        Args:
            k: Trajectory window length (paper default: 5)
            eta: Threshold for detecting continuity changes
        """
        self.k = k
        self.eta = eta
        self.action_history: Deque = deque(maxlen=k + 1)
        self.continuity_history: Deque = deque(maxlen=3)
    
    def update(self, action: torch.Tensor) -> float:
        """Add new action and compute current continuity score."""
        self.action_history.append(action.detach())
        
        if len(self.action_history) < 2:
            self.continuity_history.append(0.0)
            return 0.0
        
        history = list(self.action_history)
        diffs = [torch.norm(history[j] - history[j - 1]).item()
                 for j in range(max(1, len(history) - self.k), len(history))]
        
        C_t = -(sum(diffs) / len(diffs)) if diffs else 0.0
        self.continuity_history.append(C_t)
        return C_t
    
    def continuity_change(self) -> float:
        """Return delta_C_t = C_t - C_{t-1}."""
        if len(self.continuity_history) < 2:
            return 0.0
        hist = list(self.continuity_history)
        return hist[-1] - hist[-2]
    
    def is_critical_transition(self) -> bool:
        """
        Detect first moment of continuity decrease (Eq. delta_C_t < -eta and delta_C_{t-1} > eta).
        This triggers post-skip verification.
        """
        if len(self.continuity_history) < 3:
            return False
        hist = list(self.continuity_history)
        delta_t   = hist[-1] - hist[-2]
        delta_tm1 = hist[-2] - hist[-3]
        return delta_t < -self.eta and delta_tm1 > self.eta


# ─── Section 3: Skipping-Allow Point Manager ─────────────────────────────────

class SkippingAllowPoint:
    """
    Manages the skipping-allow point l_i between two adjacent static layers.
    
    The allow point determines where between two static layers the skipping
    controller becomes active. Before the point, layers are forced to execute.
    
    Moves FORWARD (restricting more skipping) when continuity drops.
    Moves BACKWARD (allowing more skipping) when continuity improves.
    """
    
    def __init__(self, s_prev: int, s_next: int, eta: float = 1e-3):
        self.s_prev = s_prev   # Index of preceding static layer
        self.s_next = s_next   # Index of next static layer
        self.eta = eta
        self.l_i = s_prev + 1  # Initially allow skipping from first dynamic layer
    
    def update(self, delta_C: float):
        """
        Move allow point based on continuity change (Eq. 2 and 3 from paper).
        
        Args:
            delta_C: Change in continuity score (C_t - C_{t-1})
        """
        if delta_C < -self.eta and self.l_i < self.s_next:
            # Continuity dropping → move forward (less skipping)
            delta_l = math.ceil(abs(delta_C) / self.eta)
            self.l_i = min(self.l_i + delta_l, self.s_next)
        elif delta_C > self.eta and self.l_i > self.s_prev + 1:
            # Continuity improving → move backward by 1 (restore skipping gradually)
            self.l_i -= 1
    
    def layer_is_active(self, layer_idx: int) -> bool:
        """Returns True if this layer must execute (before allow point)."""
        return layer_idx < self.l_i


# ─── Section 4: DySL-VLA Inference Engine ────────────────────────────────────

class DySLVLAInference:
    """
    Production inference engine implementing DySL-VLA's dynamic-static
    layer skipping with prior-post skipping guidance.
    """
    
    def __init__(
        self,
        vla_backbone: nn.Module,
        static_layers: List[int],
        dynamic_layers: List[int],
        adapters: nn.ModuleList,
        controllers: nn.ModuleList,
        eta: float = 1e-3,
        k: int = 5
    ):
        self.backbone = vla_backbone
        self.static_layers = set(static_layers)
        self.dynamic_layers = dynamic_layers
        self.adapters = adapters
        self.controllers = controllers
        self.continuity = TrajectoryContinuityMonitor(k=k, eta=eta)
        self.allow_points = self._init_allow_points(static_layers)
        self.eta = eta
    
    def _init_allow_points(self, static_layers: List[int]) -> List[SkippingAllowPoint]:
        """Create a SkippingAllowPoint manager between each pair of adjacent static layers."""
        points = []
        for i in range(len(static_layers) - 1):
            points.append(SkippingAllowPoint(
                s_prev=static_layers[i],
                s_next=static_layers[i + 1],
                eta=self.eta
            ))
        return points
    
    def predict_action(
        self,
        observation: torch.Tensor,
        instruction_tokens: torch.Tensor
    ) -> torch.Tensor:
        """
        Full inference step with dynamic-static skipping and prior-post guidance.
        
        Args:
            observation: Current image observation
            instruction_tokens: Tokenized language instruction
            
        Returns:
            action: Predicted robot action
        """
        # Update allow points from latest continuity reading
        delta_C = self.continuity.continuity_change()
        for ap in self.allow_points:
            ap.update(delta_C)
        
        # Check if post-skip verification is needed
        needs_verification = self.continuity.is_critical_transition()
        
        if needs_verification:
            # Post-skip verification: re-run with ALL layers (no skipping)
            action = self._run_full_model(observation, instruction_tokens)
            verified_C = self.continuity.update(action)
            # Recompute delta with verified action for better future estimates
            delta_C_verified = self.continuity.continuity_change()
            for ap in self.allow_points:
                ap.update(delta_C_verified)
        else:
            action = self._run_with_skipping(observation, instruction_tokens)
            self.continuity.update(action)
        
        return action
    
    def _run_with_skipping(
        self,
        observation: torch.Tensor,
        instruction_tokens: torch.Tensor
    ) -> torch.Tensor:
        """Execute backbone with dynamic-static layer skipping."""
        x = self.backbone.embed(observation, instruction_tokens)
        
        i = 0
        allow_point_idx = 0
        
        while i < len(self.backbone.layers):
            layer = self.backbone.layers[i]
            
            if i in self.static_layers:
                # Static layer: always execute
                x = layer(x)
                i += 1
                allow_point_idx = min(allow_point_idx + 1, len(self.allow_points) - 1)
            else:
                # Dynamic layer: check if before allow point
                ap = self.allow_points[min(allow_point_idx, len(self.allow_points) - 1)]
                
                if ap.layer_is_active(i):
                    # Before allow point: force execution
                    x = layer(x)
                    i += 1
                else:
                    # After allow point: consult controller
                    controller_idx = self.dynamic_layers.index(i)
                    skip_prob = self.controllers[controller_idx](x).sigmoid().item()
                    
                    if skip_prob > 0.5:
                        # Skip to next static layer using adapter
                        adapter_idx = controller_idx
                        x = self.adapters[adapter_idx](x)
                        # Jump to next static layer
                        next_static = next((s for s in self.static_layers if s > i), i + 1)
                        i = next_static
                    else:
                        x = layer(x)
                        i += 1
        
        return self.backbone.action_head(x)
    
    def _run_full_model(
        self,
        observation: torch.Tensor,
        instruction_tokens: torch.Tensor
    ) -> torch.Tensor:
        """Execute all layers with no skipping (for post-skip verification)."""
        with torch.no_grad():
            x = self.backbone.embed(observation, instruction_tokens)
            for layer in self.backbone.layers:
                x = layer(x)
            return self.backbone.action_head(x)


# ─── Section 5: Two-Stage Knowledge Distillation ─────────────────────────────

class SkipAwareDistillation:
    """
    Implements the two-stage training protocol for DySL-VLA.
    
    Stage 1: Train adapters in isolation with forced skipping.
    Stage 2: Joint adapter + controller training with harmonic layer selection.
    """
    
    def __init__(self, model: DySLVLAInference, lambda_norm: float = 0.01):
        self.model = model
        self.lambda_norm = lambda_norm
    
    def stage1_adapter_loss(
        self,
        x_i: torch.Tensor,
        layer_idx: int,
        next_static_idx: int
    ) -> torch.Tensor:
        """
        Stage 1 loss: adapter output should match running all dynamic layers.
        (Eq. 4 from paper)
        """
        adapter_idx = self.model.dynamic_layers.index(layer_idx)
        adapter_out = self.model.adapters[adapter_idx](x_i)
        
        # Target: run all dynamic layers up to next static layer
        target = x_i.clone()
        with torch.no_grad():
            for j in range(layer_idx, next_static_idx):
                target = self.model.backbone.layers[j](target)
        
        return torch.norm(adapter_out - target, p='fro')
    
    def stage2_joint_loss(
        self,
        x_i: torch.Tensor,
        selected_layer_idx: int,
        next_static_idx: int,
        task_loss: torch.Tensor
    ) -> torch.Tensor:
        """
        Stage 2 loss: task loss + normalization loss encouraging skipping.
        (Eq. 5 and 6 from paper)
        """
        adapter_idx = self.model.dynamic_layers.index(selected_layer_idx)
        controller_prob = self.model.controllers[adapter_idx](x_i).sigmoid()
        adapter_out = self.model.adapters[adapter_idx](x_i)
        
        # Full path: run all dynamic layers up to next static
        full_path = x_i.clone()
        for j in range(selected_layer_idx, next_static_idx):
            full_path = self.model.backbone.layers[j](full_path)
        
        # Soft combination (differentiable forward, Eq. 5)
        x_next_static = controller_prob * adapter_out + (1 - controller_prob) * full_path
        
        # Normalization loss: encourage controller to skip more
        layers_to_skip = next_static_idx - selected_layer_idx
        norm_loss = (1 - controller_prob) * layers_to_skip
        
        return task_loss + self.lambda_norm * norm_loss.mean()
    
    def harmonic_layer_selection(
        self,
        dynamic_layers_between: List[int]
    ) -> int:
        """
        Select a dynamic layer for training using harmonic decay probability.
        Earlier layers are selected more often (they summarize more layers).
        """
        n = len(dynamic_layers_between)
        weights = [1.0 / (i + 1) for i in range(n)]  # Harmonic weights
        total = sum(weights)
        probs = [w / total for w in weights]
        
        return torch.multinomial(torch.tensor(probs), 1).item()


# ─── Section 6: Demo ─────────────────────────────────────────────────────────

if __name__ == "__main__":
    print("=" * 70)
    print("DySL-VLA: Dynamic-Static Layer Skipping for VLA Models")
    print("arXiv:2602.22896v2 [cs.RO] 2026")
    print("=" * 70)
    
    # Demonstrate trajectory continuity monitor
    print("\n[1] Trajectory Continuity Monitor Demo")
    monitor = TrajectoryContinuityMonitor(k=5, eta=1e-3)
    
    # Simulate smooth approach actions (low variance)
    print("  Approach phase (smooth motion):")
    for t in range(8):
        action = torch.tensor([0.1, 0.1, 0.02]) + torch.randn(3) * 0.001
        C = monitor.update(action)
        print(f"    Step {t}: C_t = {C:.5f}")
    
    # Simulate grasp onset (high variance)
    print("\n  Grasp phase (variable, critical):")
    for t in range(4):
        action = torch.tensor([0.1, 0.1, 0.02]) + torch.randn(3) * 0.05
        C = monitor.update(action)
        critical = monitor.is_critical_transition()
        print(f"    Step {t+8}: C_t = {C:.5f}  → critical={critical}")
    
    print("\n[2] Key Results Summary")
    print("  Calvin D→D: DySL-VLA 2.89 vs DeeR-VLA 2.83 vs baseline 2.92")
    print("  Latency: 13.6ms (DySL-VLA) vs 51.0ms (RoboFlamingo) → 3.75×")
    print("  Training: 14M params, 7 GPU-hours vs DeeR-VLA 1.2B, 112 hrs")
    print("  Jetson Orin: 23.2 Hz effective control frequency")
    
    print("\n[3] Component Importance (Ablation)")
    components = [
        ("Full DySL-VLA",                  2.89, "13.6ms"),
        ("w/o Post-skip Verification",     2.79, "13.4ms"),
        ("w/o Pre-skip Prediction",        2.42, "20.1ms"),
        ("w/o Skip-aware KD",             2.82, "74.0ms"),
        ("w/o Dynamic-static Skipping",    1.87, "27.6ms"),
    ]
    for name, avg_len, latency in components:
        print(f"  {name:<40s} AvgLen={avg_len}  Latency={latency}")
    
    print("\n" + "=" * 70)
    print("DySL-VLA achieves efficient VLA inference by respecting the fact")
    print("that not all robot actions demand the same computational attention.")
    print("=" * 70)