






import logging
import os
from typing import List, Tuple

import numpy as np
import psutil
import torch
import torch.distributed as dist

from transformers import RagRetriever


__UpperCamelCase   :   Union[str, Any]      =    logging.getLogger(__name__)


class 							__UpperCamelCase     (  _lowerCAmelCase ):







	def __init__(	self   :				Any							,					_lowerCAmelCase   :				str							,					_lowerCAmelCase   :				str							,					_lowerCAmelCase   :				Optional[int]							,					_lowerCAmelCase   :				str=None      )		->			int:
		"""simple docstring"""







		super().__init__(
		    _lowerCAmelCase							,					question_encoder_tokenizer=_lowerCAmelCase							,					generator_tokenizer=_lowerCAmelCase							,					index=_lowerCAmelCase							,					init_retrieval=_lowerCAmelCase							,					)
		__lowercase   							=	None







	def 			_a (	self   :				int							,					_lowerCAmelCase   :				int      )		->			Any:
		"""simple docstring"""







		logger.info("""initializing retrieval"""      )

		# initializing a separate process group for retrieval as the default
		# nccl backend doesn't support gather/scatter operations while gloo
		# is too slow to replace nccl for the core gpu communication
		if dist.is_initialized():
			logger.info("""dist initialized"""      )
			# needs to be set manually
			__lowercase   							=	self._infer_socket_ifname()
			# avoid clash with the NCCL port
			__lowercase   							=	str(distributed_port + 1      )
			__lowercase   							=	dist.new_group(ranks=_lowerCAmelCase							,					backend="""gloo"""      )

		# initialize retriever only on the main worker
		if not dist.is_initialized() or self._is_main():
			logger.info("""dist not initialized / main"""      )
			self.index.init_index()

		# all processes wait untill the retriever is initialized by the main process
		if dist.is_initialized():
			torch.distributed.barrier(group=self.process_group      )







	def 			_a (	self   :				Tuple      )		->			List[str]:
		"""simple docstring"""







		return dist.get_rank(group=self.process_group      ) == 0







	def 			_a (	self   :				Tuple							,					_lowerCAmelCase   :				Tuple							,					_lowerCAmelCase   :				Optional[Any]							,					_lowerCAmelCase   :				Optional[Any]=torch.floataa      )		->			Tuple:
		"""simple docstring"""







		__lowercase   							=	torch.empty(_lowerCAmelCase							,					dtype=_lowerCAmelCase      )
		dist.scatter(_lowerCAmelCase							,					src=0							,					scatter_list=_lowerCAmelCase							,					group=self.process_group      )
		return target_tensor







	def 			_a (	self   :				Union[str, Any]      )		->			Any:
		"""simple docstring"""







		__lowercase   							=	psutil.net_if_addrs()
		# a hacky way to deal with varying network interface names
		__lowercase   							=	next((addr for addr in addrs if addr.startswith("""e"""      ))							,					_lowerCAmelCase      )
		return ifname







	def 			_a (	self   :				str							,					_lowerCAmelCase   :				np.ndarray							,					_lowerCAmelCase   :				int      )		->			Tuple[np.ndarray, List[dict]]:
		"""simple docstring"""







		if not dist.is_initialized():
			__lowercase							,    __lowercase   							=	self._main_retrieve(_lowerCAmelCase							,					_lowerCAmelCase      )
			return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(_lowerCAmelCase      )

		# distributed training
		__lowercase   							=	dist.get_world_size(group=self.process_group      )

		# gather logic
		__lowercase   							=	None
		if self._is_main():
			__lowercase   							=	[torch.empty(question_hidden_states.shape							,					dtype=torch.floataa      ) for _ in range(_lowerCAmelCase      )]
		dist.gather(torch.tensor(_lowerCAmelCase      )							,					dst=0							,					gather_list=_lowerCAmelCase							,					group=self.process_group      )

		# scatter logic
		__lowercase   							=	question_hidden_states.shape[0]
		__lowercase   							=	[]
		__lowercase   							=	[]
		if self._is_main():
			assert len(_lowerCAmelCase      ) == world_size
			__lowercase							,    __lowercase   							=	self._main_retrieve(torch.cat(_lowerCAmelCase      ).numpy()							,					_lowerCAmelCase      )
			__lowercase							,    __lowercase   							=	torch.tensor(_lowerCAmelCase      ), torch.tensor(_lowerCAmelCase      )
			__lowercase   							=	self._chunk_tensor(_lowerCAmelCase							,					_lowerCAmelCase      )
			__lowercase   							=	self._chunk_tensor(_lowerCAmelCase							,					_lowerCAmelCase      )
		__lowercase   							=	self._scattered(_lowerCAmelCase							,					[n_queries, n_docs]							,					target_type=torch.intaa      )
		__lowercase   							=	self._scattered(_lowerCAmelCase							,					[n_queries, n_docs, question_hidden_states.shape[1]]      )

		return retrieved_doc_embeds.numpy(), doc_ids.numpy(), self.index.get_doc_dicts(_lowerCAmelCase      )








# 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.
import re

from ..utils import cached_file


# docstyle-ignore
UpperCamelCase__           =			"\nHuman: <<task>>\n\nAssistant: "


UpperCamelCase__           =			"huggingface-tools/default-prompts"
UpperCamelCase__           =			{"chat": "chat_prompt_template.txt", "run": "run_prompt_template.txt"}





def      _UpperCamelCase  (a__ :List[Any] ,      a__ :Optional[Any] ,      a__ :List[str]="run"							):






							"""simple docstring"""





							if prompt_or_repo_id is None:
														UpperCamelCase__					  =      DEFAULT_PROMPTS_REPO

							# prompt is considered a repo ID when it does not contain any kind of space
							if re.search("""\\s""" ,      a__							) is not None:
														return prompt_or_repo_id

							UpperCamelCase__					  =      cached_file(
							    a__ ,      PROMPT_FILES[mode] ,      repo_type="""dataset""" ,      user_agent={"""agent""": agent_name}							)
							with open(a__ ,      """r""" ,      encoding="""utf-8"""							) as f:
														return f.read()




