




'''simple docstring'''
import logging
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional

import numpy as np
import torch
from utils_hans import HansDataset, InputFeatures, hans_processors, hans_tasks_num_labels

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import is_main_process


A							:				List[str]			     =	logging.getLogger(__name__)






@dataclass
class 	SCREAMING_SNAKE_CASE:
       snake_case_			:					Optional[Any]   = field(
           metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}					)
       snake_case_			:					Any   = field(
           default=__A       ,				metadata={"""help""": """Pretrained config name or path if not the same as model_name"""}					)
       snake_case_			:					Optional[Any]   = field(
           default=__A       ,				metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""}					)
       snake_case_			:					str   = field(
           default=__A       ,				metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}       ,				)






@dataclass
class 	SCREAMING_SNAKE_CASE:
       snake_case_			:					Optional[int]   = field(
           metadata={"""help""": """The name of the task to train selected in the list: """ + """, """.join(hans_processors.keys()					)}					)
       snake_case_			:					Tuple   = field(
           metadata={"""help""": """The input data dir. Should contain the .tsv files (or other data files) for the task."""}					)
       snake_case_			:					Union[str, Any]   = field(
           default=128       ,				metadata={
               """help""": (
                   """The maximum total input sequence length after tokenization. Sequences longer """
                   """than this will be truncated, sequences shorter will be padded."""
               )
           }       ,				)
       snake_case_			:					Union[str, Any]   = field(
           default=__A       ,				metadata={"""help""": """Overwrite the cached training and evaluation sets"""}					)





def     snake_case_  (       a__		:     Optional[Any]      ):



    """simple docstring"""




    __lowercase          =						default_data_collator(_a      )
    __lowercase          =						batch.pop("""pairID"""   ,_a      )
    return batch
def     snake_case_  (       ):



    """simple docstring"""




    __lowercase          =						HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)      )
    __lowercase          =						parser.parse_args_into_dataclasses()

    if (
        os.path.exists(training_args.output_dir      )
        and os.listdir(training_args.output_dir      )
        and training_args.do_train
        and not training_args.overwrite_output_dir
    ):
        raise ValueError(
            f'Output directory ({training_args.output_dir}) already exists and is not empty. Use'
            """ --overwrite_output_dir to overcome."""      )

    # Setup logging
    logging.basicConfig(
        format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s"""   ,datefmt="""%m/%d/%Y %H:%M:%S"""   ,level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN   ,)
    logger.warning(
        """Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"""   ,training_args.local_rank   ,training_args.device   ,training_args.n_gpu   ,bool(training_args.local_rank != -1      )   ,training_args.fpaa   ,)
    # Set the verbosity to info of the Transformers logger (on main process only):
    if is_main_process(training_args.local_rank      ):
        transformers.utils.logging.set_verbosity_info()
        transformers.utils.logging.enable_default_handler()
        transformers.utils.logging.enable_explicit_format()
    logger.info("""Training/evaluation parameters %s"""   ,_a      )

    # Set seed
    set_seed(training_args.seed      )

    try:
        __lowercase          =						hans_tasks_num_labels[data_args.task_name]
    except KeyError:
        raise ValueError("""Task not found: %s""" % (data_args.task_name)      )

    # Load pretrained model and tokenizer
    #
    # Distributed training:
    # The .from_pretrained methods guarantee that only one local process can concurrently
    # download model & vocab.

    __lowercase          =						AutoConfig.from_pretrained(
        model_args.config_name if model_args.config_name else model_args.model_name_or_path   ,num_labels=_a   ,finetuning_task=data_args.task_name   ,cache_dir=model_args.cache_dir   ,)
    __lowercase          =						AutoTokenizer.from_pretrained(
        model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path   ,cache_dir=model_args.cache_dir   ,)
    __lowercase          =						AutoModelForSequenceClassification.from_pretrained(
        model_args.model_name_or_path   ,from_tf=bool(""".ckpt""" in model_args.model_name_or_path      )   ,config=_a   ,cache_dir=model_args.cache_dir   ,)

    # Get datasets
    __lowercase          =						(
        HansDataset(
            data_dir=data_args.data_dir   ,tokenizer=_a   ,task=data_args.task_name   ,max_seq_length=data_args.max_seq_length   ,overwrite_cache=data_args.overwrite_cache   ,)
        if training_args.do_train
        else None
    )
    __lowercase          =						(
        HansDataset(
            data_dir=data_args.data_dir   ,tokenizer=_a   ,task=data_args.task_name   ,max_seq_length=data_args.max_seq_length   ,overwrite_cache=data_args.overwrite_cache   ,evaluate=_a   ,)
        if training_args.do_eval
        else None
    )

    # Initialize our Trainer
    __lowercase          =						Trainer(
        model=_a   ,args=_a   ,train_dataset=_a   ,eval_dataset=_a   ,data_collator=_a   ,)

    # Training
    if training_args.do_train:
        trainer.train(
            model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path      ) else None      )
        trainer.save_model()
        # For convenience, we also re-save the tokenizer to the same directory,
        # so that you can share your model easily on huggingface.co/models =)
        if trainer.is_world_master():
            tokenizer.save_pretrained(training_args.output_dir      )

    # Evaluation
    if training_args.do_eval:
        logger.info("""*** Evaluate ***"""      )

        __lowercase          =						trainer.predict(_a      )
        __lowercase          =						output.predictions
        __lowercase          =						np.argmax(_a   ,axis=1      )

        __lowercase          =						[ex.pairID for ex in eval_dataset]
        __lowercase          =						os.path.join(training_args.output_dir   ,"""hans_predictions.txt"""      )
        __lowercase          =						eval_dataset.get_labels()
        if trainer.is_world_master():
            with open(_a   ,"""w"""      ) as writer:
                writer.write("""pairID,gold_label\n"""      )
                for pid, pred in zip(_a   ,_a      ):
                    writer.write("""ex""" + str(_a      ) + """,""" + label_list[int(_a      )] + """\n"""      )

        trainer._log(output.metrics      )
def     snake_case_  (       a__		:     Optional[Any]      ):



    """simple docstring"""




    main()


if __name__ == "__main__":
  main()










'''simple docstring'''
# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.auto import AutoModelForSeqaSeqLM, AutoTokenizer
from .base import PipelineTool


A							:				Dict			     =	"""Here is a text containing a lot of information: '''{text}'''.

Can you answer this question about the text: '{question}'"""






class 	SCREAMING_SNAKE_CASE(    __A					):
       snake_case_			:					List[Any]   = """google/flan-t5-base"""
       snake_case_			:					int   = (
           """This is a tool that answers questions related to a text. It takes two arguments named `text`, which is the """
           """text where to find the answer, and `question`, which is the question, and returns the answer to the question."""
       )
       snake_case_			:					Optional[int]   = """text_qa"""
       snake_case_			:					List[Any]   = AutoTokenizer
       snake_case_			:					List[str]   = AutoModelForSeqaSeqLM

       snake_case_			:					int   = ["""text""", """text"""]
       snake_case_			:					Dict   = ["""text"""]

       def    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__					)   -> Optional[int]:

           """simple docstring"""



           __lowercase          =						QA_PROMPT.format(text=lowerCamelCase__ ,		question=lowerCamelCase__					)
           return self.pre_processor(lowerCamelCase__ ,		return_tensors="""pt"""					)

       def    snake_case__      (   self ,		lowerCamelCase__					)   -> int:

           """simple docstring"""



           __lowercase          =						self.model.generate(**lowerCamelCase__					)

           __lowercase    ,__lowercase          =						inputs["""input_ids"""].shape
           __lowercase          =						output_ids.shape[0]

           return output_ids.reshape(lowerCamelCase__ ,		out_b // in_b ,		*output_ids.shape[1:]					)[0][0]




       def    snake_case__      (   self ,		lowerCamelCase__					)   -> Tuple:

           """simple docstring"""



           return self.pre_processor.decode(lowerCamelCase__ ,		skip_special_tokens=lowerCamelCase__ ,		clean_up_tokenization_spaces=lowerCamelCase__					)




