Ray Serve 모델 서빙 플랫폼 구축 가이드 — 오토스케일링, 멀티모델, 프로덕션 배포

Client (HTTP/gRPC)
        |
        v
+-------------------+
|   HTTP Proxy      |  (Ingress, 라우팅, 로드밸런싱)
+-------------------+
        |
        v
+-------------------+     +-------------------+
|  Deployment A     |     |  Deployment B     |
|  - Replica 1      |     |  - Replica 1      |
|  - Replica 2      |     |  - Replica 2      |
|  - Replica 3      |     |  - Replica 3      |
+-------------------+     +-------------------+
        |                         |
        v                         v
+-------------------------------------------+
|          Ray Cluster (분산 런타임)           |
|  - Head Node (Serve Controller)            |
|  - Worker Node 1 (GPU)                     |
|  - Worker Node 2 (GPU)                     |
+-------------------------------------------+

import ray
from ray import serve
from transformers import pipeline

@serve.deployment(
    num_replicas=2,
    ray_actor_options={"num_gpus": 1},
)
class SentimentModel:
    def __init__(self):
        self.model = pipeline(
            "sentiment-analysis",
            model="distilbert-base-uncased-finetuned-sst-2-english",
            device=0,
        )

    async def __call__(self, request):
        data = await request.json()
        result = self.model(data["text"])
        return {"prediction": result}

@serve.deployment(num_replicas=1)
class Router:
    def __init__(self, sentiment_handle):
        self.sentiment = sentiment_handle

    async def __call__(self, request):
        # DeploymentHandle을 통한 내부 호출
        result = await self.sentiment.remote(request)
        return result

sentiment_app = SentimentModel.bind()
router_app = Router.bind(sentiment_app)
serve.run(router_app, route_prefix="/predict")

from ray import serve
from ray.serve.llm import LLMServer, LLMConfig, build_openai_app

# LLM 설정 정의
llm_config = LLMConfig(
    model_id="meta-llama/Llama-3.1-8B-Instruct",
    engine_kwargs={
        "tensor_parallel_size": 2,
        "max_model_len": 4096,
        "gpu_memory_utilization": 0.85,
        "enforce_eager": False,
    },
    # 오토스케일링 설정
    deployment_config={
        "autoscaling_config": {
            "min_replicas": 1,
            "max_replicas": 4,
            "target_ongoing_requests": 5,
        }
    },
    accelerator_type="A100",
)

# OpenAI 호환 앱 빌드 및 배포
app = build_openai_app(llm_config)
serve.run(app, host="0.0.0.0", port=8000)

# 헬스체크
curl http://localhost:8000/v1/models

# Chat Completions API 호출
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "meta-llama/Llama-3.1-8B-Instruct",
    "messages": [
      {"role": "user", "content": "Ray Serve의 장점을 설명해주세요."}
    ],
    "max_tokens": 512,
    "temperature": 0.7,
    "stream": true
  }'

from ray.serve.llm import LLMConfig, build_openai_app

# 여러 모델 설정
configs = [
    LLMConfig(
        model_id="meta-llama/Llama-3.1-8B-Instruct",
        engine_kwargs={
            "tensor_parallel_size": 1,
            "gpu_memory_utilization": 0.85,
        },
        accelerator_type="A100",
    ),
    LLMConfig(
        model_id="Qwen/Qwen2.5-7B-Instruct",
        engine_kwargs={
            "tensor_parallel_size": 1,
            "gpu_memory_utilization": 0.85,
        },
        accelerator_type="A100",
    ),
]

# 모든 모델을 단일 앱으로 빌드
app = build_openai_app(configs)

from ray import serve

@serve.deployment(
    # 오토스케일링 상세 설정
    autoscaling_config={
        "min_replicas": 1,          # 최소 Replica 수
        "max_replicas": 10,         # 최대 Replica 수 (피크 대비 20% 여유)
        "target_ongoing_requests": 5,  # Replica당 목표 진행중 요청 수
        "upscale_delay_s": 30,      # 스케일업 판단 대기 시간
        "downscale_delay_s": 300,   # 스케일다운 판단 대기 시간 (보수적)
        "upscaling_factor": 1.0,    # 스케일업 비율
        "downscaling_factor": 0.5,  # 스케일다운 비율
        "smoothing_factor": 0.5,    # 메트릭 스무딩 팩터
        "metrics_interval_s": 10,   # 메트릭 수집 주기
    },
    max_ongoing_requests=10,  # Replica당 최대 동시 처리 요청 수
    ray_actor_options={
        "num_gpus": 1,
        "num_cpus": 4,
        "memory": 16 * 1024 * 1024 * 1024,  # 16GB
    },
    health_check_period_s=10,
    health_check_timeout_s=30,
    graceful_shutdown_timeout_s=60,
)
class GPUModel:
    def __init__(self):
        # 모델 로드
        pass

    async def __call__(self, request):
        # 추론 로직
        pass

desired_replicas = ceil(total_ongoing_requests / target_ongoing_requests)

@serve.deployment(
    autoscaling_config={
        "min_replicas": 0,   # 트래픽 없으면 0으로 축소
        "max_replicas": 5,
    },
)
class BatchModel:
    pass

@serve.deployment(
    ray_actor_options={"num_gpus": 0.25}  # GPU의 25%만 사용
)
class LightModel:
    pass

from ray import serve

@serve.deployment(num_replicas=2)
class Preprocessor:
    def preprocess(self, text: str) -> dict:
        # 텍스트 전처리: 토크나이징, 정규화
        tokens = text.lower().split()
        return {"tokens": tokens, "length": len(tokens)}

@serve.deployment(
    num_replicas=3,
    ray_actor_options={"num_gpus": 1},
)
class Classifier:
    def __init__(self):
        from transformers import pipeline
        self.model = pipeline("text-classification", device=0)

    def classify(self, preprocessed: dict) -> dict:
        text = " ".join(preprocessed["tokens"])
        result = self.model(text)
        return {"label": result[0]["label"], "score": result[0]["score"]}

@serve.deployment(num_replicas=1)
class Postprocessor:
    def format_response(self, classification: dict) -> dict:
        return {
            "result": classification["label"],
            "confidence": round(classification["score"], 4),
            "status": "success",
        }

@serve.deployment(num_replicas=1)
class Pipeline:
    def __init__(self, preprocessor, classifier, postprocessor):
        self.preprocessor = preprocessor
        self.classifier = classifier
        self.postprocessor = postprocessor

    async def __call__(self, request):
        data = await request.json()
        text = data["text"]

        # 순차적으로 파이프라인 실행
        preprocessed = await self.preprocessor.preprocess.remote(text)
        classified = await self.classifier.classify.remote(preprocessed)
        response = await self.postprocessor.format_response.remote(classified)
        return response

# 배포 그래프 구성
preprocessor = Preprocessor.bind()
classifier = Classifier.bind()
postprocessor = Postprocessor.bind()
app = Pipeline.bind(preprocessor, classifier, postprocessor)

from ray import serve

@serve.deployment(
    num_replicas=2,
    max_ongoing_requests=10,
)
class MultiplexedModel:
    def __init__(self):
        self.models = {}

    def _load_model(self, model_id: str):
        if model_id not in self.models:
            # 모델 동적 로드 (캐시 관리)
            import joblib
            self.models[model_id] = joblib.load(f"/models/{model_id}.pkl")
            # 메모리 관리: LRU 캐시로 최대 10개 모델 유지
            if len(self.models) > 10:
                oldest = next(iter(self.models))
                del self.models[oldest]
        return self.models[model_id]

    async def __call__(self, request):
        data = await request.json()
        model_id = data["model_id"]
        features = data["features"]

        model = self._load_model(model_id)
        prediction = model.predict([features])
        return {"model_id": model_id, "prediction": prediction.tolist()}

import numpy as np
from ray import serve

@serve.deployment(
    num_replicas=2,
    ray_actor_options={"num_gpus": 1},
)
class BatchClassifier:
    def __init__(self):
        from transformers import AutoTokenizer, AutoModel
        import torch

        self.tokenizer = AutoTokenizer.from_pretrained(
            "bert-base-uncased"
        )
        self.model = AutoModel.from_pretrained(
            "bert-base-uncased"
        ).cuda().eval()

    @serve.batch(max_batch_size=32, batch_wait_timeout_s=0.1)
    async def handle_batch(self, texts: list[str]) -> list[dict]:
        import torch

        # 배치 토크나이징
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=512,
            return_tensors="pt",
        ).to("cuda")

        # 배치 추론
        with torch.no_grad():
            outputs = self.model(**inputs)
            embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()

        # 결과를 개별 응답으로 분리
        results = []
        for i, emb in enumerate(embeddings):
            results.append({
                "embedding": emb.tolist()[:10],  # 일부만 반환
                "batch_size": len(texts),
            })
        return results

    async def __call__(self, request):
        data = await request.json()
        return await self.handle_batch(data["text"])

apiVersion: ray.io/v1
kind: RayService
metadata:
  name: llm-serving
  namespace: ml-serving
spec:
  # Serve 애플리케이션 설정
  serveConfigV2: |
    applications:
      - name: llm-app
        route_prefix: /
        import_path: serve_app:app
        runtime_env:
          working_dir: "https://github.com/my-org/serve-apps/archive/main.zip"
          pip:
            - transformers>=4.40.0
            - torch>=2.2.0
            - vllm>=0.6.0
            - ray[serve]>=2.40.0
        deployments:
          - name: LLMDeployment
            num_replicas: auto
            autoscaling_config:
              min_replicas: 1
              max_replicas: 4
              target_ongoing_requests: 5
              upscale_delay_s: 30
              downscale_delay_s: 300
            max_ongoing_requests: 10
            ray_actor_options:
              num_gpus: 1
              num_cpus: 4

  # Ray 클러스터 설정
  rayClusterConfig:
    rayVersion: '2.40.0'
    headGroupSpec:
      rayStartParams:
        dashboard-host: '0.0.0.0'
        num-gpus: '0'
      template:
        spec:
          containers:
            - name: ray-head
              image: rayproject/ray-ml:2.40.0-py310-gpu
              ports:
                - containerPort: 6379
                  name: gcs-server
                - containerPort: 8265
                  name: dashboard
                - containerPort: 8000
                  name: serve
              resources:
                limits:
                  cpu: '4'
                  memory: '16Gi'
                requests:
                  cpu: '2'
                  memory: '8Gi'

    workerGroupSpecs:
      - replicas: 2
        minReplicas: 1
        maxReplicas: 8
        groupName: gpu-workers
        rayStartParams:
          num-gpus: '1'
        template:
          spec:
            containers:
              - name: ray-worker
                image: rayproject/ray-ml:2.40.0-py310-gpu
                resources:
                  limits:
                    cpu: '8'
                    memory: '32Gi'
                    nvidia.com/gpu: '1'
                  requests:
                    cpu: '4'
                    memory: '16Gi'
                    nvidia.com/gpu: '1'
            tolerations:
              - key: 'nvidia.com/gpu'
                operator: 'Exists'
                effect: 'NoSchedule'
            nodeSelector:
              cloud.google.com/gke-accelerator: nvidia-tesla-a100

# KubeRay Operator 설치
helm repo add kuberay https://ray-project.github.io/kuberay-helm/
helm repo update
helm install kuberay-operator kuberay/kuberay-operator \
  --namespace kuberay-system \
  --create-namespace \
  --version 1.2.2

# RayService 배포
kubectl apply -f ray-service-llm.yaml

# 상태 확인
kubectl get rayservice llm-serving -n ml-serving
kubectl describe rayservice llm-serving -n ml-serving

# 서비스 로그 확인
kubectl logs -l ray.io/serve=llm-serving -n ml-serving --tail=100

# 포트 포워딩으로 로컬 접근
kubectl port-forward svc/llm-serving-serve-svc 8000:8000 -n ml-serving

# ServiceMonitor로 Ray 메트릭 수집
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: ray-serve-monitor
  namespace: ml-serving
spec:
  selector:
    matchLabels:
      ray.io/serve: llm-serving
  endpoints:
    - port: metrics
      interval: 15s
      path: /metrics
  namespaceSelector:
    matchNames:
      - ml-serving

import logging
from ray import serve

logger = logging.getLogger("ray.serve")

@serve.deployment
class MonitoredModel:
    def __init__(self):
        self.request_count = 0

    async def __call__(self, request):
        self.request_count += 1
        data = await request.json()

        logger.info(
            f"Request #{self.request_count} received",
            extra={
                "request_id": request.headers.get("X-Request-ID", "unknown"),
                "model": "monitored-model",
            },
        )

        try:
            result = self._predict(data)
            logger.info(f"Request #{self.request_count} completed successfully")
            return result
        except Exception as e:
            logger.error(
                f"Request #{self.request_count} failed: {str(e)}",
                exc_info=True,
            )
            raise

    def _predict(self, data):
        # 추론 로직
        return {"status": "ok"}

# 메모리 프로파일링 활성화
export RAY_SERVE_ENABLE_MEMORY_PROFILING=1

# 프로파일링 결과 확인
ls /tmp/ray/session_latest/logs/serve/*.bin

# 레이턴시 히스토그램 버킷 커스터마이징
export RAY_SERVE_REQUEST_LATENCY_BUCKETS_MS="5,10,25,50,100,250,500,1000,2500,5000"

# 예방 전략: 리소스 제한 명시
@serve.deployment(
    ray_actor_options={
        "num_gpus": 1,
        "memory": 16 * 1024 * 1024 * 1024,  # 16GB 명시
    },
    # 헬스체크로 비정상 상태 조기 감지
    health_check_period_s=10,
    health_check_timeout_s=30,
)
class SafeModel:
    def __init__(self):
        import torch
        # GPU 메모리 사용량 로깅
        if torch.cuda.is_available():
            allocated = torch.cuda.memory_allocated() / 1e9
            reserved = torch.cuda.memory_reserved() / 1e9
            logger.info(
                f"GPU memory - allocated: {allocated:.2f}GB, "
                f"reserved: {reserved:.2f}GB"
            )

# gpu_memory_utilization 조절
llm_config = LLMConfig(
    model_id="meta-llama/Llama-3.1-8B-Instruct",
    engine_kwargs={
        "gpu_memory_utilization": 0.80,  # 90%에서 80%로 낮춤
        "max_model_len": 2048,           # 컨텍스트 길이 제한
    },
)

# NCCL 환경 변수 확인
export NCCL_DEBUG=INFO
export NCCL_SOCKET_IFNAME=eth0
export NCCL_IB_DISABLE=1  # InfiniBand 없는 환경에서

# 실패 상태 확인
kubectl get rayservice llm-serving -n ml-serving -o jsonpath='{.status.serviceStatus}'

# 복구: 삭제 후 재생성
kubectl delete rayservice llm-serving -n ml-serving
kubectl apply -f ray-service-llm.yaml

# 진동 방지 설정
autoscaling_config={
    "upscale_delay_s": 60,        # 스케일업 대기 시간 증가
    "downscale_delay_s": 600,     # 스케일다운 대기 시간 크게 증가
    "smoothing_factor": 0.3,      # 메트릭 변화에 둔감하게
    "downscaling_factor": 0.3,    # 한 번에 30%만 축소
}