LLM 멀티모달 비전-언어 모델 서빙과 최적화 실전 가이드

# Qwen2.5-VL-7B 단일 GPU 서빙
vllm serve Qwen/Qwen2.5-VL-7B-Instruct \
  --host 0.0.0.0 \
  --port 8000 \
  --max-model-len 8192 \
  --gpu-memory-utilization 0.92 \
  --limit-mm-per-prompt '{"image": 4}' \
  --dtype bfloat16 \
  --enable-prefix-caching

# Qwen2.5-VL-72B 멀티 GPU 서빙 (ViT DP + LLM TP)
vllm serve Qwen/Qwen2.5-VL-72B-Instruct \
  --host 0.0.0.0 \
  --port 8000 \
  --tensor-parallel-size 4 \
  --mm-encoder-tp-mode data \
  --max-model-len 16384 \
  --gpu-memory-utilization 0.95 \
  --limit-mm-per-prompt '{"image": 8}' \
  --dtype bfloat16 \
  --enable-prefix-caching

from openai import OpenAI
import base64
from pathlib import Path


def encode_image_to_base64(image_path: str) -> str:
    """이미지 파일을 base64로 인코딩"""
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")


client = OpenAI(
    api_key="EMPTY",
    base_url="http://localhost:8000/v1",
)

# URL 기반 이미지 입력
response = client.chat.completions.create(
    model="Qwen/Qwen2.5-VL-7B-Instruct",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "이 이미지에서 텍스트를 추출해주세요."},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/document.png",
                        "detail": "high",
                    },
                },
            ],
        }
    ],
    max_tokens=2048,
    temperature=0.1,
)
print(response.choices[0].message.content)

# base64 인코딩 이미지 입력
b64_image = encode_image_to_base64("/data/images/receipt.jpg")
response = client.chat.completions.create(
    model="Qwen/Qwen2.5-VL-7B-Instruct",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "이 영수증의 총 금액은?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/jpeg;base64,{b64_image}",
                    },
                },
            ],
        }
    ],
    max_tokens=512,
)
print(response.choices[0].message.content)

# Qwen3-VL-235B-A22B MoE 서빙 (8 GPU 필수)
vllm serve Qwen/Qwen3-VL-235B-A22B-Instruct \
  --host 0.0.0.0 \
  --port 8000 \
  --tensor-parallel-size 8 \
  --mm-encoder-tp-mode data \
  --enable-expert-parallel \
  --max-model-len 32768 \
  --gpu-memory-utilization 0.95 \
  --limit-mm-per-prompt '{"image": 4, "video": 1}' \
  --dtype bfloat16

# docker-compose.yml - 전처리 워커 + VLM 서버 분리
version: '3.8'
services:
  image-preprocessor:
    image: vlm-preprocessor:latest
    deploy:
      replicas: 4
    environment:
      - MAX_IMAGE_SIZE=2048
      - OUTPUT_FORMAT=base64
      - REDIS_URL=redis://redis:6379/0
    volumes:
      - /data/image-cache:/cache
    cpus: '4'
    mem_limit: 8g

  vlm-server:
    image: vllm/vllm-openai:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 4
              capabilities: [gpu]
    command: >
      --model Qwen/Qwen2.5-VL-72B-Instruct
      --tensor-parallel-size 4
      --mm-encoder-tp-mode data
      --max-model-len 16384
      --gpu-memory-utilization 0.95
    ports:
      - '8000:8000'

  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'

import hashlib
from io import BytesIO
from PIL import Image
import redis
import base64


class ImagePreprocessor:
    """VLM 서빙을 위한 이미지 전처리 파이프라인"""

    def __init__(
        self,
        max_size: int = 2048,
        min_size: int = 28,
        redis_url: str = "redis://localhost:6379/0",
    ):
        self.max_size = max_size
        self.min_size = min_size
        self.cache = redis.from_url(redis_url)
        self.cache_ttl = 3600  # 1시간

    def _compute_hash(self, image_bytes: bytes) -> str:
        return hashlib.sha256(image_bytes).hexdigest()

    def _smart_resize(self, image: Image.Image) -> Image.Image:
        """해상도를 VLM 패치 크기에 맞게 조정 (28의 배수)"""
        w, h = image.size

        # 최대 크기 제한
        if max(w, h) > self.max_size:
            ratio = self.max_size / max(w, h)
            w = int(w * ratio)
            h = int(h * ratio)

        # 28의 배수로 맞춤 (Qwen-VL 패치 크기)
        w = max(self.min_size, (w // 28) * 28)
        h = max(self.min_size, (h // 28) * 28)

        return image.resize((w, h), Image.LANCZOS)

    def process(self, image_bytes: bytes) -> str:
        """이미지를 전처리하고 캐시 적용, base64 반환"""
        img_hash = self._compute_hash(image_bytes)

        # 캐시 히트 확인
        cached = self.cache.get(f"img:{img_hash}")
        if cached:
            return cached.decode("utf-8")

        # 이미지 처리
        image = Image.open(BytesIO(image_bytes)).convert("RGB")
        image = self._smart_resize(image)

        # JPEG 품질 85로 압축 (품질과 크기 균형)
        buffer = BytesIO()
        image.save(buffer, format="JPEG", quality=85)
        b64_result = base64.b64encode(buffer.getvalue()).decode("utf-8")

        # 캐시 저장
        self.cache.setex(f"img:{img_hash}", self.cache_ttl, b64_result)

        return b64_result

# AWQ 양자화 모델로 서빙 (비전 인코더는 BF16 유지)
vllm serve Qwen/Qwen2.5-VL-72B-Instruct-AWQ \
  --host 0.0.0.0 \
  --port 8000 \
  --tensor-parallel-size 2 \
  --mm-encoder-tp-mode data \
  --max-model-len 8192 \
  --quantization awq \
  --gpu-memory-utilization 0.95 \
  --kv-cache-dtype fp8_e5m2 \
  --enable-prefix-caching

from prometheus_client import Histogram, Counter, Gauge

# VLM 전용 메트릭 정의
vlm_image_preprocess_duration = Histogram(
    "vlm_image_preprocess_seconds",
    "이미지 전처리 소요 시간",
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0],
)

vlm_encoder_duration = Histogram(
    "vlm_encoder_inference_seconds",
    "비전 인코더 추론 시간",
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0],
)

vlm_total_image_tokens = Counter(
    "vlm_total_image_tokens",
    "처리된 총 이미지 토큰 수",
)

vlm_image_cache_hits = Counter(
    "vlm_image_cache_hits_total",
    "이미지 인코딩 캐시 히트 수",
)

vlm_image_cache_misses = Counter(
    "vlm_image_cache_misses_total",
    "이미지 인코딩 캐시 미스 수",
)

vlm_active_image_count = Gauge(
    "vlm_active_image_count",
    "현재 처리 중인 이미지 수",
)

vlm_kv_cache_image_tokens_ratio = Gauge(
    "vlm_kv_cache_image_tokens_ratio",
    "KV Cache 중 이미지 토큰 비율",
)

import asyncio
import time
import statistics
from openai import AsyncOpenAI


async def benchmark_vlm_serving(
    base_url: str,
    model: str,
    image_urls: list[str],
    num_requests: int = 100,
    concurrency: int = 10,
):
    """VLM 서빙 벤치마크: TTFT, 처리량, 이미지별 레이턴시 측정"""
    client = AsyncOpenAI(api_key="EMPTY", base_url=base_url)
    ttfts = []
    total_latencies = []
    semaphore = asyncio.Semaphore(concurrency)

    async def single_request(image_url: str):
        async with semaphore:
            start = time.monotonic()
            first_token_time = None

            stream = await client.chat.completions.create(
                model=model,
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": "이 이미지를 설명해주세요."},
                            {
                                "type": "image_url",
                                "image_url": {"url": image_url},
                            },
                        ],
                    }
                ],
                max_tokens=256,
                stream=True,
            )

            async for chunk in stream:
                if first_token_time is None and chunk.choices[0].delta.content:
                    first_token_time = time.monotonic()
                    ttfts.append(first_token_time - start)

            end = time.monotonic()
            total_latencies.append(end - start)

    # 벤치마크 실행
    tasks = [
        single_request(image_urls[i % len(image_urls)])
        for i in range(num_requests)
    ]
    bench_start = time.monotonic()
    await asyncio.gather(*tasks)
    bench_duration = time.monotonic() - bench_start

    # 결과 출력
    print(f"총 요청: {num_requests}, 동시성: {concurrency}")
    print(f"총 소요시간: {bench_duration:.2f}s")
    print(f"처리량: {num_requests / bench_duration:.2f} req/s")
    print(f"TTFT P50: {statistics.median(ttfts)*1000:.0f}ms")
    print(f"TTFT P99: {sorted(ttfts)[int(len(ttfts)*0.99)]*1000:.0f}ms")
    print(f"총 레이턴시 P50: {statistics.median(total_latencies)*1000:.0f}ms")
    print(f"총 레이턴시 P99: {sorted(total_latencies)[int(len(total_latencies)*0.99)]*1000:.0f}ms")