Kyutai REKI 2.4B VLA模型解析：32个内存寄存器、流匹配与LIBERO-Plus性能表现

import time
import math
import json
import asyncio
import logging
from collections import deque
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("kyutai_reki_vla")

@dataclass
class ProprioceptiveState:
    joint_angles_rad: List[float]
    gripper_opening_mm: float
    robot_morphology: str = "ur5_6dof"

@dataclass
class ActionChunkResponse:
    horizon_steps: int
    trajectory_deltas: List[List[float]]
    inference_latency_ms: float
    register_cache_depth: int
    oscillation_risk_score: float

class AdaptiveRMSNorm:
    def __init__(self, dim: int, eps: float = 1e-6):
        self.dim = dim
        self.eps = eps

    def forward(self, x: List[float], cond_scale: float, cond_shift: float) -> List[float]:
        mean_sq = sum(v * v for v in x) / max(len(x), 1)
        rms = math.sqrt(mean_sq + self.eps)
        return [((v / rms) * (1.0 + cond_scale)) + cond_shift for v in x]

class RollingRegisterCache:
    def __init__(self, max_frames: int = 3, num_registers: int = 32, reg_dim: int = 16):
        self.max_frames = max_frames
        self.num_registers = num_registers
        self.reg_dim = reg_dim
        self.buffer: deque = deque(maxlen=max_frames)

    def reset(self) -> None:
        self.buffer.clear()

    def compute_and_store_registers(self, visual_features: List[float], prompt_hash: int) -> List[List[float]]:
        current_regs: List[List[float]] = []
        past_summary = 0.0
        if len(self.buffer) > 0:
            past_summary = sum(frame[0][0] for frame in self.buffer) / len(self.buffer)
        for r_idx in range(self.num_registers):
            base_val = visual_features[r_idx % len(visual_features)]
            phase = ((prompt_hash + r_idx) % 97) * 0.01
            reg_vec = [math.tanh(base_val + phase + 0.15 * past_summary + (d * 0.02)) for d in range(self.reg_dim)]
            current_regs.append(reg_vec)
        self.buffer.append(current_regs)
        return [reg for frame in self.buffer for reg in frame]

class KyutaiRekiActionExpert:
    def __init__(self, action_dim: int = 7, chunk_horizon: int = 16, euler_steps: int = 10):
        self.action_dim = action_dim
        self.chunk_horizon = chunk_horizon
        self.euler_steps = euler_steps
        self.ada_norm = AdaptiveRMSNorm(dim=action_dim)
        self.register_cache = RollingRegisterCache(max_frames=3, num_registers=32, reg_dim=16)
        self.last_action_head: Optional[List[float]] = None

    def reset_episode(self) -> None:
        self.register_cache.reset()
        self.last_action_head = None
        logger.info("32寄存器缓存已重置，准备迎接新的操作回合。")

    def _encode_dual_cameras(self, cam_top_px: List[float], cam_wrist_px: List[float]) -> List[float]:
        return [(t * 0.6 + w * 0.4) for t, w in zip(cam_top_px, cam_wrist_px)]

    def _state_mlp_conditioning(self, state: ProprioceptiveState, tau: float) -> Tuple[float, float]:
        joint_energy = sum(abs(a) for a in state.joint_angles_rad) / max(len(state.joint_angles_rad), 1)
        scale = 0.1 * math.sin(tau * math.pi) + 0.05 * joint_energy
        shift = 0.02 * (state.gripper_opening_mm / 85.0) - 0.01 * tau
        return scale, shift

    async def predict_action_chunk(self, cam_top: List[float], cam_wrist: List[float], state: ProprioceptiveState, task_prompt: str) -> ActionChunkResponse:
        t0 = time.perf_counter()
        vis_tokens = self._encode_dual_cameras(cam_top, cam_wrist)
        prompt_hash = sum(ord(c) for c in task_prompt)
        cached_regs = self.register_cache.compute_and_store_registers(vis_tokens, prompt_hash)
        reg_context_scalar = sum(r[0] for r in cached_regs) / max(len(cached_regs), 1)
        
        action_volume: List[List[float]] = [[0.0 for _ in range(self.action_dim)] for _ in range(self.chunk_horizon)]
        dt = 1.0 / float(self.euler_steps)
        
        for step_idx in range(self.euler_steps):
            tau = step_idx * dt
            c_scale, c_shift = self._state_mlp_conditioning(state, tau)
            for h in range(self.chunk_horizon):
                normed = self.ada_norm.forward(action_volume[h], c_scale, c_shift)
                velocity = [math.tanh(normed[d] + reg_context_scalar * 0.25 + (h + 1) * 0.01 - tau * 0.05) for d in range(self.action_dim)]
                action_volume[h] = [round(action_volume[h][d] + dt * velocity[d], 5) for d in range(self.action_dim)]
        
        osc_score = 0.041
        if self.last_action_head is not None:
            dot_prod = sum(a * b for a, b in zip(action_volume[0], self.last_action_head))
            if dot_prod < 0.0:
                osc_score = 0.125
        self.last_action_head = action_volume[0]
        
        latency = (time.perf_counter() - t0) * 1000.0 + 21.7
        return ActionChunkResponse(
            horizon_steps=self.chunk_horizon,
            trajectory_deltas=action_volume,
            inference_latency_ms=round(latency, 2),
            register_cache_depth=len(self.register_cache.buffer),
            oscillation_risk_score=osc_score
        )

async def simulate_lerobot_control_loop():
    expert = KyutaiRekiActionExpert(action_dim=7, chunk_horizon=16, euler_steps=10)
    expert.reset_episode()
    state = ProprioceptiveState(joint_angles_rad=[0.12, -1.34, 1.85, -0.72, 1.57, 0.04], gripper_opening_mm=42.5)
    
    for frame_id in range(1, 4):
        cam_top = [0.25 * frame_id + (i * 0.01) for i in range(32)]
        cam_wrist = [0.18 * frame_id + (i * 0.015) for i in range(32)]
        resp = await expert.predict_action_chunk(cam_top, cam_wrist, state, task_prompt="将陶瓷碗放到金属盘子上")
        logger.info(f"帧 {frame_id} | 缓存深度: {resp.register_cache_depth} | 推理延迟: {resp.inference_latency_ms} ms | 增量[0]: {resp.trajectory_deltas[0][:3]}")

if __name__ == "__main__":
    asyncio.run(simulate_lerobot_control_loop())