MINI

- Pea : Executor는 데이터를 처리할 수 있는 Driver가 필요하기 때문에 둘 다 Flow에서 마이크로 서비스의 필수 구성 요소이다. 따라서 우리는 Encoder Microservice를 얻기 위해 Pea를 사용하여 Executor와 Driver를 함께 wrapping 한다. Pea는 게이트웨이 또는 Flow의 다른 Peas에서 들어오는 메시지를 지속적으로 수신 대기하고 메시지를 수신할 때 Driver를 호출하는 마이크로 서비스이다.

- Pod : 신경 검색 애플리케이션을 최적화하기 위해 Jina는 기본적으로 병렬화를 제공한다. 단일 인코더를 사용하는 대신 여러 프로세스로 분할할 수 있다. multiple Encoder microservice가 하나의 Encoder처럼 기능적으로 작동하도록 하기 위해 Peas 그룹을 Pod에 wrapping 한다. Pod는 load balancing, further control, context management를 담당하는 microservice 그룹이다.

- 이 디자인의 장점은 Pod가 로컬 호스트 또는 네트워크를 통해 서로 다른 컴퓨터에서 실행될 때, 애플리케이션을 분산시키고, 효율적이며 확장 가능하다는 것이다.

- Jina가 제공하는 building block을 사용하여 아래 2가지를 할 수 있다.

   1) design an Index Flow

   2) create an Encoder Pod with two simple YAML files   

1. Create a Pod for the Encoder

   - 먼저 pods 폴더 안에 encode.yml 파일을 생성한 후, Jina Hub에서 사용할 Encoder이름을 지정한다.

!사용할 인코더 이름
with:
  pool_strategy: auto
  pretrained_model_name_or_path: model/모델이름
  max_length: 512

2. Add the Encoder to the Index Flow

  - 위에서 Encoder가 준비되었으므로, Index Flow를 만들어 본다.

  - flows 폴더 안에 index.yml 파일을 생성한 후, Index Flow에서 첫 번째 Pod를 지정한다.

  - 매개변수를 사용하여 Encoder를 분할할 프로세스의 수를 지정할 수 있다.

!Flow
pods:
  encoder:
    uses: pods/encode.yml
    parallel: $JINA_PARALLEL
    timeout_ready: 600000
    read_only: true

1. Create a Pod for the Indexer

- 먼저 pods 폴더 안에 doc.yml 파일을 생성한 후, NumpyIndexer인 Vector Indexer와 BinaryPbIndexer인 KV Indexer로 구성된 복합 Indexer인 CompoundIndexer 를 정의한다.

- 인덱싱된 데이터는 vec.gz와 doc.gz에 각각 저장된다.

- workspace는 인덱스가 저장될 디렉토리이다.

!CompoundIndexer
components:
  - !NumpyIndexer
    with:
      index_filename: vec.gz
    metas:
      name: vecidx
      workspace: $WORKDIR
  - !BinaryPbIndexer
    with:
      index_filename: doc.gz
    metas:
      name: docidx
      workspace: $WORKDIR
metas:
  name: doc_compound_indexer
  workspace: $WORKDIR

2. Add the Indexer to the Index Flow

  - flows/index.yml로 돌아가서 Index Flow에 doc_indexer로 Indexer를 추가한다.

  - 만약 데이터가 크다면, 최적화를 위해 애플리케이션에 sharding을 추가할 수 있다.

  - app.py에서 환경 변수로 사용된다.

!Flow
pods:
  encoder:
    uses: pods/encode.yml
    parallel: $JINA_PARALLEL
    timeout_ready: 600000
    read_only: true
  doc_indexer:
    uses: pods/doc.yml
    shards: $JINA_SHARDS
    separated_workspace: true

Jina의 Flow API를 사용하여 Index Flow를 시각화 할 수 있다.

from jina.flow import Flow

f = Flow.load_config('flows/index.yml')
f.plot()

1) Set up

- Executor type 중 4번 Ranker를 선택한다.

jina hub new

2) Fill in the logit for reranking

- __init__.py 파일에 아래와 같은 로직을 구현한다.

 from jina.executors.rankers import BaseRanker
            
 class FinBertQARanker(BaseRanker):
     """
     :class:`FinBertQARanker` recomputes match scores using FinBERT-QA.
     """
        
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # your customized __init__ below
         raise NotImplementedError
        
     def score(self, *args, **kwargs):
         raise NotImplementedError

- Jina는 다양한 Executor 클래스를 포함하고 있으며, 여기서 사용할 기본 Ranker 클래스는 Match2DocRankers라고 하며, 일치 점수를 다시 계산하는 기능이 있다.

- 먼저 BaseRanker의 기본 클래스를 Match2DocRanker로 변경한다. 또한 Jina와 현재 디렉토리를 정의할 뿐만 아니라 필요한 다른 모듈을 사용하여 PyTorch를 가져온다.

 import os
 from typing import Dict
 import numpy as np
    
 from jina.executors.devices import TorchDevice
 from jina.executors.rankers import Match2DocRanker
    
 cur_dir = os.path.dirname(os.path.abspath(__file__))

- 로직은 Jina의 TorchDevice 및 Match2DocRanker를 사용하는 FinBertQARanker 클래스에서 구현된다. 나중에 Dockerfile에 필요한 모델을 다운로드한다.

- models 폴더에 (1) bert-qa 및 (2) 2_finbert-qa-50_512_16_3e6.pt 라는 두 가지 모델이 있다고 가정해보자.

(1) bert-qa : BERT를 사용한 Passage Re-ranking의 MS 매크로 데이터 세트에서 미세 조정된 bert-base-uncased

(2) 2_finbert-qa-50_512_16_3e6.pt: FinBERT-QA 모델 - FiQA 데이터 세트에서 미세 조정된 bert-qa

- 먼저 초기화에 사용할 사전 훈련된 모델로 bert-qa/를 지정하고

- QA relevancy score를 계산하는 데 사용할 모델로 2_finbert-qa-50_512_16_3e6.pt를 지정하고

- QA의 최대 시퀀스 길이를 지정한다.

class FinBertQARanker(TorchDevice, Match2DocRanker):
     """
     :class:`FinBertQARanker` Compute QA relevancy scores using a fine-tuned BERT model.
     """
    
     required_keys = {"text"}
    
     def __init__(
             self,
             pretrained_model_name_or_path: str = os.path.join(cur_dir, "models/bert-qa"),
             model_path: str = os.path.join(cur_dir, "models/2_finbert-qa-50_512_16_3e6.pt"),
             max_length: int = 512,
             *args, **kwargs):
         """
         :param pretrained_model_name_or_path: the name of the pre-trained model.
         :param model_path: the path of the fine-tuned model.
         :param max_length: the max length to truncate the tokenized sequences to.
         """
    
         super().__init__(*args, **kwargs)
         self.pretrained_model_name_or_path = pretrained_model_name_or_path
         self.model_path = model_path
         self.max_length = max_length

- 그런 다음 Hugging Face transformer을 사용하여 binary classification task에 대한 모델을 로드하기 위해 클래스에 post_init 함수를 추가한다. evaluation mode에서 모델을 설정해야 한다.

     def post_init(self):
         super().post_init()
         import torch
         from transformers import BertForSequenceClassification, AutoTokenizer
    
         self.device = torch.device("cpu")
         self.tokenizer = AutoTokenizer.from_pretrained(self.pretrained_model_name_or_path, do_lower_case=True)
         self.model = BertForSequenceClassification.from_pretrained(self.pretrained_model_name_or_path, cache_dir=None, num_labels=2)
         self.model.load_state_dict(torch.load(self.model_path, map_location=self.device), strict=False)
         self.to_device(self.model)
         self.model.eval()

-  _get_score 함수를 구현하여 질문의 relevancy score와 Top-k개 매칭된 답변을 각각 계산한다. 먼저 질문과 각 top-k 답변을 연결하고 변환하여 tokenizer를 사용하여 모델에 필요한 입력( input_ids, token_type_ids, att_mask)을 얻기 위해 인코딩한다.

- 그런 다음 input을 모델에 입력하고 QA 쌍이 관련이 있다는 prediction score를 얻는다(label = 1). prediction score를 0과 1 사이의 확률로 변환하기 위해 점수에 softmax 함수를 적용한다. output은 QA 쌍에 대한 확률 형태의 relevancy score가 된다.

     def _get_score(self, query, answer):
         import torch
         from torch.nn.functional import softmax
    
         # Create input embeddings for the model
         encoded_seq = self.tokenizer.encode_plus(query, answer,
                                             max_length=self.max_length,
                                             pad_to_max_length=True,
                                             return_token_type_ids=True,
                                             return_attention_mask=True)
         # Numericalized, padded, clipped seq with special tokens
         input_ids = torch.tensor([encoded_seq['input_ids']]).to(self.device)
         # Specify which position of the embedding is the question or answer
         token_type_ids = torch.tensor([encoded_seq['token_type_ids']]).to(self.device)
         # Specify which position of the embedding is padded
         att_mask = torch.tensor([encoded_seq['attention_mask']]).to(self.device)
         # Don't calculate gradients
         with torch.no_grad():
             # Forward pass, calculate logit predictions for each QA pair
             outputs = self.model(input_ids, token_type_ids=token_type_ids, attention_mask=att_mask)
         # Get the predictions
         logits = outputs[0]
         # Apply activation function to get the relevancy score
         rel_score = softmax(logits, dim=1)
         rel_score = rel_score.numpy()
         # Probability that the QA pair is relevant
         rel_score = rel_score[:, 1][0]
    
         return rel_score

- 마지막으로, 사용자와 Jina의 match score를 입력으로 가져와서 _get_scores를 사용하여 new score를 다시 계산하는 scoring function 를 만든다.

     def score(
             self, query_meta: Dict, old_match_scores: Dict, match_meta: Dict
     ) -> "np.ndarray":
    
         new_scores = [
             (
                 match_id,
                 self._get_score(query_meta['text'], match_meta[match_id]['text']),
              )
             for match_id, old_score in old_match_scores.items()
         ]
         return np.array(
             new_scores
             ,
             dtype=[(self.COL_MATCH_ID, np.int64), (self.COL_SCORE, np.float64)],
         )

3) Write a Unit Test

- 새로운 Executor를 생성하고 Jina Hub API를 사용하여 Docker 이미지를 빌드하려면 단위 테스트를 작성해야 한다. tests/test_finbertqaranker.py에서 이에 대한 템플릿을 찾을 수 있다. 쿼리가 주어진 두 개의 답변 일치에 대한 관련성 확률을 계산하고 FinBertQARanker가 우리의 예상과 동일한 점수를 계산하는지 확인하기 위해 간단한 검사를 작성한다.

 import copy
 import json
 import numpy as np
 import torch
 from torch.nn.functional import softmax
 from transformers import BertTokenizer, BertForSequenceClassification
    
 from jina.executors.rankers import Match2DocRanker
    
 from .. import FinBertQARanker
    
 def test_finbertqaranker():
     """here is my test code
    
     https://docs.pytest.org/en/stable/getting-started.html#create-your-first-test
     """
     query_meta = {"text": "Why are big companies like Apple or Google not included in the Dow Jones Industrial "
                           "Average (DJIA) index?"}
     query_meta_json = json.dumps(query_meta, sort_keys=True)
     old_match_scores = {1: 5, 2: 7}
     old_match_scores_json = json.dumps(old_match_scores, sort_keys=True)
     match_meta = {1: {"text": "That is a pretty exclusive club and for the most part they are not interested in "
                               "highly volatile companies like Apple and Google. Sure, IBM is part of the DJIA, "
                               "but that is about as stalwart as you can get these days. The typical profile for a "
                               "DJIA stock would be one that pays fairly predictable dividends, has been around since "
                               "money was invented, and are not going anywhere unless the apocalypse really happens "
                               "this year. In summary, DJIA is the boring reliable company index."},
                   2: {"text": "In  most  cases  you  cannot  do  reverse  lookup  on  tax  id  in  the  US.  You  can "
                               " verify ,  but  for  that  you  need  to  have  more  than  just  the  FEIN/SSN.  You  "
                               "should  also  have  a  name ,  and  some  times  address.  Non-profits ,  specifically "
                               ",  have  to  publish  their  EIN  to  donors ,  so  it  may  be  easier  than  others  "
                               "to  identify  those.  Other  businesses  may  not  be  as  easy  to  find  just  by  "
                               "EIN."}}
     match_meta_json = json.dumps(match_meta, sort_keys=True)
    
     pretrained_model = 'models/bert-qa'
     model_path = "models/2_finbert-qa-50_512_16_3e6.pt"
    
     ranker = FinBertQARanker(pretrained_model_name_or_path=pretrained_model, model_path=model_path)
    
     new_scores = ranker.score(
         copy.deepcopy(query_meta),
         copy.deepcopy(old_match_scores),
         copy.deepcopy(match_meta)
     )
    
     # new_scores = [(1, 0.7607551217079163), (2, 0.0001482228108216077)]
     np.testing.assert_approx_equal(new_scores[0][1], 0.7607, significant=4)
     np.testing.assert_approx_equal(round(new_scores[1][1], 4), 0.0001, significant=4)
    
     # Guarantee no side-effects happen
     assert query_meta_json == json.dumps(query_meta, sort_keys=True)
     assert old_match_scores_json == json.dumps(old_match_scores, sort_keys=True)
     assert match_meta_json == json.dumps(match_meta, sort_keys=True)

4) Add Requirements

- Jina 외에도 FinBertQARanker용 PyTorch 및 transformer를 사용하고 있으므로 FinBertQARanker/requirements.txt에 추가한다.

torch==1.7.1
transformers==4.0.1

** Dockerfile 관련해서는 아래 내용 참고!

FROM jinaai/jina
    
 RUN apt-get update && \
     apt-get -y install wget && \
     apt-get -y install unzip
    
 COPY ./requirements.txt /
 RUN pip install -r requirements.txt
 RUN pip uninstall -y dataclasses
 RUN pip install pytest
    
 RUN wget https://www.dropbox.com/s/sh2h9o5yd7v4ku6/bert-qa.zip
 RUN wget https://www.dropbox.com/s/12uiuumz4vbqvhk/2_finbert-qa-50_512_16_3e6.pt
    
 # setup the workspace
 COPY  . /workspace
 WORKDIR /workspace
    
 RUN if [ ! -d models ]; then \
     mkdir models/ && cd models/ && \
     mv /bert-qa.zip                   . && \
     mv /2_finbert-qa-50_512_16_3e6.pt . && \
     mkdir bert-qa && \
     unzip bert-qa.zip -d bert-qa/; \
 fi
    
 RUN pytest
    
 ENTRYPOINT ["jina", "pod", "--uses", "config.yml"]

jina hub build FinBertQARanker/ --pull --test-uses --timeout-ready 60000