地平线开源 HoloMotion-1：4 亿参数机器人小脑模型，MoE 稀疏激活压住推理开销

"""
模拟 HoloMotion-1 风格的 MoE 小脑模型推理流程
依赖: pip install torch
注意: 这是教学示例，展示架构思路，非官方模型代码
"""
import torch
import torch.nn as nn

class MoEExpert(nn.Module):
    """单个专家网络：输入状态+指令，输出关节角目标"""
    def __init__(self, state_dim=40, cmd_dim=16, output_dim=20, hidden_dim=128):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(state_dim + cmd_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, output_dim),
        )

    def forward(self, state, cmd):
        x = torch.cat([state, cmd], dim=-1)
        return self.net(x)


class TopKGating(nn.Module):
    """路由网络：选择 top-k 个专家"""
    def __init__(self, state_dim=40, cmd_dim=16, num_experts=8, top_k=2):
        super().__init__()
        self.gate = nn.Linear(state_dim + cmd_dim, num_experts)
        self.top_k = top_k

    def forward(self, state, cmd):
        x = torch.cat([state, cmd], dim=-1)
        logits = self.gate(x)  # [B, num_experts]
        topk_vals, topk_idx = torch.topk(logits, self.top_k, dim=-1)
        weights = torch.softmax(topk_vals, dim=-1)
        return weights, topk_idx


class CerebellumMoE(nn.Module):
    """小脑 MoE 模型：稀疏激活 + KV-cache 风格推理"""
    def __init__(self, num_experts=8, top_k=2, state_dim=40, cmd_dim=16, output_dim=20):
        super().__init__()
        self.experts = nn.ModuleList([
            MoEExpert(state_dim, cmd_dim, output_dim) for _ in range(num_experts)
        ])
        self.gating = TopKGating(state_dim, cmd_dim, num_experts, top_k)
        self.top_k = top_k

    def forward(self, state, cmd):
        """
        state: [B, state_dim]  当前关节状态
        cmd:   [B, cmd_dim]    高层动作指令编码
        返回:  [B, output_dim] 下一帧关节角目标
        """
        weights, expert_idx = self.gating(state, cmd)  # [B, top_k], [B, top_k]
        batch_size = state.shape[0]
        output = torch.zeros(batch_size, self.experts[0].net[-1].out_features, device=state.device)

        for i in range(self.top_k):
            idx = expert_idx[:, i]          # [B]
            w = weights[:, i].unsqueeze(-1) # [B, 1]
            # 逐样本路由到对应专家
            for b in range(batch_size):
                expert_out = self.experts[idx[b].item()](state[b:b+1], cmd[b:b+1])
                output[b:b+1] += w[b:b+1] * expert_out

        return output


# --- 推理演示 ---
model = CerebellumMoE(num_experts=8, top_k=2)
model.eval()

# 模拟输入：1 个机器人，40 维关节状态 + 16 维指令（"向前走"）
state = torch.randn(1, 40)   # 实际部署时来自传感器
cmd = torch.randn(1, 16)     # 实际部署时来自高层规划器

with torch.no_grad():
    joint_targets = model(state, cmd)

print(f"输出关节角目标维度: {joint_targets.shape}")  # [1, 20]
print(f"下一步关节角目标值（前5个关节）: {joint_targets[0, :5].tolist()}")

# --- 连续帧推理（模拟 KV-cache 思路）---
# 实际部署中，连续帧的注意力 KV 可以缓存复用
# 这里演示连续生成多帧动作的循环
frames = []
for step in range(10):
    with torch.no_grad():
        target = model(state, cmd)
    frames.append(target)
    # 用输出的关节角目标更新状态（简化模拟，实际需动力学模型）
    state = torch.cat([target, state[:, 20:]], dim=-1)  # 保留非关节角部分状态

print(f"生成了 {len(frames)} 帧连续动作，每帧 {frames[0].shape[-1]} 个关节角目标")