094 FastAPI로 모델 배포

키워드: FastAPI, API 배포

개요

FastAPI는 Python으로 API를 빠르게 만들 수 있는 현대적인 웹 프레임워크입니다. 학습된 PyCaret 모델을 REST API로 배포하여 실시간 예측 서비스를 구축합니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: pycaret[full]>=3.0, fastapi, uvicorn

FastAPI 기본 구조

# 094 app.py
from fastapi import FastAPI

app = FastAPI(title="ML Model API", version="1.0.0")

@app.get("/")
def root():
    return {"message": "ML Model API is running"}

@app.get("/health")
def health():
    return {"status": "healthy"}

# 094 실행: uvicorn app:app --reload

모델 준비

# 094 먼저 모델 학습 및 저장
from pycaret.classification import *
from pycaret.datasets import get_data

# 094 데이터 로드
data = get_data('diabetes')

# 094 환경 설정
clf = setup(data, target='Class variable', session_id=42, verbose=False)

# 094 모델 생성 및 저장
rf = create_model('rf')
final = finalize_model(rf)
save_model(final, 'diabetes_model')

print("모델 저장 완료: diabetes_model.pkl")

예측 API 구현

# 094 app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import pandas as pd
from pycaret.classification import load_model, predict_model

# 094 FastAPI 앱
app = FastAPI(
    title="Diabetes Prediction API",
    description="당뇨병 예측 모델 API",
    version="1.0.0"
)

# 094 모델 로드 (앱 시작 시 한 번만)
model = load_model('diabetes_model')

# 094 입력 스키마 정의
class PatientData(BaseModel):
    Pregnancies: float
    Glucose: float
    BloodPressure: float
    SkinThickness: float
    Insulin: float
    BMI: float
    DiabetesPedigreeFunction: float
    Age: float

class PredictionResponse(BaseModel):
    prediction: int
    probability: float

@app.get("/")
def root():
    return {"message": "Diabetes Prediction API"}

@app.post("/predict", response_model=PredictionResponse)
def predict(patient: PatientData):
    """단일 환자 예측"""
    try:
        # 입력 데이터를 DataFrame으로 변환
        input_df = pd.DataFrame([patient.dict()])

        # 예측
        predictions = predict_model(model, data=input_df)

        # 결과 추출
        pred_label = int(predictions['prediction_label'].values[0])
        pred_score = float(predictions['prediction_score'].values[0])

        return PredictionResponse(
            prediction=pred_label,
            probability=pred_score
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 094 실행: uvicorn app:app --host 0.0.0.0 --port 8000

배치 예측 API

# 094 app.py (확장)
from typing import List

class BatchPredictionRequest(BaseModel):
    patients: List[PatientData]

class BatchPredictionResponse(BaseModel):
    predictions: List[PredictionResponse]

@app.post("/predict/batch", response_model=BatchPredictionResponse)
def predict_batch(request: BatchPredictionRequest):
    """배치 예측"""
    try:
        # 여러 환자 데이터를 DataFrame으로 변환
        input_df = pd.DataFrame([p.dict() for p in request.patients])

        # 예측
        predictions = predict_model(model, data=input_df)

        # 결과 변환
        results = []
        for idx in range(len(predictions)):
            results.append(PredictionResponse(
                prediction=int(predictions['prediction_label'].values[idx]),
                probability=float(predictions['prediction_score'].values[idx])
            ))

        return BatchPredictionResponse(predictions=results)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

비동기 처리

# 094 app_async.py
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
import pandas as pd
import asyncio
from pycaret.classification import load_model, predict_model
import uuid

app = FastAPI()
model = load_model('diabetes_model')

# 094 작업 저장소
tasks = {}

class TaskStatus(BaseModel):
    task_id: str
    status: str
    result: dict = None

@app.post("/predict/async")
async def predict_async(patient: PatientData, background_tasks: BackgroundTasks):
    """비동기 예측 요청"""
    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "processing", "result": None}

    background_tasks.add_task(process_prediction, task_id, patient)

    return {"task_id": task_id, "status": "processing"}

def process_prediction(task_id: str, patient: PatientData):
    """백그라운드 예측 처리"""
    try:
        input_df = pd.DataFrame([patient.dict()])
        predictions = predict_model(model, data=input_df)

        tasks[task_id] = {
            "status": "completed",
            "result": {
                "prediction": int(predictions['prediction_label'].values[0]),
                "probability": float(predictions['prediction_score'].values[0])
            }
        }
    except Exception as e:
        tasks[task_id] = {"status": "failed", "result": {"error": str(e)}}

@app.get("/predict/status/{task_id}")
def get_task_status(task_id: str):
    """작업 상태 확인"""
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks[task_id]

입력 검증

from pydantic import BaseModel, validator, Field

class PatientDataValidated(BaseModel):
    Pregnancies: float = Field(..., ge=0, le=20)
    Glucose: float = Field(..., ge=0, le=200)
    BloodPressure: float = Field(..., ge=0, le=150)
    SkinThickness: float = Field(..., ge=0, le=100)
    Insulin: float = Field(..., ge=0, le=900)
    BMI: float = Field(..., ge=0, le=70)
    DiabetesPedigreeFunction: float = Field(..., ge=0, le=3)
    Age: float = Field(..., ge=0, le=120)

    @validator('BMI')
    def bmi_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('BMI must be positive')
        return v

    @validator('Age')
    def age_must_be_valid(cls, v):
        if v < 18:
            raise ValueError('Age must be at least 18')
        return v

모델 메타데이터 API

@app.get("/model/info")
def model_info():
    """모델 정보 반환"""
    return {
        "model_type": "RandomForestClassifier",
        "version": "1.0.0",
        "features": [
            "Pregnancies", "Glucose", "BloodPressure",
            "SkinThickness", "Insulin", "BMI",
            "DiabetesPedigreeFunction", "Age"
        ],
        "target": "Class variable",
        "classes": ["0 (No Diabetes)", "1 (Diabetes)"]
    }

@app.get("/model/features")
def model_features():
    """필요한 특성 정보"""
    return {
        "features": {
            "Pregnancies": {"type": "float", "description": "임신 횟수", "range": "0-20"},
            "Glucose": {"type": "float", "description": "포도당 농도", "range": "0-200"},
            "BloodPressure": {"type": "float", "description": "혈압", "range": "0-150"},
            "SkinThickness": {"type": "float", "description": "피부 두께", "range": "0-100"},
            "Insulin": {"type": "float", "description": "인슐린", "range": "0-900"},
            "BMI": {"type": "float", "description": "체질량 지수", "range": "0-70"},
            "DiabetesPedigreeFunction": {"type": "float", "description": "당뇨 가계력", "range": "0-3"},
            "Age": {"type": "float", "description": "나이", "range": "18-120"}
        }
    }

로깅 미들웨어

from fastapi import Request
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """요청 로깅"""
    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.status_code} "
        f"- Time: {process_time:.3f}s"
    )

    return response

에러 핸들링

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """전역 에러 핸들러"""
    logger.error(f"Unhandled error: {exc}")
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal Server Error",
            "detail": str(exc) if app.debug else "An error occurred"
        }
    )

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    """값 에러 핸들러"""
    return JSONResponse(
        status_code=400,
        content={"error": "Bad Request", "detail": str(exc)}
    )

테스트 클라이언트

# 094 test_api.py
import requests

BASE_URL = "http://localhost:8000"

def test_health():
    response = requests.get(f"{BASE_URL}/health")
    print(f"Health: {response.json()}")

def test_single_prediction():
    patient = {
        "Pregnancies": 6,
        "Glucose": 148,
        "BloodPressure": 72,
        "SkinThickness": 35,
        "Insulin": 0,
        "BMI": 33.6,
        "DiabetesPedigreeFunction": 0.627,
        "Age": 50
    }

    response = requests.post(f"{BASE_URL}/predict", json=patient)
    print(f"Prediction: {response.json()}")

def test_batch_prediction():
    patients = {
        "patients": [
            {
                "Pregnancies": 6, "Glucose": 148, "BloodPressure": 72,
                "SkinThickness": 35, "Insulin": 0, "BMI": 33.6,
                "DiabetesPedigreeFunction": 0.627, "Age": 50
            },
            {
                "Pregnancies": 1, "Glucose": 85, "BloodPressure": 66,
                "SkinThickness": 29, "Insulin": 0, "BMI": 26.6,
                "DiabetesPedigreeFunction": 0.351, "Age": 31
            }
        ]
    }

    response = requests.post(f"{BASE_URL}/predict/batch", json=patients)
    print(f"Batch Predictions: {response.json()}")

if __name__ == "__main__":
    test_health()
    test_single_prediction()
    test_batch_prediction()

API 문서 자동 생성

# 094 FastAPI는 자동으로 OpenAPI 문서 생성
# 094 /docs - Swagger UI
# 094 /redoc - ReDoc UI

# 094 추가 문서 설정
app = FastAPI(
    title="Diabetes Prediction API",
    description="""
    ## 당뇨병 예측 API

    이 API는 PyCaret으로 학습된 모델을 사용하여
    환자의 당뇨병 발병 가능성을 예측합니다.

    ### 사용 방법
    1. `/predict` 엔드포인트에 환자 데이터 전송
    2. 예측 결과와 확률 반환

    ### 특성 설명
    - Glucose: 경구 포도당 내성 검사 2시간 후 혈장 포도당 농도
    - BMI: 체질량 지수 (체중 kg / 키 m^2)
    """,
    version="1.0.0",
    contact={
        "name": "Data Science Team",
        "email": "[email protected]"
    }
)

전체 코드

# 094 main.py - 완전한 API 서버
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List
import pandas as pd
import logging
import time
from pycaret.classification import load_model, predict_model

# 094 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 094 FastAPI 앱
app = FastAPI(
    title="Diabetes Prediction API",
    version="1.0.0"
)

# 094 모델 로드
model = load_model('diabetes_model')

# 094 스키마 정의
class PatientData(BaseModel):
    Pregnancies: float = Field(..., ge=0, le=20)
    Glucose: float = Field(..., ge=0, le=200)
    BloodPressure: float = Field(..., ge=0, le=150)
    SkinThickness: float = Field(..., ge=0, le=100)
    Insulin: float = Field(..., ge=0, le=900)
    BMI: float = Field(..., ge=0, le=70)
    DiabetesPedigreeFunction: float = Field(..., ge=0, le=3)
    Age: float = Field(..., ge=18, le=120)

class PredictionResponse(BaseModel):
    prediction: int
    probability: float

# 094 미들웨어
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    logger.info(f"{request.method} {request.url.path} - {process_time:.3f}s")
    return response

# 094 엔드포인트
@app.get("/health")
def health():
    return {"status": "healthy"}

@app.post("/predict", response_model=PredictionResponse)
def predict(patient: PatientData):
    try:
        input_df = pd.DataFrame([patient.dict()])
        predictions = predict_model(model, data=input_df)
        return PredictionResponse(
            prediction=int(predictions['prediction_label'].values[0]),
            probability=float(predictions['prediction_score'].values[0])
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 094 실행: uvicorn main:app --host 0.0.0.0 --port 8000

정리

• FastAPI: 현대적인 Python 웹 프레임워크
• Pydantic: 입력 검증 및 스키마 정의
• 비동기 처리: 대용량 요청 처리
• 자동 문서화: /docs, /redoc
• 미들웨어: 로깅, 인증 등 추가

다음 글 예고

다음 글에서는 Docker 컨테이너화를 다룹니다.

---

PyCaret 머신러닝 마스터 시리즈 #094