hivemind/tests/test_utils/custom_networks.py

import torch
import torch.nn as nn
import torch.nn.functional as F

from hivemind.moe import register_expert_class

sample_input = lambda batch_size, hidden_dim: torch.empty((batch_size, hidden_dim))


@register_expert_class("perceptron", sample_input)
class MultilayerPerceptron(nn.Module):
    def __init__(self, hidden_dim, num_classes=10):
        super().__init__()
        self.layer1 = nn.Linear(hidden_dim, 2 * hidden_dim)
        self.layer2 = nn.Linear(2 * hidden_dim, 2 * hidden_dim)
        self.layer3 = nn.Linear(2 * hidden_dim, num_classes)

    def forward(self, x):
        x = F.relu(self.layer1(x))
        x = F.relu(self.layer2(x))
        x = self.layer3(x)
        return x


multihead_sample_input = lambda batch_size, hidden_dim: (
    torch.empty((batch_size, hidden_dim)),
    torch.empty((batch_size, 2 * hidden_dim)),
    torch.empty((batch_size, 3 * hidden_dim)),
)


@register_expert_class("multihead", multihead_sample_input)
class MultiheadNetwork(nn.Module):
    def __init__(self, hidden_dim, num_classes=10):
        super().__init__()
        self.layer1 = nn.Linear(hidden_dim, num_classes)
        self.layer2 = nn.Linear(2 * hidden_dim, num_classes)
        self.layer3 = nn.Linear(3 * hidden_dim, num_classes)

    def forward(self, x1, x2, x3):
        x = self.layer1(x1) + self.layer2(x2) + self.layer3(x3)
        return x