





import logging
import os
from dataclasses import dataclass, field
from functools import partial
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import List, Optional

import faiss
import torch
from datasets import Features, Sequence, Value, load_dataset

from transformers import (
    DPRContextEncoder,
    DPRContextEncoderTokenizerFast,
    HfArgumentParser,
    RagRetriever,
    RagSequenceForGeneration,
    RagTokenizer,
)


__lowerCAmelCase 	=      logging.getLogger(__name__)
torch.set_grad_enabled(False)
__lowerCAmelCase 	=      "cuda" if torch.cuda.is_available() else "cpu"

def      a    (   _lowercase :  Dict			,    _lowercase :  List[Any]=1_0_0			,    _lowercase :  Tuple=" "							)  ->List[str]:
		UpperCamelCase__								=						text.split(_SCREAMING_SNAKE_CASE							)
		return [character.join(text[i : i + n]							).strip() for i in range(0			,    len(_SCREAMING_SNAKE_CASE							)			,    _SCREAMING_SNAKE_CASE							)]

def      a    (   _lowercase :  List[Any]							)  ->dict:
		UpperCamelCase__								=						[], []
		for title, text in zip(documents["""title"""]			,    documents["""text"""]							):
				if text is not None:
						for passage in split_text(_SCREAMING_SNAKE_CASE							):
								titles.append(title if title is not None else """"""							)
								texts.append(_SCREAMING_SNAKE_CASE							)
		return {"title": titles, "text": texts}

def      a    (   _lowercase :  int			,    _lowercase :  Optional[Any]			,    _lowercase :  Optional[Any]							)  ->dict:
		UpperCamelCase__								=						ctx_tokenizer(
		    documents["""title"""]			,    documents["""text"""]			,    truncation=_SCREAMING_SNAKE_CASE			,    padding="""longest"""			,    return_tensors="""pt"""							)['input_ids']
		UpperCamelCase__								=						ctx_encoder(input_ids.to(device=_SCREAMING_SNAKE_CASE							)			,    return_dict=_SCREAMING_SNAKE_CASE							).pooler_output
		return {"embeddings": embeddings.detach().cpu().numpy()}

def      a    (   _lowercase :  Dict			,    _lowercase :  Dict			,    _lowercase :  Union[str, Any]			,    )  ->int:
		######################################
		logger.info("""Step 1 - Create the dataset"""							)
		######################################

		# The dataset needed for RAG must have three columns:
		# - title (string): title of the document
		# - text (string): text of a passage of the document
		# - embeddings (array of dimension d): DPR representation of the passage

		# Let's say you have documents in tab-separated csv files with columns "title" and "text"
		assert os.path.isfile(rag_example_args.csv_path							), "Please provide a valid path to a csv file"

		# You can load a Dataset object this way
		UpperCamelCase__								=						load_dataset(
		    """csv"""			,    data_files=[rag_example_args.csv_path]			,    split="""train"""			,    delimiter="""\t"""			,    column_names=["""title""", """text"""]							)

		# More info about loading csv files in the documentation: https://huggingface.co/docs/datasets/loading_datasets.html?highlight=csv#csv-files

		# Then split the documents into passages of 100 words
		UpperCamelCase__								=						dataset.map(_SCREAMING_SNAKE_CASE			,    batched=_SCREAMING_SNAKE_CASE			,    num_proc=processing_args.num_proc							)

		# And compute the embeddings
		UpperCamelCase__								=						DPRContextEncoder.from_pretrained(rag_example_args.dpr_ctx_encoder_model_name							).to(device=_SCREAMING_SNAKE_CASE							)
		UpperCamelCase__								=						DPRContextEncoderTokenizerFast.from_pretrained(rag_example_args.dpr_ctx_encoder_model_name							)
		UpperCamelCase__								=						Features(
		    {"""text""": Value("""string"""							), """title""": Value("""string"""							), """embeddings""": Sequence(Value("""float32"""							)							)}							)  # optional, save as float32 instead of float64 to save space
		UpperCamelCase__								=						dataset.map(
		    partial(_SCREAMING_SNAKE_CASE			,    ctx_encoder=_SCREAMING_SNAKE_CASE			,    ctx_tokenizer=_SCREAMING_SNAKE_CASE							)			,    batched=_SCREAMING_SNAKE_CASE			,    batch_size=processing_args.batch_size			,    features=_SCREAMING_SNAKE_CASE			,    )

		# And finally save your dataset
		UpperCamelCase__								=						os.path.join(rag_example_args.output_dir			,    """my_knowledge_dataset"""							)
		dataset.save_to_disk(_SCREAMING_SNAKE_CASE							)
		# from datasets import load_from_disk
		# dataset = load_from_disk(passages_path)  # to reload the dataset

		######################################
		logger.info("""Step 2 - Index the dataset"""							)
		######################################

		# Let's use the Faiss implementation of HNSW for fast approximate nearest neighbor search
		UpperCamelCase__								=						faiss.IndexHNSWFlat(index_hnsw_args.d			,    index_hnsw_args.m			,    faiss.METRIC_INNER_PRODUCT							)
		dataset.add_faiss_index("""embeddings"""			,    custom_index=_SCREAMING_SNAKE_CASE							)

		# And save the index
		UpperCamelCase__								=						os.path.join(rag_example_args.output_dir			,    """my_knowledge_dataset_hnsw_index.faiss"""							)
		dataset.get_index("""embeddings"""							).save(_SCREAMING_SNAKE_CASE							)
		# dataset.load_faiss_index("embeddings", index_path)  # to reload the index

		######################################
		logger.info("""Step 3 - Load RAG"""							)
		######################################

		# Easy way to load the model
		UpperCamelCase__								=						RagRetriever.from_pretrained(
		    rag_example_args.rag_model_name			,    index_name="""custom"""			,    indexed_dataset=_SCREAMING_SNAKE_CASE							)
		UpperCamelCase__								=						RagSequenceForGeneration.from_pretrained(rag_example_args.rag_model_name			,    retriever=_SCREAMING_SNAKE_CASE							)
		UpperCamelCase__								=						RagTokenizer.from_pretrained(rag_example_args.rag_model_name							)

		# For distributed fine-tuning you'll need to provide the paths instead, as the dataset and the index are loaded separately.
		# retriever = RagRetriever.from_pretrained(rag_model_name, index_name="custom", passages_path=passages_path, index_path=index_path)

		######################################
		logger.info("""Step 4 - Have fun"""							)
		######################################

		UpperCamelCase__								=						rag_example_args.question or 'What does Moses\' rod turn into ?'
		UpperCamelCase__								=						tokenizer.question_encoder(_SCREAMING_SNAKE_CASE			,    return_tensors="""pt"""							)['input_ids']
		UpperCamelCase__								=						model.generate(_SCREAMING_SNAKE_CASE							)
		UpperCamelCase__								=						tokenizer.batch_decode(_SCREAMING_SNAKE_CASE			,    skip_special_tokens=_SCREAMING_SNAKE_CASE							)[0]
		logger.info("""Q: """ + question							)
		logger.info("""A: """ + generated_string							)




@dataclass
class 	_UpperCAmelCase						:
					__A        = field(
					    default=str(Path(__lowerCAmelCase       ).parent / "test_data" / "my_knowledge_dataset.csv"       )			,							metadata={"help": "Path to a tab-separated csv file with columns 'title' and 'text'"}			,							)
					__A        = field(
					    default=__lowerCAmelCase			,							metadata={"help": "Question that is passed as input to RAG. Default is 'What does Moses' rod turn into ?'."}			,							)
					__A        = field(
					    default="facebook/rag-sequence-nq"			,							metadata={"help": "The RAG model to use. Either 'facebook/rag-sequence-nq' or 'facebook/rag-token-nq'"}			,							)
					__A        = field(
					    default="facebook/dpr-ctx_encoder-multiset-base"			,							metadata={
					        "help": (
					            "The DPR context encoder model to use. Either 'facebook/dpr-ctx_encoder-single-nq-base' or"
					            " 'facebook/dpr-ctx_encoder-multiset-base'"
					        )
					    }			,							)
					__A        = field(
					    default=__lowerCAmelCase			,							metadata={"help": "Path to a directory where the dataset passages and the index will be saved"}			,							)




@dataclass
class 	_UpperCAmelCase						:
					__A        = field(
					    default=__lowerCAmelCase			,							metadata={
					        "help": "The number of processes to use to split the documents into passages. Default is single process."
					    }			,							)
					__A        = field(
					    default=16			,							metadata={
					        "help": "The batch size to use when computing the passages embeddings using the DPR context encoder."
					    }			,							)




@dataclass
class 	_UpperCAmelCase						:
					__A        = field(
					    default=7_68			,							metadata={"help": "The dimension of the embeddings to pass to the HNSW Faiss index."}			,							)
					__A        = field(
					    default=1_28			,							metadata={
					        "help": (
					            "The number of bi-directional links created for every new element during the HNSW index construction."
					        )
					    }			,							)


if __name__ == "__main__":
				logging.basicConfig(level=logging.WARNING)
				logger.setLevel(logging.INFO)

				__lowerCAmelCase 	=      HfArgumentParser((RagExampleArguments, ProcessingArguments, IndexHnswArguments))
				__lowerCAmelCase	,	__lowerCAmelCase	,	__lowerCAmelCase 	=      parser.parse_args_into_dataclasses()
				with TemporaryDirectory() as tmp_dir:
								__lowerCAmelCase 	=      rag_example_args.output_dir or tmp_dir
								main(rag_example_args, processing_args, index_hnsw_args)






"""simple docstring"""
import importlib.util
import weakref
from functools import partialmethod

from .dependency_versions_check import dep_version_check
from .utils import is_accelerate_available, is_torch_available, logging


if is_torch_available():
 import torch

lowercase__   			=      logging.get_logger(__name__)







def 		__a  (      )      ->Dict:
  return importlib.util.find_spec('deepspeed'  ) is not None


if is_accelerate_available() and is_deepspeed_available():
 from accelerate.utils.deepspeed import HfDeepSpeedConfig as DeepSpeedConfig
else:
 # Inherits from a dummy `object` if accelerate is not available, so that python succeeds to import this file.
 # Deepspeed glue code will never inherit this dummy object as it checks if accelerate is available.
 from builtins import object as DeepSpeedConfig




class   __snake_case			(	__lowerCAmelCase							):



      def __init__(		self       ,     lowercase)      ->     int:





        '''simple docstring'''





        set_hf_deepspeed_config(self)
        dep_version_check('accelerate')
        dep_version_check('deepspeed')
        super().__init__(lowercase)







class   __snake_case			(	__lowerCAmelCase							):



      def __init__(		self       ,     lowercase)      ->     List[Any]:





        '''simple docstring'''





        super().__init__(lowercase)
        a__:	List[Any]     				=      None
        a__:	int     				=      []



      def     lowerCamelCase_  (		self)      ->     Any:





        '''simple docstring'''





        if self._dtype is None:
          raise ValueError('trainer_config_process() wasn\'t called yet to tell dtype')
        return self._dtype



      def     lowerCamelCase_  (		self       ,     lowercase)      ->     Optional[Any]:





        '''simple docstring'''





        a__:	Optional[int]     				=      self.get_value(lowercase)
        if val is None:
          return False
        else:
          return val == "auto"



      def     lowerCamelCase_  (		self       ,     lowercase       ,     lowercase       ,     lowercase=None       ,     lowercase=True)      ->     int:





        '''simple docstring'''





        a__				,       a__:	int     				=      self.find_config_node(lowercase)
        if config is None:
          return

        if config.get(lowercase) == "auto":
          a__:	Optional[int]     				=      hf_val
          return

        if not must_match:
          return

        a__:	List[Any]     				=      config.get(lowercase)
        if ds_val is not None and ds_val != hf_val:
          self.mismatches.append(f'- ds {ds_key_long}={ds_val} vs hf {hf_key}={hf_val}')

      a__								=						partialmethod(__lowerCAmelCase			, must_match=__lowerCAmelCase							)



      def     lowerCamelCase_  (		self       ,     lowercase)      ->     Dict:





        '''simple docstring'''





        a__:	str     				=      args.world_size * args.per_device_train_batch_size * args.gradient_accumulation_steps
        self.fill_match(
            'train_micro_batch_size_per_gpu'       ,     args.per_device_train_batch_size       ,     'per_device_train_batch_size')
        self.fill_match('gradient_accumulation_steps'       ,     args.gradient_accumulation_steps       ,     'gradient_accumulation_steps')
        self.fill_match('train_batch_size'       ,     lowercase       ,     'train_batch_size (calculated)')
        self.fill_match('gradient_clipping'       ,     args.max_grad_norm       ,     'max_grad_norm')

        self.fill_match('optimizer.params.lr'       ,     args.learning_rate       ,     'learning_rate')
        self.fill_match('optimizer.params.betas'       ,     [args.adam_betaa, args.adam_betaa]       ,     'adam_beta1+adam_beta2')
        self.fill_match('optimizer.params.eps'       ,     args.adam_epsilon       ,     'adam_epsilon')
        self.fill_match('optimizer.params.weight_decay'       ,     args.weight_decay       ,     'weight_decay')

        self.fill_only('scheduler.params.warmup_min_lr'       ,     0)  # not a trainer arg
        self.fill_match('scheduler.params.warmup_max_lr'       ,     args.learning_rate       ,     'learning_rate')
        # total_num_steps - will get set in trainer_config_finalize

        # fp16
        if args.fpaa or args.fpaa_full_eval:
          a__:	Tuple     				=      'apex' if args.fpaa_backend == 'apex' else 'amp'
        else:
          a__:	Optional[int]     				=      None

        if args.save_on_each_node:
          # deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
          a__:	Union[str, Any]     				=      self.config.get('checkpoint'       ,     {})
          a__:	Tuple     				=      args.save_on_each_node

        # amp: similar to the pytorch native amp - it has a bunch of optional params but we won't set
        # any here unless the user did the work
        self.fill_match(
            'fp16.enabled'       ,     ((args.fpaa or args.fpaa_full_eval) and fpaa_backend == 'amp')       ,     'fp16|fp16_full_eval+fp16_backend(amp)'       ,     )

        # apex: delegates amp work to apex (which needs to be available), but it cannot be used with any
        # ZeRO features
        self.fill_match('amp.enabled'       ,     fpaa_backend == 'apex'       ,     'fp16+fp16_backend(apex)')
        self.fill_match('amp.opt_level'       ,     args.fpaa_opt_level       ,     'fp16_opt_level')

        self.fill_match('bf16.enabled'       ,     (args.bfaa or args.bfaa_full_eval)       ,     'bf16|bf16_full_eval')

        # deepspeed's default mode is fp16 unless there is a config that says differently
        if self.is_true('bf16.enabled'):
          a__:	Optional[int]     				=      torch.bfloataa
        elif self.is_false('fp16.enabled'):
          a__:	Tuple     				=      torch.floataa
        else:
          a__:	Tuple     				=      torch.floataa



      def     lowerCamelCase_  (		self       ,     lowercase       ,     lowercase       ,     lowercase)      ->     Any:





        '''simple docstring'''





        a__:	List[Any]     				=      [
            'zero_optimization.reduce_bucket_size',
            'zero_optimization.stage3_prefetch_bucket_size',
            'zero_optimization.stage3_param_persistence_threshold',
        ]
        a__:	Union[str, Any]     				=      [x for x in hidden_size_based_keys if self.is_auto(lowercase)]

        if len(lowercase) > 0:
          if hasattr(model.config       ,     'hidden_size'):
            a__:	Any     				=      model.config.hidden_size
          elif hasattr(model.config       ,     'hidden_sizes'):
            # if there are many hidden sizes pick the largest one
            a__:	List[str]     				=      max(model.config.hidden_sizes)
          else:
            raise ValueError(
                'The model\'s config file has neither `hidden_size` nor `hidden_sizes` entry, '
                'therefore it\'s not possible to automatically fill out the following `auto` entries '
                f'in the DeepSpeed config file: {hidden_size_auto_keys}. You can fix that by replacing '
                '`auto` values for these keys with an integer value of your choice.')

          self.fill_only('zero_optimization.reduce_bucket_size'       ,     hidden_size * hidden_size)
          if self.is_zeroa():
            # automatically assign the optimal config values based on model config
            self.fill_only('zero_optimization.stage3_prefetch_bucket_size'       ,     0.9 * hidden_size * hidden_size)
            self.fill_only('zero_optimization.stage3_param_persistence_threshold'       ,     10 * hidden_size)

        # scheduler
        self.fill_match('scheduler.params.total_num_steps'       ,     lowercase       ,     'num_training_steps (calculated)')
        self.fill_match('scheduler.params.warmup_num_steps'       ,     args.get_warmup_steps(lowercase)       ,     'warmup_steps')

        if len(self.mismatches) > 0:
          a__:	int     				=      '\n'.join(self.mismatches)
          raise ValueError(
              'Please correct the following DeepSpeed config values that mismatch TrainingArguments'
              f' values:\n{mismatches}\nThe easiest method is to set these DeepSpeed config values to \'auto\'.')







lowercase__   			=      None







def 		__a  (      _SCREAMING_SNAKE_CASE  )      ->List[Any]:
  # this is a special weakref global object to allow us to get to Deepspeed config from APIs
  # that don't have an easy way to get to the Deepspeed config outside of the Trainer domain.
  global _hf_deepspeed_config_weak_ref
  # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
  a__:	Any     				=      weakref.ref(_SCREAMING_SNAKE_CASE  )







def 		__a  (      )      ->List[Any]:
  # useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
  global _hf_deepspeed_config_weak_ref
  a__:	Any     				=      None







def 		__a  (      )      ->Union[str, Any]:
  if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
    return _hf_deepspeed_config_weak_ref().is_zeroa()
  else:
    return False







def 		__a  (      )      ->Tuple:
  if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
    return _hf_deepspeed_config_weak_ref().config
  else:
    return None







def 		__a  (      _SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE  )      ->Dict:
  from accelerate.utils import DummyOptim, DummyScheduler

  a__:	Optional[Any]     				=      hf_deepspeed_config.config

  # Optimizer + Scheduler
  # Currently supported combos:
  # 1. DS scheduler + DS optimizer: Yes
  # 2. HF scheduler + HF optimizer: Yes
  # 3. DS scheduler + HF optimizer: Yes
  # 4. HF scheduler + DS optimizer: No
  #
  # Unless Offload is enabled in which case it's:
  # 1. DS scheduler + DS optimizer: Yes
  # 2. HF scheduler + HF optimizer: Mostly*
  # 3. DS scheduler + HF optimizer: Mostly*
  # 4. HF scheduler + DS optimizer: No
  #
  # Mostly*: All non-native DeepSpeed optimizers that have both CPU and GPU implementation should work (except LAMB)

  a__:	Dict     				=      None
  if "optimizer" in config:
    if args.adafactor:
      raise ValueError(
          '--adafactor was passed, but also found `optimizer` configured in the DeepSpeed config. '
          'Only one optimizer can be configured.'  )
    a__:	str     				=      DummyOptim(params=_SCREAMING_SNAKE_CASE  )
  else:
    if hf_deepspeed_config.is_offload():
      logger.info(
          'Detected ZeRO Offload and non-DeepSpeed optimizers: This combination should work as long as the'
          ' custom optimizer has both CPU and GPU implementation (except LAMB)'  )

    # ds supports Adam, OneBitAdam, and Lamb optimizers and can import other optimizers from torch.
    # But trainer uses AdamW by default.
    a__:	Any     				=      trainer.create_optimizer()
    # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
    a__:	str     				=      True

  a__:	Dict     				=      None
  if "scheduler" in config:
    a__:	List[str]     				=      DummyScheduler(_SCREAMING_SNAKE_CASE  )
  else:
    if isinstance(_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE  ):
      raise ValueError(
          'Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. '
          'Please configure a scheduler in the DeepSpeed config.'  )
    a__:	Tuple     				=      trainer.create_scheduler(num_training_steps=_SCREAMING_SNAKE_CASE						,	optimizer=_SCREAMING_SNAKE_CASE  )

  return optimizer, lr_scheduler







def 		__a  (      _SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE=False  )      ->Dict:
  from deepspeed.utils import logger as ds_logger

  a__:	Optional[int]     				=      trainer.model
  a__:	List[Any]     				=      trainer.args

  a__:	Optional[int]     				=      trainer.accelerator.state.deepspeed_plugin.hf_ds_config

  # resume config update - some bits like `model` and `num_training_steps` only become available during train
  hf_deepspeed_config.trainer_config_finalize(_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE  )

  # set the Deepspeed log level consistent with the Trainer
  ds_logger.setLevel(args.get_process_log_level()  )

  if inference:
    # only Z3 makes sense for the inference
    if not hf_deepspeed_config.is_zeroa():
      raise ValueError('ZeRO inference only makes sense with ZeRO Stage 3 - please adjust your config'  )

    # in case the training config is re-used for inference
    hf_deepspeed_config.del_config_sub_tree('optimizer'  )
    hf_deepspeed_config.del_config_sub_tree('lr_scheduler'  )
    a__				,       a__:	Optional[int]     				=      None, None
    a__:	Union[str, Any]     				=      None
  else:
    a__:	List[Any]     				=      None  # important for when deepspeed_init is used as re-init
    a__:	Union[str, Any]     				=      list(filter(lambda _SCREAMING_SNAKE_CASE  : p.requires_grad						,	model.parameters()  )  )
    a__				,       a__:	Optional[int]     				=      deepspeed_optim_sched(
        _SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE  )

  # keep for quick debug:
  # from pprint import pprint; pprint(config)

  return optimizer, lr_scheduler







def 		__a  (      _SCREAMING_SNAKE_CASE						,	_SCREAMING_SNAKE_CASE  )      ->int:
  # it's possible that the user is trying to resume from model_path, which doesn't necessarily
  # contain a deepspeed checkpoint. e.g. examples just check if the dir exists and assume it's
  # a resume from a checkpoint and not just a local pretrained weight. So we check here if the
  # path contains what looks like a deepspeed checkpoint
  import glob

  a__:	Tuple     				=      sorted(glob.glob(F'{checkpoint_path}/global_step*'  )  )

  if len(_SCREAMING_SNAKE_CASE  ) > 0:
    logger.info(F'Attempting to resume from {checkpoint_path}'  )
    # this magically updates self.optimizer and self.lr_scheduler
    a__				,       a__:	Union[str, Any]     				=      deepspeed_engine.load_checkpoint(
        _SCREAMING_SNAKE_CASE						,	load_optimizer_states=_SCREAMING_SNAKE_CASE						,	load_lr_scheduler_states=_SCREAMING_SNAKE_CASE  )
    if load_path is None:
      raise ValueError(F'[deepspeed] failed to resume from checkpoint {checkpoint_path}'  )
  else:
    raise ValueError(F'Can\'t find a valid checkpoint at {checkpoint_path}'  )


