code: 
  
  
  
  
  
  import os
  from typing import Optional, Union
  
  import numpy as np
  from huggingface_hub import hf_hub_download
  
  from ... import AutoTokenizer
  from ...utils import logging
  
  
  lowerCAmelCase_   =   "block_records.npy"
  
  
  lowerCAmelCase_   =   logging.get_logger(__name__)
  
  
  
  
  def 						A_			(  lowercase_      ,		lowercase_  )			->						np.ndarray:
         import tensorflow.compat.va as tf
  
         _snake_case      :							Dict     				=						tf.data.TFRecordDataset(lowercase_      ,		buffer_size=512 * 1024 * 1024  )
         _snake_case      :							Union[str, Any]     				=						blocks_dataset.batch(lowercase_      ,		drop_remainder=lowercase_  )
         _snake_case      :							List[Any]     				=						next(blocks_dataset.take(1  ).as_numpy_iterator()  )
  
         return np_record
  
  class 				A    :
  
  
  
        def __init__(					self     ,  lowercase_     ,  lowercase_     ,  lowercase_=2     ,  lowercase_=1000     ,  lowercase_=100     ,  lowercase_=10_0000     ,  )      ->   str:
  
  
  
               '''simple docstring'''
  
               from scann.scann_ops.py.scann_ops_pybind import builder as Builder
  
               _snake_case      :							int     				=						Builder(db=lowercase_     ,  num_neighbors=lowercase_     ,  distance_measure='''dot_product'''					)
               _snake_case      :							Optional[Any]     				=						builder.tree(
                   num_leaves=lowercase_     ,  num_leaves_to_search=lowercase_     ,  training_sample_size=lowercase_					)
               _snake_case      :							Any     				=						builder.score_ah(dimensions_per_block=lowercase_					)
  
               _snake_case      :							List[Any]     				=						builder.build()
  
  
  
        def    __a (					self     ,  lowercase_					)      ->   Dict:
  
  
  
               '''simple docstring'''
  
               _snake_case				,    _snake_case      :							str     				=						self.searcher.search_batched(question_projection.detach().cpu()					)
               return retrieved_block_ids.astype('''int64'''					)
  
  
  class 				A    :
  
  
  
        def __init__(					self     ,  lowercase_     ,  lowercase_					)      ->   Tuple:
  
  
  
               '''simple docstring'''
  
               super().__init__()
               _snake_case      :							Optional[int]     				=						block_records
               _snake_case      :							List[str]     				=						tokenizer
  
  
  
        def __call__(					self     ,  lowercase_     ,  lowercase_     ,  lowercase_     ,  lowercase_=None     ,  lowercase_="pt"					)      ->   Tuple:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Dict     				=						np.take(self.block_records     ,  indices=lowercase_     ,  axis=0					)
  
               _snake_case      :							int     				=						self.tokenizer.decode(question_input_ids[0]     ,  skip_special_tokens=lowercase_					)
  
               _snake_case      :							Optional[int]     				=						[]
               _snake_case      :							int     				=						[]
               for retrieved_block in retrieved_blocks:
                      text.append(lowercase_					)
                      text_pair.append(retrieved_block.decode()					)
  
               _snake_case      :							Dict     				=						self.tokenizer(
                   lowercase_     ,  lowercase_     ,  padding=lowercase_     ,  truncation=lowercase_     ,  return_special_tokens_mask=lowercase_     ,  max_length=lowercase_					)
               _snake_case      :							List[str]     				=						concat_inputs.convert_to_tensors(lowercase_					)
  
               if answer_ids is not None:
                      return self.block_has_answer(lowercase_     ,  lowercase_					) + (concat_inputs_tensors,)
               else:
                      return (None, None, None, concat_inputs_tensors)
  
  
  
        @classmethod
        def    __a (					cls     ,  lowercase_     ,  *lowercase_     ,  **lowercase_					)      ->   str:
  
  
  
               '''simple docstring'''
  
               if os.path.isdir(lowercase_					):
                      _snake_case      :							str     				=						os.path.join(lowercase_     ,  _REALM_BLOCK_RECORDS_FILENAME					)
               else:
                      _snake_case      :							Tuple     				=						hf_hub_download(
                          repo_id=lowercase_     ,  filename=_REALM_BLOCK_RECORDS_FILENAME     ,  **lowercase_					)
               _snake_case      :							Dict     				=						np.load(lowercase_     ,  allow_pickle=lowercase_					)
  
               _snake_case      :							Tuple     				=						AutoTokenizer.from_pretrained(lowercase_     ,  *lowercase_     ,  **lowercase_					)
  
               return cls(lowercase_     ,  lowercase_					)
  
  
  
        def    __a (					self     ,  lowercase_					)      ->   Optional[int]:
  
  
  
               '''simple docstring'''
  
               np.save(os.path.join(lowercase_     ,  _REALM_BLOCK_RECORDS_FILENAME					)     ,  self.block_records					)
               # save tokenizer
               self.tokenizer.save_pretrained(lowercase_					)
  
  
  
        def    __a (					self     ,  lowercase_     ,  lowercase_					)      ->   Any:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Optional[int]     				=						[]
               _snake_case      :							List[Any]     				=						[]
               _snake_case      :							Dict     				=						[]
               _snake_case      :							Any     				=						0
  
               for input_id in concat_inputs.input_ids:
                      _snake_case      :							List[str]     				=						input_id.tolist()
                      # Check answers between two [SEP] tokens
                      _snake_case      :							Optional[int]     				=						input_id_list.index(self.tokenizer.sep_token_id					)
                      _snake_case      :							str     				=						first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id					)
  
                      start_pos.append([]					)
                      end_pos.append([]					)
                      for answer in answer_ids:
                             for idx in range(first_sep_idx + 1     ,  lowercase_					):
                                    if answer[0] == input_id_list[idx]:
                                           if input_id_list[idx : idx + len(lowercase_					)] == answer:
                                                  start_pos[-1].append(lowercase_					)
                                                  end_pos[-1].append(idx + len(lowercase_					) - 1					)
  
                      if len(start_pos[-1]					) == 0:
                             has_answers.append(lowercase_					)
                      else:
                             has_answers.append(lowercase_					)
                             if len(start_pos[-1]					) > max_answers:
                                    _snake_case      :							Any     				=						len(start_pos[-1]					)
  
          # Pad -1 to max_answers
               for start_pos_, end_pos_ in zip(lowercase_     ,  lowercase_					):
                      if len(start_pos_					) < max_answers:
                             _snake_case      :							Optional[int]     				=						[-1] * (max_answers - len(start_pos_					))
                             start_pos_ += padded
                             end_pos_ += padded
               return has_answers, start_pos, end_pos
  
  
  
  
  
  
  
style_context: 
  
  
  
  
  
  def 						A_			(  lowercase_  )			->						list:
         if len(lowercase_  ) <= 1:
                return data
         else:
                return [
                    *quick_sort([e for e in data[1:] if e <= data[0]]  ),
                    data[0],
                    *quick_sort([e for e in data[1:] if e > data[0]]  ),
                ]
  
  
  if __name__ == "__main__":
     import doctest
  
     doctest.testmod()
  
  
  
  
  
  
  