본문 바로가기
C.W.K.
Stream
Lesson 04 of 05 · published

Lightweight Deployment

~30 min · deployment, fastapi

Level 0Scout
0 XP0/48 lessons0/11 achievements
0/120 XP to next level120 XP to go0% complete

필요한 규모에서 시작해

고전 기계학습 서비스 대부분은 로드 밸런서 뒤의 작은 FastAPI 서버나 정기 배치 작업으로 충분해. 모델이 수백 MB이고 예측이 몇 밀리초이며 담당자가 둘뿐인데 처음부터 Kubernetes와 복잡한 특성 저장소를 들이면 운영 부담이 모델 가치보다 커질 수 있어. 트래픽, 응답 시간, 격리, 규정 준수가 실제로 요구할 때만 다음 층을 추가해.

실시간이 필요한지도 먼저 물어

결과를 하루 한 번 고객 목록으로 쓰는 업무라면 배치 점수 계산이 더 단순하고 재현하기 쉬워. 사용자의 즉시 행동에 반응해야 할 때만 온라인 API가 필요해. 배치와 실시간의 지연 요구, 자료 최신성, 실패 시 영향, 재처리 방법을 비교해 제공 형태를 골라.

점수 API의 최소 계약

원시 행을 받아 입력을 검증하고 확률, 적용한 문턱, 최종 결정, 모델과 정책 버전을 반환하는 /score 같은 경로가 필요해. 요청 식별자와 예측 시각을 포함하고 잘못된 스키마에는 4xx 오류와 안전한 설명을 내. 여러 행을 받는 배치 경로가 필요하다면 최대 크기와 부분 실패 동작을 정해.

건강 상태와 준비 상태를 나눠

프로세스가 살아 있는지만 보는 liveness와 결과물을 불러 실제 예측을 할 준비가 됐는지 보는 readiness를 구분해. 버전 경로는 모델 ID, 자료·코드 버전, 문턱, 빌드 시각을 반환해 운영자가 현재 상태를 확인하게 해. 건강 검사가 비싼 외부 의존성을 매번 호출해 장애를 키우지 않게 하고.

기록은 입력 전체보다 책임에 맞게

요청 ID, 모델 점수, 결정, 버전, 응답 시간, 오류를 구조화해 남겨야 변화와 성능을 나중에 계산할 수 있어. 하지만 개인정보와 원문을 무제한 기록하면 더 큰 위험이 돼. 필요한 특성의 해시나 감사용 최소 필드만 남기고 접근 권한, 보존 기간, 삭제 요청 경로를 정해.

실패 동작을 코드로 정해

시간 초과, 결과물 로드 실패, 미지 범주, 입력 누락 때 요청을 거부할지 이전 모델이나 명시적 규칙으로 돌아갈지 결정해. 서킷 브레이커가 연속 실패를 감지해 안전한 경로로 보내고 경보를 울리게 해. 대체 경로도 같은 정도로 시험하고 버전을 기록해야 ‘안전장치’가 오래된 오류가 되지 않아.

먼저 그림자로 실제 요청을 받아

새 모델에 실제 요청 일부를 복제해 예측은 기록하되 사용자 결정은 기존 모델이 내리게 해. 입력 오류, 응답 시간, 점수 분포, 양성 판정률을 기존 버전과 비교해 오프라인에서 보지 못한 차이를 찾을 수 있어. 기간은 ‘일주일’ 같은 관습보다 주말·정산일 등 중요한 운영 주기를 포함하도록 정해.

점진 배포와 즉시 복구

그림자를 통과하면 작은 트래픽에서 새 모델의 기술·사업·안전 지표를 보고 단계적으로 늘려. 승격 조건과 자동 중단 문턱을 미리 적고 이전 불변 결과물과 정책을 즉시 다시 가리킬 수 있게 해. 배포가 성공했다는 건 서버가 뜬 게 아니라 실제 요청, 기록, 경보, 복구가 모두 검증됐다는 뜻이야.

Code

scikit-learn 파이프라인용 최소 FastAPI 서버·python
import joblib, json
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
pipe = joblib.load("artifacts/churn_v1.joblib")
meta = json.load(open("artifacts/churn_v1.json"))

class Row(BaseModel):
    payload: dict

@app.get("/health")
def health():
    return {"ok": True, "version": meta["version"]}

@app.post("/score")
def score(row: Row):
    X = pd.DataFrame([row.payload])[meta["features"]]
    p = float(pipe.predict_proba(X)[0, 1])
    return {"prob": p, "label": int(p >= meta["threshold"]), "version": meta["version"]}
구조화된 예측 기록·python
import json, datetime

def log_prediction(payload, prob, label, version, latency_ms):
    print(json.dumps({
        "ts": datetime.datetime.utcnow().isoformat(),
        "version": version,
        "latency_ms": latency_ms,
        "prob": prob,
        "label": label,
        "payload": payload,
    }))

External links

Exercise

저장한 파이프라인을 작은 FastAPI 앱으로 감싸 curl로 원시 행을 /score에 보내. 응답의 확률, 결정, 모델·정책 버전과 스키마 오류, readiness, 구조화 기록, 대체 경로가 계약대로 동작하는지 검증해.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.