Colab 모델 저장과 배포 완벽 가이드
학습한 모델을 저장하고 실서비스에 적용하기까지
Google Colab에서 모델을 학습했다면, 다음 단계는 이를 안전하게 저장하고 필요한 곳에서 활용하는 것입니다. 이 글에서는 프레임워크별 저장 방법부터 실서비스 배포까지 전체 흐름을 다룹니다.
모델 저장하기
딥러닝 프레임워크마다 모델을 저장하는 방식이 다릅니다. 각 프레임워크의 권장 방법을 살펴보겠습니다.
PyTorch 모델 저장
PyTorch는 두 가지 저장 방식을 제공합니다. 가중치만 저장하는 방식이 더 가볍고 이식성이 좋아 권장됩니다.
import torch
# 방법 1: 가중치만 저장 (권장)
torch.save(model.state_dict(), 'model_weights.pt')
# 방법 2: 전체 모델 저장 (모델 구조 + 가중치)
torch.save(model, 'model_complete.pt')
# 방법 3: 추가 정보와 함께 저장
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'loss': loss,
'accuracy': accuracy
}, 'checkpoint.pt')TensorFlow/Keras 모델 저장
TensorFlow는 여러 형식으로 모델을 저장할 수 있습니다. SavedModel 형식이 프로덕션 환경에서 가장 널리 사용됩니다.
import tensorflow as tf
# 방법 1: HDF5 형식 (.h5)
model.save('model.h5')
# 방법 2: SavedModel 형식 (권장)
model.save('saved_model_dir')
# 방법 3: 가중치만 저장
model.save_weights('model_weights.h5')
# 방법 4: 체크포인트 형식
checkpoint = tf.train.Checkpoint(model=model)
checkpoint.save('./checkpoints/ckpt')Scikit-learn 모델 저장
전통적인 머신러닝 모델은 joblib을 사용하여 직렬화합니다. pickle보다 큰 numpy 배열을 효율적으로 처리합니다.
import joblib
import pickle
# joblib 사용 (권장)
joblib.dump(model, 'model.pkl')
# 압축 저장
joblib.dump(model, 'model.pkl', compress=3)
# pickle 사용 (대안)
with open('model.pkl', 'wb') as f:
pickle.dump(model, f)파일 내보내기
Colab에서 저장한 파일을 로컬 환경으로 가져오는 방법은 크게 두 가지입니다.
방법 1: 브라우저로 직접 다운로드
간단한 파일이나 즉시 다운로드가 필요한 경우 사용합니다.
from google.colab import files
# 단일 파일 다운로드
files.download('model_weights.pt')
# 여러 파일 다운로드
files.download('checkpoint.pt')
files.download('config.json')방법 2: Google Drive 연동 (권장)
파일을 영구적으로 보관하고 언제든 접근할 수 있어 가장 안정적인 방법입니다.
from google.colab import drive
# 1. Drive 마운트 (최초 1회 인증 필요)
drive.mount('/content/drive')
# 2. 모델 저장 경로 설정
save_path = '/content/drive/MyDrive/models/'
# 3. 디렉토리 생성 (없는 경우)
!mkdir -p {save_path}
# 4. 파일 복사
!cp model_weights.pt {save_path}
# 또는 직접 Drive에 저장
import torch
torch.save(
model.state_dict(),
'/content/drive/MyDrive/models/model_weights.pt'
)Drive 마운트 과정
처음 drive.mount()를 실행하면 Google 인증 페이지가 나타납니다. 계정을 선택하고 권한을 허용하면 /content/drive 경로에 드라이브가 연결됩니다.
압축하여 저장
여러 파일을 한 번에 전송하거나 용량을 줄이고 싶을 때 압축 파일을 생성합니다.
# 여러 파일을 zip으로 압축
!zip -r models.zip \
model_weights.pt \
config.json \
tokenizer.json
# Drive로 복사
!cp models.zip /content/drive/MyDrive/
# 또는 tar.gz 사용
!tar -czf models.tar.gz \
model_weights.pt \
config.json모델 불러오기
저장한 모델을 다시 불러와서 추론이나 추가 학습에 사용할 수 있습니다.
PyTorch 모델 로드
import torch
from model import MyModel
# 방법 1: state_dict 불러오기 (권장)
model = MyModel()
model.load_state_dict(
torch.load('model_weights.pt')
)
model.eval() # 추론 모드로 전환
# 방법 2: 전체 모델 불러오기
model = torch.load('model_complete.pt')
model.eval()
# 방법 3: 체크포인트 불러오기
checkpoint = torch.load('checkpoint.pt')
model = MyModel()
model.load_state_dict(checkpoint['model_state_dict'])
optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
epoch = checkpoint['epoch']
loss = checkpoint['loss']# GPU에서 저장 → CPU에서 로드
model.load_state_dict(
torch.load('model_weights.pt', map_location='cpu')
)
# CPU에서 저장 → GPU에서 로드
model.load_state_dict(
torch.load('model_weights.pt', map_location='cuda:0')
)TensorFlow 모델 로드
import tensorflow as tf
from tensorflow import keras
# HDF5 형식 로드
model = keras.models.load_model('model.h5')
# SavedModel 형식 로드
model = keras.models.load_model('saved_model_dir')
# 가중치만 로드
model = create_model() # 모델 구조 재생성
model.load_weights('model_weights.h5')
# 체크포인트 로드
checkpoint = tf.train.Checkpoint(model=model)
checkpoint.restore('./checkpoints/ckpt-1')Scikit-learn 모델 로드
import joblib
# joblib로 저장한 모델 로드
model = joblib.load('model.pkl')
# 예측 수행
predictions = model.predict(X_test)체크포인트 관리
장시간 학습 시 중간 결과를 주기적으로 저장하면 예기치 않은 세션 종료에 대비할 수 있습니다.
기본 체크포인트 전략
import torch
import os
# 체크포인트 저장 디렉토리
checkpoint_dir = '/content/drive/MyDrive/checkpoints'
os.makedirs(checkpoint_dir, exist_ok=True)
# 학습 루프
for epoch in range(num_epochs):
# 모델 학습
train_loss = train_epoch(model, train_loader, optimizer)
val_loss = validate(model, val_loader)
# 10 epoch마다 저장
if (epoch + 1) % 10 == 0:
checkpoint_path = f'{checkpoint_dir}/checkpoint_epoch_{epoch+1}.pt'
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'train_loss': train_loss,
'val_loss': val_loss
}, checkpoint_path)
print(f'Checkpoint saved at epoch {epoch+1}')최고 성능 모델만 저장
모든 체크포인트를 저장하면 용량이 커지므로, 검증 손실이 가장 낮은 모델만 유지할 수 있습니다.
best_val_loss = float('inf')
best_model_path = '/content/drive/MyDrive/best_model.pt'
for epoch in range(num_epochs):
train_loss = train_epoch(model, train_loader, optimizer)
val_loss = validate(model, val_loader)
# 검증 손실이 개선되었을 때만 저장
if val_loss < best_val_loss:
best_val_loss = val_loss
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'best_val_loss': best_val_loss
}, best_model_path)
print(f'New best model saved! Val loss: {val_loss:.4f}')체크포인트에서 학습 재개
# 저장된 체크포인트 확인
checkpoint_path = '/content/drive/MyDrive/checkpoints/checkpoint_epoch_50.pt'
if os.path.exists(checkpoint_path):
print("Resuming from checkpoint...")
checkpoint = torch.load(checkpoint_path)
model.load_state_dict(checkpoint['model_state_dict'])
optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
start_epoch = checkpoint['epoch'] + 1
print(f"Resuming from epoch {start_epoch}")
else:
print("Starting training from scratch")
start_epoch = 0
# 이어서 학습
for epoch in range(start_epoch, num_epochs):
train_epoch(model, train_loader, optimizer)Colab 세션 관리
Colab의 무료 사용자는 런타임 제한이 있으므로 주의가 필요합니다.
세션 제한 사항
| 조건 | 무료 사용자 | Colab Pro | Colab Pro+ |
|---|---|---|---|
| 최대 런타임 | 12시간 | 24시간 | 24시간 |
| 유휴 시간 초과 | 90분 | 90분 | 90분 |
| 브라우저 닫으면 | 90분 후 종료 | 90분 후 종료 | 90분 후 종료 |
| GPU 메모리 | 16GB (T4) | ~40GB | ~80GB |
자동 저장 스크립트
일정 시간마다 자동으로 체크포인트를 저장하는 코드를 작성할 수 있습니다.
import time
from datetime import datetime
def save_checkpoint_with_timestamp(model, optimizer, epoch, loss):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f'checkpoint_{timestamp}_epoch_{epoch}.pt'
path = f'/content/drive/MyDrive/checkpoints/{filename}'
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'loss': loss,
'timestamp': timestamp
}, path)
return path
# 사용 예시
last_save_time = time.time()
save_interval = 1800 # 30분마다 저장
for epoch in range(num_epochs):
loss = train_epoch(model, train_loader, optimizer)
# 시간 기반 자동 저장
if time.time() - last_save_time > save_interval:
save_checkpoint_with_timestamp(model, optimizer, epoch, loss)
last_save_time = time.time()세션 유지 팁
# JavaScript를 통한 자동 클릭 (세션 유지)
# 주의: 남용 시 계정 제한 가능
from IPython.display import display, Javascript
def keep_colab_alive():
display(Javascript('''
function ClickConnect(){
console.log("Keeping session alive");
document.querySelector("colab-connect-button").click()
}
setInterval(ClickConnect, 60000)
'''))
# 실행
keep_colab_alive()실서비스 배포
학습한 모델을 실제 서비스에 적용하려면 추가 단계가 필요합니다. 여기서는 기본적인 배포 방법을 소개합니다.
배포 방식 비교
| 방식 | 설명 | 사용 기술 | 난이도 |
|---|---|---|---|
| REST API | HTTP 요청으로 예측 결과 반환 | FastAPI, Flask | ⭐⭐ |
| 컨테이너 | 일관된 환경에서 실행 | Docker, Kubernetes | ⭐⭐⭐ |
| 서버리스 | 트래픽에 따라 자동 확장 | AWS Lambda, Cloud Functions | ⭐⭐⭐ |
| 클라우드 ML | 관리형 서비스 활용 | SageMaker, Vertex AI | ⭐⭐⭐⭐ |
| 모델 최적화 | 경량화 후 엣지 배포 | ONNX, TensorRT, Quantization | ⭐⭐⭐⭐⭐ |
FastAPI로 간단한 API 서버 만들기
가장 빠르게 모델을 서비스화할 수 있는 방법입니다.
# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
import numpy as np
# FastAPI 앱 생성
app = FastAPI(title="ML Model API")
# 모델 로드 (서버 시작 시 1회만 실행)
class ModelLoader:
model = None
@classmethod
def get_model(cls):
if cls.model is None:
cls.model = torch.load('model_weights.pt')
cls.model.eval()
return cls.model
# 요청 데이터 스키마
class PredictionRequest(BaseModel):
features: list[float]
# 응답 데이터 스키마
class PredictionResponse(BaseModel):
prediction: float
confidence: float
@app.get("/")
def root():
return {"message": "ML Model API is running"}
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
try:
# 모델 가져오기
model = ModelLoader.get_model()
# 입력 데이터 전처리
input_tensor = torch.tensor([request.features])
# 추론
with torch.no_grad():
output = model(input_tensor)
prediction = output.item()
confidence = torch.sigmoid(output).item()
return PredictionResponse(
prediction=prediction,
confidence=confidence
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Prediction failed: {str(e)}"
)서버 실행 및 테스트
# 서버 실행
!pip install fastapi uvicorn
# 로컬에서 실행
!uvicorn main:app --reload --host 0.0.0.0 --port 8000
# API 테스트 (다른 터미널에서)
import requests
response = requests.post(
"http://localhost:8000/predict",
json={"features": [1.0, 2.0, 3.0, 4.0]}
)
print(response.json())Docker 컨테이너화
# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 애플리케이션 코드 복사
COPY . .
# 모델 파일 포함
COPY model_weights.pt .
# 포트 노출
EXPOSE 8000
# 서버 실행
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]# Docker 이미지 빌드
!docker build -t ml-model-api .
# 컨테이너 실행
!docker run -p 8000:8000 ml-model-api모델 최적화
프로덕션 환경에서는 모델의 크기와 추론 속도가 중요합니다.
# PyTorch to ONNX 변환
import torch.onnx
# 더미 입력 생성
dummy_input = torch.randn(1, 10)
# ONNX로 내보내기
torch.onnx.export(
model,
dummy_input,
"model.onnx",
export_params=True,
opset_version=11,
input_names=['input'],
output_names=['output']
)
# ONNX Runtime으로 추론
import onnxruntime as ort
session = ort.InferenceSession("model.onnx")
outputs = session.run(
None,
{'input': dummy_input.numpy()}
)배포 체크리스트
모델 준비
✓ 모델 저장 및 버전 관리
✓ 추론 코드 작성 및 테스트
✓ 필요시 모델 최적화 (ONNX, Quantization)
API 서버
✓ FastAPI/Flask 서버 구현
✓ 입력 검증 및 에러 핸들링
✓ 로깅 및 모니터링 설정
배포
✓ Docker 이미지 생성
✓ 클라우드 플랫폼 선택
✓ CI/CD 파이프라인 구성
✓ 부하 테스트 및 성능 튜닝
'AI_학습' 카테고리의 다른 글
| 딥러닝 모델 최적화 완벽 가이드: 프로덕션 환경을 위한 모델 경량화와 속도 개선 (0) | 2026.01.06 |
|---|---|
| Google Colab으로 시작하는 머신러닝: GPU 환경 설정부터 회귀분석 실습까지 (0) | 2026.01.06 |