[TensorFlow] Flask 와 Tensorflow Serving gRPC 통신

목적

• Flask 로 TensorFlow Serving gRPC Inference 서비스를 만들어서, TensorFlow Serving Server와 gRPC 통신을 통해 Inference 되는지 테스트 진행합니다.

TensorFlow Serving Server Docker 실행

• Flask로 gRPC 서비스 개발에 앞서, TensorFlow Serving Server 를 Docker를 이용하여 Container로 실행하였습니다.

docker run -t --rm --name=tf-serving \
-p 8500:8500 \
-p 8501:8501 \
-v "C:/ADC45_PythonAPI/adc45_edge/src/test_data/save_model/tensorflow/:/models/model" \
tensorflow/serving \
--model_name=mnist-model \
--model_config_file=/models/model/Config/models.config \
--model_config_file_poll_wait_seconds=60

• 위 명령어를 실행하게 되면 tf-serving 이름을 가진 TensorFlow Serving 컨테이너가 실행됩니다.

컨테이너 실행 결과

• docker ps -a 명령어를 통해 컨테이너 실행 유무 확인합니다.
• 확인 결과 tf-serving 이름을 가진 컨테이너가 현재 상태가 UP 인 것을 확인할 수 있습니다.

PS C:\Users\bh.cho> docker ps -a

CONTAINER ID   IMAGE                COMMAND                  CREATED          STATUS                    PORTS                               NAMES
046f3c33c49b   tensorflow/serving   "/usr/bin/tf_serving…"   32 seconds ago   Up 31 seconds             0.0.0.0:8500-8501->8500-8501/tcp    tf-serving

TensorFlow Serving Models.Config 파일

• TensorFlow Serving Server 컨테이너를 실행할 때, Models.Config 파일 옵션이 있었습니다.
• 해당 파일의 설정 내용은 아래와 같습니다.
• model.config 파일 안에는 총 5개의 Model 종류가 정의되어 있습니다.
  • inception_v1
  • inception_v3
  • xception
  • vgg_19
  • efficientnet_v2_m
• TensorFlow Serving Server 가 실행 되면, 총 5개의 모델을 주기적으로 Save_model 로 저장된 모델 파일을 재 로딩해주면서 Client 에서 특정 모델에 대한 Serving 요청을 기다리고 있게 됩니다.

model_config_list {
  config {
    name: 'inception_v1'
    base_path: '/models/model/inception_v1/'
    model_platform: 'tensorflow'
  }
  config {
    name: 'inception_v3'
    base_path: '/models/model/inception_v3/'
    model_platform: 'tensorflow'
  }
  config {
    name: 'xception'
    base_path: '/models/model/xception/'
    model_platform: 'tensorflow'
  }
  config {
    name: 'vgg_19'
    base_path: '/models/model/vgg_19/'
    model_platform: 'tensorflow'
  }
  config {
    name: 'efficientnet_v2_m'
    base_path: '/models/model/efficientnet_v2_m/'
    model_platform: 'tensorflow'
  }
}

Flask gRPC Inference 서비스 코드 작성

• 앞서, TensorFlow Serving Server 를 Docker Container 로 실행하였습니다.
• TensorFlow Serving Server 와 Flask Service 와 gRPC 통신을 통하여 Model Inference 되도록 예제 코드 작성 진행 하였습니다.

import grpc
import tensorflow as tf
import numpy as np
import pathlib

from flask import Flask
from flask import jsonify

from tensorflow_serving.apis import prediction_service_pb2_grpc
from tensorflow_serving.config import model_server_config_pb2
from tensorflow_serving.apis import model_service_pb2_grpc
from tensorflow_serving.apis import model_management_pb2
from tensorflow_serving.apis import get_model_metadata_pb2
from tensorflow_serving.apis import predict_pb2

app = Flask(__name__)

input_shape=[]

def get_dataset():
    dataset_url = "https://storage.googleapis.com/download.tensorflow.org/example_images/flower_photos.tgz"
    data_dir = tf.keras.utils.get_file(origin=dataset_url, 
                                        fname='flower_photos', 
                                        untar=True)
    data_dir = pathlib.Path(data_dir)

    val_ds = tf.keras.preprocessing.image_dataset_from_directory(
                                                                    data_dir,
                                                                    validation_split=0.2,
                                                                    subset="validation",
                                                                    seed=123,
                                                                    image_size=(299, 299),
                                                                    batch_size=1)

    x_test = np.concatenate([x for x, y in val_ds], axis=0)
    y_test = np.concatenate([y for x, y in val_ds], axis=0)

    x_test = x_test[:200]
    y_test = y_test[:200]

    return (x_test, y_test)

def get_input_output(datas):
    for key in datas.keys():
        key = key
    return key

def list_chunk(list, label_count):
    return [list[idx:idx+label_count] for idx in range(0, len(list), label_count)]

@app.route('/')
def run_service():
    localhost = 'localhost'
    port = 8500
    (x_test, y_test) = get_dataset()

    with grpc.insecure_channel(f'{localhost}:{port}') as channel:
        stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)

        get_model_metadata_request  = get_model_metadata_pb2.GetModelMetadataRequest()
        get_model_metadata_request.model_spec.name = 'xception'
        get_model_metadata_request.metadata_field.append('signature_def')
        resp = stub.GetModelMetadata(get_model_metadata_request, 5.0)  # 5 secs timeout

        signature_def = resp.metadata['signature_def']
        signature_map = get_model_metadata_pb2.SignatureDefMap()
        signature_map.ParseFromString(signature_def.value)

        serving_default = signature_map.ListFields()[0][1]['serving_default']
        serving_inputs = serving_default.inputs
        serving_outputs = serving_default.outputs

        input = get_input_output(datas=serving_inputs)
        output = get_input_output(datas=serving_outputs)

        request = predict_pb2.PredictRequest()
        request.model_spec.name = 'xception'
        request.model_spec.signature_name = 'serving_default'

        request.inputs[input].CopyFrom(
                tf.make_tensor_proto(x_test))
        result = stub.Predict(request, 120.0)  # 10 secs timeout

        response = np.array(result.outputs[output].float_val)

        response = list_chunk(response.tolist(), 5)

        predicts = []
        for idx, list in enumerate(response):
            prediction = np.argmax(list)
            if(y_test[idx] == prediction):
                predicts.append(f'Real Value : {y_test[idx]} / Pred Value : {prediction}')

        return jsonify({'predicts': predicts})


if __name__ == '__main__':
    app.run(debug=True)

실행 결과

• localhost:5000 또는 127.0.0.1:5000 으로 접속하여 결과 값을 확인합니다.
• 확인 결과, Model Predict 값이 웹 화면에 출력 되는 것을 확인할 수 있습니다.