Module 1

이 과정에서 만들 것

전체 과정 흐름

이 전체 사이클을 두 서비스 모두에 적용합니다

MLOps란?

• ML + DevOps = 머신러닝 모델의 개발-배포-운영 자동화
• 모델 만드는 것은 전체의 10~20%
• 나머지 80%가 운영: 배포, 모니터링, 재학습, 버전 관리

실습 환경 체크리스트

Miniconda 설치

1. https://docs.conda.io/en/latest/
   miniconda.html 에서 다운로드
2. 설치 시 "Add to PATH" 체크
3. Anaconda Prompt 실행
4. conda --version 확인
              

# 다운로드 & 설치
wget https://repo.anaconda.com/miniconda/\
Miniconda3-latest-Linux-x86_64.sh
bash Miniconda3-latest-Linux-x86_64.sh

# 확인
conda --version
              

실습 환경 세팅

# 1. conda 환경 생성 (Python 3.10)
conda create -n mlops python=3.10 -y
conda activate mlops

# 2. 레포 클론
git clone https://github.com/blockenters/mlops-edu.git
cd mlops-edu

# 3. 기본 패키지 설치
pip install notebook fastapi uvicorn scikit-learn xgboost pandas joblib

# 4. 확인
python --version            # 3.10.x
python -c "import fastapi; print(fastapi.__version__)"
          

프로젝트 구조 (최종 목표)

mlops-course/
├── service-a-loan/          # 서비스 A: 대출 심사
│   ├── app/
│   │   ├── main.py          # FastAPI 엔트리포인트
│   │   ├── model.py         # 모델 로딩 & 예측
│   │   └── schemas.py       # 요청/응답 스키마
│   ├── models/
│   │   └── loan_pipeline.pkl   # 학습된 모델
│   ├── data/
│   │   └── loan_data.csv    # 학습 데이터
│   ├── train.py             # 모델 학습 스크립트
│   ├── Dockerfile
│   ├── requirements.txt
│   └── tests/
├── service-b-review/        # 서비스 B: 리뷰 분석
│   ├── app/
│   │   ├── main.py
│   │   ├── gemini_client.py # Gemini API 연동
│   │   └── schemas.py
│   ├── Dockerfile
│   ├── requirements.txt
│   └── tests/
├── .github/workflows/       # CI/CD
└── README.md
          

서비스 A 비즈니스 시나리오

• 실제 핀테크/은행에서 사용하는 패턴
• 심사 담당자의 업무 부담 감소
• 일관된 기준으로 빠른 1차 심사

입력과 출력

{
  "approved": true,
  "probability": 0.87,
  "risk_grade": "A"
}
              

데이터 살펴보기

import pandas as pd

df = pd.read_csv("data/loan_data.csv")
print(df.shape)        # (1500, 14)
print(df.head())
print(df.describe())
print(df['승인여부'].value_counts())
          

모델 학습 흐름 (train.py)

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from xgboost import XGBClassifier
import joblib

# 1. 데이터 로드
df = pd.read_csv("data/loan_data.csv")

# 2. 범주형 → 숫자 변환
label_cols = ["성별", "주거형태", "대출목적", "상환방식"]
encoders = {}
for col in label_cols:
    le = LabelEncoder()
    df[col] = le.fit_transform(df[col])
    encoders[col] = le

feature_cols = [
    "나이", "성별", "연소득", "근속연수", "주거형태",
    "신용점수", "기존대출건수", "연간카드사용액", "부채비율",
    "대출신청액", "대출목적", "상환방식", "대출기간",
]
X = df[feature_cols].copy()
y = df["승인여부"]

# 3. 분할
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# 4. 파이프라인 (전처리 + XGBoost)
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", XGBClassifier(
        n_estimators=100, max_depth=6,
        learning_rate=0.1, random_state=42,
        eval_metric="logloss"
    ))
])

# 5. 학습 & 저장
pipeline.fit(X_train, y_train)
print(f"정확도: {pipeline.score(X_test, y_test):.4f}")

joblib.dump(pipeline, "models/loan_pipeline.pkl")
joblib.dump(encoders, "models/label_encoders.pkl")
joblib.dump(feature_cols, "models/feature_names.pkl")
print("모델 저장 완료!")
          

왜 Pipeline으로 저장하는가?

# 학습 시
scaler.fit(X_train)
X_scaled = scaler.transform(X_train)
model.fit(X_scaled, y_train)

# 저장
joblib.dump(model, "model.pkl")
# scaler는? encoder는?
              

# 파이프라인으로 묶기
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("model", XGBClassifier())
])
pipeline.fit(X_train, y_train)

# 통째로 저장
joblib.dump(pipeline, "model.pkl")
              

서비스 A 디렉토리 구조

service-a-loan/
├── app/
│   ├── __init__.py
│   ├── main.py          ← FastAPI 앱, 라우터 등록
│   ├── model.py         ← 모델 로딩, 예측 함수
│   └── schemas.py       ← Pydantic 요청/응답 스키마
├── models/
│   └── loan_pipeline.pkl   ← 학습된 파이프라인
├── data/
│   └── loan_data.csv    ← 학습 데이터
├── train.py             ← 모델 학습 스크립트
├── requirements.txt
└── tests/
    └── test_api.py
          

왜 FastAPI인가?

• 빠른 개발: 타입 힌트 기반 자동 문서화
• Swagger UI: 별도 설정 없이 API 문서 자동 생성
• 비동기 지원: async/await로 고성능 서빙
• Pydantic: 요청/응답 데이터 자동 검증
• 현업 표준: ML 서빙에서 Flask를 빠르게 대체 중

Hello World: 최소 FastAPI

# app/main.py
from fastapi import FastAPI

app = FastAPI(
    title="대출 심사 예측 API",
    description="고객 정보를 기반으로 대출 승인 여부를 예측합니다",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "대출 심사 예측 API가 실행 중입니다"}

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

서버 실행 & Swagger 확인

# 서버 실행
cd service-a-loan
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 브라우저에서 확인
# API 문서:  http://localhost:8000/docs
# 헬스체크:  http://localhost:8000/health
          

Swagger UI 활용

• /docs → Swagger UI (인터랙티브 테스트)
• /redoc → ReDoc (읽기 전용 문서)
• /openapi.json → OpenAPI 스펙 (JSON)

Path Parameter & Query Parameter

# Path Parameter
@app.get("/customers/{customer_id}")
def get_customer(customer_id: int):
    return {"customer_id": customer_id}

# Query Parameter
@app.get("/search")
def search(keyword: str, limit: int = 10):
    return {"keyword": keyword, "limit": limit}
          

POST 요청 기초

from pydantic import BaseModel

class LoanRequest(BaseModel):
    annual_income: float
    loan_amount: float

class LoanResponse(BaseModel):
    approved: bool
    message: str

@app.post("/predict", response_model=LoanResponse)
def predict(request: LoanRequest):
    # 임시 로직 (다음 차시에서 실제 모델 연동)
    is_approved = request.annual_income > request.loan_amount
    return LoanResponse(
        approved=is_approved,
        message="승인" if is_approved else "거절"
    )
          

요청/응답 스키마 설계

# app/schemas.py
from pydantic import BaseModel, Field

class LoanRequest(BaseModel):
    age: int = Field(..., ge=19, le=100, description="나이")
    gender: str = Field(..., description="성별 (남/여)")
    annual_income: float = Field(..., ge=0, description="연소득(만원)")
    employment_years: int = Field(..., ge=0, le=50, description="근속연수")
    housing_type: str = Field(..., description="주거형태 (자가/전세/월세)")
    credit_score: int = Field(..., ge=300, le=900, description="신용점수")
    existing_loan_count: int = Field(..., ge=0, description="기존대출건수")
    annual_card_usage: float = Field(..., ge=0, description="연간카드사용액(만원)")
    debt_ratio: float = Field(..., ge=0, le=100, description="부채비율(%)")
    loan_amount: float = Field(..., ge=100, description="대출신청액(만원)")
    loan_purpose: str = Field(..., description="대출목적")
    repayment_method: str = Field(..., description="상환방식")
    loan_period: int = Field(..., ge=6, le=360, description="대출기간(개월)")
          

응답 스키마

# app/schemas.py (계속)
class LoanResponse(BaseModel):
    approved: bool = Field(..., description="승인 여부")
    probability: float = Field(..., ge=0.0, le=1.0, description="승인 확률(0~1)")
    risk_grade: str = Field(..., description="리스크 등급 (A/B/C/D)")
          

예측 엔드포인트 구현

# app/main.py
from fastapi import FastAPI, HTTPException
from app.schemas import LoanRequest, LoanResponse
from app.model import LoanModel

app = FastAPI(title="대출 심사 예측 API", version="1.0.0")

loan_model = LoanModel()

@app.post("/predict", response_model=LoanResponse)
def predict(request: LoanRequest):
    try:
        result = loan_model.predict(request)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

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

cURL / Python으로 테스트

# cURL
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{
    "age": 35, "gender": "남", "annual_income": 5000.0,
    "employment_years": 5, "housing_type": "자가", "credit_score": 720,
    "existing_loan_count": 2, "annual_card_usage": 2000.0,
    "debt_ratio": 35.5, "loan_amount": 3000.0,
    "loan_purpose": "주택구입", "repayment_method": "원리금균등",
    "loan_period": 36
  }'
          

# Python requests
import requests

response = requests.post("http://localhost:8000/predict", json={
    "age": 35, "gender": "남", "annual_income": 5000.0,
    "employment_years": 5, "housing_type": "자가", "credit_score": 720,
    "existing_loan_count": 2, "annual_card_usage": 2000.0,
    "debt_ratio": 35.5, "loan_amount": 3000.0,
    "loan_purpose": "주택구입", "repayment_method": "원리금균등",
    "loan_period": 36
})
print(response.json())
          

입력 검증 확인해보기

# 잘못된 입력: credit_score가 범위 밖
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"age": 35, "credit_score": 1000, ...}'
          

{
  "detail": [
    {
      "loc": ["body", "credit_score"],
      "msg": "Input should be less than or equal to 900",
      "type": "less_than_equal"
    }
  ]
}
          

모델 로딩 클래스

# app/model.py
import joblib
import pandas as pd
from pathlib import Path
from app.schemas import LoanRequest, LoanResponse

class LoanModel:
    def __init__(self, model_path: str = "models/loan_pipeline.pkl",
                       encoder_path: str = "models/label_encoders.pkl"):
        self.pipeline = None
        self.encoders = None
        self.is_loaded = False
        self._load_model(model_path, encoder_path)

    def _load_model(self, model_path, encoder_path):
        if Path(model_path).exists() and Path(encoder_path).exists():
            self.pipeline = joblib.load(model_path)
            self.encoders = joblib.load(encoder_path)
            self.is_loaded = True
            print("모델 & 인코더 로드 완료")
        else:
            print("모델 또는 인코더 파일 없음")
          

예측 메서드

# app/model.py (계속)
    # ── 핵심: API 필드명 → CSV 컬럼명 매핑 ──
    # API는 영어 필드명(age, gender 등)을 사용하지만,
    # 학습 데이터(CSV)와 LabelEncoder는 한글 컬럼명을 사용한다.
    # 이 매핑으로 두 세계를 연결한다.
    FIELD_TO_COLUMN = {
        "age": "나이", "gender": "성별",
        "annual_income": "연소득", "employment_years": "근속연수",
        "housing_type": "주거형태", "credit_score": "신용점수",
        "existing_loan_count": "기존대출건수",
        "annual_card_usage": "연간카드사용액",
        "debt_ratio": "부채비율", "loan_amount": "대출신청액",
        "loan_purpose": "대출목적", "repayment_method": "상환방식",
        "loan_period": "대출기간",
    }

    def predict(self, data: dict) -> dict:
        if self.pipeline is None:
            raise RuntimeError("모델이 로드되지 않았습니다")

        # 1. 영어 → 한글 매핑
        mapped = {self.FIELD_TO_COLUMN.get(k, k): v
                  for k, v in data.items()}

        # 2. DataFrame 변환 + 피처 순서 맞추기
        df = pd.DataFrame([mapped])[self.feature_names]

        # 3. 범주형 인코딩 (학습 시 저장한 인코더 사용)
        for col, encoder in self.label_encoders.items():
            if col in df.columns:
                df[col] = encoder.transform(df[col])

        # 4. 예측
        probability = float(self.pipeline.predict_proba(df)[:, 1][0])
        approved = probability >= self.threshold

        # 5. 리스크 등급 (A/B/C/D)
        if probability >= 0.75:   risk_grade = "A"
        elif probability >= 0.50: risk_grade = "B"
        elif probability >= 0.25: risk_grade = "C"
        else:                     risk_grade = "D"

        return {
            "approved": approved,
            "probability": round(probability, 4),
            "risk_grade": risk_grade,
        }
          

모델 로딩 패턴: 앱 시작 시 1회

@app.post("/predict")
def predict(req: LoanRequest):
    # 매 요청마다 모델 로딩!
    model = joblib.load("model.pkl")
    return model.predict(...)
              

# 앱 시작 시 1번만 로딩
loan_model = LoanModel()

@app.post("/predict")
def predict(req: LoanRequest):
    return loan_model.predict(req)
              

Lifespan 이벤트 (고급)

from contextlib import asynccontextmanager
from fastapi import FastAPI

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 시작 시 실행
    ml_models["loan"] = LoanModel()
    print("모델 로딩 완료")
    yield
    # 종료 시 실행
    ml_models.clear()
    print("리소스 정리 완료")

app = FastAPI(lifespan=lifespan)

@app.post("/predict")
def predict(request: LoanRequest):
    return ml_models["loan"].predict(request)
          

모델 학습 실행해보기

서비스 B 목표

{
  "review_text": "배송이 너무 느려서
   실망했어요. 상품 품질은
   괜찮은데 포장이 엉망..."
}
              

{
  "sentiment": "부정",
  "category": "배송",
  "summary": "배송 지연 및 포장
   불량에 대한 불만",
  "confidence": 0.85
}
              

Gemini API 준비

# 1. Google AI Studio에서 API 키 발급
#    https://aistudio.google.com/apikey

# 2. 환경변수 설정
export GEMINI_API_KEY="your-api-key-here"

# 3. 패키지 설치
pip install google-genai
          

참고: Gemini 클라이언트 코드

# app/gemini_client.py
import os
import json
from google import genai
from google.genai import types

class ReviewAnalyzer:
    def __init__(self):
        api_key = os.environ.get("GEMINI_API_KEY")
        if not api_key:
            raise ValueError("GEMINI_API_KEY 환경변수를 설정하세요")
        self.client = genai.Client(api_key=api_key)
        self.model = "gemini-2.5-flash"

    def analyze(self, review_text: str) -> dict:
        prompt = f"""다음 고객 리뷰를 분석해주세요.

리뷰: "{review_text}"

sentiment(긍정/부정/중립), category(배송/품질/가격/CS),
summary(한줄 요약), confidence(0.0~1.0)를 분석하세요."""
        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema={
                    "type": "object",
                    "properties": {
                        "sentiment": {"type": "string"},
                        "category": {"type": "string"},
                        "summary": {"type": "string"},
                        "confidence": {"type": "number"}
                    },
                    "required": ["sentiment", "category", "summary", "confidence"]
                }
            )
        )
        return json.loads(response.text)
          

자율실습 과제

테스트용 샘플 리뷰

미니 프로젝트 목표

1. 코드 수정: 새 엔드포인트 추가 또는 기존 수정
2. 로컬 실행: uvicorn으로 서버 실행
3. 테스트: Swagger / cURL / Python으로 검증
4. 반복: 수정 → 실행 → 테스트 사이클 체험

서비스 A 추가 과제

서비스 B 추가 과제

체크리스트

Module 1 정리

• MLOps = ML 모델의 개발-배포-운영 자동화
• FastAPI로 ML 모델을 REST API로 서빙
• Pydantic으로 입력 검증 자동화
• Pipeline으로 전처리+모델 통합 저장
• 모듈 레벨 인스턴스로 모델 1회 로딩
• LLM 서비스는 프롬프트 관리 + 환경변수가 핵심