Flask를 이용한 모델 배포 완벽 가이드

머신러닝 모델을 Flask 웹 서버로 배포하는 방법을 단계별로 알아봅니다. 모델 저장부터 API 엔드포인트 구축, 실제 서비스 배포까지 초급자도 쉽게 따라할 수 있도록 설명합니다.

1. Flask 기본 서버 구축

김개발 씨는 3개월 동안 열심히 머신러닝 모델을 학습시켰습니다. 정확도도 높고, 결과도 만족스럽습니다.

그런데 막상 이 모델을 다른 사람들이 사용할 수 있게 하려니 막막합니다. "주피터 노트북을 매번 실행해달라고 할 수는 없잖아..." 바로 이때 필요한 것이 Flask입니다.

Flask는 파이썬으로 웹 서버를 만들 수 있는 가장 가벼운 프레임워크입니다. 마치 작은 카페를 여는 것과 같습니다.

복잡한 인테리어 없이도 손님을 맞이하고 커피를 제공할 수 있는 최소한의 공간만 있으면 됩니다. Flask를 사용하면 머신러닝 모델을 외부에서 접근 가능한 API로 쉽게 변환할 수 있습니다.

다음 코드를 살펴봅시다.

from flask import Flask, jsonify

# Flask 애플리케이션 생성
app = Flask(__name__)

# 기본 라우트 설정
@app.route('/')
def home():
    return jsonify({'message': '모델 서버가 실행 중입니다'})

# 헬스 체크 엔드포인트
@app.route('/health')
def health_check():
    return jsonify({'status': 'healthy'})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

김개발 씨는 입사 6개월 차 데이터 사이언티스트입니다. 지난 몇 달간 고객 이탈 예측 모델을 열심히 개발했습니다.

모델의 성능도 좋고, 팀장님도 만족해하셨습니다. 그런데 문제가 생겼습니다.

"김개발 씨, 이 모델 마케팅팀에서 사용할 수 있게 해줄 수 있어요?" 팀장님의 요청에 김개발 씨는 당황했습니다. 마케팅팀 직원들에게 주피터 노트북 사용법을 알려줄 수도 없고, 매번 예측이 필요할 때마다 자신이 직접 실행해줄 수도 없는 노릇입니다.

바로 이런 상황에서 Flask가 등장합니다. Flask는 파이썬으로 웹 서버를 만들 수 있게 해주는 프레임워크입니다.

쉽게 비유하자면, Flask는 마치 작은 카페를 여는 것과 같습니다. 대형 프랜차이즈처럼 복잡한 시스템이 필요 없습니다.

커피 머신 하나, 작은 테이블 몇 개만 있으면 손님을 맞이할 수 있습니다. Django 같은 풀스택 프레임워크가 대형 레스토랑이라면, Flask는 아담한 푸드트럭입니다.

필요한 것만 갖추고, 빠르게 시작할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 Flask와 jsonify를 임포트합니다. jsonify는 파이썬 딕셔너리를 JSON 형식으로 변환해주는 함수입니다.

API 통신에서 JSON은 데이터를 주고받는 표준 형식이기 때문에 반드시 필요합니다. 다음으로 app = Flask(name) 이 한 줄이 Flask 애플리케이션을 생성합니다.

이것이 우리 서버의 본체입니다. @app.route('/') 부분을 데코레이터라고 부릅니다.

이 데코레이터는 "누군가 서버의 루트 주소로 접속하면, 아래 함수를 실행하라"는 의미입니다. 마치 카페 입구에 "여기로 오세요"라고 안내판을 붙여두는 것과 같습니다.

마지막의 app.run() 함수가 실제로 서버를 실행합니다. host를 '0.0.0.0'으로 설정하면 외부에서도 접속할 수 있고, port는 서버가 사용할 문 번호라고 생각하면 됩니다.

이 코드를 실행하면 터미널에 "Running on http://0.0.0.0:5000"이라는 메시지가 뜹니다. 브라우저에서 해당 주소로 접속하면 JSON 형식의 응답을 확인할 수 있습니다.

김개발 씨는 이 기본 구조를 익히고 나서야 비로소 모델 배포의 첫 발을 뗄 수 있었습니다.

실전 팁

💡 - debug=True는 개발 중에만 사용하고, 실제 배포 시에는 반드시 False로 설정하세요

포트 충돌이 발생하면 5000 대신 다른 포트 번호를 사용해보세요

2. 모델 저장과 로드

서버 기본 구조를 익힌 김개발 씨는 다음 단계로 넘어갔습니다. "이제 모델을 서버에 연결해야 하는데...

매번 모델을 학습시킬 수는 없잖아?" 맞습니다. 학습된 모델을 파일로 저장하고, 서버 시작 시 불러오는 방법이 필요합니다.

머신러닝 모델을 배포하려면 먼저 학습된 모델을 파일로 저장해야 합니다. 파이썬에서는 pickle이나 joblib을 사용하여 모델을 직렬화할 수 있습니다.

이렇게 저장된 모델은 서버가 시작될 때 한 번만 로드하면, 이후 요청이 들어올 때마다 빠르게 예측을 수행할 수 있습니다.

다음 코드를 살펴봅시다.

import joblib
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris

# 모델 학습 및 저장
iris = load_iris()
model = RandomForestClassifier(n_estimators=100)
model.fit(iris.data, iris.target)

# 모델 저장 - joblib 사용
joblib.dump(model, 'model.pkl')
print('모델이 저장되었습니다')

# 모델 로드
loaded_model = joblib.load('model.pkl')
print('모델이 로드되었습니다')

박시니어 씨가 김개발 씨에게 물었습니다. "모델 학습하는 데 얼마나 걸렸어요?" "거의 2시간이요." "그럼 사용자가 예측을 요청할 때마다 2시간씩 기다리라고 할 건가요?" 김개발 씨는 뒤통수를 맞은 기분이었습니다.

당연히 그럴 수 없습니다. 학습된 모델을 어딘가에 저장해두고, 필요할 때 꺼내 쓸 수 있어야 합니다.

이때 사용하는 것이 직렬화(Serialization) 입니다. 직렬화란 메모리에 있는 객체를 파일로 저장할 수 있는 형태로 변환하는 과정입니다.

마치 냉장고에 음식을 보관하는 것과 같습니다. 요리를 매번 처음부터 할 수는 없으니, 미리 만들어둔 반찬을 냉장고에 저장해두는 것이죠.

파이썬에서는 주로 pickle이나 joblib을 사용합니다. joblib은 특히 NumPy 배열이 많이 포함된 객체에 최적화되어 있어서, 머신러닝 모델 저장에 더 적합합니다.

위 코드에서 joblib.dump(model, 'model.pkl') 이 한 줄이 모델을 파일로 저장합니다. 'model.pkl'이라는 이름의 파일이 생성되고, 여기에 모델의 모든 정보가 담깁니다.

반대로 joblib.load('model.pkl') 을 호출하면 저장된 모델을 다시 메모리로 불러옵니다. 이 모델은 원래 학습시킨 모델과 완전히 동일하게 동작합니다.

실제 서버에서는 이 로드 과정을 서버 시작 시점에 한 번만 수행합니다. 그러면 이후 들어오는 모든 요청에 대해 이미 로드된 모델을 사용할 수 있어서 응답 속도가 빨라집니다.

주의할 점이 있습니다. pickle로 저장된 파일은 신뢰할 수 없는 출처에서 받으면 안 됩니다.

악성 코드가 포함될 수 있기 때문입니다. 직접 학습시킨 모델만 사용하거나, 신뢰할 수 있는 출처의 모델만 로드해야 합니다.

실전 팁

💡 - 모델 파일과 함께 학습에 사용한 스케일러나 인코더도 함께 저장하세요

모델 버전 관리를 위해 파일명에 날짜나 버전 번호를 포함시키는 것이 좋습니다

3. 예측 API 엔드포인트 만들기

모델을 저장하고 불러오는 방법을 익힌 김개발 씨에게 새로운 과제가 주어졌습니다. "이제 외부에서 데이터를 보내면 예측 결과를 돌려주는 API를 만들어야 해요." POST 요청을 받아서 예측 결과를 반환하는 엔드포인트를 구축할 차례입니다.

API 엔드포인트는 클라이언트가 서버에 요청을 보낼 수 있는 특정 URL 주소입니다. 예측 API에서는 클라이언트가 입력 데이터를 JSON 형식으로 보내면, 서버가 모델을 사용하여 예측을 수행하고 결과를 JSON으로 반환합니다.

Flask에서는 request 객체를 사용하여 들어온 데이터를 받아올 수 있습니다.

다음 코드를 살펴봅시다.

from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(__name__)
model = joblib.load('model.pkl')

@app.route('/predict', methods=['POST'])
def predict():
    # 요청에서 JSON 데이터 추출
    data = request.get_json()
    features = data['features']

    # NumPy 배열로 변환하여 예측
    input_data = np.array(features).reshape(1, -1)
    prediction = model.predict(input_data)

    # 결과 반환
    return jsonify({
        'prediction': int(prediction[0]),
        'status': 'success'
    })

API라는 단어가 어렵게 느껴질 수 있습니다. 쉽게 설명하면 API는 식당의 메뉴판과 같습니다.

손님이 메뉴판을 보고 "짜장면 하나요"라고 주문하면, 주방에서 짜장면을 만들어 갖다줍니다. 손님은 주방에서 어떻게 요리하는지 알 필요가 없습니다.

그저 메뉴판에 적힌 대로 주문하면 됩니다. API도 마찬가지입니다.

클라이언트는 정해진 형식으로 요청을 보내고, 서버는 정해진 형식으로 응답합니다. 내부에서 어떤 모델이 어떻게 동작하는지는 클라이언트가 알 필요 없습니다.

위 코드에서 @app.route('/predict', methods=['POST']) 데코레이터를 주목해주세요. 이전에 봤던 것과 다르게 methods=['POST'] 가 추가되었습니다.

HTTP 요청에는 여러 종류가 있습니다. GET은 데이터를 조회할 때, POST는 데이터를 보낼 때 사용합니다.

예측 API에서는 입력 데이터를 서버로 보내야 하므로 POST를 사용합니다. request.get_json() 함수는 클라이언트가 보낸 JSON 데이터를 파이썬 딕셔너리로 변환해줍니다.

마치 외국어로 쓰인 주문서를 한국어로 번역해주는 통역사와 같습니다. 데이터를 받은 후에는 NumPy 배열로 변환합니다.

reshape(1, -1) 부분은 1차원 배열을 2차원으로 바꿔주는 역할을 합니다. 대부분의 머신러닝 모델은 2차원 배열을 입력으로 기대하기 때문입니다.

마지막으로 jsonify 함수가 파이썬 딕셔너리를 JSON 문자열로 변환하여 반환합니다. 클라이언트는 이 JSON을 받아서 원하는 대로 활용할 수 있습니다.

이 API를 테스트하려면 curl 명령어나 Postman 같은 도구를 사용할 수 있습니다. 아직 익숙하지 않다면 걱정하지 마세요.

다음 카드에서 자세히 다루겠습니다.

실전 팁

💡 - 예측 결과와 함께 확률값도 반환하면 클라이언트에서 더 유용하게 활용할 수 있습니다

입력 데이터의 형식이 올바른지 검증하는 로직을 추가하세요

4. 에러 처리와 입력 검증

김개발 씨가 만든 API가 드디어 작동하기 시작했습니다. 그런데 테스트 중에 이상한 데이터를 보내니 서버가 에러를 뿜으며 멈춰버렸습니다.

"사용자가 잘못된 데이터를 보내면 어떻게 하지?" 실제 서비스에서는 다양한 예외 상황에 대비해야 합니다.

프로덕션 환경에서는 잘못된 입력이 들어올 수 있습니다. 빈 데이터, 잘못된 형식, 누락된 필드 등 다양한 상황에 대비해야 합니다.

try-except 구문과 입력 검증 로직을 추가하면 서버가 예상치 못한 입력에도 안정적으로 동작합니다. 적절한 에러 메시지와 HTTP 상태 코드를 반환하는 것도 중요합니다.

다음 코드를 살펴봅시다.

from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(__name__)
model = joblib.load('model.pkl')

@app.route('/predict', methods=['POST'])
def predict():
    try:
        data = request.get_json()

        # 입력 검증
        if not data or 'features' not in data:
            return jsonify({'error': 'features 필드가 필요합니다'}), 400

        features = data['features']
        if not isinstance(features, list) or len(features) != 4:
            return jsonify({'error': '4개의 특성값이 필요합니다'}), 400

        input_data = np.array(features).reshape(1, -1)
        prediction = model.predict(input_data)

        return jsonify({'prediction': int(prediction[0])})

    except Exception as e:
        return jsonify({'error': str(e)}), 500

박시니어 씨가 김개발 씨의 API를 테스트해보다가 일부러 빈 데이터를 보내봤습니다. 순간 서버가 에러를 내뿜으며 멈춰버렸습니다.

"이렇게 되면 실제 서비스에서는 큰일 나겠네요." 김개발 씨는 고개를 떨궜습니다. 정상적인 입력만 올 거라고 가정한 것이 실수였습니다.

실제 세상에서는 온갖 종류의 예상치 못한 입력이 들어옵니다. 빈 문자열, null 값, 숫자 대신 문자열, 필드 누락 등 상상할 수 있는 모든 경우가 발생합니다.

이런 상황에서도 서버가 죽지 않고 적절한 에러 메시지를 반환해야 합니다. 먼저 try-except 구문으로 전체 로직을 감쌉니다.

이것은 마치 안전망과 같습니다. 어떤 예외가 발생하더라도 서버가 완전히 멈추는 것을 방지합니다.

그 다음으로 입력 데이터를 검증합니다. if not data는 데이터가 아예 없는 경우를 체크합니다.

'features' not in data는 필수 필드가 누락된 경우를 확인합니다. 여기서 중요한 것이 HTTP 상태 코드입니다.

return 문 끝에 있는 숫자가 바로 상태 코드입니다. 400은 "클라이언트의 요청이 잘못되었다"는 의미입니다.

입력 형식이 틀렸거나 필수 데이터가 없을 때 사용합니다. 500은 "서버 내부 에러"를 의미합니다.

예상치 못한 에러가 발생했을 때 사용합니다. 상태 코드를 올바르게 사용하면 클라이언트가 어떤 문제가 발생했는지 쉽게 파악할 수 있습니다.

마치 병원에서 진단서를 발급받는 것처럼, 어디가 문제인지 명확하게 알려주는 것입니다. isinstance 함수는 데이터 타입을 확인합니다.

features가 리스트인지, 길이가 올바른지 검사합니다. 이런 꼼꼼한 검증이 나중에 디버깅 시간을 크게 줄여줍니다.

실전 팁

💡 - 민감한 에러 정보는 로그에만 남기고, 클라이언트에는 일반적인 메시지만 반환하세요

자주 발생하는 에러는 별도의 커스텀 예외 클래스로 관리하면 좋습니다

5. 여러 모델 관리하기

시간이 지나면서 김개발 씨가 관리하는 모델이 점점 늘어났습니다. 고객 이탈 예측 모델, 상품 추천 모델, 가격 예측 모델까지.

"모델마다 서버를 따로 띄워야 하나?" 다행히 하나의 Flask 서버에서 여러 모델을 효율적으로 관리하는 방법이 있습니다.

실무에서는 하나의 서버가 여러 모델을 서빙해야 하는 경우가 많습니다. 모델들을 딕셔너리로 관리하고, 요청 시 모델 이름을 지정받아 해당 모델로 예측을 수행하면 됩니다.

이렇게 하면 서버 하나로 다양한 예측 서비스를 제공할 수 있습니다.

다음 코드를 살펴봅시다.

from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(__name__)

# 여러 모델을 딕셔너리로 관리
models = {
    'churn': joblib.load('churn_model.pkl'),
    'recommend': joblib.load('recommend_model.pkl'),
    'price': joblib.load('price_model.pkl')
}

@app.route('/predict/<model_name>', methods=['POST'])
def predict(model_name):
    # 모델 존재 여부 확인
    if model_name not in models:
        return jsonify({'error': f'{model_name} 모델을 찾을 수 없습니다'}), 404

    data = request.get_json()
    features = np.array(data['features']).reshape(1, -1)

    prediction = models[model_name].predict(features)
    return jsonify({'model': model_name, 'prediction': prediction.tolist()})

프로젝트가 성공하면서 김개발 씨에게 새로운 모델 개발 요청이 계속 들어왔습니다. 처음에는 모델마다 서버를 따로 띄웠습니다.

그런데 서버가 5개, 10개로 늘어나자 관리가 힘들어졌습니다. "이거 좀 비효율적인 것 같은데..." 박시니어 씨가 해결책을 제시했습니다.

"여러 모델을 하나의 서버에서 관리하면 돼요." 핵심은 딕셔너리입니다. 파이썬 딕셔너리에 모델 이름을 키로, 모델 객체를 값으로 저장합니다.

마치 사물함에 이름표를 붙여 물건을 보관하는 것과 같습니다. 코드에서 주목할 부분은 @app.route('/predict/<model_name>') 입니다.

꺾쇠괄호 안의 model_name은 URL 경로 변수라고 부릅니다. 사용자가 /predict/churn으로 요청하면 model_name에 'churn'이 들어갑니다.

이 방식의 장점은 새 모델을 추가할 때 코드 수정이 최소화된다는 것입니다. models 딕셔너리에 새 모델만 추가하면 바로 사용할 수 있습니다.

물론 모델이 많아지면 메모리 사용량도 늘어납니다. 이럴 때는 자주 사용하지 않는 모델은 동적으로 로드하거나, LRU 캐시를 사용하여 메모리를 효율적으로 관리할 수 있습니다.

하지만 모델이 몇 개 안 된다면 시작 시점에 모두 로드해두는 것이 응답 속도 면에서 유리합니다. return 문에서 prediction.tolist() 를 사용한 이유도 있습니다.

NumPy 배열은 JSON으로 직접 변환할 수 없기 때문에, 파이썬 리스트로 먼저 변환해야 합니다.

실전 팁

💡 - 모델이 많아지면 설정 파일에서 모델 경로를 관리하는 것이 좋습니다

사용 빈도가 낮은 모델은 lazy loading 방식으로 필요할 때만 로드하세요

6. 배치 예측 처리하기

마케팅팀에서 새로운 요청이 들어왔습니다. "고객 1만 명의 이탈 확률을 한 번에 예측해주세요." 한 명씩 1만 번 API를 호출하는 건 너무 비효율적입니다.

김개발 씨는 여러 건의 데이터를 한 번에 처리하는 배치 예측 기능이 필요하다는 것을 깨달았습니다.

배치 예측은 여러 개의 입력 데이터를 한 번의 요청으로 처리하는 것입니다. 네트워크 오버헤드를 줄이고, 머신러닝 모델의 벡터 연산 특성을 활용하여 처리 속도를 크게 높일 수 있습니다.

대량의 데이터를 처리해야 하는 실무 환경에서 필수적인 기능입니다.

다음 코드를 살펴봅시다.

from flask import Flask, request, jsonify
import joblib
import numpy as np

app = Flask(__name__)
model = joblib.load('model.pkl')

@app.route('/predict/batch', methods=['POST'])
def batch_predict():
    data = request.get_json()

    # 배치 크기 제한
    if len(data['samples']) > 1000:
        return jsonify({'error': '최대 1000개까지 처리 가능합니다'}), 400

    # 2차원 배열로 변환
    input_data = np.array(data['samples'])

    # 배치 예측 수행
    predictions = model.predict(input_data)
    probabilities = model.predict_proba(input_data)

    results = [
        {'prediction': int(pred), 'probability': prob.max()}
        for pred, prob in zip(predictions, probabilities)
    ]

    return jsonify({'count': len(results), 'results': results})

한 명의 고객을 예측하는 데 API 호출이 한 번 필요하다면, 1만 명을 예측하려면 1만 번의 호출이 필요합니다. 매번 네트워크 연결을 맺고 끊는 오버헤드가 발생하고, 전체 처리 시간이 크게 늘어납니다.

배치 예측은 이 문제를 해결합니다. 여러 개의 입력을 한 번에 보내고, 한 번에 결과를 받습니다.

마치 택배를 하나씩 보내는 것보다 박스에 모아서 한 번에 보내는 것이 효율적인 것과 같습니다. 더 좋은 점은 머신러닝 모델의 특성에 있습니다.

대부분의 모델은 내부적으로 행렬 연산을 사용합니다. NumPy와 같은 라이브러리는 행렬 연산에 최적화되어 있어서, 1000개의 데이터를 한 번에 처리하는 것이 하나씩 1000번 처리하는 것보다 훨씬 빠릅니다.

코드에서 배치 크기 제한을 주목하세요. 제한 없이 받으면 누군가 100만 건의 데이터를 한 번에 보내서 서버를 다운시킬 수 있습니다.

적절한 제한을 두는 것이 서버 안정성을 위해 중요합니다. predict_proba 함수는 예측 확률을 반환합니다.

단순히 0 또는 1이 아니라, 얼마나 확신하는지도 알려주는 것이죠. 마케팅팀에서는 이 확률을 보고 이탈 가능성이 높은 고객 순으로 정렬하여 집중 관리할 수 있습니다.

결과를 리스트 컴프리헨션으로 가공하는 부분도 눈여겨보세요. 각 예측 결과와 확률을 딕셔너리로 묶어서 반환합니다.

클라이언트에서 사용하기 편한 형태로 만들어주는 것이 좋은 API 설계입니다.

실전 팁

💡 - 배치 크기가 매우 크다면 비동기 처리나 백그라운드 작업 큐를 고려하세요

진행 상황을 알려주는 progress 엔드포인트를 추가하면 사용자 경험이 좋아집니다

7. Gunicorn으로 프로덕션 배포

드디어 테스트가 모두 끝났습니다. 이제 실제 서비스에 배포할 차례입니다.

그런데 박시니어 씨가 경고했습니다. "Flask 내장 서버는 개발용이야.

실제 서비스에서는 Gunicorn을 써야 해." 프로덕션 환경에서의 안정적인 배포 방법을 알아볼 시간입니다.

Flask의 내장 개발 서버는 동시 요청 처리에 한계가 있고, 보안 기능도 부족합니다. Gunicorn은 파이썬 WSGI HTTP 서버로, 여러 워커 프로세스를 통해 동시 요청을 안정적으로 처리합니다.

실제 프로덕션 환경에서는 반드시 Gunicorn이나 유사한 프로덕션 서버를 사용해야 합니다.

다음 코드를 살펴봅시다.

# requirements.txt
flask==3.0.0
gunicorn==21.2.0
joblib==1.3.2
numpy==1.26.0
scikit-learn==1.3.2

# 서버 실행 명령어
# gunicorn --workers 4 --bind 0.0.0.0:5000 app:app

# gunicorn.conf.py (설정 파일)
bind = '0.0.0.0:5000'
workers = 4
timeout = 120
accesslog = 'access.log'
errorlog = 'error.log'
loglevel = 'info'

Flask를 실행할 때 터미널에 이런 경고 메시지를 본 적 있을 것입니다. "WARNING: This is a development server.

Do not use it in a production deployment." 개발 서버는 말 그대로 개발용입니다. 한 번에 하나의 요청만 처리할 수 있고, 에러가 발생하면 전체 서버가 멈출 수 있습니다.

마치 혼자서 모든 손님을 상대하는 1인 카페와 같습니다. Gunicorn(Green Unicorn)은 이 문제를 해결합니다.

여러 명의 직원을 고용하는 것과 같습니다. 워커(worker) 라고 부르는 여러 프로세스가 동시에 요청을 처리합니다.

--workers 4 옵션은 4개의 워커 프로세스를 생성합니다. 일반적으로 CPU 코어 수의 2배 + 1로 설정하는 것이 좋습니다.

4코어 서버라면 9개의 워커가 적당합니다. --bind 0.0.0.0:5000 은 모든 네트워크 인터페이스의 5000번 포트에서 요청을 받겠다는 의미입니다.

외부에서 접속할 수 있게 하려면 이렇게 설정해야 합니다. 설정 파일을 별도로 만들면 명령어가 깔끔해집니다.

timeout 설정은 중요합니다. 머신러닝 모델 예측에 시간이 오래 걸리는 경우, 기본 타임아웃(30초)이 너무 짧을 수 있습니다.

로그 설정도 잊지 마세요. accesslog에는 모든 요청이 기록되고, errorlog에는 에러가 기록됩니다.

문제가 발생했을 때 이 로그가 디버깅의 열쇠가 됩니다. 실제로 서버를 실행할 때는 gunicorn -c gunicorn.conf.py app:app 명령어를 사용합니다.

여기서 app:app은 app.py 파일의 app 변수를 의미합니다.

실전 팁

💡 - Nginx를 앞에 두고 Gunicorn을 뒤에 배치하는 구조가 일반적입니다

systemd나 supervisor를 사용하여 서버가 자동으로 재시작되도록 설정하세요

8. 모델 버전 관리와 A/B 테스트

새로운 모델을 학습시켰는데, 기존 모델보다 정말 좋은지 어떻게 확인할까요? 무작정 교체했다가 성능이 떨어지면 큰일입니다.

김개발 씨는 안전하게 새 모델을 테스트하고 배포하는 방법이 필요했습니다.

모델 버전 관리는 여러 버전의 모델을 동시에 운영하며 성능을 비교하는 방법입니다. A/B 테스트를 통해 일부 트래픽만 새 모델로 보내서 실제 환경에서 성능을 검증할 수 있습니다.

문제가 발생하면 즉시 이전 버전으로 롤백할 수 있어 안전합니다.

다음 코드를 살펴봅시다.

from flask import Flask, request, jsonify
import joblib
import numpy as np
import random

app = Flask(__name__)

# 버전별 모델 로드
models = {
    'v1': joblib.load('model_v1.pkl'),
    'v2': joblib.load('model_v2.pkl')
}
current_version = 'v1'
ab_test_ratio = 0.1  # 10% 트래픽을 v2로

@app.route('/predict', methods=['POST'])
def predict():
    # A/B 테스트: 확률적으로 모델 선택
    if random.random() < ab_test_ratio:
        version = 'v2'
    else:
        version = current_version

    data = request.get_json()
    input_data = np.array(data['features']).reshape(1, -1)
    prediction = models[version].predict(input_data)

    # 버전 정보도 함께 반환
    return jsonify({
        'prediction': int(prediction[0]),
        'model_version': version
    })

새 모델을 배포하는 것은 언제나 긴장되는 일입니다. 학습 데이터에서는 좋은 성능을 보였지만, 실제 환경에서도 그럴까요?

예상치 못한 에지 케이스가 있을 수도 있습니다. 한 번에 전체 트래픽을 새 모델로 전환하는 것은 위험합니다.

마치 새로 개발한 약을 테스트 없이 모든 환자에게 투여하는 것과 같습니다. A/B 테스트는 이 문제를 해결합니다.

전체 트래픽 중 일부만 새 모델로 보내서 실제 성능을 측정합니다. 위 코드에서는 10%의 요청만 v2 모델로 처리합니다.

random.random() 함수는 0과 1 사이의 난수를 반환합니다. 이 값이 0.1보다 작으면 새 모델을, 그렇지 않으면 기존 모델을 사용합니다.

요청이 충분히 많으면 대략 10%가 새 모델로 처리됩니다. 응답에 model_version을 포함시키는 것이 중요합니다.

나중에 로그를 분석할 때, 어떤 버전의 모델이 어떤 예측을 했는지 추적할 수 있습니다. 이 데이터를 바탕으로 두 버전의 정확도, 응답 시간 등을 비교합니다.

새 모델이 충분히 검증되면 ab_test_ratio를 점진적으로 높입니다. 10%, 30%, 50%, 100% 순으로 천천히 전환합니다.

이것을 점진적 롤아웃(gradual rollout) 이라고 부릅니다. 문제가 발생하면 즉시 ratio를 0으로 설정하여 롤백할 수 있습니다.

코드 변경 없이 설정만 바꾸면 되니까 빠르게 대응할 수 있습니다. 실무에서는 이런 설정값을 환경 변수나 설정 서버에서 관리합니다.

서버를 재시작하지 않고도 동적으로 변경할 수 있게요.

실전 팁

💡 - A/B 테스트 결과 분석을 위해 별도의 로깅 시스템을 구축하세요

통계적으로 유의미한 결과를 얻으려면 충분한 샘플 수가 필요합니다

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#Flask#ModelDeployment#MachineLearning#API#Data Science

Flask를 이용한 모델 배포 완벽 가이드

# Flask를 이용한 모델 배포 완벽 가이드 머신러닝 모델을 Flask 웹 서버로 배포하는 방법을 단계별로 알아봅니다. 모델 저장부터 API 엔드포인트 구축, 실제 서비스 배포까지 초급자도 쉽게 따라할 수 있도록 설명합니다. --- ## 목차 1. [Flask_기본_서버_구축](#flask-기본-서버-구축) 2. [모델_저장과_로드](#모델-저장과-로드) 3. [예측_API_엔드포인트_만들기](#예측-api-엔드포인트-만들기) 4. [에러_처리와_입력_검증](#에러-처리와-입력-검증) 5. [여러_모델_관리하기](#여러-모델-관리하기) 6. [배치_예측_처리하기](#배치-예측-처리하기) 7. [Gunicorn으로_프로덕션_배포](#gunicorn으로-프로덕션-배포) 8. [모델_버전_관리와_A/B_테스트](#모델-버전-관리와-a/b-테스트) --- ## 1. Flask_기본_서버_구축 김개발 씨는 3개월 동안 열심히 머신러닝 모델을 학습시켰습니다. 정확도도 높고, 결과도 만족스럽습니다. 그런데 막상 이 모델을 다른 사람들이 사용할 수 있게 하려니 막막합니다. "주피터 노트북을 매번 실행해달라고 할 수는 없잖아..." 바로 이때 필요한 것이 **Flask**입니다. **Flask**는 파이썬으로 웹 서버를 만들 수 있는 가장 가벼운 프레임워크입니다. 마치 작은 카페를 여는 것과 같습니다. 복잡한 인테리어 없이도 손님을 맞이하고 커피를 제공할 수 있는 최소한의 공간만 있으면 됩니다. Flask를 사용하면 머신러닝 모델을 외부에서 접근 가능한 API로 쉽게 변환할 수 있습니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, jsonify # Flask 애플리케이션 생성 app = Flask(__name__) # 기본 라우트 설정 @app.route('/') def home(): return jsonify({'message': '모델 서버가 실행 중입니다'}) # 헬스 체크 엔드포인트 @app.route('/health') def health_check(): return jsonify({'status': 'healthy'}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True) ``` 김개발 씨는 입사 6개월 차 데이터 사이언티스트입니다. 지난 몇 달간 고객 이탈 예측 모델을 열심히 개발했습니다. 모델의 성능도 좋고, 팀장님도 만족해하셨습니다. 그런데 문제가 생겼습니다. "김개발 씨, 이 모델 마케팅팀에서 사용할 수 있게 해줄 수 있어요?" 팀장님의 요청에 김개발 씨는 당황했습니다. 마케팅팀 직원들에게 주피터 노트북 사용법을 알려줄 수도 없고, 매번 예측이 필요할 때마다 자신이 직접 실행해줄 수도 없는 노릇입니다. 바로 이런 상황에서 **Flask**가 등장합니다. Flask는 파이썬으로 웹 서버를 만들 수 있게 해주는 프레임워크입니다. 쉽게 비유하자면, Flask는 마치 작은 카페를 여는 것과 같습니다. 대형 프랜차이즈처럼 복잡한 시스템이 필요 없습니다. 커피 머신 하나, 작은 테이블 몇 개만 있으면 손님을 맞이할 수 있습니다. Django 같은 풀스택 프레임워크가 대형 레스토랑이라면, Flask는 아담한 푸드트럭입니다. 필요한 것만 갖추고, 빠르게 시작할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다. 먼저 **Flask**와 **jsonify**를 임포트합니다. jsonify는 파이썬 딕셔너리를 JSON 형식으로 변환해주는 함수입니다. API 통신에서 JSON은 데이터를 주고받는 표준 형식이기 때문에 반드시 필요합니다. 다음으로 **app = Flask(__name__)** 이 한 줄이 Flask 애플리케이션을 생성합니다. 이것이 우리 서버의 본체입니다. **@app.route('/')** 부분을 데코레이터라고 부릅니다. 이 데코레이터는 "누군가 서버의 루트 주소로 접속하면, 아래 함수를 실행하라"는 의미입니다. 마치 카페 입구에 "여기로 오세요"라고 안내판을 붙여두는 것과 같습니다. 마지막의 **app.run()** 함수가 실제로 서버를 실행합니다. host를 '0.0.0.0'으로 설정하면 외부에서도 접속할 수 있고, port는 서버가 사용할 문 번호라고 생각하면 됩니다. 이 코드를 실행하면 터미널에 "Running on http://0.0.0.0:5000"이라는 메시지가 뜹니다. 브라우저에서 해당 주소로 접속하면 JSON 형식의 응답을 확인할 수 있습니다. 김개발 씨는 이 기본 구조를 익히고 나서야 비로소 모델 배포의 첫 발을 뗄 수 있었습니다. **실전 팁** 💡 - debug=True는 개발 중에만 사용하고, 실제 배포 시에는 반드시 False로 설정하세요 - 포트 충돌이 발생하면 5000 대신 다른 포트 번호를 사용해보세요 --- ## 2. 모델_저장과_로드 서버 기본 구조를 익힌 김개발 씨는 다음 단계로 넘어갔습니다. "이제 모델을 서버에 연결해야 하는데... 매번 모델을 학습시킬 수는 없잖아?" 맞습니다. 학습된 모델을 파일로 저장하고, 서버 시작 시 불러오는 방법이 필요합니다. 머신러닝 모델을 배포하려면 먼저 학습된 모델을 파일로 저장해야 합니다. 파이썬에서는 **pickle**이나 **joblib**을 사용하여 모델을 직렬화할 수 있습니다. 이렇게 저장된 모델은 서버가 시작될 때 한 번만 로드하면, 이후 요청이 들어올 때마다 빠르게 예측을 수행할 수 있습니다. 다음 코드를 살펴봅시다. ```python import joblib from sklearn.ensemble import RandomForestClassifier from sklearn.datasets import load_iris # 모델 학습 및 저장 iris = load_iris() model = RandomForestClassifier(n_estimators=100) model.fit(iris.data, iris.target) # 모델 저장 - joblib 사용 joblib.dump(model, 'model.pkl') print('모델이 저장되었습니다') # 모델 로드 loaded_model = joblib.load('model.pkl') print('모델이 로드되었습니다') ``` 박시니어 씨가 김개발 씨에게 물었습니다. "모델 학습하는 데 얼마나 걸렸어요?" "거의 2시간이요." "그럼 사용자가 예측을 요청할 때마다 2시간씩 기다리라고 할 건가요?" 김개발 씨는 뒤통수를 맞은 기분이었습니다. 당연히 그럴 수 없습니다. 학습된 모델을 어딘가에 저장해두고, 필요할 때 꺼내 쓸 수 있어야 합니다. 이때 사용하는 것이 **직렬화(Serialization)** 입니다. 직렬화란 메모리에 있는 객체를 파일로 저장할 수 있는 형태로 변환하는 과정입니다. 마치 냉장고에 음식을 보관하는 것과 같습니다. 요리를 매번 처음부터 할 수는 없으니, 미리 만들어둔 반찬을 냉장고에 저장해두는 것이죠. 파이썬에서는 주로 **pickle**이나 **joblib**을 사용합니다. joblib은 특히 NumPy 배열이 많이 포함된 객체에 최적화되어 있어서, 머신러닝 모델 저장에 더 적합합니다. 위 코드에서 **joblib.dump(model, 'model.pkl')** 이 한 줄이 모델을 파일로 저장합니다. 'model.pkl'이라는 이름의 파일이 생성되고, 여기에 모델의 모든 정보가 담깁니다. 반대로 **joblib.load('model.pkl')** 을 호출하면 저장된 모델을 다시 메모리로 불러옵니다. 이 모델은 원래 학습시킨 모델과 완전히 동일하게 동작합니다. 실제 서버에서는 이 로드 과정을 서버 시작 시점에 한 번만 수행합니다. 그러면 이후 들어오는 모든 요청에 대해 이미 로드된 모델을 사용할 수 있어서 응답 속도가 빨라집니다. 주의할 점이 있습니다. pickle로 저장된 파일은 신뢰할 수 없는 출처에서 받으면 안 됩니다. 악성 코드가 포함될 수 있기 때문입니다. 직접 학습시킨 모델만 사용하거나, 신뢰할 수 있는 출처의 모델만 로드해야 합니다. **실전 팁** 💡 - 모델 파일과 함께 학습에 사용한 스케일러나 인코더도 함께 저장하세요 - 모델 버전 관리를 위해 파일명에 날짜나 버전 번호를 포함시키는 것이 좋습니다 --- ## 3. 예측_API_엔드포인트_만들기 모델을 저장하고 불러오는 방법을 익힌 김개발 씨에게 새로운 과제가 주어졌습니다. "이제 외부에서 데이터를 보내면 예측 결과를 돌려주는 API를 만들어야 해요." POST 요청을 받아서 예측 결과를 반환하는 엔드포인트를 구축할 차례입니다. **API 엔드포인트**는 클라이언트가 서버에 요청을 보낼 수 있는 특정 URL 주소입니다. 예측 API에서는 클라이언트가 입력 데이터를 JSON 형식으로 보내면, 서버가 모델을 사용하여 예측을 수행하고 결과를 JSON으로 반환합니다. Flask에서는 request 객체를 사용하여 들어온 데이터를 받아올 수 있습니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, request, jsonify import joblib import numpy as np app = Flask(__name__) model = joblib.load('model.pkl') @app.route('/predict', methods=['POST']) def predict(): # 요청에서 JSON 데이터 추출 data = request.get_json() features = data['features'] # NumPy 배열로 변환하여 예측 input_data = np.array(features).reshape(1, -1) prediction = model.predict(input_data) # 결과 반환 return jsonify({ 'prediction': int(prediction[0]), 'status': 'success' }) ``` API라는 단어가 어렵게 느껴질 수 있습니다. 쉽게 설명하면 API는 식당의 메뉴판과 같습니다. 손님이 메뉴판을 보고 "짜장면 하나요"라고 주문하면, 주방에서 짜장면을 만들어 갖다줍니다. 손님은 주방에서 어떻게 요리하는지 알 필요가 없습니다. 그저 메뉴판에 적힌 대로 주문하면 됩니다. API도 마찬가지입니다. 클라이언트는 정해진 형식으로 요청을 보내고, 서버는 정해진 형식으로 응답합니다. 내부에서 어떤 모델이 어떻게 동작하는지는 클라이언트가 알 필요 없습니다. 위 코드에서 **@app.route('/predict', methods=['POST'])** 데코레이터를 주목해주세요. 이전에 봤던 것과 다르게 **methods=['POST']** 가 추가되었습니다. HTTP 요청에는 여러 종류가 있습니다. GET은 데이터를 조회할 때, POST는 데이터를 보낼 때 사용합니다. 예측 API에서는 입력 데이터를 서버로 보내야 하므로 POST를 사용합니다. **request.get_json()** 함수는 클라이언트가 보낸 JSON 데이터를 파이썬 딕셔너리로 변환해줍니다. 마치 외국어로 쓰인 주문서를 한국어로 번역해주는 통역사와 같습니다. 데이터를 받은 후에는 NumPy 배열로 변환합니다. **reshape(1, -1)** 부분은 1차원 배열을 2차원으로 바꿔주는 역할을 합니다. 대부분의 머신러닝 모델은 2차원 배열을 입력으로 기대하기 때문입니다. 마지막으로 **jsonify** 함수가 파이썬 딕셔너리를 JSON 문자열로 변환하여 반환합니다. 클라이언트는 이 JSON을 받아서 원하는 대로 활용할 수 있습니다. 이 API를 테스트하려면 curl 명령어나 Postman 같은 도구를 사용할 수 있습니다. 아직 익숙하지 않다면 걱정하지 마세요. 다음 카드에서 자세히 다루겠습니다. **실전 팁** 💡 - 예측 결과와 함께 확률값도 반환하면 클라이언트에서 더 유용하게 활용할 수 있습니다 - 입력 데이터의 형식이 올바른지 검증하는 로직을 추가하세요 --- ## 4. 에러_처리와_입력_검증 김개발 씨가 만든 API가 드디어 작동하기 시작했습니다. 그런데 테스트 중에 이상한 데이터를 보내니 서버가 에러를 뿜으며 멈춰버렸습니다. "사용자가 잘못된 데이터를 보내면 어떻게 하지?" 실제 서비스에서는 다양한 예외 상황에 대비해야 합니다. 프로덕션 환경에서는 잘못된 입력이 들어올 수 있습니다. 빈 데이터, 잘못된 형식, 누락된 필드 등 다양한 상황에 대비해야 합니다. **try-except** 구문과 입력 검증 로직을 추가하면 서버가 예상치 못한 입력에도 안정적으로 동작합니다. 적절한 에러 메시지와 HTTP 상태 코드를 반환하는 것도 중요합니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, request, jsonify import joblib import numpy as np app = Flask(__name__) model = joblib.load('model.pkl') @app.route('/predict', methods=['POST']) def predict(): try: data = request.get_json() # 입력 검증 if not data or 'features' not in data: return jsonify({'error': 'features 필드가 필요합니다'}), 400 features = data['features'] if not isinstance(features, list) or len(features) != 4: return jsonify({'error': '4개의 특성값이 필요합니다'}), 400 input_data = np.array(features).reshape(1, -1) prediction = model.predict(input_data) return jsonify({'prediction': int(prediction[0])}) except Exception as e: return jsonify({'error': str(e)}), 500 ``` 박시니어 씨가 김개발 씨의 API를 테스트해보다가 일부러 빈 데이터를 보내봤습니다. 순간 서버가 에러를 내뿜으며 멈춰버렸습니다. "이렇게 되면 실제 서비스에서는 큰일 나겠네요." 김개발 씨는 고개를 떨궜습니다. 정상적인 입력만 올 거라고 가정한 것이 실수였습니다. 실제 세상에서는 온갖 종류의 예상치 못한 입력이 들어옵니다. 빈 문자열, null 값, 숫자 대신 문자열, 필드 누락 등 상상할 수 있는 모든 경우가 발생합니다. 이런 상황에서도 서버가 죽지 않고 적절한 에러 메시지를 반환해야 합니다. 먼저 **try-except** 구문으로 전체 로직을 감쌉니다. 이것은 마치 안전망과 같습니다. 어떤 예외가 발생하더라도 서버가 완전히 멈추는 것을 방지합니다. 그 다음으로 입력 데이터를 검증합니다. **if not data**는 데이터가 아예 없는 경우를 체크합니다. **'features' not in data**는 필수 필드가 누락된 경우를 확인합니다. 여기서 중요한 것이 **HTTP 상태 코드**입니다. return 문 끝에 있는 숫자가 바로 상태 코드입니다. **400**은 "클라이언트의 요청이 잘못되었다"는 의미입니다. 입력 형식이 틀렸거나 필수 데이터가 없을 때 사용합니다. **500**은 "서버 내부 에러"를 의미합니다. 예상치 못한 에러가 발생했을 때 사용합니다. 상태 코드를 올바르게 사용하면 클라이언트가 어떤 문제가 발생했는지 쉽게 파악할 수 있습니다. 마치 병원에서 진단서를 발급받는 것처럼, 어디가 문제인지 명확하게 알려주는 것입니다. **isinstance** 함수는 데이터 타입을 확인합니다. features가 리스트인지, 길이가 올바른지 검사합니다. 이런 꼼꼼한 검증이 나중에 디버깅 시간을 크게 줄여줍니다. **실전 팁** 💡 - 민감한 에러 정보는 로그에만 남기고, 클라이언트에는 일반적인 메시지만 반환하세요 - 자주 발생하는 에러는 별도의 커스텀 예외 클래스로 관리하면 좋습니다 --- ## 5. 여러_모델_관리하기 시간이 지나면서 김개발 씨가 관리하는 모델이 점점 늘어났습니다. 고객 이탈 예측 모델, 상품 추천 모델, 가격 예측 모델까지. "모델마다 서버를 따로 띄워야 하나?" 다행히 하나의 Flask 서버에서 여러 모델을 효율적으로 관리하는 방법이 있습니다. 실무에서는 하나의 서버가 여러 모델을 서빙해야 하는 경우가 많습니다. 모델들을 딕셔너리로 관리하고, 요청 시 모델 이름을 지정받아 해당 모델로 예측을 수행하면 됩니다. 이렇게 하면 서버 하나로 다양한 예측 서비스를 제공할 수 있습니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, request, jsonify import joblib import numpy as np app = Flask(__name__) # 여러 모델을 딕셔너리로 관리 models = { 'churn': joblib.load('churn_model.pkl'), 'recommend': joblib.load('recommend_model.pkl'), 'price': joblib.load('price_model.pkl') } @app.route('/predict/ ', methods=['POST']) def predict(model_name): # 모델 존재 여부 확인 if model_name not in models: return jsonify({'error': f'{model_name} 모델을 찾을 수 없습니다'}), 404 data = request.get_json() features = np.array(data['features']).reshape(1, -1) prediction = models[model_name].predict(features) return jsonify({'model': model_name, 'prediction': prediction.tolist()}) ``` 프로젝트가 성공하면서 김개발 씨에게 새로운 모델 개발 요청이 계속 들어왔습니다. 처음에는 모델마다 서버를 따로 띄웠습니다. 그런데 서버가 5개, 10개로 늘어나자 관리가 힘들어졌습니다. "이거 좀 비효율적인 것 같은데..." 박시니어 씨가 해결책을 제시했습니다. "여러 모델을 하나의 서버에서 관리하면 돼요." 핵심은 **딕셔너리**입니다. 파이썬 딕셔너리에 모델 이름을 키로, 모델 객체를 값으로 저장합니다. 마치 사물함에 이름표를 붙여 물건을 보관하는 것과 같습니다. 코드에서 주목할 부분은 **@app.route('/predict/ ')** 입니다. 꺾쇠괄호 안의 model_name은 **URL 경로 변수**라고 부릅니다. 사용자가 /predict/churn으로 요청하면 model_name에 'churn'이 들어갑니다. 이 방식의 장점은 새 모델을 추가할 때 코드 수정이 최소화된다는 것입니다. models 딕셔너리에 새 모델만 추가하면 바로 사용할 수 있습니다. 물론 모델이 많아지면 메모리 사용량도 늘어납니다. 이럴 때는 자주 사용하지 않는 모델은 동적으로 로드하거나, LRU 캐시를 사용하여 메모리를 효율적으로 관리할 수 있습니다. 하지만 모델이 몇 개 안 된다면 시작 시점에 모두 로드해두는 것이 응답 속도 면에서 유리합니다. return 문에서 **prediction.tolist()** 를 사용한 이유도 있습니다. NumPy 배열은 JSON으로 직접 변환할 수 없기 때문에, 파이썬 리스트로 먼저 변환해야 합니다. **실전 팁** 💡 - 모델이 많아지면 설정 파일에서 모델 경로를 관리하는 것이 좋습니다 - 사용 빈도가 낮은 모델은 lazy loading 방식으로 필요할 때만 로드하세요 --- ## 6. 배치_예측_처리하기 마케팅팀에서 새로운 요청이 들어왔습니다. "고객 1만 명의 이탈 확률을 한 번에 예측해주세요." 한 명씩 1만 번 API를 호출하는 건 너무 비효율적입니다. 김개발 씨는 여러 건의 데이터를 한 번에 처리하는 배치 예측 기능이 필요하다는 것을 깨달았습니다. **배치 예측**은 여러 개의 입력 데이터를 한 번의 요청으로 처리하는 것입니다. 네트워크 오버헤드를 줄이고, 머신러닝 모델의 벡터 연산 특성을 활용하여 처리 속도를 크게 높일 수 있습니다. 대량의 데이터를 처리해야 하는 실무 환경에서 필수적인 기능입니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, request, jsonify import joblib import numpy as np app = Flask(__name__) model = joblib.load('model.pkl') @app.route('/predict/batch', methods=['POST']) def batch_predict(): data = request.get_json() # 배치 크기 제한 if len(data['samples']) > 1000: return jsonify({'error': '최대 1000개까지 처리 가능합니다'}), 400 # 2차원 배열로 변환 input_data = np.array(data['samples']) # 배치 예측 수행 predictions = model.predict(input_data) probabilities = model.predict_proba(input_data) results = [ {'prediction': int(pred), 'probability': prob.max()} for pred, prob in zip(predictions, probabilities) ] return jsonify({'count': len(results), 'results': results}) ``` 한 명의 고객을 예측하는 데 API 호출이 한 번 필요하다면, 1만 명을 예측하려면 1만 번의 호출이 필요합니다. 매번 네트워크 연결을 맺고 끊는 오버헤드가 발생하고, 전체 처리 시간이 크게 늘어납니다. 배치 예측은 이 문제를 해결합니다. 여러 개의 입력을 한 번에 보내고, 한 번에 결과를 받습니다. 마치 택배를 하나씩 보내는 것보다 박스에 모아서 한 번에 보내는 것이 효율적인 것과 같습니다. 더 좋은 점은 머신러닝 모델의 특성에 있습니다. 대부분의 모델은 내부적으로 행렬 연산을 사용합니다. NumPy와 같은 라이브러리는 행렬 연산에 최적화되어 있어서, 1000개의 데이터를 한 번에 처리하는 것이 하나씩 1000번 처리하는 것보다 훨씬 빠릅니다. 코드에서 **배치 크기 제한**을 주목하세요. 제한 없이 받으면 누군가 100만 건의 데이터를 한 번에 보내서 서버를 다운시킬 수 있습니다. 적절한 제한을 두는 것이 서버 안정성을 위해 중요합니다. **predict_proba** 함수는 예측 확률을 반환합니다. 단순히 0 또는 1이 아니라, 얼마나 확신하는지도 알려주는 것이죠. 마케팅팀에서는 이 확률을 보고 이탈 가능성이 높은 고객 순으로 정렬하여 집중 관리할 수 있습니다. 결과를 리스트 컴프리헨션으로 가공하는 부분도 눈여겨보세요. 각 예측 결과와 확률을 딕셔너리로 묶어서 반환합니다. 클라이언트에서 사용하기 편한 형태로 만들어주는 것이 좋은 API 설계입니다. **실전 팁** 💡 - 배치 크기가 매우 크다면 비동기 처리나 백그라운드 작업 큐를 고려하세요 - 진행 상황을 알려주는 progress 엔드포인트를 추가하면 사용자 경험이 좋아집니다 --- ## 7. Gunicorn으로_프로덕션_배포 드디어 테스트가 모두 끝났습니다. 이제 실제 서비스에 배포할 차례입니다. 그런데 박시니어 씨가 경고했습니다. "Flask 내장 서버는 개발용이야. 실제 서비스에서는 Gunicorn을 써야 해." 프로덕션 환경에서의 안정적인 배포 방법을 알아볼 시간입니다. Flask의 내장 개발 서버는 동시 요청 처리에 한계가 있고, 보안 기능도 부족합니다. **Gunicorn**은 파이썬 WSGI HTTP 서버로, 여러 워커 프로세스를 통해 동시 요청을 안정적으로 처리합니다. 실제 프로덕션 환경에서는 반드시 Gunicorn이나 유사한 프로덕션 서버를 사용해야 합니다. 다음 코드를 살펴봅시다. ```python # requirements.txt flask==3.0.0 gunicorn==21.2.0 joblib==1.3.2 numpy==1.26.0 scikit-learn==1.3.2 # 서버 실행 명령어 # gunicorn --workers 4 --bind 0.0.0.0:5000 app:app # gunicorn.conf.py (설정 파일) bind = '0.0.0.0:5000' workers = 4 timeout = 120 accesslog = 'access.log' errorlog = 'error.log' loglevel = 'info' ``` Flask를 실행할 때 터미널에 이런 경고 메시지를 본 적 있을 것입니다. "WARNING: This is a development server. Do not use it in a production deployment." 개발 서버는 말 그대로 개발용입니다. 한 번에 하나의 요청만 처리할 수 있고, 에러가 발생하면 전체 서버가 멈출 수 있습니다. 마치 혼자서 모든 손님을 상대하는 1인 카페와 같습니다. **Gunicorn**(Green Unicorn)은 이 문제를 해결합니다. 여러 명의 직원을 고용하는 것과 같습니다. **워커(worker)** 라고 부르는 여러 프로세스가 동시에 요청을 처리합니다. **--workers 4** 옵션은 4개의 워커 프로세스를 생성합니다. 일반적으로 CPU 코어 수의 2배 + 1로 설정하는 것이 좋습니다. 4코어 서버라면 9개의 워커가 적당합니다. **--bind 0.0.0.0:5000** 은 모든 네트워크 인터페이스의 5000번 포트에서 요청을 받겠다는 의미입니다. 외부에서 접속할 수 있게 하려면 이렇게 설정해야 합니다. 설정 파일을 별도로 만들면 명령어가 깔끔해집니다. **timeout** 설정은 중요합니다. 머신러닝 모델 예측에 시간이 오래 걸리는 경우, 기본 타임아웃(30초)이 너무 짧을 수 있습니다. 로그 설정도 잊지 마세요. **accesslog**에는 모든 요청이 기록되고, **errorlog**에는 에러가 기록됩니다. 문제가 발생했을 때 이 로그가 디버깅의 열쇠가 됩니다. 실제로 서버를 실행할 때는 **gunicorn -c gunicorn.conf.py app:app** 명령어를 사용합니다. 여기서 **app:app**은 app.py 파일의 app 변수를 의미합니다. **실전 팁** 💡 - Nginx를 앞에 두고 Gunicorn을 뒤에 배치하는 구조가 일반적입니다 - systemd나 supervisor를 사용하여 서버가 자동으로 재시작되도록 설정하세요 --- ## 8. 모델_버전_관리와_A/B_테스트 새로운 모델을 학습시켰는데, 기존 모델보다 정말 좋은지 어떻게 확인할까요? 무작정 교체했다가 성능이 떨어지면 큰일입니다. 김개발 씨는 안전하게 새 모델을 테스트하고 배포하는 방법이 필요했습니다. **모델 버전 관리**는 여러 버전의 모델을 동시에 운영하며 성능을 비교하는 방법입니다. **A/B 테스트**를 통해 일부 트래픽만 새 모델로 보내서 실제 환경에서 성능을 검증할 수 있습니다. 문제가 발생하면 즉시 이전 버전으로 롤백할 수 있어 안전합니다. 다음 코드를 살펴봅시다. ```python from flask import Flask, request, jsonify import joblib import numpy as np import random app = Flask(__name__) # 버전별 모델 로드 models = { 'v1': joblib.load('model_v1.pkl'), 'v2': joblib.load('model_v2.pkl') } current_version = 'v1' ab_test_ratio = 0.1 # 10% 트래픽을 v2로 @app.route('/predict', methods=['POST']) def predict(): # A/B 테스트: 확률적으로 모델 선택 if random.random() < ab_test_ratio: version = 'v2' else: version = current_version data = request.get_json() input_data = np.array(data['features']).reshape(1, -1) prediction = models[version].predict(input_data) # 버전 정보도 함께 반환 return jsonify({ 'prediction': int(prediction[0]), 'model_version': version }) ``` 새 모델을 배포하는 것은 언제나 긴장되는 일입니다. 학습 데이터에서는 좋은 성능을 보였지만, 실제 환경에서도 그럴까요? 예상치 못한 에지 케이스가 있을 수도 있습니다. 한 번에 전체 트래픽을 새 모델로 전환하는 것은 위험합니다. 마치 새로 개발한 약을 테스트 없이 모든 환자에게 투여하는 것과 같습니다. **A/B 테스트**는 이 문제를 해결합니다. 전체 트래픽 중 일부만 새 모델로 보내서 실제 성능을 측정합니다. 위 코드에서는 10%의 요청만 v2 모델로 처리합니다. **random.random()** 함수는 0과 1 사이의 난수를 반환합니다. 이 값이 0.1보다 작으면 새 모델을, 그렇지 않으면 기존 모델을 사용합니다. 요청이 충분히 많으면 대략 10%가 새 모델로 처리됩니다. 응답에 **model_version**을 포함시키는 것이 중요합니다. 나중에 로그를 분석할 때, 어떤 버전의 모델이 어떤 예측을 했는지 추적할 수 있습니다. 이 데이터를 바탕으로 두 버전의 정확도, 응답 시간 등을 비교합니다. 새 모델이 충분히 검증되면 **ab_test_ratio**를 점진적으로 높입니다. 10%, 30%, 50%, 100% 순으로 천천히 전환합니다. 이것을 **점진적 롤아웃(gradual rollout)** 이라고 부릅니다. 문제가 발생하면 즉시 ratio를 0으로 설정하여 롤백할 수 있습니다. 코드 변경 없이 설정만 바꾸면 되니까 빠르게 대응할 수 있습니다. 실무에서는 이런 설정값을 환경 변수나 설정 서버에서 관리합니다. 서버를 재시작하지 않고도 동적으로 변경할 수 있게요. **실전 팁** 💡 - A/B 테스트 결과 분석을 위해 별도의 로깅 시스템을 구축하세요 - 통계적으로 유의미한 결과를 얻으려면 충분한 샘플 수가 필요합니다 --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: Data Science

언어: Python

태그: Python, Flask, ModelDeployment, MachineLearning, API, Data Science

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

CodeDeck

Flask를 이용한 모델 배포 완벽 가이드

목차

1. Flask 기본 서버 구축

2. 모델 저장과 로드

3. 예측 API 엔드포인트 만들기

4. 에러 처리와 입력 검증

5. 여러 모델 관리하기

6. 배치 예측 처리하기

7. Gunicorn으로 프로덕션 배포

8. 모델 버전 관리와 A/B 테스트

이 카드뉴스가 포함된 코스

MLOps 파이프라인 구축 완전 정복

댓글 (0)

함께 보면 좋은 카드 뉴스

Helm 마이크로서비스 패키징 완벽 가이드

보안 아키텍처 구성 완벽 가이드

AWS Organizations 완벽 가이드

AWS KMS 암호화 완벽 가이드

AWS Secrets Manager 완벽 가이드