Antigravity から自前 vLLM サーバーで Gemma 4 を本番運用する — 高スループット推論基盤を組み立てる実装ガイド

# docker-compose.yml
# Gemma 4 9B を vLLM で起動する最小構成
# - OpenAI 互換 API が http://localhost:8000/v1 で公開される
# - モデルは初回のみ Hugging Face から DL され、.cache に永続化される
# - GPU がない環境ではこのまま動かないので、--dtype と --device の変更が必要
 
services:
  vllm:
    image: vllm/vllm-openai:latest
    container_name: vllm-gemma4
    restart: unless-stopped
    runtime: nvidia
    environment:
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
      - NVIDIA_VISIBLE_DEVICES=all
    volumes:
      - ./cache:/root/.cache/huggingface
    ports:
      - "8000:8000"
    ipc: host
    command:
      - --model
      - google/gemma-4-9b-it
      - --served-model-name
      - gemma-4-9b
      - --dtype
      - bfloat16
      - --max-model-len
      - "8192"
      - --gpu-memory-utilization
      - "0.90"
      - --api-key
      - ${VLLM_API_KEY}
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 15s
      timeout: 5s
      retries: 10
      start_period: 120s

# .env
HF_TOKEN=hf_YOUR_HUGGINGFACE_TOKEN
VLLM_API_KEY=sk-local-dev-ANY_RANDOM_STRING

docker compose up -d
docker compose logs -f vllm   # "Application startup complete" が出たら準備完了

curl http://localhost:8000/v1/chat/completions \
  -H "Authorization: Bearer sk-local-dev-ANY_RANDOM_STRING" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gemma-4-9b",
    "messages": [{"role": "user", "content": "自己紹介を一文でお願いします"}],
    "max_tokens": 64
  }'

# verify_vllm.py
# Antigravity 設定と同じエンドポイントを Python 側からも検証するスクリプト
# 期待: ネットワーク到達性・認証・モデル ID の 3 点を一度に確認できる
import os
from openai import OpenAI
from openai import APIConnectionError, AuthenticationError, NotFoundError
 
client = OpenAI(
    base_url=os.environ.get("VLLM_BASE_URL", "http://localhost:8000/v1"),
    api_key=os.environ["VLLM_API_KEY"],
)
 
def verify() -> None:
    try:
        resp = client.chat.completions.create(
            model="gemma-4-9b",
            messages=[{"role": "user", "content": "ping"}],
            max_tokens=8,
            timeout=30,
        )
        print("[OK] response:", resp.choices[0].message.content)
    except APIConnectionError as e:
        # 大抵は docker compose up していない or ファイアウォール
        print("[NG] 接続失敗 — vLLM が起動しているか、URL が正しいかを確認:", e)
    except AuthenticationError:
        # API キーがずれている
        print("[NG] 認証エラー — VLLM_API_KEY と起動時の --api-key が一致しているか確認")
    except NotFoundError:
        # served-model-name が違う
        print("[NG] モデル未検出 — --served-model-name と model が一致しているか確認")
 
if __name__ == "__main__":
    verify()

# vllm_benchmark.py
# 複数ワーカーから同時にリクエストを投げ、スループットとレイテンシ分布を測る
# - 本番で想定する「同時接続数」を CONCURRENCY に入れて評価する
# - OOM が出る場合は --gpu-memory-utilization を 0.85 に下げるか、max_model_len を縮める
import os
import time
import asyncio
import statistics
from openai import AsyncOpenAI
 
PROMPTS = [
    "Explain PagedAttention in three sentences.",
    "Write a haiku about self-hosted LLMs.",
    "List 5 reasons to use vLLM.",
    "Summarize why continuous batching matters.",
    "Translate '本番運用' into natural English.",
]
CONCURRENCY = int(os.environ.get("CONCURRENCY", "8"))
TOTAL = int(os.environ.get("TOTAL", "80"))
 
client = AsyncOpenAI(
    base_url=os.environ["VLLM_BASE_URL"],
    api_key=os.environ["VLLM_API_KEY"],
)
 
async def one_call(prompt: str) -> tuple[float, int]:
    start = time.perf_counter()
    resp = await client.chat.completions.create(
        model="gemma-4-9b",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=128,
        temperature=0.7,
    )
    elapsed = time.perf_counter() - start
    tokens = resp.usage.completion_tokens if resp.usage else 0
    return elapsed, tokens
 
async def main() -> None:
    semaphore = asyncio.Semaphore(CONCURRENCY)
 
    async def guarded(i: int):
        async with semaphore:
            return await one_call(PROMPTS[i % len(PROMPTS)])
 
    t0 = time.perf_counter()
    results = await asyncio.gather(*[guarded(i) for i in range(TOTAL)])
    wall = time.perf_counter() - t0
 
    latencies = [r[0] for r in results]
    total_tokens = sum(r[1] for r in results)
    p50 = statistics.median(latencies)
    p95 = statistics.quantiles(latencies, n=20)[18]
 
    print(f"concurrency: {CONCURRENCY} / total: {TOTAL}")
    print(f"wall: {wall:.2f}s  throughput: {total_tokens/wall:.1f} tok/s")
    print(f"latency p50: {p50:.2f}s  p95: {p95:.2f}s")
 
if __name__ == "__main__":
    asyncio.run(main())

    command:
      - --model
      - google/gemma-4-9b-it
      - --quantization
      - awq              # "awq" / "fp8" / "bitsandbytes" / "gptq" などから選ぶ
      - --dtype
      - auto

# proxy.py
# Antigravity -> このプロキシ -> vLLM の順で経由させ、レート制限とタイムアウトを集中管理する
# 運用ポイント:
# - VLLM_UPSTREAM は内部ネットワークのアドレスを推奨（外部に vLLM を直接さらさない）
# - Authorization ヘッダは clients_keys に登録された API キーのみ通す
# - 429 は必ず Retry-After を返し、Antigravity 側のリトライを健全に動かす
import os
import time
import asyncio
from collections import defaultdict, deque
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse
import httpx
 
VLLM_UPSTREAM = os.environ["VLLM_UPSTREAM"]  # 例: http://vllm:8000
VLLM_API_KEY = os.environ["VLLM_API_KEY"]
CLIENT_KEYS = set(os.environ["CLIENT_KEYS"].split(","))  # カンマ区切りで配布
RATE_PER_MIN = int(os.environ.get("RATE_PER_MIN", "60"))
MAX_PROMPT_CHARS = int(os.environ.get("MAX_PROMPT_CHARS", "32000"))
UPSTREAM_TIMEOUT = float(os.environ.get("UPSTREAM_TIMEOUT", "60"))
 
app = FastAPI()
bucket: dict[str, deque[float]] = defaultdict(deque)
lock = asyncio.Lock()
 
def _check_and_record(key: str) -> None:
    now = time.time()
    window = bucket[key]
    while window and now - window[0] > 60:
        window.popleft()
    if len(window) >= RATE_PER_MIN:
        retry = int(60 - (now - window[0]))
        raise HTTPException(429, "rate limit", headers={"Retry-After": str(retry)})
    window.append(now)
 
async def _guard(request: Request) -> str:
    # Authorization: Bearer <token> を想定
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer ") or auth[7:] not in CLIENT_KEYS:
        raise HTTPException(401, "bad client key")
    key = auth[7:]
    body = await request.body()
    if len(body) > MAX_PROMPT_CHARS * 4:  # 大雑把な上限
        raise HTTPException(413, "prompt too large")
    async with lock:
        _check_and_record(key)
    return key
 
@app.post("/v1/chat/completions")
async def proxy(request: Request):
    await _guard(request)
    headers = {
        "Authorization": f"Bearer {VLLM_API_KEY}",
        "Content-Type": "application/json",
    }
    body = await request.body()
    async with httpx.AsyncClient(timeout=UPSTREAM_TIMEOUT) as client:
        try:
            r = await client.post(
                f"{VLLM_UPSTREAM}/v1/chat/completions",
                content=body,
                headers=headers,
            )
        except httpx.ReadTimeout:
            raise HTTPException(504, "upstream timeout")
    return StreamingResponse(iter([r.content]), media_type=r.headers.get("content-type", "application/json"), status_code=r.status_code)
 
@app.get("/health")
async def health():
    async with httpx.AsyncClient(timeout=5) as client:
        r = await client.get(f"{VLLM_UPSTREAM}/health")
        return {"vllm": r.status_code}

# NG: 既定値（KV キャッシュの余裕がなくなる）
--max-model-len 32768
--gpu-memory-utilization 0.95
 
# OK: 実際のプロンプトサイズに合わせて縮める
--max-model-len 8192
--gpu-memory-utilization 0.88

    command:
      - --model
      - google/gemma-4-9b-it
      - --enable-lora
      - --max-loras
      - "4"
      - --max-lora-rank
      - "32"
      - --lora-modules
      - review=/adapters/code-review-lora
      - summary=/adapters/summary-lora
    volumes:
      - ./cache:/root/.cache/huggingface
      - ./adapters:/adapters

# lora_switch.py
# 同じベースモデルに対して、タスクごとに異なる LoRA アダプタを指名する
# - model="gemma-4-9b" なら素のベースモデル
# - model="review" なら code-review-lora が乗る
from openai import OpenAI
import os
 
client = OpenAI(
    base_url=os.environ["VLLM_BASE_URL"],
    api_key=os.environ["VLLM_API_KEY"],
)
 
def review(code: str) -> str:
    resp = client.chat.completions.create(
        model="review",  # LoRA 名をそのまま指定
        messages=[
            {"role": "system", "content": "あなたは厳格だが丁寧な日本語のコードレビュアーです。"},
            {"role": "user", "content": f"以下のコードをレビューしてください:\n\n{code}"},
        ],
        max_tokens=512,
        temperature=0.2,
    )
    return resp.choices[0].message.content or ""
 
def summary(text: str) -> str:
    resp = client.chat.completions.create(
        model="summary",
        messages=[{"role": "user", "content": f"3 行で要約してください:\n{text}"}],
        max_tokens=256,
        temperature=0.4,
    )
    return resp.choices[0].message.content or ""
 
if __name__ == "__main__":
    print(review("def add(a, b): return a - b"))