FastAPI로 ML 모델 REST API 구축

머신러닝 모델을 실제 서비스로 배포하기 위한 FastAPI 기반 REST API 구축 방법을 다룹니다. 모델 로딩부터 예측 엔드포인트 생성, 입출력 검증까지 단계별로 살펴봅니다.

---

목차

1. FastAPI_시작하기
2. Pydantic_스키마_정의
3. ML_모델_로딩
4. 예측_엔드포인트_구현
5. 배치_예측_구현
6. 에러_처리와_검증
7. 비동기_처리와_성능_최적화
8. API_문서화와_메타데이터
9. 헬스체크와_모니터링
10. 배포와_서버_실행

---

1. FastAPI 시작하기

김개발 씨는 데이터 사이언스 팀에서 열심히 만든 머신러닝 모델을 웹 서비스로 배포해야 하는 첫 미션을 받았습니다. "모델은 만들었는데, 이걸 어떻게 다른 팀에서 쓸 수 있게 하지?" 막막한 표정의 김개발 씨에게 박시니어 씨가 다가왔습니다.

FastAPI는 Python으로 REST API를 빠르게 구축할 수 있는 현대적인 웹 프레임워크입니다. 마치 레스토랑의 주문 시스템처럼, 외부 요청을 받아 적절한 응답을 돌려주는 역할을 합니다.

자동 문서화와 타입 검증이 내장되어 있어 ML 모델 서빙에 특히 적합합니다.

다음 코드를 살펴봅시다.

# FastAPI 기본 구조
from fastapi import FastAPI

# FastAPI 애플리케이션 인스턴스 생성
app = FastAPI(
    title="ML Model API",
    description="머신러닝 모델 서빙을 위한 API",
    version="1.0.0"
)

# 헬스 체크 엔드포인트
@app.get("/health")
def health_check():
    return {"status": "healthy", "message": "API가 정상 동작 중입니다"}

# 루트 엔드포인트
@app.get("/")
def root():
    return {"message": "ML Model API에 오신 것을 환영합니다"}

김개발 씨는 입사 6개월 차 ML 엔지니어입니다. 팀에서 고객 이탈 예측 모델을 만들었는데, 이제 이 모델을 마케팅 팀에서 실시간으로 사용할 수 있도록 API로 만들어야 합니다.

"Flask를 써볼까요?" 김개발 씨가 물었습니다. 박시니어 씨가 고개를 저었습니다.

"요즘은 FastAPI를 많이 써요. 더 빠르고, 자동으로 API 문서도 만들어주거든요." 그렇다면 FastAPI란 정확히 무엇일까요?

쉽게 비유하자면, FastAPI는 마치 식당의 주문 카운터와 같습니다. 손님이 원하는 메뉴를 말하면, 카운터에서 주문을 받아 주방에 전달하고, 완성된 음식을 손님에게 돌려줍니다.

FastAPI도 마찬가지로 외부에서 들어오는 요청을 받아 내부 로직(우리의 ML 모델)을 실행하고, 결과를 응답으로 전달합니다. FastAPI가 없던 시절에는 어땠을까요?

개발자들은 Flask나 Django 같은 프레임워크를 사용했습니다. 물론 훌륭한 도구들이지만, 타입 검증을 위해 추가 라이브러리를 설치해야 했고, API 문서도 직접 작성해야 했습니다.

ML 모델의 입출력 형식이 복잡해질수록 이런 작업은 점점 부담이 되었습니다. 바로 이런 문제를 해결하기 위해 FastAPI가 등장했습니다.

FastAPI를 사용하면 Python의 타입 힌트만으로 자동 검증이 가능해집니다. 또한 /docs 경로로 접속하면 Swagger UI가 자동 생성됩니다.

무엇보다 비동기 처리를 기본 지원하여 동시에 여러 요청을 효율적으로 처리할 수 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 FastAPI() 클래스로 애플리케이션 인스턴스를 생성합니다. title, description, version은 자동 생성되는 문서에 표시됩니다.

다음으로 @app.get() 데코레이터로 GET 요청을 처리할 엔드포인트를 정의합니다. 함수가 반환하는 딕셔너리는 자동으로 JSON 응답으로 변환됩니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 금융 서비스를 개발한다고 가정해봅시다.

대출 심사 모델을 API로 만들면, 웹사이트, 모바일 앱, 상담원 시스템 등 다양한 채널에서 같은 모델을 호출할 수 있습니다. 이것이 바로 ML 모델 서빙의 핵심입니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 app 인스턴스를 여러 개 만드는 것입니다.

하나의 애플리케이션에는 하나의 FastAPI 인스턴스만 있어야 합니다. 또한 개발 환경에서는 uvicorn의 reload 옵션을 사용하지만, 프로덕션에서는 반드시 꺼야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 터미널에 uvicorn main:app --reload를 입력했습니다.

브라우저에서 /docs로 접속하자 멋진 API 문서가 나타났습니다. FastAPI의 기본 구조를 이해하면 ML 모델 서빙의 첫 발을 내딛은 것입니다.

다음 단계에서는 실제 모델을 연결해 보겠습니다.

실전 팁

💡 - uvicorn main:app --reload 명령으로 개발 서버를 실행하세요

• /docs와 /redoc 경로에서 자동 생성된 API 문서를 확인할 수 있습니다

---

2. Pydantic 스키마 정의

김개발 씨가 API를 만들기 시작했는데, 문제가 생겼습니다. 프론트엔드 팀에서 잘못된 형식의 데이터를 보내자 모델이 에러를 뿜어냈습니다.

"입력값 검증을 어떻게 하면 좋을까요?" 박시니어 씨가 웃으며 Pydantic을 소개해주었습니다.

Pydantic은 Python의 타입 힌트를 활용해 데이터 검증을 자동으로 수행하는 라이브러리입니다. 마치 건물 입구의 보안 검색대처럼, 들어오는 데이터가 올바른 형식인지 확인합니다.

FastAPI와 함께 사용하면 요청과 응답의 구조를 명확하게 정의할 수 있습니다.

다음 코드를 살펴봅시다.

from pydantic import BaseModel, Field
from typing import Optional, List

# 예측 요청 스키마
class PredictionRequest(BaseModel):
    age: int = Field(..., ge=0, le=150, description="고객 나이")
    income: float = Field(..., gt=0, description="연간 소득")
    tenure: int = Field(..., ge=0, description="서비스 이용 기간(월)")
    features: Optional[List[float]] = Field(None, description="추가 특성")

# 예측 응답 스키마
class PredictionResponse(BaseModel):
    prediction: int = Field(..., description="예측 결과 (0: 유지, 1: 이탈)")
    probability: float = Field(..., ge=0, le=1, description="이탈 확률")
    model_version: str = Field(..., description="모델 버전")

김개발 씨가 만든 첫 번째 API는 잘 동작했습니다. 하지만 실제 서비스에 연결하자 문제가 터졌습니다.

프론트엔드에서 나이를 문자열로 보내거나, 소득에 음수를 넣는 경우가 있었던 것입니다. "모델이 이상한 값을 받으면 예측할 수 없잖아요." 김개발 씨가 한숨을 쉬었습니다.

박시니어 씨가 말했습니다. "그래서 입력 검증이 중요해요.

Pydantic을 쓰면 자동으로 처리됩니다." Pydantic이란 정확히 무엇일까요? 쉽게 비유하자면, Pydantic은 마치 공항의 보안 검색대와 같습니다.

탑승객이 가져온 짐을 검사해서 허용된 물품만 통과시키듯이, Pydantic은 들어오는 데이터를 검사해서 올바른 형식의 데이터만 애플리케이션 내부로 들여보냅니다. Pydantic이 없다면 어떻게 될까요?

개발자가 직접 모든 입력값을 검증해야 합니다. 나이가 정수인지, 소득이 양수인지, 필수 필드가 있는지 일일이 확인하는 코드를 작성해야 합니다.

코드가 길어지고, 한 군데라도 검증을 빼먹으면 버그가 발생합니다. Pydantic을 사용하면 이 모든 것이 자동화됩니다.

BaseModel을 상속받아 클래스를 정의하면, 해당 클래스가 데이터 스키마가 됩니다. 각 필드에 타입을 지정하면 자동으로 타입 변환과 검증이 이루어집니다.

Field 함수를 사용하면 더 세밀한 제약 조건을 추가할 수 있습니다. 위의 코드를 자세히 살펴보겠습니다.

**age: int = Field(..., ge=0, le=150)**에서 ge는 "greater than or equal"의 약자로 최솟값을, le는 "less than or equal"로 최댓값을 의미합니다. **...**은 필수 필드임을 나타냅니다.

Optional로 감싸면 선택적 필드가 됩니다. 실제 ML 서비스에서 Pydantic은 어떻게 활용될까요?

추천 시스템을 예로 들어봅시다. 사용자 ID, 현재 보고 있는 상품 ID, 이전 구매 이력 등 다양한 정보가 API로 들어옵니다.

Pydantic으로 스키마를 정의하면, 잘못된 형식의 요청은 모델에 도달하기도 전에 자동으로 거부됩니다. 에러 메시지도 친절하게 어떤 필드가 왜 잘못되었는지 알려줍니다.

주의할 점도 있습니다. Field의 첫 번째 인자인 **...**과 None의 차이를 명확히 이해해야 합니다.

**...**은 필수, None은 선택이며 기본값이 None입니다. 또한 복잡한 중첩 구조를 가진 스키마는 별도의 클래스로 분리하는 것이 좋습니다.

김개발 씨가 PredictionRequest 스키마를 정의하자, API 문서에 각 필드의 설명과 제약 조건이 자동으로 표시되었습니다. "이제 프론트엔드 팀에서 뭘 보내야 하는지 한눈에 알겠네요!"

실전 팁

💡 - Field의 description 파라미터는 API 문서에 그대로 표시되므로 상세하게 작성하세요

• Config 클래스를 정의하여 스키마 예시나 추가 설정을 지정할 수 있습니다

---

3. ML 모델 로딩

스키마 정의를 마친 김개발 씨가 다음 고민에 빠졌습니다. "학습된 모델 파일을 어떻게 불러와야 하지?

매 요청마다 로딩하면 느리지 않을까?" 박시니어 씨가 말했습니다. "서버 시작할 때 한 번만 로딩하면 됩니다.

lifespan 이벤트를 활용하세요."

ML 모델은 서버 시작 시점에 한 번만 메모리에 로딩하는 것이 효율적입니다. 마치 요리사가 영업 시작 전에 재료를 미리 준비해두는 것과 같습니다.

FastAPI의 lifespan 컨텍스트 매니저를 사용하면 서버 시작과 종료 시점에 필요한 작업을 정의할 수 있습니다.

다음 코드를 살펴봅시다.

import joblib
from contextlib import asynccontextmanager
from fastapi import FastAPI

# 모델을 저장할 전역 변수
ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 시 모델 로딩
    print("모델을 로딩합니다...")
    ml_models["churn_model"] = joblib.load("models/churn_model.pkl")
    ml_models["version"] = "1.0.0"
    print("모델 로딩 완료!")
    yield
    # 서버 종료 시 정리 작업
    ml_models.clear()
    print("리소스를 정리했습니다.")

app = FastAPI(lifespan=lifespan)

김개발 씨가 처음 작성한 코드는 이랬습니다. 예측 함수가 호출될 때마다 **joblib.load()**로 모델을 불러왔습니다.

테스트할 때는 문제없어 보였지만, 부하 테스트를 돌리자 응답 시간이 급격히 늘어났습니다. "모델 파일이 50MB인데, 매번 디스크에서 읽어오니까 느린 거예요." 박시니어 씨가 문제를 짚었습니다.

모델 로딩 문제는 왜 발생할까요? 머신러닝 모델 파일은 보통 수십 MB에서 수 GB까지 다양합니다.

이런 큰 파일을 매 요청마다 디스크에서 읽어 메모리에 올리면 심각한 병목이 됩니다. 마치 손님이 올 때마다 냉장고에서 재료를 꺼내 해동하는 것과 같습니다.

해결책은 서버 시작 시점에 미리 로딩하는 것입니다. FastAPI의 lifespan 컨텍스트 매니저는 서버의 생명주기를 관리합니다.

yield 전에 작성된 코드는 서버 시작 시 실행되고, yield 후에 작성된 코드는 서버 종료 시 실행됩니다. 이 패턴을 사용하면 모델을 메모리에 한 번만 올려두고 계속 재사용할 수 있습니다.

코드를 자세히 살펴보겠습니다. ml_models라는 딕셔너리에 로딩된 모델을 저장합니다.

@asynccontextmanager 데코레이터는 비동기 컨텍스트 매니저를 만듭니다. **joblib.load()**는 scikit-learn 모델을 불러오는 표준적인 방법입니다.

버전 정보도 함께 저장해두면 디버깅에 유용합니다. 실제 서비스에서는 어떤 점을 고려해야 할까요?

대규모 서비스에서는 여러 개의 모델을 동시에 서빙하는 경우가 많습니다. 추천 모델, 분류 모델, 회귀 모델 등을 모두 ml_models 딕셔너리에 저장하고, 요청에 따라 적절한 모델을 선택하면 됩니다.

모델 업데이트 시에는 서버를 재시작하거나, 동적 로딩 메커니즘을 구현합니다. 주의할 점이 있습니다.

모델 파일 경로는 상대 경로보다 절대 경로를 사용하는 것이 안전합니다. 서버 실행 위치에 따라 상대 경로가 달라질 수 있기 때문입니다.

또한 모델 파일이 존재하지 않을 경우를 대비해 예외 처리를 추가하는 것이 좋습니다. 김개발 씨가 lifespan을 적용하자, 첫 요청부터 응답이 빨라졌습니다.

"모델 로딩은 서버 시작할 때 한 번만, 예측은 메모리에서 빠르게!" 박시니어 씨가 엄지를 들어 보였습니다.

실전 팁

💡 - 모델 파일은 반드시 버전 관리하고, API 응답에 모델 버전을 포함시키세요

• 여러 모델을 사용할 경우 딕셔너리로 관리하면 편리합니다

---

4. 예측 엔드포인트 구현

모델 로딩까지 완료한 김개발 씨가 드디어 핵심 기능에 도전합니다. 바로 예측 결과를 반환하는 API 엔드포인트입니다.

"POST 요청으로 데이터를 받아서 예측 결과를 돌려주면 되는 거죠?" 박시니어 씨가 고개를 끄덕였습니다.

예측 엔드포인트는 ML 모델 API의 핵심입니다. 마치 자판기에 동전을 넣고 버튼을 누르면 음료가 나오듯이, 입력 데이터를 받아 예측 결과를 반환합니다.

앞서 정의한 Pydantic 스키마와 로딩된 모델을 연결하면 완성입니다.

다음 코드를 살펴봅시다.

from fastapi import FastAPI, HTTPException
import numpy as np

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # 입력 데이터를 모델 입력 형태로 변환
        features = np.array([[
            request.age,
            request.income,
            request.tenure
        ]])

        # 모델 예측 수행
        model = ml_models["churn_model"]
        prediction = model.predict(features)[0]
        probability = model.predict_proba(features)[0][1]

        return PredictionResponse(
            prediction=int(prediction),
            probability=float(probability),
            model_version=ml_models["version"]
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

김개발 씨가 드디어 가장 기다리던 순간에 도달했습니다. 지금까지 준비한 모든 것들이 하나로 합쳐지는 순간입니다.

"이제 실제로 예측하는 부분을 만들어볼까요?" 박시니어 씨가 말했습니다. 예측 엔드포인트는 ML API의 심장과 같습니다.

비유하자면, 지금까지 만든 것들은 무대 설치와 같습니다. FastAPI라는 무대를 세우고, Pydantic으로 입장 검사 시스템을 만들고, 모델이라는 배우를 대기시켜 놓았습니다.

이제 실제 공연이 시작됩니다. 코드의 흐름을 따라가 봅시다.

@app.post("/predict") 데코레이터는 POST 요청을 처리합니다. response_model=PredictionResponse는 응답이 해당 스키마를 따라야 함을 명시합니다.

잘못된 형식의 응답을 반환하면 FastAPI가 에러를 발생시킵니다. 함수 파라미터 request: PredictionRequest가 핵심입니다.

FastAPI는 요청 본문을 자동으로 파싱하여 PredictionRequest 객체로 변환합니다. Pydantic이 타입 검증을 수행하고, 잘못된 데이터가 들어오면 422 에러를 반환합니다.

내부 로직을 살펴보겠습니다. 먼저 **np.array()**로 입력 데이터를 numpy 배열로 변환합니다.

대부분의 scikit-learn 모델은 2차원 배열을 입력으로 받으므로 이중 대괄호를 사용합니다. **predict()**는 예측 클래스를, **predict_proba()**는 확률을 반환합니다.

응답은 PredictionResponse 스키마에 맞춰 반환합니다. numpy 타입은 JSON 직렬화가 안 되므로 **int()**와 **float()**로 Python 기본 타입으로 변환해야 합니다.

이 부분을 빠뜨리면 직렬화 에러가 발생합니다. 예외 처리도 중요합니다.

모델 예측 중 예상치 못한 에러가 발생할 수 있습니다. HTTPException을 사용하면 에러를 적절한 HTTP 상태 코드와 함께 반환할 수 있습니다.

500은 서버 내부 에러를 의미합니다. 실제 서비스에서는 더 세밀한 에러 처리가 필요합니다.

입력값이 모델의 학습 범위를 벗어나는 경우, 특정 필드 조합이 유효하지 않은 경우 등 다양한 상황에 대응해야 합니다. 각 상황에 맞는 HTTP 상태 코드(400, 422 등)를 사용하면 API 사용자가 문제를 쉽게 파악할 수 있습니다.

김개발 씨가 Postman으로 테스트 요청을 보냈습니다. 나이 35, 소득 50000, 이용 기간 24개월.

결과가 돌아왔습니다. 이탈 확률 23%.

"드디어 작동하네요!" 김개발 씨의 얼굴에 미소가 번졌습니다.

실전 팁

💡 - numpy 타입은 반드시 Python 기본 타입으로 변환하여 반환하세요

• 예측 로직과 데이터 변환 로직을 별도 함수로 분리하면 테스트가 쉬워집니다

---

5. 배치 예측 구현

API가 잘 동작하자 마케팅 팀에서 새로운 요청이 들어왔습니다. "고객 1만 명을 한꺼번에 예측해야 하는데, 하나씩 호출하면 너무 오래 걸려요." 김개발 씨가 머리를 긁적였습니다.

"배치 처리를 지원해야겠네요."

배치 예측은 여러 개의 입력을 한 번에 처리하는 방식입니다. 마치 택배를 하나씩 배송하는 것보다 트럭에 가득 실어 한 번에 배송하는 것이 효율적인 것처럼, 네트워크 호출 횟수를 줄여 전체 처리 시간을 단축할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import List

# 배치 요청 스키마
class BatchPredictionRequest(BaseModel):
    instances: List[PredictionRequest] = Field(
        ...,
        min_length=1,
        max_length=1000,
        description="예측할 데이터 목록"
    )

# 배치 응답 스키마
class BatchPredictionResponse(BaseModel):
    predictions: List[PredictionResponse]
    total_count: int

@app.post("/predict/batch", response_model=BatchPredictionResponse)
async def predict_batch(request: BatchPredictionRequest):
    # 모든 입력을 하나의 배열로 변환
    features = np.array([
        [inst.age, inst.income, inst.tenure]
        for inst in request.instances
    ])

    model = ml_models["churn_model"]
    predictions = model.predict(features)
    probabilities = model.predict_proba(features)[:, 1]

    results = [
        PredictionResponse(
            prediction=int(pred),
            probability=float(prob),
            model_version=ml_models["version"]
        )
        for pred, prob in zip(predictions, probabilities)
    ]

    return BatchPredictionResponse(
        predictions=results,
        total_count=len(results)
    )

마케팅 팀의 요청을 받은 김개발 씨가 계산을 해봤습니다. 단건 API를 1만 번 호출하면 네트워크 왕복 시간만 해도 엄청납니다.

각 요청당 50ms라고 해도 총 500초, 약 8분이 걸립니다. "배치로 처리하면 훨씬 빨라요." 박시니어 씨가 설명했습니다.

배치 처리가 왜 효율적일까요? 택배에 비유하면 이해가 쉽습니다.

물건 100개를 배송한다고 할 때, 오토바이로 하나씩 배송하면 100번 왕복해야 합니다. 하지만 트럭에 100개를 실어 한 번에 가면 1번만 왕복하면 됩니다.

API도 마찬가지로, 네트워크 왕복 횟수를 줄이는 것이 핵심입니다. 더불어 대부분의 ML 모델은 벡터화 연산을 지원합니다.

numpy와 scikit-learn은 내부적으로 최적화된 행렬 연산을 수행합니다. 1000개의 데이터를 하나씩 예측하는 것보다 한꺼번에 예측하는 것이 훨씬 빠릅니다.

이것이 배치 처리의 두 번째 장점입니다. 코드를 살펴봅시다.

BatchPredictionRequest는 여러 개의 PredictionRequest를 리스트로 받습니다. min_length와 max_length로 한 번에 처리할 수 있는 양을 제한합니다.

너무 많은 데이터를 한꺼번에 받으면 서버 메모리가 부족해질 수 있기 때문입니다. 리스트 컴프리헨션으로 모든 입력을 2차원 numpy 배열로 변환합니다.

모델의 **predict()**와 **predict_proba()**는 배열 전체에 대해 한 번에 예측을 수행합니다. **zip()**으로 예측값과 확률을 묶어 응답 객체를 생성합니다.

실무에서는 어떻게 활용될까요? 야간에 전체 고객 데이터를 분석하는 배치 작업, 대량의 이미지를 한꺼번에 분류하는 작업 등에 활용됩니다.

단, 배치 크기가 너무 크면 타임아웃이 발생할 수 있으므로 적절한 최대 크기를 설정해야 합니다. 주의할 점이 있습니다.

배치 요청 중 일부 데이터만 실패하는 경우를 고려해야 합니다. 전체를 실패 처리할지, 성공한 것만 반환할지 정책을 정해야 합니다.

또한 배치 처리는 메모리를 많이 사용하므로, 서버 리소스에 맞게 max_length를 조정해야 합니다. 마케팅 팀에서 1000건의 데이터로 테스트했습니다.

단건으로 처리하면 50초 걸리던 작업이 배치로 처리하니 0.5초 만에 끝났습니다. "100배나 빨라졌어요!"

실전 팁

💡 - 배치 크기 제한은 서버 메모리와 타임아웃 설정을 고려하여 결정하세요

• 부분 실패 처리 전략을 미리 정의하고 문서화해두세요

---

6. 에러 처리와 검증

서비스가 안정화되자 김개발 씨가 안심하고 있었는데, 어느 날 모니터링 시스템에서 알람이 울렸습니다. 일부 요청에서 500 에러가 발생하고 있었습니다.

"에러 메시지가 너무 불친절해요. 뭐가 문제인지 모르겠어요." 사용자 피드백이 들어왔습니다.

체계적인 에러 처리는 API의 신뢰성을 높이고 디버깅을 쉽게 만듭니다. 마치 병원에서 증상에 따라 다른 진료과로 안내하듯이, 에러 유형에 따라 적절한 상태 코드와 메시지를 반환해야 합니다.

사용자가 문제를 이해하고 스스로 해결할 수 있도록 안내하는 것이 목표입니다.

다음 코드를 살펴봅시다.

from fastapi import HTTPException, status
from fastapi.responses import JSONResponse
from pydantic import ValidationError

# 커스텀 예외 클래스
class ModelNotLoadedError(Exception):
    pass

class InvalidFeatureRangeError(Exception):
    def __init__(self, field: str, value: float, valid_range: tuple):
        self.field = field
        self.value = value
        self.valid_range = valid_range

# 전역 예외 핸들러
@app.exception_handler(ModelNotLoadedError)
async def model_not_loaded_handler(request, exc):
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={"error": "모델이 로딩되지 않았습니다", "retry_after": 30}
    )

@app.exception_handler(InvalidFeatureRangeError)
async def invalid_feature_handler(request, exc: InvalidFeatureRangeError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "입력값이 유효 범위를 벗어났습니다",
            "field": exc.field,
            "provided_value": exc.value,
            "valid_range": exc.valid_range
        }
    )

김개발 씨가 에러 로그를 분석해보니 다양한 원인이 있었습니다. 모델 파일 경로 오류, 입력값 범위 초과, 메모리 부족 등.

하지만 모든 에러가 똑같이 "Internal Server Error"로만 표시되고 있었습니다. "사용자 입장에서는 내 잘못인지 서버 잘못인지도 모르겠네요." 김개발 씨가 말했습니다.

에러 처리가 왜 중요할까요? 병원에 비유해 봅시다.

환자가 "아파요"라고만 말하면 의사도 어디를 치료해야 할지 모릅니다. "오른쪽 무릎이 계단 내려갈 때 아파요"라고 구체적으로 말해야 정확한 진단과 치료가 가능합니다.

API 에러 메시지도 마찬가지입니다. HTTP 상태 코드의 의미를 이해해야 합니다.

400은 클라이언트의 잘못된 요청, 422는 데이터 형식은 맞지만 값이 유효하지 않음, 500은 서버 내부 오류, 503은 일시적으로 서비스 불가를 의미합니다. 상태 코드만 봐도 어디서 문제가 생겼는지 알 수 있어야 합니다.

코드를 살펴보겠습니다. 커스텀 예외 클래스를 정의하면 특정 에러 상황을 명확하게 표현할 수 있습니다.

ModelNotLoadedError는 모델 로딩 실패를, InvalidFeatureRangeError는 입력값 범위 오류를 나타냅니다. @app.exception_handler() 데코레이터로 특정 예외를 전역적으로 처리합니다.

애플리케이션 어디서든 해당 예외가 발생하면 등록된 핸들러가 실행됩니다. JSONResponse로 일관된 형식의 에러 응답을 반환합니다.

실무에서는 어떤 에러들을 고려해야 할까요? 모델 서빙에서 자주 발생하는 에러로는 입력 데이터 형식 오류, 모델 로딩 실패, 예측 중 수치 오류(NaN, Infinity), 메모리 부족, 타임아웃 등이 있습니다.

각각에 대해 명확한 에러 코드와 메시지를 정의해두면 운영이 훨씬 수월해집니다. 에러 응답에는 해결 방법도 포함시키면 좋습니다.

단순히 "에러 발생"보다 "입력값이 유효 범위를 벗어났습니다. age 필드는 0-150 사이여야 합니다"라고 안내하면 사용자가 스스로 문제를 해결할 수 있습니다.

503 에러의 경우 retry_after를 포함시켜 언제 다시 시도하면 좋을지 알려주는 것도 좋은 방법입니다. 에러 처리를 개선한 후, 고객 문의가 크게 줄었습니다.

"에러 메시지가 친절해져서 직접 해결할 수 있게 됐어요." 사용자들의 만족도가 올라갔습니다.

실전 팁

💡 - 모든 에러 응답은 동일한 JSON 구조를 따르도록 일관성을 유지하세요

• 민감한 내부 정보(스택 트레이스 등)는 프로덕션 환경에서 노출하지 마세요

---

7. 비동기 처리와 성능 최적화

서비스 사용자가 늘어나자 새로운 문제가 생겼습니다. 동시 접속자가 많아지면 응답 시간이 느려지는 것입니다.

"100명이 동시에 요청하면 어떻게 되는 거죠?" 김개발 씨가 물었습니다. 박시니어 씨가 async/await에 대해 설명하기 시작했습니다.

비동기 처리는 여러 요청을 동시에 효율적으로 처리하는 기법입니다. 마치 혼자서 여러 냄비를 돌보는 요리사처럼, 하나의 작업이 끝나기를 기다리는 동안 다른 작업을 처리할 수 있습니다.

FastAPI는 비동기를 기본 지원하여 높은 동시성을 달성할 수 있습니다.

다음 코드를 살펴봅시다.

import asyncio
from concurrent.futures import ThreadPoolExecutor

# CPU 바운드 작업을 위한 스레드 풀
executor = ThreadPoolExecutor(max_workers=4)

def sync_predict(features):
    """동기 예측 함수 (CPU 집약적 작업)"""
    model = ml_models["churn_model"]
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features)[0][1]
    return prediction, probability

@app.post("/predict/async")
async def predict_async(request: PredictionRequest):
    features = np.array([[request.age, request.income, request.tenure]])

    # CPU 바운드 작업을 스레드 풀에서 실행
    loop = asyncio.get_event_loop()
    prediction, probability = await loop.run_in_executor(
        executor,
        sync_predict,
        features
    )

    return PredictionResponse(
        prediction=int(prediction),
        probability=float(probability),
        model_version=ml_models["version"]
    )

김개발 씨가 부하 테스트를 돌려봤습니다. 동시 접속자 100명 기준으로 평균 응답 시간이 2초를 넘어갔습니다.

서버 CPU 사용률을 보니 한 코어만 100%이고 나머지는 놀고 있었습니다. "Python의 GIL 때문에 그래요." 박시니어 씨가 설명했습니다.

GIL(Global Interpreter Lock)이란 무엇일까요? Python은 한 번에 하나의 스레드만 Python 코드를 실행할 수 있습니다.

이것이 GIL입니다. 마치 좁은 문 하나로만 통과할 수 있는 것처럼, 아무리 많은 스레드가 있어도 한 번에 하나씩만 일을 할 수 있습니다.

하지만 방법이 있습니다. async/await는 I/O 바운드 작업(네트워크, 파일 읽기 등)에 효과적입니다.

데이터베이스 응답을 기다리는 동안 다른 요청을 처리할 수 있습니다. 그러나 ML 예측처럼 CPU 바운드 작업에는 ThreadPoolExecutor가 더 적합합니다.

코드를 분석해봅시다. ThreadPoolExecutor는 별도의 스레드 풀을 생성합니다.

max_workers=4는 최대 4개의 스레드를 사용한다는 의미입니다. 보통 CPU 코어 수에 맞추거나 약간 적게 설정합니다.

**run_in_executor()**는 동기 함수를 비동기적으로 실행합니다. 메인 이벤트 루프를 블로킹하지 않고 스레드 풀에서 작업을 수행합니다.

await를 사용하면 결과가 준비될 때까지 다른 요청을 처리할 수 있습니다. 언제 어떤 방식을 써야 할까요?

데이터베이스 쿼리, 외부 API 호출 같은 I/O 작업은 async/await만으로 충분합니다. ML 예측, 이미지 처리 같은 CPU 작업은 ThreadPoolExecutor나 ProcessPoolExecutor를 사용합니다.

둘을 적절히 조합하는 것이 최선입니다. 프로덕션 환경에서는 더 고려할 것이 있습니다.

Gunicorn이나 Uvicorn의 worker 수를 조절하여 여러 프로세스로 요청을 분산할 수 있습니다. 또한 Redis 같은 캐시를 도입하면 반복 요청에 대한 응답 속도를 크게 높일 수 있습니다.

비동기 처리를 적용한 후 동시 접속자 100명 기준 평균 응답 시간이 200ms로 줄었습니다. "10배나 빨라졌네요!" 김개발 씨가 놀라워했습니다.

실전 팁

💡 - CPU 바운드 작업에는 ThreadPoolExecutor, I/O 바운드 작업에는 async/await를 사용하세요

• worker 수는 (CPU 코어 수 x 2) + 1을 기본으로 시작하여 조절하세요

---

8. API 문서화와 메타데이터

API가 완성되자 프론트엔드 팀과 외부 파트너사에서 연동 문의가 쏟아졌습니다. "API 사용법을 어디서 확인하나요?" "각 필드가 무슨 의미인가요?" 김개발 씨는 매일 같은 질문에 답하느라 정작 개발에 집중할 수가 없었습니다.

API 문서화는 개발자 경험(DX)의 핵심입니다. 마치 가전제품의 사용 설명서처럼, 잘 정리된 문서가 있으면 사용자가 스스로 API를 이해하고 활용할 수 있습니다.

FastAPI는 OpenAPI 스펙 기반의 문서를 자동 생성하며, 추가 메타데이터로 더욱 풍부하게 만들 수 있습니다.

다음 코드를 살펴봅시다.

from fastapi import FastAPI, Query, Path

app = FastAPI(
    title="고객 이탈 예측 API",
    description="""
    ## 소개
    머신러닝 기반 고객 이탈 예측 서비스입니다.

    ## 기능
    * 단건 예측: 개별 고객의 이탈 확률 예측
    * 배치 예측: 다수 고객 일괄 예측

    ## 연락처
    문의사항은 ml-team@company.com으로 연락주세요.
    """,
    version="2.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# 엔드포인트에 상세 설명 추가
@app.post(
    "/predict",
    summary="단건 이탈 예측",
    description="개별 고객 정보를 받아 이탈 확률을 예측합니다",
    response_description="예측 결과와 확률",
    tags=["예측"]
)
async def predict(request: PredictionRequest):
    ...

김개발 씨가 하루에 받는 질문이 줄어들지 않았습니다. 문서가 있긴 한데, 너무 기본적인 내용만 있어서 실제 사용법을 알기 어려웠던 것입니다.

"문서에 예제랑 설명을 더 추가해야겠어요." 김개발 씨가 결심했습니다. 좋은 API 문서란 무엇일까요?

사용 설명서에 비유해 봅시다. 최고의 사용 설명서는 그림과 예시가 풍부하고, 목적별로 찾기 쉽게 정리되어 있습니다.

API 문서도 마찬가지입니다. 각 엔드포인트의 용도, 입력 형식, 예상 응답, 에러 케이스가 명확해야 합니다.

FastAPI는 OpenAPI(구 Swagger) 스펙을 자동 생성합니다. 코드에서 타입 힌트와 Pydantic 모델을 분석하여 API 스펙을 만들어냅니다.

/docs로 접속하면 Swagger UI가, /redoc으로 접속하면 ReDoc이 표시됩니다. 둘 다 같은 스펙을 다른 UI로 보여주는 것입니다.

메타데이터를 추가하면 문서가 더 풍부해집니다. FastAPI() 생성자에 title, description, version을 전달하면 API 전체에 대한 설명이 추가됩니다.

description에는 마크다운 문법을 사용할 수 있어서 제목, 목록, 링크 등을 자유롭게 작성할 수 있습니다. 각 엔드포인트에도 상세 정보를 추가할 수 있습니다.

summary는 짧은 한 줄 설명, description은 상세 설명입니다. tags로 엔드포인트를 그룹화하면 문서가 체계적으로 정리됩니다.

예를 들어 "예측", "관리", "모니터링" 등의 태그로 분류할 수 있습니다. Pydantic 모델의 Field 설명도 문서에 반영됩니다.

앞서 작성한 description 파라미터가 API 문서의 필드 설명으로 표시됩니다. 또한 Config 클래스에 schema_extra를 정의하면 예제 값을 보여줄 수 있습니다.

문서를 개선한 후, 질문이 80% 줄었습니다. "문서 보고 직접 테스트해봤는데 바로 됐어요!" 프론트엔드 팀에서 긍정적인 피드백이 왔습니다.

실전 팁

💡 - 태그를 활용하여 관련 엔드포인트를 그룹화하면 문서 탐색이 쉬워집니다

• Config.schema_extra에 현실적인 예제 값을 제공하면 사용자 이해도가 높아집니다

---

9. 헬스체크와 모니터링

어느 날 새벽, 장애 알람이 울렸습니다. 서버는 살아 있는데 모델이 응답을 안 하는 상황이었습니다.

"서버가 떠 있으면 정상인 줄 알았는데..." 김개발 씨가 당황했습니다. 박시니어 씨가 말했습니다.

"헬스체크가 제대로 안 되어 있네요."

헬스체크는 서비스가 정상적으로 동작하는지 확인하는 엔드포인트입니다. 마치 병원의 정기 검진처럼, 시스템의 각 구성 요소가 제대로 작동하는지 주기적으로 확인합니다.

단순히 서버가 응답하는지뿐 아니라, 모델이 로딩되었는지, 의존 서비스에 연결되었는지까지 점검해야 합니다.

다음 코드를 살펴봅시다.

from datetime import datetime
from typing import Dict

@app.get("/health")
async def health_check() -> Dict:
    """기본 헬스체크 - 서버 응답 확인"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

@app.get("/health/ready")
async def readiness_check() -> Dict:
    """준비 상태 체크 - 모델 로딩 확인"""
    if "churn_model" not in ml_models:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="모델이 아직 로딩되지 않았습니다"
        )

    # 간단한 예측 테스트
    test_input = np.array([[30, 50000, 12]])
    try:
        ml_models["churn_model"].predict(test_input)
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"모델 예측 실패: {str(e)}"
        )

    return {
        "status": "ready",
        "model_version": ml_models.get("version", "unknown"),
        "timestamp": datetime.now().isoformat()
    }

장애 원인을 분석해보니, 메모리 부족으로 모델이 언로드된 상태였습니다. 하지만 기존 헬스체크는 서버 프로세스만 확인했기 때문에 "정상"으로 표시되고 있었습니다.

"헬스체크에도 종류가 있어요." 박시니어 씨가 설명했습니다. Liveness와 Readiness의 차이가 뭘까요?

사람에 비유하면 이해가 쉽습니다. Liveness는 "살아 있나요?"입니다.

심장이 뛰는지 확인하는 것입니다. Readiness는 "일할 수 있나요?"입니다.

살아 있어도 아프면 일을 못 합니다. Kubernetes 같은 오케스트레이터는 이 두 체크를 구분해서 사용합니다.

ML 서비스에서 Readiness 체크는 특히 중요합니다. 서버가 시작되어도 모델 로딩에 시간이 걸립니다.

50MB 모델은 수 초, 수 GB 모델은 수 분이 걸릴 수 있습니다. 모델이 로딩되기 전에 트래픽이 들어오면 에러가 발생합니다.

Readiness 체크가 통과해야만 로드밸런서가 트래픽을 보내도록 설정합니다. 코드를 살펴봅시다.

/health는 단순히 서버가 응답하는지 확인합니다. 가장 빠르게 응답해야 하므로 아무 로직도 없습니다.

/health/ready는 모델이 로딩되었는지, 실제 예측이 가능한지까지 확인합니다. 실제 예측을 수행해보는 것이 핵심입니다.

모델 객체가 존재해도 내부적으로 손상되었을 수 있습니다. 더미 데이터로 예측을 시도하여 정상 동작을 검증합니다.

이 과정에서 에러가 발생하면 503을 반환합니다. 실무에서는 더 다양한 항목을 체크합니다.

데이터베이스 연결, Redis 연결, 외부 API 연결 등 의존 서비스의 상태도 확인합니다. 각 항목의 상태를 개별적으로 반환하면 어디서 문제가 생겼는지 빠르게 파악할 수 있습니다.

헬스체크를 개선한 후, 모니터링 시스템이 문제를 즉시 감지하게 되었습니다. "모델 로딩 실패하면 바로 알람이 오니까 새벽에 깨지 않아도 되겠네요." 김개발 씨가 안심했습니다.

실전 팁

💡 - Kubernetes 환경에서는 livenessProbe와 readinessProbe를 분리하여 설정하세요

• 헬스체크 응답에 버전 정보를 포함시키면 배포 상태 확인이 쉬워집니다

---

10. 배포와 서버 실행

모든 기능이 완성되어 드디어 프로덕션 배포 날이 다가왔습니다. "개발 환경에서는 uvicorn --reload로 실행했는데, 실제 서비스도 이렇게 하면 되나요?" 김개발 씨의 질문에 박시니어 씨가 고개를 저었습니다.

"프로덕션에서는 다르게 해야 해요."

프로덕션 배포는 개발 환경과 다른 설정이 필요합니다. 마치 연습 경기와 본 경기가 다르듯이, 안정성과 성능을 위한 최적화가 필수입니다.

Gunicorn과 Uvicorn을 조합하고, 환경 변수로 설정을 관리하며, 적절한 worker 수를 설정해야 합니다.

다음 코드를 살펴봅시다.

# config.py - 환경별 설정 관리
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "ML Model API"
    debug: bool = False
    model_path: str = "models/churn_model.pkl"
    max_batch_size: int = 1000
    workers: int = 4

    class Config:
        env_file = ".env"

settings = Settings()

# main.py - 설정 적용
app = FastAPI(
    title=settings.app_name,
    debug=settings.debug
)

# 실행 명령어 (프로덕션)
# gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

# Dockerfile
"""
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]
"""

개발 환경에서 --reload 옵션은 코드 변경을 자동 감지하여 서버를 재시작합니다. 편리하지만 성능 오버헤드가 있습니다.

프로덕션에서는 절대 사용하면 안 됩니다. "프로덕션 배포에는 뭐가 다른가요?" 김개발 씨가 물었습니다.

가장 큰 차이는 멀티 워커입니다. Uvicorn 단독으로는 단일 프로세스로 실행됩니다.

멀티코어 CPU를 활용하려면 여러 워커 프로세스가 필요합니다. Gunicorn이 워커들을 관리하고, 각 워커는 UvicornWorker를 사용하여 비동기 처리를 합니다.

pydantic_settings로 환경별 설정을 관리합니다. 개발, 스테이징, 프로덕션 환경마다 설정이 다릅니다.

모델 경로, 디버그 모드, 워커 수 등을 환경 변수로 관리하면 코드 수정 없이 환경만 바꿔서 배포할 수 있습니다. .env 파일에 설정을 저장하고 gitignore에 추가합니다.

워커 수는 어떻게 정해야 할까요? 일반적인 공식은 (CPU 코어 수 x 2) + 1입니다.

4코어 서버라면 9개의 워커가 적당합니다. 하지만 ML 모델은 메모리를 많이 사용하므로 메모리도 고려해야 합니다.

모델이 500MB라면 워커 4개가 2GB를 사용합니다. Docker 컨테이너로 배포하면 환경 일관성이 보장됩니다.

Dockerfile에서 기본 이미지, 의존성 설치, 실행 명령을 정의합니다. slim 이미지를 사용하면 이미지 크기를 줄일 수 있습니다.

모델 파일은 이미지에 포함시키거나 볼륨으로 마운트할 수 있습니다. 실무에서는 추가로 고려할 것들이 있습니다.

SSL/TLS 인증서 설정, 로드밸런서 연결, 로그 수집 시스템 연동, 메트릭 모니터링 설정 등이 필요합니다. 클라우드 서비스를 사용한다면 AWS ECS, GCP Cloud Run, Azure Container Apps 등을 활용할 수 있습니다.

첫 프로덕션 배포가 완료되었습니다. "드디어 실서비스에 올라갔네요!" 김개발 씨의 ML 모델이 전사에서 사용되기 시작했습니다.

실전 팁

💡 - 환경 변수와 시크릿은 절대 코드에 하드코딩하지 마세요

• 프로덕션 로그는 JSON 형식으로 출력하면 로그 분석 시스템과 연동이 쉽습니다

---

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