| 21 | |
| 22 | @dataclass |
| 23 | class RouterCheckpoint: |
| 24 | routers: Dict[int, TokenRouter] |
| 25 | hidden_dim: int |
| 26 | bottleneck_dim: int |
| 27 | |
| 28 | def save(self, path: str | Path) -> None: |
| 29 | path = Path(path) |
| 30 | path.parent.mkdir(parents=True, exist_ok=True) |
| 31 | state = { |
| 32 | "hidden_dim": self.hidden_dim, |
| 33 | "bottleneck_dim": self.bottleneck_dim, |
| 34 | "router_layers": {}, |
| 35 | } |
| 36 | for layer_idx, router in self.routers.items(): |
| 37 | state["router_layers"][layer_idx] = router.state_dict() |
| 38 | torch.save(state, path) |
| 39 | |
| 40 | @classmethod |
| 41 | def load(cls, path: str | Path, device: str = "cpu") -> RouterCheckpoint: |
| 42 | state = torch.load(path, map_location=device, weights_only=True) |
| 43 | hidden_dim = state["hidden_dim"] |
| 44 | bottleneck_dim = state["bottleneck_dim"] |
| 45 | routers = {} |
| 46 | for layer_idx_str, router_state in state["router_layers"].items(): |
| 47 | layer_idx = int(layer_idx_str) if isinstance(layer_idx_str, str) else layer_idx_str |
| 48 | router = TokenRouter(hidden_dim, bottleneck_dim) |
| 49 | router.load_state_dict(router_state) |
| 50 | router.to(device) |
| 51 | routers[layer_idx] = router |
| 52 | return cls(routers=routers, hidden_dim=hidden_dim, bottleneck_dim=bottleneck_dim) |
no outgoing calls