File size: 4,633 Bytes
7c174eb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
import torch
import torch.nn as nn
import math


class PerceiverModel(nn.Module):
    '''

    perceiver model with sparse attention.

    Scale: tiny (dim=128, layers=3, heads=2)

    Fusion: cross_attention, Task: contrastive

    '''

    def __init__(self, embed_dim=128, num_layers=3, num_heads=2, num_classes=10):
        super().__init__()
        self.embed_dim = embed_dim
        self.num_layers = num_layers
        self.num_heads = num_heads

        # image patch embedding
        self.patch_proj = nn.Conv2d(3, embed_dim, kernel_size=16, stride=16)
        self.cls_token = nn.Parameter(torch.randn(1, 1, embed_dim) * 0.02)
        self.pos_embed = nn.Parameter(torch.randn(1, 197, embed_dim) * 0.02)
        self.dropout = nn.Dropout(0.1)

        # image transformer blocks
        self.image_blocks = nn.ModuleList([
            nn.TransformerEncoderLayer(
                embed_dim, num_heads, embed_dim * 4, 0.1,
                activation='gelu', batch_first=True, norm_first=True
            )
            for _ in range(num_layers)
        ])
        self.image_norm = nn.LayerNorm(embed_dim)

        # text embedding
        self.text_embed = nn.Embedding(30522, embed_dim, padding_idx=0)
        self.text_pos = nn.Parameter(torch.randn(1, 128, embed_dim) * 0.02)
        self.text_blocks = nn.ModuleList([
            nn.TransformerEncoderLayer(
                embed_dim, num_heads, embed_dim * 4, 0.1,
                activation='gelu', batch_first=True, norm_first=True
            )
            for _ in range(num_layers)
        ])
        self.text_norm = nn.LayerNorm(embed_dim)

        # fusion
        self.fusion_blocks = nn.ModuleList([
            nn.TransformerEncoderLayer(
                embed_dim, num_heads, embed_dim * 4, 0.1,
                activation='gelu', batch_first=True, norm_first=True
            )
            for _ in range(2)
        ])
        self.fusion_norm = nn.LayerNorm(embed_dim)

        # task head
        self.classifier = nn.Sequential(
            nn.Linear(embed_dim, embed_dim),
            nn.GELU(),
            nn.Dropout(0.1),
            nn.Linear(embed_dim, num_classes),
        )

        self._initialize_weights()

    def _initialize_weights(self):
        for m in self.modules():
            if isinstance(m, nn.Linear):
                nn.init.kaiming_normal_(m.weight, mode='fan_out')
                if m.bias is not None:
                    nn.init.zeros_(m.bias)
            elif isinstance(m, nn.Embedding):
                nn.init.trunc_normal_(m.weight, std=0.02)
                if m.padding_idx is not None:
                    m.weight[m.padding_idx].zero_()
            elif isinstance(m, nn.LayerNorm):
                nn.init.ones_(m.weight)
                nn.init.zeros_(m.bias)

    def encode_image(self, images):
        x = self.patch_proj(images)
        x = x.flatten(2).transpose(1, 2)
        cls = self.cls_token.expand(x.size(0), -1, -1)
        x = torch.cat([cls, x], dim=1)
        x = x + self.pos_embed
        x = self.dropout(x)
        for block in self.image_blocks:
            x = block(x)
        return self.image_norm(x)

    def encode_text(self, input_ids, attention_mask=None):
        x = self.text_embed(input_ids)
        x = x + self.text_pos[:, :input_ids.size(1)]
        x = self.dropout(x)
        padding = (attention_mask == 0) if attention_mask is not None else None
        for block in self.text_blocks:
            x = block(x, src_key_padding_mask=padding)
        return self.text_norm(x)

    def forward(self, images, input_ids, attention_mask=None, labels=None):
        image_features = self.encode_image(images)
        text_features = self.encode_text(input_ids, attention_mask)

        fused = text_features
        for block in self.fusion_blocks:
            fused = block(fused)
        fused = self.fusion_norm(fused[:, 0])
        logits = self.classifier(fused)

        loss = None
        if labels is not None:
            loss = nn.functional.cross_entropy(logits, labels)

        return {'logits': logits, 'loss': loss}


if __name__ == '__main__':
    model = PerceiverModel()
    total = sum(p.numel() for p in model.parameters())
    print(f'PerceiverModel: {total:,} params ({total/1e6:.2f}M)')
    img = torch.randn(2, 3, 224, 224)
    ids = torch.randint(0, 30522, (2, 128))
    mask = torch.ones(2, 128)
    out = model(img, ids, mask, torch.tensor([0, 1]))
    print(f'Output: {out["logits"].shape}, Loss: {out["loss"].item():.4f}')