8장: 엔터프라이즈 통합

프로덕션 배포, MLOps 및 거버넌스

弘益人間 · 널리 인간을 이롭게 하라

8.1 엔터프라이즈 AI 배포 환경

엔터프라이즈 AI 배포는 강력한 인프라, 모니터링, 거버넌스 및 규정 준수가 필요합니다. WIA-AI-014 신경망 형식 표준은 엔터프라이즈 MLOps 플랫폼과의 원활한 통합을 가능하게 하여 모델이 프로덕션 준비, 보안 및 대규모 관리 가능하도록 보장합니다.

8.2 MLOps 파이프라인 아키텍처

개발에서 프로덕션까지의 완전한 ML 파이프라인:

WIA-AI-014를 사용한 MLOps 파이프라인:

1. 개발
   ├── 모델 훈련 (PyTorch/TensorFlow)
   ├── 실험 추적 (MLflow/W&B)
   └── 버전 관리 (Git + DVC)

2. 검증
   ├── WIA-AI-014 형식으로 내보내기
   ├── 모델 검증 테스트
   ├── 성능 벤치마킹
   └── 보안 스캐닝

3. 등록
   ├── 모델 레지스트리 (MLflow/Custom)
   ├── 메타데이터 보강
   ├── 거버넌스 승인
   └── 아티팩트 저장소 (S3/GCS)

4. 배포
   ├── A/B 테스팅 인프라
   ├── Blue-Green 배포
   ├── 카나리 릴리스
   └── 멀티 클라우드 지원

5. 모니터링
   ├── 성능 메트릭
   ├── 데이터 드리프트 감지
   ├── 모델 드리프트 감지
   └── 알림 및 페이징

6. 유지보수
   ├── 모델 재훈련
   ├── 롤백 절차
   └── 지속적 개선

8.3 모델 서빙 인프라

Kubernetes 배포

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-server
  labels:
    app: ml-model
    version: v2.1.0
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model
  template:
    metadata:
      labels:
        app: ml-model
        version: v2.1.0
    spec:
      containers:
      - name: model-server
        image: company/model-server:latest
        ports:
        - containerPort: 8080
        env:
        - name: MODEL_PATH
          value: "/models/resnet50_v2.1.0.onnx"
        - name: BATCH_SIZE
          value: "32"
        resources:
          limits:
            memory: "4Gi"
            cpu: "2"
            nvidia.com/gpu: "1"
          requests:
            memory: "2Gi"
            cpu: "1"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 5
        volumeMounts:
        - name: model-storage
          mountPath: /models
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc

---
apiVersion: v1
kind: Service
metadata:
  name: model-service
spec:
  selector:
    app: ml-model
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8080
  type: LoadBalancer

다중 모델 서버

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import onnxruntime as ort
from typing import Dict, List
import numpy as np

app = FastAPI()

class ModelRegistry:
    """모델 레지스트리"""

    def __init__(self):
        self.models: Dict[str, ort.InferenceSession] = {}

    def load_model(self, name: str, path: str):
        """레지스트리에 모델 로드"""
        session = ort.InferenceSession(path)
        self.models[name] = session
        return True

    def get_model(self, name: str):
        """레지스트리에서 모델 가져오기"""
        if name not in self.models:
            raise HTTPException(status_code=404, detail="모델을 찾을 수 없습니다")
        return self.models[name]

    def list_models(self) -> List[str]:
        """로드된 모든 모델 나열"""
        return list(self.models.keys())

registry = ModelRegistry()

# 시작 시 모델 로드
@app.on_event("startup")
async def load_models():
    registry.load_model("resnet50_v2", "/models/resnet50_v2.1.0.onnx")
    registry.load_model("bert_base", "/models/bert_base_v1.0.0.onnx")

class PredictionRequest(BaseModel):
    model_name: str
    input_data: List[List[float]]

class PredictionResponse(BaseModel):
    model_name: str
    predictions: List[float]
    latency_ms: float

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    import time

    start = time.perf_counter()

    # 모델 가져오기
    model = registry.get_model(request.model_name)

    # 입력 준비
    input_array = np.array(request.input_data, dtype=np.float32)
    input_name = model.get_inputs()[0].name

    # 추론 실행
    outputs = model.run(None, {input_name: input_array})

    latency = (time.perf_counter() - start) * 1000

    return PredictionResponse(
        model_name=request.model_name,
        predictions=outputs[0].tolist(),
        latency_ms=latency
    )

@app.get("/models")
async def list_models():
    return {"models": registry.list_models()}

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

# 실행: uvicorn server:app --host 0.0.0.0 --port 8080

8.4 A/B 테스팅 및 카나리 배포

import random
from enum import Enum

class DeploymentStrategy(Enum):
    """배포 전략"""
    BLUE_GREEN = "blue_green"
    CANARY = "canary"
    AB_TEST = "ab_test"

class ModelRouter:
    """모델 라우팅 시스템"""

    def __init__(self):
        self.models = {}
        self.traffic_split = {}

    def register_model(self, name: str, version: str,
                      session, weight: float = 0.0):
        """트래픽 가중치로 모델 변형 등록"""
        key = f"{name}:{version}"
        self.models[key] = session
        self.traffic_split[key] = weight

    def route_request(self, model_name: str):
        """요청을 적절한 모델 버전으로 라우팅"""
        # 이 모델의 모든 버전 가져오기
        versions = {k: v for k, v in self.traffic_split.items()
                   if k.startswith(f"{model_name}:")}

        if not versions:
            raise ValueError(f"{model_name}의 버전을 찾을 수 없습니다")

        # 가중치 기반 랜덤 선택
        total_weight = sum(versions.values())
        rand = random.uniform(0, total_weight)

        cumsum = 0
        for version_key, weight in versions.items():
            cumsum += weight
            if rand <= cumsum:
                return self.models[version_key]

# 사용 예
router = ModelRouter()

# Blue-Green: v2에 100%, v1은 웜업 유지
router.register_model("resnet50", "v1", model_v1, weight=0.0)
router.register_model("resnet50", "v2", model_v2, weight=1.0)

# 카나리: v1에 95%, v2에 5%
router.register_model("bert", "v1", bert_v1, weight=0.95)
router.register_model("bert", "v2", bert_v2, weight=0.05)

# A/B 테스트: 50/50 분할
router.register_model("gpt", "variant_a", gpt_a, weight=0.5)
router.register_model("gpt", "variant_b", gpt_b, weight=0.5)

# 요청 라우팅
model = router.route_request("resnet50")
output = model.run(None, input_data)

8.5 모니터링 및 관찰성

Prometheus 메트릭

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 메트릭 정의
prediction_requests = Counter(
    'model_prediction_requests_total',
    '총 예측 요청 수',
    ['model_name', 'version', 'status']
)

prediction_latency = Histogram(
    'model_prediction_latency_seconds',
    '예측 지연 시간 (초)',
    ['model_name', 'version'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)

model_memory_usage = Gauge(
    'model_memory_usage_bytes',
    '모델 메모리 사용량 (바이트)',
    ['model_name', 'version']
)

active_requests = Gauge(
    'model_active_requests',
    '활성 요청 수',
    ['model_name', 'version']
)

class MonitoredModel:
    """모니터링되는 모델"""

    def __init__(self, model, name, version):
        self.model = model
        self.name = name
        self.version = version

    def predict(self, input_data):
        # 활성 요청 증가
        active_requests.labels(
            model_name=self.name,
            version=self.version
        ).inc()

        start = time.perf_counter()

        try:
            # 추론 실행
            output = self.model.run(None, input_data)

            # 성공 기록
            prediction_requests.labels(
                model_name=self.name,
                version=self.version,
                status='success'
            ).inc()

            return output

        except Exception as e:
            # 실패 기록
            prediction_requests.labels(
                model_name=self.name,
                version=self.version,
                status='error'
            ).inc()
            raise

        finally:
            # 지연 시간 기록
            latency = time.perf_counter() - start
            prediction_latency.labels(
                model_name=self.name,
                version=self.version
            ).observe(latency)

            # 활성 요청 감소
            active_requests.labels(
                model_name=self.name,
                version=self.version
            ).dec()

# 메트릭 서버 시작
start_http_server(9090)

# 모니터링되는 모델 사용
monitored = MonitoredModel(onnx_session, "resnet50", "v2.1.0")
output = monitored.predict(input_data)

8.6 데이터 드리프트 감지

import numpy as np
from scipy import stats

class DataDriftDetector:
    """데이터 드리프트 감지 시스템"""

    def __init__(self, reference_data, significance=0.05):
        self.reference_data = reference_data
        self.significance = significance
        self.reference_stats = self.compute_stats(reference_data)

    def compute_stats(self, data):
        """통계 계산"""
        return {
            'mean': np.mean(data, axis=0),
            'std': np.std(data, axis=0),
            'min': np.min(data, axis=0),
            'max': np.max(data, axis=0)
        }

    def detect_drift_ks_test(self, current_data):
        """Kolmogorov-Smirnov 테스트로 분포 이동 감지"""
        drifted_features = []

        for i in range(current_data.shape[1]):
            statistic, pvalue = stats.ks_2samp(
                self.reference_data[:, i],
                current_data[:, i]
            )

            if pvalue < self.significance:
                drifted_features.append({
                    'feature_index': i,
                    'statistic': statistic,
                    'pvalue': pvalue
                })

        return {
            'drifted': len(drifted_features) > 0,
            'num_drifted_features': len(drifted_features),
            'details': drifted_features
        }

    def detect_drift_psi(self, current_data, num_bins=10):
        """Population Stability Index"""
        psi_values = []

        for i in range(current_data.shape[1]):
            ref_counts, bin_edges = np.histogram(
                self.reference_data[:, i],
                bins=num_bins
            )
            curr_counts, _ = np.histogram(
                current_data[:, i],
                bins=bin_edges
            )

            # 정규화
            ref_pct = ref_counts / len(self.reference_data)
            curr_pct = curr_counts / len(current_data)

            # 0으로 나누기 방지
            ref_pct = np.where(ref_pct == 0, 0.0001, ref_pct)
            curr_pct = np.where(curr_pct == 0, 0.0001, curr_pct)

            # PSI 계산
            psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
            psi_values.append(psi)

        avg_psi = np.mean(psi_values)

        # PSI 해석: <0.1 안정, 0.1-0.2 보통, >0.2 심각
        return {
            'psi': avg_psi,
            'status': '안정' if avg_psi < 0.1 else '보통' if avg_psi < 0.2 else '심각',
            'feature_psi': psi_values
        }

# 사용 예
detector = DataDriftDetector(reference_data=training_data)

# 정기적으로 프로덕션 데이터 확인
drift_result = detector.detect_drift_ks_test(production_data)
if drift_result['drifted']:
    print(f"⚠️  {drift_result['num_drifted_features']}개 특성에서 데이터 드리프트 감지")
    # 재훈련 또는 알림 트리거

8.7 보안 및 규정 준수

모델 암호화

from cryptography.fernet import Fernet
import hashlib

class SecureModelStorage:
    """보안 모델 저장소"""

    def __init__(self, encryption_key=None):
        if encryption_key is None:
            encryption_key = Fernet.generate_key()
        self.cipher = Fernet(encryption_key)
        self.key = encryption_key

    def encrypt_model(self, model_path, output_path):
        """모델 파일 암호화"""
        with open(model_path, 'rb') as f:
            model_data = f.read()

        encrypted_data = self.cipher.encrypt(model_data)

        with open(output_path, 'wb') as f:
            f.write(encrypted_data)

        # 무결성을 위한 해시 계산
        model_hash = hashlib.sha256(model_data).hexdigest()
        return model_hash

    def decrypt_model(self, encrypted_path, output_path):
        """모델 파일 복호화"""
        with open(encrypted_path, 'rb') as f:
            encrypted_data = f.read()

        decrypted_data = self.cipher.decrypt(encrypted_data)

        with open(output_path, 'wb') as f:
            f.write(decrypted_data)

        return True

# 사용 예
secure_storage = SecureModelStorage()

# 저장 전 암호화
model_hash = secure_storage.encrypt_model(
    'model.onnx',
    'encrypted_model.enc'
)

# 로드를 위해 복호화
secure_storage.decrypt_model(
    'encrypted_model.enc',
    'decrypted_model.onnx'
)

8.8 비용 최적화

class CostOptimizer:
    """비용 최적화 도구"""

    def __init__(self):
        self.request_costs = {
            'cpu': 0.0001,        # 요청당
            'gpu': 0.001,         # 요청당
            'memory_gb_hour': 0.01
        }

    def estimate_monthly_cost(self, requests_per_day, avg_latency_ms,
                             memory_gb, use_gpu=False):
        """월간 인프라 비용 추정"""
        monthly_requests = requests_per_day * 30
        compute_type = 'gpu' if use_gpu else 'cpu'

        # 컴퓨팅 비용
        compute_cost = monthly_requests * self.request_costs[compute_type]

        # 메모리 비용 (항상 켜짐 가정)
        memory_hours = 24 * 30  # 월간 시간
        memory_cost = memory_gb * memory_hours * self.request_costs['memory_gb_hour']

        total_cost = compute_cost + memory_cost

        return {
            'compute_cost': compute_cost,
            'memory_cost': memory_cost,
            'total_cost': total_cost,
            'cost_per_request': total_cost / monthly_requests
        }

    def recommend_optimization(self, current_cost):
        """비용 최적화 권장사항 제공"""
        recommendations = []

        if current_cost['compute_cost'] > current_cost['memory_cost'] * 2:
            recommendations.append({
                'type': '배칭',
                'description': '요청 배칭을 구현하여 요청당 오버헤드 감소',
                'estimated_savings': current_cost['compute_cost'] * 0.3
            })

        if current_cost['memory_cost'] > 100:
            recommendations.append({
                'type': '양자화',
                'description': 'INT8 양자화 적용하여 메모리 사용량 감소',
                'estimated_savings': current_cost['memory_cost'] * 0.5
            })

        return recommendations

# 사용 예
optimizer = CostOptimizer()
cost = optimizer.estimate_monthly_cost(
    requests_per_day=1000000,
    avg_latency_ms=10,
    memory_gb=4,
    use_gpu=True
)
print(f"월간 비용: ${cost['total_cost']:.2f}")
print(f"요청당 비용: ${cost['cost_per_request']:.6f}")

recommendations = optimizer.recommend_optimization(cost)
for rec in recommendations:
    print(f"\n{rec['type']}: {rec['description']}")
    print(f"예상 절감액: ${rec['estimated_savings']:.2f}/월")

8.9 재해 복구

8.10 WIA-AI-014 엔터프라이즈 모범 사례