Foundation Model API

import mlflow.deployments

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

# OpenAI GPT-4o를 Databricks에서 프록시
endpoint = client.create_endpoint(
    name="gpt-4o-proxy",
    config={
        "served_entities": [{
            "external_model": {
                "name": "gpt-4o",
                "provider": "openai",
                "openai_config": {
                    "openai_api_key": "{{secrets/llm-keys/openai-key}}"
                }
            }
        }]
    }
)

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": 1000,
        "temperature": 0.1
    }
)

print(response["choices"][0]["message"]["content"])

response = client.predict(
    endpoint="databricks-meta-llama-3-3-70b-instruct",
    inputs={
        "prompt": "다음 Python 코드를 완성하세요:\ndef calculate_revenue(orders):\n",
        "max_tokens": 200,
        "temperature": 0.0
    }
)

# 텍스트를 벡터로 변환
response = client.predict(
    endpoint="databricks-gte-large-en",
    inputs={
        "input": [
            "Delta Lake is an open-source storage layer",
            "Apache Spark is a distributed processing engine"
        ]
    }
)

# 결과: 각 텍스트의 1024차원 벡터
embeddings = [item["embedding"] for item in response["data"]]
print(f"벡터 차원: {len(embeddings[0])}")  # 1024

-- 고객 리뷰를 한국어로 요약
SELECT
    review_id,
    review_text,
    ai_query(
        'databricks-meta-llama-3-3-70b-instruct',
        CONCAT('다음 리뷰를 한국어로 한 문장으로 요약해 주세요: ', review_text)
    ) AS summary
FROM customer_reviews
LIMIT 100;

-- 텍스트 감성 분석
SELECT
    review_id,
    ai_query(
        'databricks-meta-llama-3-3-70b-instruct',
        CONCAT('다음 리뷰의 감성을 "긍정", "부정", "중립" 중 하나로 분류하세요: ', review_text)
    ) AS sentiment
FROM customer_reviews;

-- 임베딩 생성
SELECT
    doc_id,
    ai_query(
        'databricks-gte-large-en',
        content,
        returnType => 'ARRAY<FLOAT>'
    ) AS embedding
FROM documents;

예시 계산:
입력: 500 토큰 × $0.001/1K토큰 = $0.0005
출력: 200 토큰 × $0.002/1K토큰 = $0.0004
합계: $0.0009/요청

월 10만 건 요청 시: 약 $90/월

예약: 특정 모델에 대해 토큰/초 처리량 보장
비용: 시간당 고정 DBU (예: 50 DBU/시간)
장점: 일정한 응답 시간 보장, 대량 처리 시 단가 저렴
특징: 커스텀 파인튜닝 모델도 배포 가능

import time
import mlflow.deployments

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

def call_with_retry(endpoint, inputs, max_retries=5):
    """지수 백오프(Exponential Backoff)로 재시도합니다."""
    for attempt in range(max_retries):
        try:
            return client.predict(endpoint=endpoint, inputs=inputs)
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait_time = 2 **attempt  # 1, 2, 4, 8, 16초
                print(f"Rate limited. {wait_time}초 후 재시도... ({attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise

다음 중 3개 이상 해당하면 Provisioned Throughput으로 전환을 고려하세요:

☐ 월 50만 건 이상의 API 요청이 발생한다
☐ P95 응답 시간 SLA (예: 3초 이내)가 필요하다
☐ 트래픽이 예측 가능하고 일정하다 (하루 종일 균등)
☐ 커스텀 파인튜닝 모델을 서빙해야 한다
☐ Pay-per-token에서 Rate Limiting에 자주 걸린다

[시나리오: 대량 배치 처리]
- 100만 건의 고객 리뷰를 ai_query()로 감성 분석
- 분당 토큰 제한에 걸려서 처리가 수 시간 → 수일로 늘어남
- 중간에 실패하면 어디까지 처리했는지 추적 어려움

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_process_with_rate_limit(records, endpoint, max_workers=5, delay=0.1):
    """
    Rate Limiting을 고려한 대량 배치 처리 패턴
    - max_workers: 동시 요청 수 (너무 높으면 429 에러)
    - delay: 요청 간 대기 시간 (Rate Limit 여유)
    """
    results = []
    failed = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {}
        for i, record in enumerate(records):
            future = executor.submit(call_with_retry, endpoint, record)
            futures[future] = i
            time.sleep(delay)  # 요청 간격 조절

        for future in as_completed(futures):
            idx = futures[future]
            try:
                results.append((idx, future.result()))
            except Exception as e:
                failed.append((idx, str(e)))

    print(f"성공: {len(results)}, 실패: {len(failed)}")
    return results, failed