코드로 이해하는 Swin Transformer 리뷰

microsoft에서 고안한 Swin Transformer: Hierarchical Vision Transformer using Shifted Windows 논문에 대한 내용을 담고 있다.

 

본 논문만으로는 한번에 이해 안되는 부분이 몇몇 존재 했다. (예를들어 각 윈도우 내에서의 Self-attention 결과를 어떻게 다음 단계로 넘겨주는지?) 따라서 오피셜 코드를 살펴볼 수 밖에 없었고, 논문의 이해를 돕기위한 코드에 대한 설명을 포스팅으로 공유하고자 한다.

 

Paper, Official Github

microsoft/Swin-Transformer

Swin-Transformer 특징

- Window Partitioning (M=7) -> 각 로컬 윈도우 내에서 self-attention을 적용하여 계산 복잡도 줄임

- Shifted Windows -> 항상 정해진 윈도우 내에서만 self-attention 적용하지 않도록 방지(cross-window connection)

- Hierarchical Transformer -> Patch Merging을 통해 다양한 스케일을 고려

Swin-Transformer 구조

Figure in Paper

논문에서 제안한 Swin-Transformer의 구조는 위와 같다.

Patch Partition(+Embedding) -> Swin Transformer Block -> Patch Merging -> Swin Transformer Block -> ..

 

Transformer Block과 Patch Merging이 반복되어지는, 어떻게 보면 참 쉬운 구조이다. 더 정확한 구조 및 원리 파악을 위해 코드를 참고하였다. 아래에 첨부한 코드들은 모두 official github의 models/swin_transformer.py 코드이며, 설명에 불필요한 부분은 #... 로 생략하였다.

microsoft/Swin-Transformer

SwinTransformer

class SwinTransformer(nn.Module):
    def __init__(self, img_size=224, patch_size=4, in_chans=3, num_classes=1000,
                 embed_dim=96, depths=[2, 2, 6, 2], num_heads=[3, 6, 12, 24],
                 window_size=7, mlp_ratio=4., qkv_bias=True, qk_scale=None,
                 drop_rate=0., attn_drop_rate=0., drop_path_rate=0.1,
                 norm_layer=nn.LayerNorm, ape=False, patch_norm=True,
                 use_checkpoint=False, **kwargs):
        super().__init__()
		
        # ...
        
        self.patch_embed = PatchEmbed(
            img_size=img_size, patch_size=patch_size, in_chans=in_chans, embed_dim=embed_dim,
            norm_layer=norm_layer if self.patch_norm else None)
            
        # ...

        self.layers = nn.ModuleList()
        for i_layer in range(self.num_layers):
            layer = BasicLayer(dim=int(embed_dim * 2 ** i_layer),
                               input_resolution=(patches_resolution[0] // (2 ** i_layer),
                                                 patches_resolution[1] // (2 ** i_layer)),
                               depth=depths[i_layer],
                               num_heads=num_heads[i_layer],
                               window_size=window_size,
                               mlp_ratio=self.mlp_ratio,
                               qkv_bias=qkv_bias, qk_scale=qk_scale,
                               drop=drop_rate, attn_drop=attn_drop_rate,
                               drop_path=dpr[sum(depths[:i_layer]):sum(depths[:i_layer + 1])],
                               norm_layer=norm_layer,
                               downsample=PatchMerging if (i_layer < self.num_layers - 1) else None,
                               use_checkpoint=use_checkpoint)
            self.layers.append(layer)
        
        self.norm = norm_layer(self.num_features)
        self.avgpool = nn.AdaptiveAvgPool1d(1)
        self.head = nn.Linear(self.num_features, num_classes) if num_classes > 0 else nn.Identity()
		
        # ...
    
    # ...

    def forward_features(self, x):
        x = self.patch_embed(x)
        if self.ape:
            x = x + self.absolute_pos_embed
        x = self.pos_drop(x)

        for layer in self.layers:
            x = layer(x)

        x = self.norm(x)  # B L C
        x = self.avgpool(x.transpose(1, 2))  # B C 1
        x = torch.flatten(x, 1)
        return x

    def forward(self, x):
        x = self.forward_features(x)
        x = self.head(x)
        return x
	
    # ...

 

Swint-Transformer 전체 model implementation 코드이다. forward_features(self, x) method를 보면 전체 모델의 흐름을 알 수 있다.

PatchEmbed module -> BasicLayer modules(nn.ModuleList) -> norm&avgpool -> nn.Linear

 

이 중 PatchEmbed module과 BasicLayer module를 중점으로 각 module의 직관적인(?) 의미와 x(input)의 dimension 변화를 설명하려고 한다.

PatchEmbed (Patch Partition + Linear Embedding)

class PatchEmbed(nn.Module):
    r""" Image to Patch Embedding
    Args:
        img_size (int): Image size.  Default: 224.
        patch_size (int): Patch token size. Default: 4.
        in_chans (int): Number of input image channels. Default: 3.
        embed_dim (int): Number of linear projection output channels. Default: 96.
        norm_layer (nn.Module, optional): Normalization layer. Default: None
    """

    def __init__(self, img_size=224, patch_size=4, in_chans=3, embed_dim=96, norm_layer=None):
        super().__init__()
        img_size = to_2tuple(img_size)
        patch_size = to_2tuple(patch_size)
        patches_resolution = [img_size[0] // patch_size[0], img_size[1] // patch_size[1]]
        self.img_size = img_size
        self.patch_size = patch_size
        self.patches_resolution = patches_resolution
        self.num_patches = patches_resolution[0] * patches_resolution[1]

        self.in_chans = in_chans
        self.embed_dim = embed_dim

        self.proj = nn.Conv2d(in_chans, embed_dim, kernel_size=patch_size, stride=patch_size)
        if norm_layer is not None:
            self.norm = norm_layer(embed_dim)
        else:
            self.norm = None

    def forward(self, x):
        B, C, H, W = x.shape
        # FIXME look at relaxing size constraints
        assert H == self.img_size[0] and W == self.img_size[1], \
            f"Input image size ({H}*{W}) doesn't match model ({self.img_size[0]}*{self.img_size[1]})."
        x = self.proj(x).flatten(2).transpose(1, 2)  # B Ph*Pw C
        if self.norm is not None:
            x = self.norm(x)
        return x

    # ...

 

이미지를 패치로 나누고 각 패치를 임베딩하는 module이다. 논문과 코드에서도 볼 수 있듯이, image의 shape은 (3, 224, 224)이며, patch size는 4, 그리고 embedding size는 96이다.

 

Convolution layer(위 코드에서 self.proj = nn.Conv2d(..) 부분)에 kernel size와 stride를 모두 patch size 만큼 주기 때문에, patch partition과 embedidng을 동시에 진행할 수 있다.

 

처음 x는 (B, 3, 224, 224) 였지만 self.proj(x)를 통해 (B, 96, 56, 56), .flatten(2)를 통해 (B, 96, 56x56), .transpose(1, 2)를 통해 (B, 56x56, 96)이 되었다.

(B, 3, 224, 224) -> self.proj(x).flatten(2).transpose(1, 2) -> (B, 56x56, 96)

Swin Transforer Block

class BasicLayer(nn.Module):
    """ A basic Swin Transformer layer for one stage.
    Args:
        dim (int): Number of input channels.
        input_resolution (tuple[int]): Input resolution.
        depth (int): Number of blocks.
        num_heads (int): Number of attention heads.
        # ...
    """

    def __init__(self, dim, input_resolution, depth, num_heads, window_size,
                 mlp_ratio=4., qkv_bias=True, qk_scale=None, drop=0., attn_drop=0.,
                 drop_path=0., norm_layer=nn.LayerNorm, downsample=None, use_checkpoint=False):

        super().__init__()
        # ...

        # build blocks
        self.blocks = nn.ModuleList([
            SwinTransformerBlock(dim=dim, input_resolution=input_resolution,
                                 num_heads=num_heads, window_size=window_size,
                                 shift_size=0 if (i % 2 == 0) else window_size // 2,
                                 mlp_ratio=mlp_ratio,
                                 qkv_bias=qkv_bias, qk_scale=qk_scale,
                                 drop=drop, attn_drop=attn_drop,
                                 drop_path=drop_path[i] if isinstance(drop_path, list) else drop_path,
                                 norm_layer=norm_layer)
            for i in range(depth)])

        # patch merging layer
        if downsample is not None:
            self.downsample = downsample(input_resolution, dim=dim, norm_layer=norm_layer)
        else:
            self.downsample = None

    def forward(self, x):
        for blk in self.blocks:
            if self.use_checkpoint:
                x = checkpoint.checkpoint(blk, x)
            else:
                x = blk(x)
        if self.downsample is not None:
            x = self.downsample(x)
        return x

    # ...

 

BasicLayer Module의 코드 부분이다. depth개의 SwinTransformerBlock과 downsample로 구성되어 있다. 홀수번째 SwinTransforemrBlock은 shift_size를 window_size//2로 설정되며, downsample은 뒤에 설명하는 Patch Merging 과정이다.

 

class SwinTransformerBlock(nn.Module):
    r""" Swin Transformer Block.
    Args:
        dim (int): Number of input channels.
        input_resolution (tuple[int]): Input resulotion.
        num_heads (int): Number of attention heads.
        window_size (int): Window size.
        shift_size (int): Shift size for SW-MSA.
        # ...
    """

    def __init__(self, dim, input_resolution, num_heads, window_size=7, shift_size=0,
                 mlp_ratio=4., qkv_bias=True, qk_scale=None, drop=0., attn_drop=0., drop_path=0.,
                 act_layer=nn.GELU, norm_layer=nn.LayerNorm):
        super().__init__()
        self.dim = dim
        self.input_resolution = input_resolution
        self.num_heads = num_heads
        self.window_size = window_size
        self.shift_size = shift_size
        self.mlp_ratio = mlp_ratio
        if min(self.input_resolution) <= self.window_size:
            # if window size is larger than input resolution, we don't partition windows
            self.shift_size = 0
            self.window_size = min(self.input_resolution)
        assert 0 <= self.shift_size < self.window_size, "shift_size must in 0-window_size"

        self.norm1 = norm_layer(dim)
        self.attn = WindowAttention(
            dim, window_size=to_2tuple(self.window_size), num_heads=num_heads,
            qkv_bias=qkv_bias, qk_scale=qk_scale, attn_drop=attn_drop, proj_drop=drop)

        self.drop_path = DropPath(drop_path) if drop_path > 0. else nn.Identity()
        self.norm2 = norm_layer(dim)
        mlp_hidden_dim = int(dim * mlp_ratio)
        self.mlp = Mlp(in_features=dim, hidden_features=mlp_hidden_dim, act_layer=act_layer, drop=drop)

        if self.shift_size > 0:
            # calculate attention mask for SW-MSA
            H, W = self.input_resolution
            img_mask = torch.zeros((1, H, W, 1))  # 1 H W 1
            h_slices = (slice(0, -self.window_size),
                        slice(-self.window_size, -self.shift_size),
                        slice(-self.shift_size, None))
            w_slices = (slice(0, -self.window_size),
                        slice(-self.window_size, -self.shift_size),
                        slice(-self.shift_size, None))
            cnt = 0
            for h in h_slices:
                for w in w_slices:
                    img_mask[:, h, w, :] = cnt
                    cnt += 1

            mask_windows = window_partition(img_mask, self.window_size)  # nW, window_size, window_size, 1
            mask_windows = mask_windows.view(-1, self.window_size * self.window_size)
            attn_mask = mask_windows.unsqueeze(1) - mask_windows.unsqueeze(2)
            attn_mask = attn_mask.masked_fill(attn_mask != 0, float(-100.0)).masked_fill(attn_mask == 0, float(0.0))
        else:
            attn_mask = None

        self.register_buffer("attn_mask", attn_mask)

    def forward(self, x):
        H, W = self.input_resolution
        B, L, C = x.shape
        assert L == H * W, "input feature has wrong size"

        shortcut = x
        x = self.norm1(x)
        x = x.view(B, H, W, C)

        # cyclic shift
        if self.shift_size > 0:
            shifted_x = torch.roll(x, shifts=(-self.shift_size, -self.shift_size), dims=(1, 2))
        else:
            shifted_x = x

        # partition windows
        x_windows = window_partition(shifted_x, self.window_size)  # nW*B, window_size, window_size, C
        x_windows = x_windows.view(-1, self.window_size * self.window_size, C)  # nW*B, window_size*window_size, C

        # W-MSA/SW-MSA
        attn_windows = self.attn(x_windows, mask=self.attn_mask)  # nW*B, window_size*window_size, C

        # merge windows
        attn_windows = attn_windows.view(-1, self.window_size, self.window_size, C)
        shifted_x = window_reverse(attn_windows, self.window_size, H, W)  # B H' W' C

        # reverse cyclic shift
        if self.shift_size > 0:
            x = torch.roll(shifted_x, shifts=(self.shift_size, self.shift_size), dims=(1, 2))
        else:
            x = shifted_x
        x = x.view(B, H * W, C)

        # FFN
        x = shortcut + self.drop_path(x)
        x = x + self.drop_path(self.mlp(self.norm2(x)))

        return x

    # ...

 

이 논문에 핵심인 SwinTransformerBlock Module 코드이다.

 

forward method의 input x는 현재(B, 56x56, 96) size를 갖고 있다. 이는 .view를 통해 (B, 56, 56, 96)으로 바뀐다. 이 x는 각 윈도우에 self-attention을 적용 하기위해 windows_partition 함수를 거치게된다.

(B, 56x56, 96) -> (B, 56, 56, 96) -> windows_partition

 

def window_partition(x, window_size):
    """
    Args:
        x: (B, H, W, C)
        window_size (int): window size
    Returns:
        windows: (num_windows*B, window_size, window_size, C)
    """
    B, H, W, C = x.shape
    x = x.view(B, H // window_size, window_size, W // window_size, window_size, C)
    windows = x.permute(0, 1, 3, 2, 4, 5).contiguous().view(-1, window_size, window_size, C)
    return windows

 

window_partition은 정해진 window_size(논문/코드에서는 7로 고정)로 패치들을 나눈 후, tensor을 윈도우 기준으로 생성 해준다.

 

우선 x = x.view(B, H // window_size, window_size, W // window_size, window_size, C) 부분을 통해 x를 window_size로 나누어 새로운 차원을 생성해준다. 현재 x의 shape은 (B, 8, 7, 8, 7, 96) 이다. 그 후 permute을 통해 (B, 8, 8, 7, 7, 96), .view을 통해 (Bx8x8, 7, 7, 96) shape이 된다.

(B, 56, 56, 96) -> (B, 8, 7, 8, 7, 96) -> (B, 8, 8, 7, 7, 96) -> (Bx8x8, 7, 7, 96)

다시 위 SwinTransformerBlock으로 돌아가서 (Bx8x8, 7, 7, 96) tensor 속 각 윈도우는 self-attention 처리를 위해 이차원 tensor가 되어야한다. 즉 (Bx8x8, 7x7, 96)이 되어 window 내에 있는 패치끼리 self-attention이 이루어진다.

 

window partition 함수에서 다른 이미지의 window지만 Bx8x8로 묶은 이유는 계산 편의성에 있다. window 내부 패치들끼리만 이루어지는 계산이기 때문에 다른 이미지 window와 독립적이다.

 

self-attention을 통해 나온 output은 self-attention 특성상 input과 동일한 shape을 갖게 된다. 그리고나서 window_reverse 함수를 통해 기존 shape (B, 56x56, 96)으로 복구가 된다.

(Bx8x8, 7, 7, 96) -> (Bx8x8, 7x7, 96) -> self-attention -> (Bx8x8, 7x7, 96) -> (B, 56x56, 96)

위에서 언급했듯이, shift된 window를 사용할 때에도 고려 해야한다. 위 figure 4를 보면 알 수 있듯이, 논문에서는 window를 그대로 shift 한 후, 겹치지 않은 부분(A, B, C)를 빈 공간에 넣어주는 방법이 고안되었다.

 

이는 torch.roll 함수를 통해 간단히 해결된다. torch.roll은 딱 저 기능을 처리해준다.

 

#torch.roll examples
>>> x = torch.tensor(
    [
        [1, 2, 3, 4],
        [5, 6, 7, 8]
    ]
)

>>> x = torch.roll(x, 1)
>>> x
tensor(
    [8, 1, 2, 3],
    [4, 5, 6, 7]
)

 

torch.roll을 통해 patch sfhit 후, window_partion -> self-attention -> window_reverse 진행한다.

그리고 다시 반대 방향으로 torch.roll을 통해 patch를 원상복귀하면 된다(위 코드에서 두개의 torch.roll 부분).

 

이후 논문에 나와있듯, shortcut connection이나 drop_path 등 여러 기법이 적용된다.

Patch Merging

Hierarchical 한 특성을 위해 한 인풋에 존재하는 Patch의 개수를 점점 줄이며 학습 한다. 이를 통해 이미지의 작은 물체부터 큰 물체까지 모든 정보가 학습에 사용하게된다.

 

class PatchMerging(nn.Module):
    r""" Patch Merging Layer.
    Args:
        input_resolution (tuple[int]): Resolution of input feature.
        dim (int): Number of input channels.
        norm_layer (nn.Module, optional): Normalization layer.  Default: nn.LayerNorm
    """

    def __init__(self, input_resolution, dim, norm_layer=nn.LayerNorm):
        super().__init__()
        # ...

    def forward(self, x):
        """
        x: B, H*W, C
        """
        H, W = self.input_resolution
        B, L, C = x.shape
        assert L == H * W, "input feature has wrong size"
        assert H % 2 == 0 and W % 2 == 0, f"x size ({H}*{W}) are not even."

        x = x.view(B, H, W, C)

        x0 = x[:, 0::2, 0::2, :]  # B H/2 W/2 C
        x1 = x[:, 1::2, 0::2, :]  # B H/2 W/2 C
        x2 = x[:, 0::2, 1::2, :]  # B H/2 W/2 C
        x3 = x[:, 1::2, 1::2, :]  # B H/2 W/2 C
        x = torch.cat([x0, x1, x2, x3], -1)  # B H/2 W/2 4*C
        x = x.view(B, -1, 4 * C)  # B H/2*W/2 4*C

        x = self.norm(x)
        x = self.reduction(x)

        return x

    # ...

 

x0 = x[:, 0::2, 0::2, :]  # B H/2 W/2 C
x1 = x[:, 1::2, 0::2, :]  # B H/2 W/2 C
x2 = x[:, 0::2, 1::2, :]  # B H/2 W/2 C
x3 = x[:, 1::2, 1::2, :]  # B H/2 W/2 C
x = torch.cat([x0, x1, x2, x3], -1)  # B H/2 W/2 4*C
x = x.view(B, -1, 4 * C)  # B H/2*W/2 4*C

x = self.norm(x)
x = self.reduction(x)

 

인풋 x에 대해 width, height를 각각 한칸 씩 띄어서 새로운 이미지(x0, x1, x2, x3)를 만들고, 임베딩 차원을 기준으로 concat한다. 그리고 self.reduction을 통해 임베딩 차원을 조정해준다. 말로 설명하기 어려우니... 아래 그림을 제작해보았다..

(B, 56x56, 96) -> (B, 56, 56, 96) -> (B, 56/2, 56/2, 96x4) -> (B, 56/2, 56/2, 96x2)

 

이렇게 patch merging(downsample)이 완료된 tensor은 다음 BasicLayer의 input으로 사용된다. 모든 BasicLayer를 거친 tensor은 norm/avgpool/flatten/nn.Linear을 통해 class 개수에 맞춘 tensor가 되어 loss가 계산된다.