Model Serving 개요

# Pay-per-token 사용 예시
import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")

response = client.predict(
    endpoint="databricks-meta-llama-3-3-70b-instruct",
    inputs={
        "messages": [
            {"role": "system", "content": "당신은 데이터 전문가입니다. 한국어로 답변해 주세요."},
            {"role": "user", "content": "Delta Lake의 ACID 트랜잭션이 왜 중요한가요?"}
        ],
        "max_tokens": 500,
        "temperature": 0.1
    }
)
print(response["choices"][0]["message"]["content"])

# External Model 엔드포인트 생성 (OpenAI 프록시)
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput, ServedEntityInput, ExternalModel, OpenAiConfig
)

w = WorkspaceClient()

w.serving_endpoints.create(
    name="openai-gpt4-proxy",
    config=EndpointCoreConfigInput(
        served_entities=[ServedEntityInput(
            name="gpt4",
            external_model=ExternalModel(
                name="gpt-4",
                provider="openai",
                openai_config=OpenAiConfig(
                    openai_api_key="{{secrets/scope/openai-key}}"
                ),
                task="llm/v1/chat"
            )
        )]
    )
)

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import (
    EndpointCoreConfigInput, ServedEntityInput, AutoCaptureConfigInput
)

w = WorkspaceClient()

w.serving_endpoints.create(
    name="fraud-detection-v2",
    config=EndpointCoreConfigInput(
        served_entities=[ServedEntityInput(
            entity_name="catalog.schema.fraud_model",  # UC 등록 모델
            entity_version="3",
            workload_size="Small",                     # Small, Medium, Large
            scale_to_zero_enabled=True,                # 유휴 시 0으로 스케일다운
            workload_type="CPU"                        # CPU 또는 GPU
        )],
        auto_capture_config=AutoCaptureConfigInput(    # Inference Table 자동 활성화
            enabled=True,
            catalog_name="catalog",
            schema_name="schema",
            table_name_prefix="fraud_model"
        )
    )
)

import requests

# REST API로 호출
url = f"https://{workspace_url}/serving-endpoints/fraud-detection-v2/invocations"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# 단건 예측
response = requests.post(url, json={
    "dataframe_records": [
        {"amount": 50000, "merchant_category": "online", "hour": 3, "is_foreign": True}
    ]
}, headers=headers)

print(response.json())  # {"predictions": [0.87]}  (사기 확률 87%)

# Databricks SDK로 호출
w = WorkspaceClient()
response = w.serving_endpoints.query(
    name="fraud-detection-v2",
    dataframe_records=[
        {"amount": 50000, "merchant_category": "online", "hour": 3}
    ]
)

w.serving_endpoints.update_config(
    name="fraud-detection-v2",
    served_entities=[
        ServedEntityInput(
            name="champion",
            entity_name="catalog.schema.fraud_model",
            entity_version="3",
            workload_size="Small",
            scale_to_zero_enabled=True
        ),
        ServedEntityInput(
            name="challenger",
            entity_name="catalog.schema.fraud_model",
            entity_version="4",
            workload_size="Small",
            scale_to_zero_enabled=True
        )
    ],
    traffic_config={
        "routes": [
            {"served_model_name": "champion", "traffic_percentage": 80},
            {"served_model_name": "challenger", "traffic_percentage": 20}
        ]
    }
)

-- Inference Table에서 모니터링 쿼리
SELECT
    DATE(timestamp) AS day,
    COUNT(*) AS total_requests,
    AVG(latency_ms) AS avg_latency,
    SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END) AS error_count
FROM catalog.schema.fraud_model_inference_log
GROUP BY DATE(timestamp)
ORDER BY day DESC;

월간 비용 비교 (Llama 3.3 70B 기준, 예시):

Pay-per-token:
  - 입력: $0.24 / 1M tokens
  - 출력: $0.24 / 1M tokens
  - 월 1억 토큰 사용 시: ~$24/월

Provisioned Throughput:
  - 1 단위: ~$6~8/시간 (모델, 리전에 따라 다름)
  - 24/7 운영: ~$4,300~5,800/월
  - 손익분기점: 대략 월 1.5~2억 토큰 이상 사용 시 Provisioned가 유리

# 비용 최적화된 엔드포인트 설정
w.serving_endpoints.create(
    name="fraud-detection-prod",
    config=EndpointCoreConfigInput(
        served_entities=[ServedEntityInput(
            entity_name="catalog.schema.fraud_model",
            entity_version="3",
            workload_size="Small",          # 필요한 최소 사이즈 선택
            scale_to_zero_enabled=False,    # 프로덕션: 콜드스타트 방지
            workload_type="CPU",            # GPU가 불필요하면 CPU 사용
            min_provisioned_throughput=0,
            max_provisioned_throughput=1000  # 최대 처리량 제한으로 비용 상한 설정
        )]
    )
)

# Step 1: Canary 시작 — 새 버전에 5% 트래픽 할당
w.serving_endpoints.update_config(
    name="fraud-detection-prod",
    served_entities=[
        ServedEntityInput(name="v3-champion", entity_name="catalog.schema.fraud_model",
                          entity_version="3", workload_size="Small"),
        ServedEntityInput(name="v4-canary", entity_name="catalog.schema.fraud_model",
                          entity_version="4", workload_size="Small"),
    ],
    traffic_config={"routes": [
        {"served_model_name": "v3-champion", "traffic_percentage": 95},
        {"served_model_name": "v4-canary", "traffic_percentage": 5},
    ]}
)

# Step 2: 24시간 모니터링 후 문제 없으면 → 25%로 증가
# Step 3: 48시간 모니터링 후 문제 없으면 → 50%로 증가
# Step 4: 안정 확인 후 → 100%로 전환 (이전 버전 제거)

# 즉시 롤백: 이전 버전으로 100% 트래픽 전환
w.serving_endpoints.update_config(
    name="fraud-detection-prod",
    served_entities=[
        ServedEntityInput(name="v3-champion", entity_name="catalog.schema.fraud_model",
                          entity_version="3", workload_size="Small"),
    ],
    traffic_config={"routes": [
        {"served_model_name": "v3-champion", "traffic_percentage": 100},
    ]}
)
# 롤백 소요시간: 트래픽 설정 변경은 ~30초 이내 반영

# 캐싱 활성화 (AI Gateway 설정)
# temperature=0으로 설정하면 캐시 적중률이 높아집니다
response = client.predict(
    endpoint="databricks-meta-llama-3-3-70b-instruct",
    inputs={
        "messages": [
            {"role": "system", "content": "...긴 시스템 프롬프트..."},
            {"role": "user", "content": "매출 요약해줘"}
        ],
        "temperature": 0  # 결정적 응답 → 캐시 적중률 ↑
    }
)

사용자 (한국) → Azure Korea Central Workspace → 해당 리전 엔드포인트
사용자 (미국) → AWS us-east-1 Workspace → 해당 리전 엔드포인트

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

ENDPOINTS = [
    "https://primary-workspace.cloud.databricks.com/serving-endpoints/model/invocations",
    "https://secondary-workspace.cloud.databricks.com/serving-endpoints/model/invocations",
]

def predict_with_failover(payload, headers):
    for endpoint in ENDPOINTS:
        try:
            response = requests.post(endpoint, json=payload, headers=headers, timeout=5)
            if response.status_code == 200:
                return response.json()
        except (requests.Timeout, requests.ConnectionError):
            continue  # 다음 엔드포인트로 Failover
    raise Exception("All endpoints failed")