[App Server] Flask로 AI 백엔드 서버 구축해 보기

지난 포스팅(https://projectlog-eraser.tistory.com/m/57)에서 받은 피드백을 통해, 앱 서버의 구조를 변경해 개발을 진행하였다.

 

---

 

# 구조

 

 

 백엔드에서 따로 모델을 돌리는 프로세스를 생성하지 않고, socket 통신도 하지 않는다. 따라서 불필요하게 모델을 돌리기 위한 프로세스를 새로 생성하지 않아도 되고, Flask app 프로세스와 모델 프로세스 간에 통신으로 인해 (혹시라도 발생할 수 있는) 서버 부하를 줄일 수 있다.

 또한, 모델별로 설계를 진행했던 것과 달리, 새로운 설계에서는 컴포넌트별로 레이어를 나누었다. Controller, Service, Model이 그것이다. Controller는 클라이언트의 요청을 받아, Service로 전달한다. Service는 모델링에 필요한 데이터를 Model에 input으로 전달하고, Model에서 모델링 결과를 반환하도록 inference 메소드를 호출한다. 모델마다 필요한 데이터가 다르기 때문에, 데이터베이스와 연결해서 데이터를 가져와야 하는 경우 datastore를, 미리 저장된 캐시 데이터나 외부 API를 이용해야 하는 경우, 해당 메소드를 구현하도록 한다.

 이러한 설계를 통해 백엔드의 각 컴포넌트가 수행해야 하는 역할을 계층 별로 명확히 분리할 수 있다. AI 모델 서비스는 총 3가지로, 각각의 엔드포인트 별로 Controller → Service → Model ( → datastore / external API etc. ) → Service → Controller의 흐름을 거쳐 클라이언트의 요청을 처리할 수 있다.

 

---

# 구현

 

 구현에 사용한 (주요) 라이브러리는 다음과 같다.

• flask: Flask app은 Controller 역할 담당
• dependency-injector: 백엔드 주요 컴포넌트(Service, Model, Datastore 등) object를 생성하고 주입. 실행에 필요한 환경 설정 정보도 함께 로드
• pyodbc: Datastore의 Tibero DB 접근 담당
• pandas

 

---

 

## Controller

 

 Controller 역할을 하는 Flask app이다. 실제 서비스가 되고 있는 코드이므로, 전체 구조를 보여줄 수 있는 예시만 첨부한다. 예시의 API 엔드 포인트 URI는 관례에 어긋난 것으로, 예시로서만 남긴다.

 

from flask import Flask, request, jsonify
from model2 import APIError, NotExecutableError
from container import Container
import logging

# Flask 앱
app = Flask(__name__)

# service, datastore 등 object 주입
container = Container()
container.init_resources()

model1_service = container.model1_service()
model2_service = container.model2_service()
model3_service = container.model3_service()

model1_datastore = container.model1_datastore()
model1_model = container.model1_model()


# model1 엔드포인트
@app.route("/model1/fill", methods=['POST'])
def fill_model1_score():

    resp = {}
    
    # 클라이언트의 요청 query parameter 오지 않은 경우
    date = request.args.get('date')
    if not date:
        resp['status'] = 400
        resp['msg'] = '`date` field should be in request query'
        return jsonify(resp), 400
  	
    # model1 서비스 호출
    try:
        model1_service.fill_model1_data(date)
    except:
        # TODO: model1 에러 핸들링
        pass
    else:
        resp['status'] = 200
        resp['msg'] = f'model1 score fill task on date {date} OK'
    
    return jsonify(resp)

# model2 엔드포인트
@app.route("/model2", methods=['POST'])
def model2():

    resp = {}

    # 클라이언트의 요청 body 데이터가 json 형태가 아니거나 비어 있는 경우
    if not (request.is_json and request.json):
        logging.debug('request body data not json format or empty')
        resp['status'] = 400
        resp['msg'] = 'request body data should be json format'
        return jsonify(resp), 400

    # 클라이언트의 요청 body 데이터가 API 명세에 정의된 payload와 맞지 않는 경우
    try:
        input = request.json    
        model2_input_dao = model2DAO()
        model2_input_dao.depot = input['depot']
        model2_input_dao.start_time = input['startTime']
        model2_input_dao.maxtime = input['maxTime']
    except KeyError as ex:
        logging.debug(f'field {ex} missing in request body data')
        resp['status'] = 400
        resp['msg'] = f'request body data should have {ex} field' 
        return jsonify(resp), 400
    
    # model2 서비스 호출
    try:
        result = model2_service.get_model2_data(model2_input_dao)
    except APIError as ex:
        '''
            - 클라이언트의 요청에는 문제가 없으나,
            - 외부 서비스 API 호출 단계에서 APIError가 발생한 경우
        '''
        resp['status'] = 200
        resp['msg'] = str(ex)
    except NotExecutableError as ex:
        '''
            - 클라이언트의 요청에는 문제가 없으나,
            - 클라이언트의 요청 데이터로 모델에서 최적해를 찾아낼 수 없는 경우
        '''
        resp['status'] = 200
        resp['msg'] = str(ex)
    else:
        resp['status'] = 200
        resp['msg'] = 'OK'
        resp['data'] = result

    return jsonify(resp)

# reloader 옵션 해제
if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=4000, use_reloader=False)

 

 서비스 운영 중 로그를 확인하다 Flask reload와 관련된 문제를 알게 되었다. model1 서비스에서 똑같은 메소드가 2번씩 호출되고 있었던 것이다. Container를 통해 의존성을 주입했기 때문에 Model, Service 등 object가 2번 호출될 일이 없는데도 말이다. 서버에서 실행되고 있는 프로세스를 확인해 보니, app.py를 한 번 실행했음에도 해당 명령어로 실행한 프로세스가 2개임을 발견하게 되었다.

 Flask 공식 문서(https://flask.palletsprojects.com/en/1.1.x/api/?highlight=use_reloader) 및 Stackoverflow 글(https://stackoverflow.com/questions/25504149/why-does-running-the-flask-dev-server-run-itself-twice)을 통해 문제를 해결할 수 있었다. 내가 위에서 작성했던 것과 같이, 파이썬 스크립트에서 app.run()으로 Flask app을 실행할 시, flask 라이브러리에서 개발 서버를 띄우도록 설계되어 있는데, 이 경우 개발 편의 목적으로 코드 변경 감지를 위해 child process를 하나 더 띄운다고 한다. 이 때문에 flask run 커맨드를 통해 Flask app을 실행하거나, 위와 같이 코드 내에서 실행하고자 할 경우, use_reloader=False 옵션을 주어야 한다고 한다.

 

---

 

## Service

 

 Service 인터페이스는 다음과 같다. 각 모델 별로 inference 메소드를 호출할 메소드를 구현하도록 설계한다.

from abc import *


class Model2ServiceInterface(metaclass=ABCMeta):

    @abstractmethod
    def get_model2_data(self, input: object) -> object:
        raise NotImplementedError

 

 Model2에 해당하는 Service 구현체는 다음과 같다.

from abc import *
import logging

from common.Model2ServiceInterface import Model2ServiceInterface
from common.model import Model


class Model2Service(Model2ServiceInterface):
    def __init__(self, model: Model):
        self._model = model

    def get_model2_data(self, input):
        result = self._model.inference(input)
        return result

 

---

 

## Model, Datastore

 

 Model 인터페이스는 다음과 같다. 모든 Model 구현체는 이 인터페이스를 상속하는데, Service가 호출할 `inference` 메소드만 구현하면 된다. 구현체는 AI 모델링에 대한 부분으로, 서비스와 밀접한 연관이 있어 생략한다.

import abc


class Model(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def inference(self, input_data: object) -> object:
        raise NotImplementedError

 

 Datastore 인터페이스는 다음과 같다. Model1에서만 Datastore를 이용하는데, 다음의 인터페이스를 상속해 구현한다.

from abc import *


class Model1DatastoreInterface(metaclass=ABCMeta):

    def __init__(self):
        pass

    @abstractmethod
    def update_status(self: object, sensor_id: str) -> None:
        raise NotImplementedError

    @abstractmethod
    def findall_by_date(self: object, date: str) -> object:
        raise NotImplementedError
 
    @abstractmethod
    def store_data(self: object, data: object) -> None:
        raise NotImplementedError

 

---

## Container

 

 그 외에 의존성 주입을 위한 Container는 다음과 같이 구현한다.

 

import logging.config
import json
import os
import sys
from core.model1Service import model1Service
from core.model2Service import model2Service
from core.model3Service import model3Service
from core.model1Model import model1Model
from core.model2Model import model2Model
from core.model3Model import model3Model
from core.model1DataStore import model1DataStore


class Container(containers.DeclarativeContainer):

    # 모델링에 필요한 설정 정보
    config = providers.Configuration()
    config.from_yaml(f'{os.environ["APP_ROOT"]}/resource/config.yaml', required=True)

    # 로깅 설정 정보
    with open(os.path.join(sys.model2[0], "logging.json"), 'rt') as f:
        logging = providers.Resource(logging.config.dictConfig, json.load(f))

    # 백엔드 object 생성 및 주입
    model1_datastore = providers.Factory(model1DataStore)
    model1_model = providers.Factory(model1Model, config=config.model1())
    model2_model = providers.Factory(
        model2Model, url=config.model2.url(), key=config.model2.key())
    model3_model = providers.Factory(model3Model, config=config.model3())
    model1_service = providers.Factory(
        model1Service, datastore=model1_datastore, model=model1_model)
    model2_service = providers.Factory(model2Service, model=model2_model)
    model3_service = providers.Factory(model3Service, model=model3_model)

 

 dependency-injector 라이브러리는 특정 프레임워크에 종속되지 않고 사용할 수 있는 Python의 의존성 주입 라이브러리이다. 사용법은 공식 문서(https://python-dependency-injector.ets-labs.org)에서 확인할 수 있다.

Dependency Injector — Dependency injection framework for Python — Dependency Injector 4.38.0 documentation

 

 

---

 

# 배운 점

 

 지난 피드백 이후 여러 곳에서 배우며 처음으로 AI 서비스에 활용할 수 있는 백엔드를 구축해 보았다. 백엔드 구축에 있어 계층을 어떻게 분리하여 설계할 수 있는지, 의존성 관리, 인터페이스 작성이 왜 필요한지 배울 수 있었다. 클래스 및 인터페이스의 작성에 익숙하지 않았는데, AI 서비스 백엔드 설계에도 dao, controller, service 등과 같은 백엔드 컴포넌트 개념을 적용할 수 있음을 느낄 수 있었다.

 아직 구현되지 않아, 더 연구해 보고 싶은 부분은 Model에서의 에러 처리 부분이다. 위의 코드 중 #TODO로 남겨진 부분인데, 사실 해당 부분은 특별한 문제가 있지 않으면 에러가 나지 않는(나면 안 되는) 부분이기 때문에 에러 처리를 어떻게 해야 할지 모르겠어 연구가 필요할 듯하다.