File size: 12,626 Bytes
56fe6da
 
 
 
 
 
 
 
96e41b8
56fe6da
83560ca
56fe6da
 
 
 
 
 
 
 
 
 
 
 
 
 
92ddc0b
56fe6da
cdebfc7
 
56fe6da
 
 
 
 
 
 
 
 
 
cdebfc7
 
56fe6da
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6d5d4fd
 
96e41b8
 
83560ca
56fe6da
 
 
 
 
 
 
 
6d5d4fd
 
96e41b8
 
56fe6da
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
83560ca
 
 
 
 
 
 
 
 
7d2a362
 
 
 
 
83560ca
 
56fe6da
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
# coding=utf-8
#
# Code mainly copied from:
# https://github.com/huggingface/transformers/blob/main/src/transformers/models/clip/configuration_clip.py
# and adjusted for Jina CLIP

import os
from copy import deepcopy
from typing import Any, Dict, List, Optional, Union

import torch
from transformers import PretrainedConfig, logging

logger = logging.get_logger(__name__)


""" Jina CLIP model configuration """


class JinaCLIPTextConfig(PretrainedConfig):
    model_type = 'jina_clip_text'

    def __init__(
        self,
        embed_dim: int = 768,
        hf_model_name_or_path: str = 'jinaai/jina-bert-flash-implementation',
        hf_model_config_kwargs: Optional[Dict[str, Any]] = None,
        default_instruction_task: Optional[str] = None,
        default_lora_task: Optional[str] = None,
        pooler_type: Optional[str] = None,
        proj_type: Optional[str] = None,
        proj_bias: bool = False,
        **kwargs,
    ):
        super().__init__(**kwargs)

        self.embed_dim = embed_dim
        self.hf_model_name_or_path = hf_model_name_or_path
        self.hf_model_config_kwargs = hf_model_config_kwargs or {}
        self.default_instruction_task = default_instruction_task
        self.default_lora_task = default_lora_task
        self.pooler_type = pooler_type
        self.proj_type = proj_type
        self.proj_bias = proj_bias

    @classmethod
    def from_pretrained(
        cls, pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs
    ) -> 'PretrainedConfig':
        cls._set_token_in_kwargs(kwargs)

        configdict, kwargs = cls.get_config_dict(
            pretrained_model_name_or_path, **kwargs
        )
        # get the text config dict if we are loading from JinaCLIPConfig
        if configdict.get('model_type') == 'jina_clip':
            configdict = configdict['text_config']
        if (
            'model_type' in configdict
            and hasattr(cls, 'model_type')
            and configdict['model_type'] != cls.model_type
        ):
            logger.warning(
                f'You are using a model of type {configdict["model_type"]} to '
                f'instantiate a model of type {cls.model_type}. This is not supported '
                'for all configurations of models and can yield errors.'
            )
        return cls.from_dict(configdict, **kwargs)


class JinaCLIPVisionConfig(PretrainedConfig):
    model_type = 'jina_clip_vision'

    def __init__(
        self,
        embed_dim: int = 768,
        width: int = 768,
        image_size: int = 224,
        patch_size: int = 16,
        layers: int = 12,
        head_width: int = 64,
        mlp_ratio: float = 4.0,
        ls_init_value: Optional[float] = None,
        patch_dropout: float = 0.0,
        qkv_bias: bool = True,
        fused_layer_norm: bool = False,
        x_attention: bool = False,
        post_norm: bool = False,
        rope_embeddings: bool = False,
        pt_hw_seq_len: int = 16,
        intp_freq: bool = False,
        naive_swiglu: bool = False,
        subln: bool = False,
        drop_path_rate: float = 0.0,
        proj_type: Optional[str] = None,
        **kwargs,
    ):
        super().__init__(**kwargs)

        self.layers = layers
        self.embed_dim = embed_dim
        self.width = width
        self.head_width = head_width
        self.mlp_ratio = mlp_ratio
        self.image_size = image_size
        self.patch_size = patch_size
        self.ls_init_value = ls_init_value
        self.patch_dropout = patch_dropout
        self.qkv_bias = qkv_bias
        self.fused_layer_norm = fused_layer_norm
        self.x_attention = x_attention
        self.post_norm = post_norm
        self.rope_embeddings = rope_embeddings
        self.pt_hw_seq_len = pt_hw_seq_len
        self.intp_freq = intp_freq
        self.naive_swiglu = naive_swiglu
        self.subln = subln
        self.drop_path_rate = drop_path_rate
        self.proj_type = proj_type

    @classmethod
    def from_pretrained(
        cls, pretrained_model_name_or_path: Union[str, os.PathLike], **kwargs
    ) -> 'PretrainedConfig':
        cls._set_token_in_kwargs(kwargs)

        configdict, kwargs = cls.get_config_dict(
            pretrained_model_name_or_path, **kwargs
        )
        # get the vision config dict if we are loading from JinaCLIPConfig
        if configdict.get('model_type') == 'jina_clip':
            configdict = configdict['vision_config']
        if (
            'model_type' in configdict
            and hasattr(cls, 'model_type')
            and configdict['model_type'] != cls.model_type
        ):
            logger.warning(
                f'You are using a model of type {configdict["model_type"]} to '
                f'instantiate a model of type {cls.model_type}. This is not supported '
                'for all configurations of models and can yield errors.'
            )
        return cls.from_dict(configdict, **kwargs)


class JinaCLIPConfig(PretrainedConfig):
    model_type = 'jina_clip'
    is_composition = True

    def __init__(
        self,
        text_config: Optional[Dict] = None,
        vision_config: Optional[Dict] = None,
        add_projections: bool = False,
        projection_dim: int = 768,
        logit_scale_init_value: float = 2.6592,
        use_text_flash_attn: Optional[bool] = None,
        use_vision_xformers: Optional[bool] = None,
        matryoshka_dimensions: Optional[List[int]] = None,
        truncate_dim: Optional[int] = None,
        torch_dtype: Optional[Union[str, torch.dtype]] = None,
        **kwargs,
    ):
        # If `_config_dict` exist, we use them for the backward compatibility.
        # We pop out these 2 attributes before calling `super().__init__` to avoid
        # them being saved (which causes a lot of confusion!).

        text_config_dict: Optional[Dict] = kwargs.pop('text_config_dict', None)
        vision_config_dict: Optional[Dict] = kwargs.pop('vision_config_dict', None)
        self.use_text_flash_attn = use_text_flash_attn
        self.use_vision_xformers = use_vision_xformers
        self.matryoshka_dimensions = matryoshka_dimensions
        self.truncate_dim = truncate_dim

        super().__init__(**kwargs)

        if text_config_dict is not None:
            if text_config is None:
                text_config = {}

            # This is the complete result when using `text_config_dict`.
            _text_config_dict = JinaCLIPTextConfig(**text_config_dict).to_dict()

            # Give a warning if the values exist in both `_text_config_dict` and
            # `text_config` but being different.
            for key, value in _text_config_dict.items():
                if (
                    key in text_config
                    and value != text_config[key]
                    and key not in ['transformers_version']
                ):
                    # If specified in `text_config_dict`
                    if key in text_config_dict:
                        message = (
                            f'`{key}` is found in both `text_config_dict` and '
                            f'`text_config` but with different values. '
                            f'The value `text_config_dict["{key}"]` will be used '
                            f'instead.'
                        )
                    # If inferred from default argument values (
                    # just to be super careful)
                    else:
                        message = (
                            f'`text_config_dict` is provided which will be used to '
                            f'initialize `JinaCLIPTextConfig`. The '
                            f'value `text_config["{key}"]` will be overriden.'
                        )
                    logger.info(message)

            # Update all values in `text_config` with the ones in `_text_config_dict`.
            text_config.update(_text_config_dict)

        if vision_config_dict is not None:
            if vision_config is None:
                vision_config = {}

            # This is the complete result when using `vision_config_dict`.
            _vision_config_dict = JinaCLIPVisionConfig(**vision_config_dict).to_dict()
            # convert keys to string instead of integer
            if 'id2label' in _vision_config_dict:
                _vision_config_dict['id2label'] = {
                    str(key): value
                    for key, value in _vision_config_dict['id2label'].items()
                }

            # Give a warning if the values exist in both `_vision_config_dict`
            # and `vision_config` but being different.
            for key, value in _vision_config_dict.items():
                if (
                    key in vision_config
                    and value != vision_config[key]
                    and key not in ['transformers_version']
                ):
                    # If specified in `vision_config_dict`
                    if key in vision_config_dict:
                        message = (
                            f'`{key}` is found in both `vision_config_dict` and '
                            f'`vision_config` but with different '
                            f'values. The value `vision_config_dict["{key}"]` will '
                            f'be used instead.'
                        )
                    # If inferred from default argument values
                    # (just to be super careful)
                    else:
                        message = (
                            f'`vision_config_dict` is provided which will be used to '
                            f'initialize `JinaCLIPVisionConfig`. '
                            f'The value `vision_config["{key}"]` will be overriden.'
                        )
                    logger.info(message)

            # Update all values in `vision_config` with the ones in
            # `_vision_config_dict`.
            vision_config.update(_vision_config_dict)

        if text_config is None:
            text_config = {}
            logger.info(
                '`text_config` is `None`. Initializing the `JinaCLIPTextConfig` with '
                'default values.'
            )

        if vision_config is None:
            vision_config = {}
            logger.info(
                '`vision_config` is `None`. initializing the `JinaCLIPVisionConfig` '
                'with default values.'
            )

        self.text_config = JinaCLIPTextConfig(**text_config)
        self.vision_config = JinaCLIPVisionConfig(**vision_config)

        self.add_projections = add_projections
        self.projection_dim = projection_dim
        self.logit_scale_init_value = logit_scale_init_value
        self.initializer_factor = 1.0

        if not self.add_projections:
            if self.text_config.embed_dim != self.vision_config.embed_dim:
                raise ValueError(
                    'When projections are disabled (`add_projections=False`), text '
                    'and vision towers need to have the same embedding dimensionality. '
                    f'Currently text embedding dim is {self.text_config.embed_dim} != '
                    f'{self.vision_config.embed_dim} of the vision tower. '
                    'Either set the same output dim for both towers, or enable '
                    'projections with `add_projections=True`.'
                )

        if (
            torch_dtype
            and hasattr(torch, torch_dtype)
            and type(getattr(torch, torch_dtype)) is torch.dtype
        ):
            self.torch_dtype = getattr(torch, torch_dtype)
        else:
            self.torch_dtype = torch_dtype

        use_text_flash_attn = (
            self.use_text_flash_attn if self.use_text_flash_attn is not None
            else self.text_config.hf_model_config_kwargs.get('use_flash_attn', False)
        )
        if not use_text_flash_attn or not torch.cuda.is_available():
            self.torch_dtype = torch.float32

    @classmethod
    def from_text_vision_configs(
        cls,
        text_config: JinaCLIPTextConfig,
        vision_config: JinaCLIPVisionConfig,
        **kwargs,
    ):
        return cls(
            text_config=text_config.to_dict(),
            vision_config=vision_config.to_dict(),
            projection_dim=text_config.projection_dim,
            **kwargs,
        )

    def to_dict(self):
        output = deepcopy(self.__dict__)
        output['text_config'] = self.text_config.to_dict()
        output['vision_config'] = self.vision_config.to_dict()
        output['model_type'] = self.__class__.model_type
        return output