




'''simple docstring'''




import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest import TestCase
from unittest.mock import patch

import faiss
import numpy as np
from datasets import Dataset

from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
from transformers.integrations import is_ray_available
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_ray


sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip

if is_torch_available():
							from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
else:
							snake_case__          =						None

if is_ray_available():
							import ray  # noqa: E402 # isort:skip
							from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
else:
							snake_case__          =						None
							snake_case__          =						None
							snake_case__          =						None




def 						snake_case__    ( lowerCamelCase__				:   List[Any]			)   ->   Dict:
					if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
										A_ :     str =							unittest.skip('''test requires Datasets, Faiss, psutil'''			)(lowerCamelCase__			)
					return test_case






@require_distributed_retrieval
class     UpperCamelCase_     (a__     ):







						"""simple docstring"""


						def       _a    (     self      :       Tuple    ):





											"""simple docstring"""







											A_ :     Tuple =							tempfile.mkdtemp()
											A_ :     Union[str, Any] =							8

											# DPR tok
											A_ :     Dict =							[
											    '''[UNK]''',
											    '''[CLS]''',
											    '''[SEP]''',
											    '''[PAD]''',
											    '''[MASK]''',
											    '''want''',
											    '''##want''',
											    '''##ed''',
											    '''wa''',
											    '''un''',
											    '''runn''',
											    '''##ing''',
											    ''',''',
											    '''low''',
											    '''lowest''',
											]
											A_ :     List[str] =							os.path.join(self.tmpdirname							, '''dpr_tokenizer'''    )
											os.makedirs(_lowerCamelCase							, exist_ok=_lowerCamelCase    )
											A_ :     Tuple =							os.path.join(_lowerCamelCase							, DPR_VOCAB_FILES_NAMES['''vocab_file''']    )
											with open(self.vocab_file							, '''w'''							, encoding='''utf-8'''    ) as vocab_writer:
																vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]    )    )

											# BART tok
											A_ :     Optional[Any] =							[
											    '''l''',
											    '''o''',
											    '''w''',
											    '''e''',
											    '''r''',
											    '''s''',
											    '''t''',
											    '''i''',
											    '''d''',
											    '''n''',
											    '''\u0120''',
											    '''\u0120l''',
											    '''\u0120n''',
											    '''\u0120lo''',
											    '''\u0120low''',
											    '''er''',
											    '''\u0120lowest''',
											    '''\u0120newer''',
											    '''\u0120wider''',
											    '''<unk>''',
											]
											A_ :     Any =							dict(zip(_lowerCamelCase							, range(len(_lowerCamelCase    )    )    )    )
											A_ :     Union[str, Any] =							['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
											A_ :     int =							{'''unk_token''': '''<unk>'''}

											A_ :     List[str] =							os.path.join(self.tmpdirname							, '''bart_tokenizer'''    )
											os.makedirs(_lowerCamelCase							, exist_ok=_lowerCamelCase    )
											A_ :     Dict =							os.path.join(_lowerCamelCase							, BART_VOCAB_FILES_NAMES['''vocab_file''']    )
											A_ :     Optional[Any] =							os.path.join(_lowerCamelCase							, BART_VOCAB_FILES_NAMES['''merges_file''']    )
											with open(self.vocab_file							, '''w'''							, encoding='''utf-8'''    ) as fp:
																fp.write(json.dumps(_lowerCamelCase    ) + '''\n'''    )
											with open(self.merges_file							, '''w'''							, encoding='''utf-8'''    ) as fp:
																fp.write('''\n'''.join(_lowerCamelCase    )    )







						def       _a    (     self      :       str    ):





											"""simple docstring"""







											return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname							, '''dpr_tokenizer'''    )    )







						def       _a    (     self      :       Optional[int]    ):





											"""simple docstring"""







											return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname							, '''bart_tokenizer'''    )    )







						def       _a    (     self      :       List[str]    ):





											"""simple docstring"""







											shutil.rmtree(self.tmpdirname    )







						def       _a    (     self      :       int    ):





											"""simple docstring"""







											A_ :     Union[str, Any] =							Dataset.from_dict(
											    {
											        '''id''': ['''0''', '''1'''],
											        '''text''': ['''foo''', '''bar'''],
											        '''title''': ['''Foo''', '''Bar'''],
											        '''embeddings''': [np.ones(self.retrieval_vector_size    ), 2 * np.ones(self.retrieval_vector_size    )],
											    }    )
											dataset.add_faiss_index('''embeddings'''							, string_factory='''Flat'''							, metric_type=faiss.METRIC_INNER_PRODUCT    )
											return dataset







						def       _a    (     self      :       str							, _lowerCamelCase      :       bool							, _lowerCamelCase      :       List[Any]=12345    ):





											"""simple docstring"""







											A_ :     Any =							self.get_dummy_dataset()
											A_ :     List[str] =							RagConfig(
											    retrieval_vector_size=self.retrieval_vector_size							, question_encoder=DPRConfig().to_dict()							, generator=BartConfig().to_dict()							, )
											with patch('''transformers.models.rag.retrieval_rag.load_dataset'''    ) as mock_load_dataset:
																A_ :     List[Any] =							dataset
																A_ :     int =							RagPyTorchDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, )
																if init_retrieval:
																					retriever.init_retrieval(_lowerCamelCase    )
											return retriever







						def       _a    (     self      :       Optional[Any]							, _lowerCamelCase      :       bool    ):





											"""simple docstring"""







											ray.init(local_mode=_lowerCamelCase    )
											A_ :     Union[str, Any] =							RagConfig(
											    retrieval_vector_size=self.retrieval_vector_size							, question_encoder=DPRConfig().to_dict()							, generator=BartConfig().to_dict()							, )
											A_ :     Dict =							ray.remote(_lowerCamelCase    )
											A_ :     Optional[Any] =							[remote_cls.remote() for _ in range(1    )]
											with patch('''transformers.models.rag.retrieval_rag.load_dataset'''    ) as mock_load_dataset:
																A_ :     Optional[Any] =							self.get_dummy_dataset()
																A_ :     Optional[int] =							RagRayDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, retrieval_workers=_lowerCamelCase							, )
																if init_retrieval:
																					retriever.init_retrieval()
											return retriever







						def       _a    (     self      :       Optional[int]							, _lowerCamelCase      :       bool							, _lowerCamelCase      :       bool							, _lowerCamelCase      :       List[Any]=12345    ):





											"""simple docstring"""







											A_ :     Union[str, Any] =							self.get_dummy_dataset()
											A_ :     Tuple =							RagConfig(
											    retrieval_vector_size=self.retrieval_vector_size							, question_encoder=DPRConfig().to_dict()							, generator=BartConfig().to_dict()							, index_name='''custom'''							, )
											if from_disk:
																A_ :     int =							os.path.join(self.tmpdirname							, '''dataset'''    )
																A_ :     Union[str, Any] =							os.path.join(self.tmpdirname							, '''index.faiss'''    )
																dataset.get_index('''embeddings'''    ).save(os.path.join(self.tmpdirname							, '''index.faiss'''    )    )
																dataset.drop_index('''embeddings'''    )
																dataset.save_to_disk(os.path.join(self.tmpdirname							, '''dataset'''    )    )
																del dataset
																A_ :     int =							RagPyTorchDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, )
											else:
																A_ :     List[Any] =							RagPyTorchDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, index=CustomHFIndex(config.retrieval_vector_size							, _lowerCamelCase    )							, )
											if init_retrieval:
																retriever.init_retrieval(_lowerCamelCase    )
											return retriever







						def       _a    (     self      :       List[Any]							, _lowerCamelCase      :       bool							, _lowerCamelCase      :       bool    ):





											"""simple docstring"""







											ray.init(local_mode=_lowerCamelCase    )
											A_ :     int =							self.get_dummy_dataset()
											A_ :     Tuple =							RagConfig(
											    retrieval_vector_size=self.retrieval_vector_size							, question_encoder=DPRConfig().to_dict()							, generator=BartConfig().to_dict()							, index_name='''custom'''							, )
											A_ :     Union[str, Any] =							ray.remote(_lowerCamelCase    )
											A_ :     int =							[remote_cls.remote() for _ in range(1    )]
											if from_disk:
																A_ :     List[Any] =							os.path.join(self.tmpdirname							, '''dataset'''    )
																A_ :     str =							os.path.join(self.tmpdirname							, '''index.faiss'''    )
																dataset.get_index('''embeddings'''    ).save(os.path.join(self.tmpdirname							, '''index.faiss'''    )    )
																dataset.drop_index('''embeddings'''    )
																dataset.save_to_disk(os.path.join(self.tmpdirname							, '''dataset'''    )    )
																del dataset
																A_ :     Optional[Any] =							RagRayDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, retrieval_workers=_lowerCamelCase							, index=CustomHFIndex.load_from_disk(
																        vector_size=config.retrieval_vector_size							, dataset_path=config.passages_path							, index_path=config.index_path							, )							, )
											else:
																A_ :     int =							RagRayDistributedRetriever(
																    _lowerCamelCase							, question_encoder_tokenizer=self.get_dpr_tokenizer()							, generator_tokenizer=self.get_bart_tokenizer()							, retrieval_workers=_lowerCamelCase							, index=CustomHFIndex(config.retrieval_vector_size							, _lowerCamelCase    )							, )
											if init_retrieval:
																retriever.init_retrieval()
											return retriever







						def       _a    (     self      :       Dict							, _lowerCamelCase      :       RagRetriever							, _lowerCamelCase      :       np.array							, _lowerCamelCase      :       int    ):





											"""simple docstring"""







											A_ ,A_ ,A_ :     List[Any] =							retriever.retrieve(_lowerCamelCase							, n_docs=_lowerCamelCase    )
											self.assertEqual(retrieved_doc_embeds.shape							, (2, n_docs, self.retrieval_vector_size)    )
											self.assertEqual(len(_lowerCamelCase    )							, 2    )
											self.assertEqual(sorted(doc_dicts[0]    )							, ['''embeddings''', '''id''', '''text''', '''title''']    )
											self.assertEqual(len(doc_dicts[0]['''id''']    )							, _lowerCamelCase    )
											self.assertEqual(doc_dicts[0]['''id'''][0]							, '''1'''    )  # max inner product is reached with second doc
											self.assertEqual(doc_dicts[1]['''id'''][0]							, '''0'''    )  # max inner product is reached with first doc
											self.assertListEqual(doc_ids.tolist()							, [[1], [0]]    )







						def       _a    (     self      :       List[Any]    ):





											"""simple docstring"""







											A_ :     Tuple =							1
											A_ :     Union[str, Any] =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )

											self.distributed_retriever_check(
											    self.get_dummy_pytorch_distributed_retriever(init_retrieval=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase    )







						def       _a    (     self      :       Tuple    ):





											"""simple docstring"""







											A_ :     List[str] =							1
											A_ :     Optional[Any] =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )

											self.distributed_retriever_check(
											    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=_lowerCamelCase							, from_disk=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase							, )







						def       _a    (     self      :       Union[str, Any]    ):





											"""simple docstring"""







											A_ :     int =							1
											A_ :     List[Any] =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )

											self.distributed_retriever_check(
											    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=_lowerCamelCase							, from_disk=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase							, )







						@require_ray
						def       _a    (     self      :       Tuple    ):





											"""simple docstring"""







											A_ :     Optional[Any] =							1
											A_ :     Dict =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )

											self.distributed_retriever_check(
											    self.get_dummy_ray_distributed_retriever(init_retrieval=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase    )
											ray.shutdown()







						@require_ray
						def       _a    (     self      :       Optional[int]    ):





											"""simple docstring"""







											A_ :     int =							1
											A_ :     Any =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )
											with self.assertRaises(_lowerCamelCase    ):
																self.distributed_retriever_check(
																    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=_lowerCamelCase							, from_disk=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase							, )
											ray.shutdown()







						@require_ray
						def       _a    (     self      :       Optional[int]    ):





											"""simple docstring"""







											A_ :     Tuple =							1
											A_ :     str =							np.array(
											    [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							, dtype=np.floataa    )

											self.distributed_retriever_check(
											    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=_lowerCamelCase							, from_disk=_lowerCamelCase    )							, _lowerCamelCase							, _lowerCamelCase    )
											ray.shutdown()









'''simple docstring'''




import argparse
import logging
import sys
import time

import tensorflow as tf
from datasets import load_dataset

from transformers import AutoTokenizer, TFAutoModelForSequenceClassification


if __name__ == "__main__":
							snake_case__          =						argparse.ArgumentParser()

							# Hyperparameters sent by the client are passed as command-line arguments to the script.
							parser.add_argument("""--epochs""", type=int, default=1)
							parser.add_argument("""--per_device_train_batch_size""", type=int, default=16)
							parser.add_argument("""--per_device_eval_batch_size""", type=int, default=8)
							parser.add_argument("""--model_name_or_path""", type=str)
							parser.add_argument("""--learning_rate""", type=str, default=5e-5)
							parser.add_argument("""--do_train""", type=bool, default=True)
							parser.add_argument("""--do_eval""", type=bool, default=True)
							parser.add_argument("""--output_dir""", type=str)

							snake_case__       ,						snake_case__          =						parser.parse_known_args()

							# overwrite batch size until we have tf_glue.py
							snake_case__          =						16
							snake_case__          =						16

							# Set up logging
							snake_case__          =						logging.getLogger(__name__)

							logging.basicConfig(
							    level=logging.getLevelName("""INFO"""),
							    handlers=[logging.StreamHandler(sys.stdout)],
							    format="""%(asctime)s - %(name)s - %(levelname)s - %(message)s""",
							)

							# Load model and tokenizer
							snake_case__          =						TFAutoModelForSequenceClassification.from_pretrained(args.model_name_or_path)
							snake_case__          =						AutoTokenizer.from_pretrained(args.model_name_or_path)

							# Load dataset
							snake_case__       ,						snake_case__          =						load_dataset("""imdb""", split=["""train""", """test"""])
							snake_case__          =						train_dataset.shuffle().select(range(50_00))  # smaller the size for train dataset to 5k
							snake_case__          =						test_dataset.shuffle().select(range(5_00))  # smaller the size for test dataset to 500

							# Preprocess train dataset
							snake_case__          =						train_dataset.map(
							    lambda e: tokenizer(e["""text"""], truncation=True, padding="""max_length"""), batched=True
							)
							train_dataset.set_format(type="""tensorflow""", columns=["""input_ids""", """attention_mask""", """label"""])

							snake_case__          =						{
							    x: train_dataset[x].to_tensor(default_value=0, shape=[None, tokenizer.model_max_length])
							    for x in ["""input_ids""", """attention_mask"""]
							}
							snake_case__          =						tf.data.Dataset.from_tensor_slices((train_features, train_dataset["""label"""])).batch(
							    args.per_device_train_batch_size
							)

							# Preprocess test dataset
							snake_case__          =						test_dataset.map(
							    lambda e: tokenizer(e["""text"""], truncation=True, padding="""max_length"""), batched=True
							)
							test_dataset.set_format(type="""tensorflow""", columns=["""input_ids""", """attention_mask""", """label"""])

							snake_case__          =						{
							    x: test_dataset[x].to_tensor(default_value=0, shape=[None, tokenizer.model_max_length])
							    for x in ["""input_ids""", """attention_mask"""]
							}
							snake_case__          =						tf.data.Dataset.from_tensor_slices((test_features, test_dataset["""label"""])).batch(
							    args.per_device_eval_batch_size
							)

							# fine optimizer and loss
							snake_case__          =						tf.keras.optimizers.Adam(learning_rate=args.learning_rate)
							snake_case__          =						tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
							snake_case__          =						[tf.keras.metrics.SparseCategoricalAccuracy()]
							model.compile(optimizer=optimizer, loss=loss, metrics=metrics)

							snake_case__          =						time.time()
							snake_case__          =						model.fit(tf_train_dataset, epochs=args.epochs, batch_size=args.per_device_train_batch_size)
							snake_case__          =						time.time() - start_train_time

							logger.info("""*** Train ***""")
							logger.info(F'train_runtime = {end_train_time}')
							for key, value in train_results.history.items():
														logger.info(F'  {key} = {value}')



