File size: 2,505 Bytes
8822739 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | from __future__ import annotations
import torch
from torch import nn
from torch.nn import functional as F
class PocketMoE(nn.Module):
def __init__(self, experts: int = 4, top_k: int = 2) -> None:
super().__init__()
self.expert_count = experts
self.top_k = top_k
self.encoder = nn.Sequential(
nn.Linear(64, 32),
nn.GELU(),
)
self.router = nn.Linear(32, experts)
self.experts = nn.ModuleList(
[
nn.Sequential(
nn.Linear(32, 16),
nn.GELU(),
nn.Linear(16, 10),
)
for _ in range(experts)
]
)
def forward(
self,
pixels: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
hidden = self.encoder(pixels)
router_probabilities = F.softmax(self.router(hidden), dim=1)
top_probabilities, top_indices = router_probabilities.topk(
self.top_k,
dim=1,
)
sparse_weights = torch.zeros_like(router_probabilities).scatter(
1,
top_indices,
top_probabilities,
)
sparse_weights = sparse_weights / sparse_weights.sum(dim=1, keepdim=True)
expert_logits = torch.stack(
[expert(hidden) for expert in self.experts],
dim=1,
)
logits = (expert_logits * sparse_weights.unsqueeze(-1)).sum(dim=1)
return logits, router_probabilities, sparse_weights
class DenseControl(nn.Module):
def __init__(self) -> None:
super().__init__()
self.network = nn.Sequential(
nn.Linear(64, 48),
nn.GELU(),
nn.Linear(48, 40),
nn.GELU(),
nn.Linear(40, 10),
)
def forward(self, pixels: torch.Tensor) -> torch.Tensor:
return self.network(pixels)
def parameter_count(model: nn.Module) -> int:
return sum(parameter.numel() for parameter in model.parameters())
def active_parameter_count(model: PocketMoE) -> int:
shared = sum(parameter.numel() for parameter in model.encoder.parameters())
router = sum(parameter.numel() for parameter in model.router.parameters())
experts = sorted(
[
sum(parameter.numel() for parameter in expert.parameters())
for expert in model.experts
],
reverse=True,
)
return shared + router + sum(experts[: model.top_k])
|