[TF 2.x] tf.keras prediction 결과 custom하기(feat. GCP ai-platform)

tf.estimator를 사용하는 사람들은 tf.estimator의 Estimatorspec 을 이용하여 predictoutput의 형태를 지정해줄 수 있다.

 

가령 classification 문제의 경우, 상품 id, 구매자 이름 등의 모델에 태우지 않은 info 정보들을 포함하고 있는 것을 말한다.

        if mode == tf.estimator.ModeKeys.PREDICT:
            in_out_dist = input_layer - output_layer
            predictions = {
                'in_out_dist': in_out_dist,
                'hint_idx': tf.argmax(tf.abs(in_out_dist), axis=1),
                'score': tf.norm(input_layer - output_layer, axis=1)
            }
            info_columns_tensor = [tf.identity(col) for col in info_column_list]
            predictions.update(dict((k,v) for k,v in zip(self.INFO_COLUMNS,info_columns_tensor)))
            export_outputs = {
                'predict': tf.estimator.export.PredictOutput(predictions)
            }

            return tf.estimator.EstimatorSpec(mode,
                                              predictions=predictions,
                                              export_outputs=export_outputs)

 

 

근데 tensorflow 2.x으로 넘어가면서 tf.keras를 쓰게 되는데...keras에는 이런 기능을 찾아보기가 어렵다.

이럴때는 어떻게 해야할까?

 

답은...

모델을 저장한 후 저장된 모델을 다시 불러와 tf.function으로 싸서 serving_signature를 지정해준 뒤 다시 모델을 저장하는 방법이다.

 

처음에는 custom keras 모델을 만들어 predict 함수를 overriding 하여 사용할 생각을 했지만...그것은 에러가 나며 잘 되지 않았다.

--> 모델을 계속 로드중일때는 오버라이딩한  predict 가 동작하나, 모델을 저장후 다시 로드하여 사용할때는 동작하지 않는다.

왜냐면 모델을 저장할때 모델의 graph와 가중치가 저장되고, 오버라이딩한 method는 저장되지 않기 때문이다.

 

이렇게 정의한 method도 함께 저장하기 위해 사용하는 것이 @tf.function 이다.

 

import os
import urllib

import pandas as pd

import tensorflow as tf
from tensorflow.keras import Input, Model
from tensorflow.keras import optimizers
from tensorflow.keras.layers import (
    Dense,
    Embedding,
    GRU
)
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.utils import to_categorical

print(tf.__version__)

from sklearn.model_selection import train_test_split



## 데이터
info_cols=['이름','나이']
x = [
    [1,2,3,4],
    [2,3,4,5],
    [3,1,2,3],
    [3,1,2,2]
]
y = [
    [1],
    [0],
    [1],
    [1]
]
x_pred = [
    ['김씨',12, 1,2,3,4],
    ['방씨',21, 1,2,5,4]
]


class MyKerasModel(tf.keras.Model):

    def __init__(self, info_cols):
        super().__init__()
        self.dense1 = tf.keras.layers.Dense(4, activation=tf.nn.relu)
        self.dense2 = tf.keras.layers.Dense(2, activation=tf.nn.softmax)
        self.dropout = tf.keras.layers.Dropout(0.5)
        self.info_cols = info_cols
        print('if you have any problems or issue, ask bdh for solution')
        
    @tf.function    
    def call(self, inputs, training=False):
        x = self.dense1(inputs)        
        re = self.dense2(x)
        return re
    
    
    # method overiding
    def predict(self, x, batch_size=None,verbose=0,steps=None,callbacks=None,max_queue_size=10,workers=1,use_multiprocessing=False):
        print('predict method of keras overiding')
        
        infos = []
        pred_data =[]
        result = []
        
        for data in x:
            print('data:',data)
            infos.append(
                {info_cols[ind] : data[ind] for ind in range(len(info_cols))} 
            )
            pred_data.append(
                data[len(info_cols):]
            )
            
        print('infos:',infos)
        print('pred_data:',pred_data)
        
        outputs =  super().predict(pred_data, batch_size, verbose, steps, callbacks, max_queue_size, workers, use_multiprocessing)    
        
        
        for info, output in zip(infos, outputs):
            info['result']=output
            result.append(info)

        return result
        
    @tf.function
    def check_test(self,x):
        print('123checkcheck')
        return x
        
model = MyKerasModel(info_cols)
   
model.compile(
        optimizer=tf.keras.optimizers.Adam(
            learning_rate=0.0005, beta_1=0.9, beta_2=0.999, epsilon=1e-07, amsgrad=False,
            name='Adam'
        ),
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    model.fit(x,y)
    
result = model.predict(x_pred)
result

========
predict method of keras overiding
data: ['김씨', 12, 1, 2, 3, 4]
data: ['방씨', 21, 1, 2, 5, 4]
infos: [{'이름': '김씨', '나이': 12}, {'이름': '방씨', '나이': 21}]
pred_data: [[1, 2, 3, 4], [1, 2, 5, 4]]
[{'result': array([1.21742676e-04, 9.99878287e-01], dtype=float32),
  '나이': 12,
  '이름': '김씨'},
 {'result': array([1.0611235e-05, 9.9998939e-01], dtype=float32),
  '나이': 21,
  '이름': '방씨'}]

tf.function으로 오버라이딩한 predict 메서드를 저장하려 하면 안되는 것을 알 수 있다. 이것은 predict 메서드 자체가 tf.function으로 쌀 수 없게 되어 있기 때문인데...

이것 때문에

 

모델을 저장후 다시 불러와 serving_signature를 tf.function으로 덮어준후 다시 저장하여 사용하는 방법이 있다.

 

MODEL_EXPORT_PATH = './test_rnn_model/'
tf.saved_model.save(rnn_model, MODEL_EXPORT_PATH)

loaded_model = tf.keras.models.load_model(MODEL_EXPORT_PATH)

 

컴파일한 모델 VS 저장후 다시 불러온 모델
- 컴파일한 오리지날 모델은 serving signature를 가지고 있지 않는다.
- 일단 모델을 저장해야 serving signature를 갖게 되어서 저장후 다시 불러온 다음 serving_signature를 지정하는 것이다.

loaded_model
<tensorflow.python.keras.saving.saved_model.load.Model at 0x7f117873d048>

rnn_model
<tensorflow.python.keras.engine.training.Model at 0x7f11ec0cfef0>

- loaded model 에서는 inference_function이 predict 의 역할을 대신한다. 그리고 이것은 keras의 Model.predict() 함수와 비슷하다.
- 주목할 것은 output tensor의 name이 serving signature에 매칭된다는 것이다

inference_function = loaded_model.signatures['serving_default']
inference_function # similar to keras.Model.predict()
<tensorflow.python.saved_model.load._WrapperFunction at 0x7f11787223c8>


result = inference_function(tf.convert_to_tensor(X_valid))
print(result)

{'output': <tf.Tensor: id=9997, shape=(200, 2), dtype=float32, numpy=
array([[0.50841326, 0.4915868 ],
       [0.5067442 , 0.49325582],
       [0.5192482 , 0.4807518 ],
       [0.50487745, 0.49512258],
       [0.5148147 , 0.4851853 ],
       [0.50024205, 0.49975792],
       [0.5185044 , 0.48149565],
       [0.5157503 , 0.48424968],
       [0.51632905, 0.4836709 ], dtype=float32)>}
       

Keyed Serving Function

• Now we'll create a new serving function that accepts and outputs a unique instance key.

• We use the fact that a Keras Model(x) call actually runs a prediction.

• The training=False parameter is included only for clarity. Then we save the model as before but provide this function as our new serving signature

@tf.function(input_signature=[tf.TensorSpec([None], dtype=tf.string), tf.TensorSpec([None, MAX_LEN], dtype=tf.int32)])
def keyed_prediction(key, data):
    pred = loaded_model(data, training=False)
    return {
        'output': pred,
        'key': key
    }

여기서, serving_default를 내가 만든 함수로 바꿔준다.

원래 serving_default는 아래와 같이 wrapperfunction으로 싸져 있음

_SignatureMap({'serving_default': <tensorflow.python.saved_model.load._WrapperFunction object at 0x7f441473b5d0>})

KEYED_EXPORT_PATH = './keyed_test_rnn_model/'
loaded_model.save(KEYED_EXPORT_PATH, signatures={'serving_default': keyed_prediction})

keyed_model = tf.keras.models.load_model(KEYED_EXPORT_PATH)

keyed_model.predict(
    {
    'input_1': tf.convert_to_tensor([X_valid[0]], dtype=tf.int32),
    'key': tf.constant("1번유저")
    }
)

array([[0.50841326, 0.49158677]], dtype=float32)

 

이렇게 loaded한 모델을 다시 불러와 tf.function을 만들어 저장해주면 아주 손쉽게 모델에 태운 결과와 태운 데이터의 정보를 같이 출력할 수 있다.

 

 

GCP ai-platform 적용

!gcloud ai-platform models create test_keyed_model \
  --regions us-central1
  
!gcloud ai-platform versions create v2 \
       --model test_keyed_model --origin  ${MODEL_LOCATION} --staging-bucket gs://daehwan \
       --runtime-version 2.1
       
       
       
with open("keyed_txt_input.json", "w") as file:
    print('{"data": [494, 9251, 9252,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], "key": "id_1234"}', file=file)
    
!gcloud ai-platform predict --model test_keyed_model \
                              --json-instances keyed_txt_input.json \
                              --version v2 \
                              --signature-name serving_default
    
    
==================    
    KEY      OUTPUT
id_1234  [0.49977055191993713, 0.5002294182777405]

--signature-name=SIGNATURE_NAME

The name of the signature defined in the SavedModel to use for this job. Defaults to DEFAULT_SERVING_SIGNATURE_DEF_KEY in

https://www.tensorflow.org/api_docs/python/tf/saved_model/signature_constants, which is "serving_default". Only applies to TensorFlow models.

 

 

참고자료: https://towardsdatascience.com/how-to-extend-a-keras-model-5effc083265c

How to extend a Keras Model

https://github.com/GoogleCloudPlatform/training-data-analyst/blob/master/blogs/batch_predictions/batch_predictions_keras.ipynb

GoogleCloudPlatform/training-data-analyst

https://dodonam.tistory.com/184

[tf 2.x] tf.keras 로 predict 결과 custom 하기 --> GCP ai-platform ( keyed model, serving_signature)