


import gc
import importlib.metadata
import tempfile
import unittest

from packaging import version

from transformers import (
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    BitsAndBytesConfig,
    pipeline,
)
from transformers.testing_utils import (
    is_torch_available,
    require_accelerate,
    require_bitsandbytes,
    require_torch,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)





def lowerCamelCase__      (						__lowerCamelCase	:							str							):







   '''simple docstring'''




   if model.config.model_type == "gpt2":
      return model.transformer.h[0].mlp.c_fc
   return model.transformer.h[0].mlp.dense_ah_to_h


if is_torch_available():
 import torch
 import torch.nn as nn






 class  __magic_name__							(		nn.Module    ):






        def __init__(      self  ,							snake_case  ,							snake_case)						->  Dict:




           '''simple docstring'''
           super().__init__()
           _UpperCAmelCase       :       List[str]							   =module
           _UpperCAmelCase       :       Optional[int]							   =nn.Sequential(
               nn.Linear(module.in_features  ,							_SCREAMING_SNAKE_CASE  ,							bias=_SCREAMING_SNAKE_CASE)  ,							nn.Linear(_SCREAMING_SNAKE_CASE  ,							module.out_features  ,							bias=_SCREAMING_SNAKE_CASE)  ,							)
           _UpperCAmelCase       :       Tuple							   =(2.0 / (5 * min(module.in_features  ,							module.out_features))) ** 0.5
           nn.init.normal_(self.adapter[0].weight  ,							std=_SCREAMING_SNAKE_CASE)
           nn.init.zeros_(self.adapter[1].weight)
           self.adapter.to(module.weight.device)






        def       lowerCAmelCase		(      self  ,							snake_case  ,							*snake_case  ,							**snake_case)						->  Tuple:




           '''simple docstring'''
           return self.module(_SCREAMING_SNAKE_CASE  ,							*_SCREAMING_SNAKE_CASE  ,							**_SCREAMING_SNAKE_CASE) + self.adapter(_SCREAMING_SNAKE_CASE)






@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  __magic_name__							(		unittest.TestCase    ):
       UpperCAmelCase										='bigscience/bloom-1b7'

       # Constant values
       UpperCAmelCase										=2.109659552692574

       UpperCAmelCase										='Hello my name is'
       UpperCAmelCase										=set()
       EXPECTED_OUTPUTS.add("Hello my name is John and I am a professional photographer. I"    )
       EXPECTED_OUTPUTS.add("Hello my name is John.\nI am a friend of your father.\n"    )
       EXPECTED_OUTPUTS.add("Hello my name is John Doe, I am a student at the University"    )
       UpperCAmelCase										=1_0






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          # Models and tokenizer
          _UpperCAmelCase       :       Tuple							   =AutoTokenizer.from_pretrained(self.model_name)






class  __magic_name__							(		_UpperCamelCase    ):






       def       lowerCAmelCase		(      self)						->  List[Any]:




          '''simple docstring'''
          super().setUp()

          # Models and tokenizer
          _UpperCAmelCase       :       Union[str, Any]							   =AutoModelForCausalLM.from_pretrained(
              self.model_name  ,							torch_dtype=torch.floataa  ,							device_map='auto')
          _UpperCAmelCase       :       List[str]							   =AutoModelForCausalLM.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          del self.model_fpaa
          del self.model_abit

          gc.collect()
          torch.cuda.empty_cache()






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          _UpperCAmelCase       :       Tuple							   =self.model_abit.config

          self.assertTrue(hasattr(_SCREAMING_SNAKE_CASE  ,							'quantization_config'))

          _UpperCAmelCase       :       Union[str, Any]							   =config.to_dict()
          _UpperCAmelCase       :       Union[str, Any]							   =config.to_diff_dict()

          _UpperCAmelCase       :       Optional[Any]							   =config.to_json_string()






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          from bitsandbytes.nn import Paramsabit

          _UpperCAmelCase       :       Optional[Any]							   =self.model_fpaa.get_memory_footprint()
          _UpperCAmelCase       :       Tuple							   =self.model_abit.get_memory_footprint()

          self.assertAlmostEqual(mem_fpaa / mem_abit  ,							self.EXPECTED_RELATIVE_DIFFERENCE)
          _UpperCAmelCase       :       Optional[Any]							   =get_some_linear_layer(self.model_abit)
          self.assertTrue(linear.weight.__class__ == Paramsabit)






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          from transformers import TaPreTrainedModel

          self.model_fpaa.get_memory_footprint()
          self.model_abit.get_memory_footprint()

          for name, module in self.model_abit.named_modules():
             if isinstance(_SCREAMING_SNAKE_CASE  ,							torch.nn.Linear):
                if name not in ["lm_head"] + TaPreTrainedModel._keep_in_fpaa_modules:
                   # 4-bit parameters are packed in uint8 variables
                   self.assertTrue(module.weight.dtype == torch.uinta)






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Tuple							   =self.tokenizer(self.input_text  ,							return_tensors='pt')
          _UpperCAmelCase       :       int							   =self.model_abit.generate(input_ids=encoded_input['input_ids'].to(0)  ,							max_new_tokens=1_0)

          self.assertIn(self.tokenizer.decode(output_sequences[0]  ,							skip_special_tokens=_SCREAMING_SNAKE_CASE)  ,							self.EXPECTED_OUTPUTS)






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       List[Any]							   =BitsAndBytesConfig()
          _UpperCAmelCase       :       Dict							   =True

          _UpperCAmelCase       :       Any							   =AutoModelForCausalLM.from_pretrained(
              self.model_name  ,							quantization_config=_SCREAMING_SNAKE_CASE  ,							device_map='auto')

          _UpperCAmelCase       :       List[str]							   =self.tokenizer(self.input_text  ,							return_tensors='pt')
          _UpperCAmelCase       :       Optional[Any]							   =model_abit_from_config.generate(
              input_ids=encoded_input['input_ids'].to(0)  ,							max_new_tokens=1_0)

          self.assertIn(self.tokenizer.decode(output_sequences[0]  ,							skip_special_tokens=_SCREAMING_SNAKE_CASE)  ,							self.EXPECTED_OUTPUTS)






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          with self.assertRaises(_SCREAMING_SNAKE_CASE), tempfile.TemporaryDirectory() as tmpdirname:
             self.model_abit.save_pretrained(_SCREAMING_SNAKE_CASE)






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          _UpperCAmelCase       :       Optional[Any]							   =BitsAndBytesConfig()

          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             _UpperCAmelCase       :       int							   =AutoModelForCausalLM.from_pretrained(
                 self.model_name  ,							quantization_config=_SCREAMING_SNAKE_CASE  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto'  ,							bnb_abit_quant_type='nf4'  ,							)






       def       lowerCAmelCase		(      self)						->  Dict:




          '''simple docstring'''
          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             # Tries with `str`
             self.model_abit.to('cpu')

          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             # Tries with a `dtype``
             self.model_abit.to(torch.floataa)

          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             # Tries with a `device`
             self.model_abit.to(torch.device('cuda:0'))

          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             # Tries with a `device`
             self.model_abit.float()

          with self.assertRaises(_SCREAMING_SNAKE_CASE):
             # Tries with a `device`
             self.model_abit.half()

          # Test if we did not break anything
          _UpperCAmelCase       :       Optional[int]							   =self.tokenizer(self.input_text  ,							return_tensors='pt')

          _UpperCAmelCase       :       List[str]							   =self.model_fpaa.to(torch.floataa)
          _UpperCAmelCase       :       str							   =self.model_fpaa.generate(input_ids=encoded_input['input_ids'].to(0)  ,							max_new_tokens=1_0)

          # Check this does not throw an error
          _UpperCAmelCase       :       str							   =self.model_fpaa.to('cpu')

          # Check this does not throw an error
          _UpperCAmelCase       :       Tuple							   =self.model_fpaa.half()

          # Check this does not throw an error
          _UpperCAmelCase       :       Optional[Any]							   =self.model_fpaa.float()






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Optional[Any]							   =AutoModelForSeqaSeqLM.from_pretrained('t5-small'  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          self.assertTrue(model.decoder.block[0].layer[2].DenseReluDense.wo.weight.dtype == torch.floataa)






@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class  __magic_name__							(		unittest.TestCase    ):






       @classmethod
       def       lowerCAmelCase		(      cls)						->  Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       Dict							   ='t5-small'
          _UpperCAmelCase       :       Dict							   ='google/flan-t5-small'  # flan-t5 uses dense-act instead of dense-relu-dense
          _UpperCAmelCase       :       List[Any]							   =AutoTokenizer.from_pretrained(cls.model_name)
          _UpperCAmelCase       :       Tuple							   ='Translate in German: Hello, my dog is cute'






       def       lowerCAmelCase		(      self)						->  Any:




          '''simple docstring'''
          gc.collect()
          torch.cuda.empty_cache()






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          from transformers import TaForConditionalGeneration

          _UpperCAmelCase       :       Any							   =TaForConditionalGeneration._keep_in_fpaa_modules
          _UpperCAmelCase       :       int							   =None

          # test with `t5-small`
          _UpperCAmelCase       :       str							   =TaForConditionalGeneration.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          _UpperCAmelCase       :       Tuple							   =self.tokenizer(self.input_text  ,							return_tensors='pt').to(0)
          _UpperCAmelCase       :       Optional[int]							   =model.generate(**_SCREAMING_SNAKE_CASE)

          # test with `flan-t5-small`
          _UpperCAmelCase       :       Optional[Any]							   =TaForConditionalGeneration.from_pretrained(
              self.dense_act_model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          _UpperCAmelCase       :       int							   =self.tokenizer(self.input_text  ,							return_tensors='pt').to(0)
          _UpperCAmelCase       :       Dict							   =model.generate(**_SCREAMING_SNAKE_CASE)
          _UpperCAmelCase       :       str							   =modules






       def       lowerCAmelCase		(      self)						->  Dict:




          '''simple docstring'''
          import bitsandbytes as bnb

          from transformers import TaForConditionalGeneration

          # test with `t5-small`
          _UpperCAmelCase       :       Any							   =TaForConditionalGeneration.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')

          # there was a bug with decoders - this test checks that it is fixed
          self.assertTrue(isinstance(model.decoder.block[0].layer[0].SelfAttention.q  ,							bnb.nn.Linearabit))

          _UpperCAmelCase       :       Dict							   =self.tokenizer(self.input_text  ,							return_tensors='pt').to(0)
          _UpperCAmelCase       :       Union[str, Any]							   =model.generate(**_SCREAMING_SNAKE_CASE)

          # test with `flan-t5-small`
          _UpperCAmelCase       :       str							   =TaForConditionalGeneration.from_pretrained(
              self.dense_act_model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          _UpperCAmelCase       :       Union[str, Any]							   =self.tokenizer(self.input_text  ,							return_tensors='pt').to(0)
          _UpperCAmelCase       :       List[str]							   =model.generate(**_SCREAMING_SNAKE_CASE)






class  __magic_name__							(		_UpperCamelCase    ):






       def       lowerCAmelCase		(      self)						->  int:




          '''simple docstring'''
          super().setUp()
          # model_name
          _UpperCAmelCase       :       str							   ='bigscience/bloom-560m'
          _UpperCAmelCase       :       List[str]							   ='t5-small'

          # Different types of model

          _UpperCAmelCase       :       Optional[int]							   =AutoModel.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          # Sequence classification model
          _UpperCAmelCase       :       List[Any]							   =AutoModelForSequenceClassification.from_pretrained(
              self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          # CausalLM model
          _UpperCAmelCase       :       str							   =AutoModelForCausalLM.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')
          # Seq2seq model
          _UpperCAmelCase       :       str							   =AutoModelForSeqaSeqLM.from_pretrained(
              self.seq_to_seq_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='auto')






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          del self.base_model
          del self.sequence_model
          del self.model_abit
          del self.seq_to_seq_model

          gc.collect()
          torch.cuda.empty_cache()






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          from bitsandbytes.nn import Paramsabit

          self.assertTrue(self.base_model.h[-1].mlp.dense_ah_to_h.weight.__class__ == Paramsabit)

          # Other heads should be nn.Parameter
          self.assertTrue(self.model_abit.lm_head.weight.__class__ == torch.nn.Parameter)
          self.assertTrue(self.sequence_model.score.weight.__class__ == torch.nn.Parameter)
          self.assertTrue(self.seq_to_seq_model.lm_head.weight.__class__ == torch.nn.Parameter)






class  __magic_name__							(		_UpperCamelCase    ):






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          super().setUp()






       def       lowerCAmelCase		(      self)						->  List[Any]:




          '''simple docstring'''
          del self.pipe

          gc.collect()
          torch.cuda.empty_cache()






       def       lowerCAmelCase		(      self)						->  Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       Any							   =pipeline(
              'text-generation'  ,							model=self.model_name  ,							model_kwargs={'device_map': 'auto', 'load_in_4bit': True, 'torch_dtype': torch.floataa}  ,							max_new_tokens=self.MAX_NEW_TOKENS  ,							)

          # Real second forward pass
          _UpperCAmelCase       :       Tuple							   =self.pipe(self.input_text)
          self.assertIn(pipeline_output[0]['generated_text']  ,							self.EXPECTED_OUTPUTS)






@require_torch_multi_gpu
class  __magic_name__							(		_UpperCamelCase    ):






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          super().setUp()






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Tuple							   =AutoModelForCausalLM.from_pretrained(
              self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE  ,							device_map='balanced')

          # Check correct device map
          self.assertEqual(set(model_parallel.hf_device_map.values())  ,							{0, 1})

          # Check that inference pass works on the model
          _UpperCAmelCase       :       int							   =self.tokenizer(self.input_text  ,							return_tensors='pt')

          # Second real batch
          _UpperCAmelCase       :       Optional[Any]							   =model_parallel.generate(input_ids=encoded_input['input_ids'].to(0)  ,							max_new_tokens=1_0)
          self.assertIn(self.tokenizer.decode(output_parallel[0]  ,							skip_special_tokens=_SCREAMING_SNAKE_CASE)  ,							self.EXPECTED_OUTPUTS)






class  __magic_name__							(		_UpperCamelCase    ):






       def       lowerCAmelCase		(      self)						->  str:




          '''simple docstring'''
          _UpperCAmelCase       :       Tuple							   ='facebook/opt-350m'
          super().setUp()






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          if version.parse(importlib.metadata.version('bitsandbytes')) < version.parse('0.37.0'):
             return

          # Step 1: freeze all parameters
          _UpperCAmelCase       :       int							   =AutoModelForCausalLM.from_pretrained(self.model_name  ,							load_in_abit=_SCREAMING_SNAKE_CASE)

          self.assertEqual(set(model.hf_device_map.values())  ,							{torch.cuda.current_device()})

          for param in model.parameters():
             _UpperCAmelCase       :       Dict							   =False  # freeze the model - train adapters later
             if param.ndim == 1:
                # cast the small parameters (e.g. layernorm) to fp32 for stability
                _UpperCAmelCase       :       Union[str, Any]							   =param.data.to(torch.floataa)

        # Step 2: add adapters
          for _, module in model.named_modules():
             if "OPTAttention" in repr(type(_SCREAMING_SNAKE_CASE)):
                _UpperCAmelCase       :       Union[str, Any]							   =LoRALayer(module.q_proj  ,							rank=1_6)
                _UpperCAmelCase       :       List[str]							   =LoRALayer(module.k_proj  ,							rank=1_6)
                _UpperCAmelCase       :       Optional[Any]							   =LoRALayer(module.v_proj  ,							rank=1_6)

        # Step 3: dummy batch
          _UpperCAmelCase       :       Dict							   =self.tokenizer('Test batch '  ,							return_tensors='pt').to(0)

          # Step 4: Check if the gradient is not None
          with torch.cuda.amp.autocast():
             _UpperCAmelCase       :       Tuple							   =model.forward(**_SCREAMING_SNAKE_CASE)
             out.logits.norm().backward()

          for module in model.modules():
             if isinstance(_SCREAMING_SNAKE_CASE  ,							_SCREAMING_SNAKE_CASE):
                self.assertTrue(module.adapter[1].weight.grad is not None)
                self.assertTrue(module.adapter[1].weight.grad.norm().item() > 0)
             elif isinstance(_SCREAMING_SNAKE_CASE  ,							nn.Embedding):
                self.assertTrue(module.weight.grad is None)






class  __magic_name__							(		_UpperCamelCase    ):
       UpperCAmelCase										='gpt2-xl'
       UpperCAmelCase										=3.3191854854152187





'''simple docstring'''






from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

lowercase							=logging.get_logger(__name__)






@add_end_docstrings(lowerCAmelCase    )
class  __magic_name__							(		lowerCAmelCase    ):






       def __init__(      self  ,							**snake_case)						->  Optional[int]:




          '''simple docstring'''
          super().__init__(**snake_case)

          requires_backends(self  ,							'vision')
          self.check_model_type(
              TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == 'tf'
              else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING)






       def __call__(      self  ,							snake_case  ,							**snake_case)						->  str:




          '''simple docstring'''
          return super().__call__(snake_case  ,							**snake_case)






       def       lowerCAmelCase		(      self  ,							**snake_case)						->  int:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   ={}
          if "candidate_labels" in kwargs:
             _UpperCAmelCase       :       Union[str, Any]							   =kwargs['candidate_labels']
          if "hypothesis_template" in kwargs:
             _UpperCAmelCase       :       List[Any]							   =kwargs['hypothesis_template']

          return preprocess_params, {}, {}






       def       lowerCAmelCase		(      self  ,							snake_case  ,							snake_case=None  ,							snake_case="This is a photo of {}.")						->  Any:




          '''simple docstring'''
          _UpperCAmelCase       :       Optional[Any]							   =load_image(snake_case)
          _UpperCAmelCase       :       Union[str, Any]							   =self.image_processor(images=[image]  ,							return_tensors=self.framework)
          _UpperCAmelCase       :       Union[str, Any]							   =candidate_labels
          _UpperCAmelCase       :       List[Any]							   =[hypothesis_template.format(snake_case) for x in candidate_labels]
          _UpperCAmelCase       :       str							   =self.tokenizer(snake_case  ,							return_tensors=self.framework  ,							padding=snake_case)
          _UpperCAmelCase       :       Any							   =[text_inputs]
          return inputs






       def       lowerCAmelCase		(      self  ,							snake_case)						->  str:




          '''simple docstring'''
          _UpperCAmelCase       :       List[str]							   =model_inputs.pop('candidate_labels')
          _UpperCAmelCase       :       Tuple							   =model_inputs.pop('text_inputs')
          if isinstance(text_inputs[0]  ,							snake_case):
             _UpperCAmelCase       :       Any							   =text_inputs[0]
          else:
             # Batching case.
             _UpperCAmelCase       :       str							   =text_inputs[0][0]

          _UpperCAmelCase       :       Any							   =self.model(**snake_case  ,							**snake_case)

          _UpperCAmelCase       :       List[str]							   ={
              'candidate_labels': candidate_labels,
              'logits': outputs.logits_per_image,
          }
          return model_outputs






       def       lowerCAmelCase		(      self  ,							snake_case)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   =model_outputs.pop('candidate_labels')
          _UpperCAmelCase       :       Union[str, Any]							   =model_outputs['logits'][0]
          if self.framework == "pt":
             _UpperCAmelCase       :       Dict							   =logits.softmax(dim=-1).squeeze(-1)
             _UpperCAmelCase       :       Union[str, Any]							   =probs.tolist()
             if not isinstance(snake_case  ,							snake_case):
                _UpperCAmelCase       :       Union[str, Any]							   =[scores]
          elif self.framework == "tf":
             _UpperCAmelCase       :       Dict							   =stable_softmax(snake_case  ,							axis=-1)
             _UpperCAmelCase       :       str							   =probs.numpy().tolist()
          else:
             raise ValueError(f"Unsupported framework: {self.framework}")

          _UpperCAmelCase       :       List[str]							   =[
              {'score': score, 'label': candidate_label}
              for score, candidate_label in sorted(zip(snake_case  ,							snake_case)  ,							key=lambda snake_case: -x[0])
          ]
          return result

