LLM Deployment Strategies: Стратегии развёртывания LLM

Введение

Развёртывание LLM в production — сложная задача, требующая учёта latency, throughput, cost и reliability. Рассмотрим основные стратегии.

---

Deployment Options Overview

Deployment Strategy    | Latency    | Cost      | Scalability | Control
-----------------------|------------|-----------|-------------|--------
Cloud API (GPT-4)      | Low        | High      | High        | Low
Cloud GPU (RunPod)     | Medium     | Medium    | Medium      | Medium
Self-hosted           | Low        | High      | High        | High
Edge/On-device         | Very Low   | Low       | Low         | High
Hybrid                 | Variable   | Variable  | Variable    | Variable

---

Cloud API Deployment

Простая интеграция

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

def generate(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": text}],
        temperature=0.7
    )
    return response.choices[0].message.content

Multi-provider для reliability

import litellm

def reliable_generate(text: str) -> str:
    # Primary: GPT-4o
    try:
        response = litellm.completion(
            model="gpt-4o",
            messages=[{"role": "user", "content": text}]
        )
        return response.choices[0].message.content
    except Exception:
        pass
    
    # Fallback: Claude
    try:
        response = litellm.completion(
            model="claude-sonnet-4",
            messages=[{"role": "user", "content": text}]
        )
        return response.choices[0].message.content
    except Exception:
        pass
    
    # Last resort: Gemini
    response = litellm.completion(
        model="gemini-2.0-flash",
        messages=[{"role": "user", "content": text}]
    )
    return response.choices[0].message.content

---

Cloud GPU Deployment

RunPod / Lambda Labs

# Dockerfile для vLLM
FROM vllm/vllm-serving:latest

RUN pip install transformers

COPY config.json /model/config.json
COPY model.safetensors /model/model.safetensors

ENV VLLM_MODEL=/model
ENV VLLM_PORT=8000

# Запуск на RunPod
runpod template deploy llama-3-70b \
  --gpu A100-80GB \
  --min-gpus 1 \
  --max-gpus 4 \
  --docker-image vllm/vllm-serving:latest

AWS SageMaker

import sagemaker
from sagemaker.hugging_model import HuggingFaceModel

# Регистрация модели
huggingface_model = HuggingFaceModel(
    role="SageMakerRole",
    model_url="https://huggingface.co/meta-llama/Llama-3-70b-instruct",
    env={
        'HF_MODEL_ID': "/opt/ml/model",
        'MAX_BATCH_SIZE': '8',
        'MAX_INPUT_LENGTH': '2048'
    }
)

# Deployment
predictor = huggingface_model.deploy(
    initial_instance_count=2,
    instance_type="ml.g5.12xlarge",
    container_startup_health_check_timeout=300
)

Azure AI

from azure.ai.ml import MLClient
from azure.ai.ml.entities import Endpoint, Deployment

ml_client = MLClient.from_config()

# Create endpoint
endpoint = Endpoint(
    name="llama-3-endpoint",
    description="Llama 3 inference endpoint"
)
ml_client.endpoints.create(endpoint)

# Create deployment
deployment = Deployment(
    name="llama-3-70b-v1",
    model="azureml://models/llama-3-70b-instruct",
    instance_type="Standard_NC48ads_A100_v4",
    instance_count=2,
    app_settings={
        "AZURE_ML_MODEL_REVISION": "v1"
    }
)
ml_client.deployments.create(deployment)

---

Self-hosted Deployment

vLLM — high-throughput serving

from vllm import LLM, SamplingParams

# Load model
llm = LLM(
    model="meta-llama/Llama-3-70b-instruct",
    tensor_parallel_size=4,  # GPU count
    max_model_len=4096,
    gpu_memory_utilization=0.9
)

# Sampling params
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=1024,
    stop=["\n\n"]
)

# Batch inference
prompts = [
    "Explain quantum computing",
    "Write a Python function",
    "Translate to French"
]

outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    generated_text = output.outputs[0].text
    print(generated_text)

FastAPI + vLLM

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from vllm import LLM, SamplingParams

app = FastAPI(title="LLM Serving API")

# Initialize model
llm = LLM(
    model="meta-llama/Llama-3-70b-instruct",
    tensor_parallel_size=4
)

class ChatRequest(BaseModel):
    messages: list
    temperature: float = 0.7
    max_tokens: int = 1024
    top_p: float = 0.9

@app.post("/v1/chat/completions")
async def chat(request: ChatRequest):
    try:
        sampling_params = SamplingParams(
            temperature=request.temperature,
            top_p=request.top_p,
            max_tokens=request.max_tokens
        )
        
        outputs = llm.generate(
            [format_messages_for_prompt(request.messages)],
            sampling_params
        )
        
        return {
            "choices": [{
                "message": {
                    "role": "assistant",
                    "content": outputs[0].outputs[0].text
                }
            }]
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

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

TGI (Text Generation Inference)

# Docker Compose для TGI
version: '3.8'

services:
  tgi:
    image: ghcr.io/huggingface/text-generation-inference:2.0
    ports:
      - "8080:80"
    volumes:
      - ./models:/models
    environment:
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
      - MODEL_ID=meta-llama/Llama-3-70b-instruct
      - NUM_SHARD=4
      - MAX_INPUT_TOKENS=4096
      - MAX_TOTAL_TOKENS=8192
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 4
              capabilities: [gpu]

---

Edge / On-device Deployment

ONNX Runtime

import onnxruntime as ort
import numpy as np

# Load ONNX model
session = ort.InferenceSession("llama3-onnx/model.onnx")

# Prepare input
input_ids = np.array([[1, 2, 3, 4, 5]])
attention_mask = np.array([[1, 1, 1, 1, 1]])

# Inference
outputs = session.run(
    output_names=["logits", "past_key_values"],
    input_feed={
        "input_ids": input_ids,
        "attention_mask": attention_mask
    }
)

print(outputs[0])

llama.cpp — CPU inference

from llama_cpp import Llama

# Load model
llm = Llama(
    model_path="./models/llama-3-7b-q4_k_m.gguf",
    n_gpu_layers=-1,  # Use GPU
    n_ctx=4096,
    n_threads=8
)

# Generate
output = llm(
    "Explain deep learning:",
    max_tokens=512,
    temperature=0.7,
    stop=["\n\n"]
)

print(output["choices"][0]["text"])

Core ML (Apple Silicon)

import coremltools as ct
from transformers import AutoModelForCausalLM

# Convert to Core ML
model = AutoModelForCausalLM.from_pretrained("microsoft/Phi-3-mini-4k-instruct")
traced_model = ct.models.detection_problems.DetectionOutput(
    score_threshold=0.5,
    iou_threshold=0.5
)

core_ml_model = ct.convert(
    model,
    inputs=[ct.TensorType(shape=(1, 128))],
    convert_to="mlprogram",
    compute_precision=ct.precision.FLOAT16
)

core_ml_model.save("phi3.mlmodel")

---

Kubernetes Deployment

Helm chart для LLM serving

# charts/llm-serving/values.yaml
replicaCount: 3

image:
  repository: vllm/vllm-serving
  tag: latest

model:
  name: "meta-llama/Llama-3-70b-instruct"
  revision: "main"

resources:
  limits:
    nvidia.com/gpu: 4
  requests:
    memory: "16Gi"
    cpu: "8"

autoscaling:
  enabled: true
  minReplicas: 2
  maxReplicas: 10
  targetGPUUtilization: 75

service:
  type: ClusterIP
  port: 8000

persistence:
  enabled: true
  storageClass: "nvidia-gpu"
  size: 500Gi

K8s deployment manifest

apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-serving
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-serving
  template:
    metadata:
      labels:
        app: llm-serving
    spec:
      containers:
        - name: vllm
          image: vllm/vllm-serving:latest
          ports:
            - containerPort: 8000
          env:
            - name: VLLM_MODEL
              value: "meta-llama/Llama-3-70b-instruct"
            - name: VLLMtensor_parallel_size
              value: "4"
          resources:
            limits:
              nvidia.com/gpu: 4
          volumeMounts:
            - name: model-cache
              mountPath: /model
      volumes:
        - name: model-cache
          persistentVolumeClaim:
            claimName: model-pvc

---

Auto-scaling strategies

HPA с GPU metrics

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-serving
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Pods
      pods:
        metric:
          name: gpu_utilization
        target:
          type: AverageValue
          averageValue: "75"
    - type: Pods
      pods:
        metric:
          name: requests_per_second
        target:
          type: AverageValue
          averageValue: "100"

KEDA для event-driven scaling

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: llm-scaledobject
spec:
  scaleTargetRef:
    name: llm-serving
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: kafka
      metadata:
        bootstrapServers: kafka:9092
        topic: llm-requests
        lagThreshold: "100"

---

Monitoring and Observability

Prometheus metrics

from prometheus_client import Counter, Histogram, generate_latest

# Metrics
request_counter = Counter(
    'llm_requests_total',
    'Total LLM requests',
    ['model', 'status']
)

request_latency = Histogram(
    'llm_request_latency_seconds',
    'LLM request latency'
)

generation_time = Histogram(
    'llm_generation_time_seconds',
    'Time to generate tokens'
)

@request_latency.time()
def generate_with_metrics(prompt):
    request_counter.labels(model="llama-3", status="success").inc()
    return generate(prompt)

Grafana dashboard JSON

{
  "dashboard": {
    "title": "LLM Serving Monitor",
    "panels": [
      {
        "title": "Requests per Second",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(llm_requests_total[5m])"
          }
        ]
      },
      {
        "title": "P50/P95/P99 Latency",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.5, rate(llm_request_latency_seconds_bucket[5m]))"
          },
          {
            "expr": "histogram_quantile(0.95, rate(llm_request_latency_seconds_bucket[5m]))"
          },
          {
            "expr": "histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m]))"
          }
        ]
      },
      {
        "title": "GPU Utilization",
        "type": "gauge",
        "targets": [
          {
            "expr": "gpu_utilization"
          }
        ]
      }
    ]
  }
}

---

Cost comparison

Provider           | $/hour    | $/M tokens  | Min cost
-------------------|-----------|-------------|----------
OpenAI GPT-4o      | N/A      | $5.00       | $0
OpenAI GPT-4o-mini | N/A      | $0.15       | $0
RunPod A100        | $1.50     | Self-hosted | $1.50/h
Lambda A100        | $1.20     | Self-hosted | $1.20/h
AWS p4d            | $32.00    | Self-hosted | $32/h
Self-hosted A100   | $0.10*    | Self-hosted | $500+ GPU

*Amortized GPU cost

---

Best practices

1. Model caching

import os
from pathlib import Path

MODEL_CACHE = os.environ.get("MODEL_CACHE", "/model-cache")

def ensure_model_cached(model_id: str):
    cache_dir = Path(MODEL_CACHE) / model_id
    if not cache_dir.exists():
        from huggingface_hub import snapshot_download
        snapshot_download(
            model_id,
            local_dir=str(cache_dir),
            local_dir_use_symlinks=False
        )
    return str(cache_dir)

2. Health checks

@app.get("/health")
async def health_check():
    try:
        # Check model is loaded
        llm.llm_engine.model_executor.is_model_ready
        
        # Check GPU memory
        free_mem = torch.cuda.mem_get_info()[0]
        if free_mem < 1e9:  # Less than 1GB free
            return {"status": "degraded", "reason": "low memory"}
        
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

3. Graceful shutdown

import signal
import sys

def graceful_shutdown(signum, frame):
    print("Shutting down gracefully...")
    llm.close()
    sys.exit(0)

signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)

---

Итоги

Выбор стратегии deployment зависит от требований к latency, budget и control. Для production используйте self-hosted с auto-scaling, для prototyping — cloud API. Всегда implement monitoring и health checks.