AI エンジニアリングデプロイ

ノートPCでモデルを動かし、応答生成が速くて品質も良い状態を実現していたのに、それをサーバーにデプロイして100人のユーザーが同時に呼び出すと、状況は一変します。あるユーザーは返信を受け取るまでに10秒待たされ、あるリクエストはそのままタイムアウトエラーになり、GPU メモリはあっという間にいっぱいになります。請求額は、心が痛むほど跳ね上がります。

これが AI デプロイが解決すべき問題です。動作するモデルを、使いやすいサービスに変えることです。。

従来の Web サービスのボトルネックは通常、CPU とデータベースです。

AI サービスのボトルネックは主に GPU にあります。GPU のメモリが十分かどうか、計算が速いかどうか、同時リクエストをどうキューイングするかです。

AI デプロイの中心的な課題は、遅延、スループット、コストの3つの言葉にまとめられます。

指標	意味	ユーザー体感
遅延（Latency）	ユーザーがリクエストを送信してから最初の文字を受け取るまでの時間	速いかどうか
スループット（Throughput）	1秒間に処理できるリクエスト数	同時に多くの人にサービスを提供できるか
コスト（Cost）	サービス運用にかかる GPU/サーバー費用	高価かどうか

優れた AI デプロイとは、この3つの間でバランスを取ることです。遅延は十分低く、スループットは十分高く、コストは制御可能な状態に保ちます。

モデルサービス化フレームワーク

学習済みモデルを API サービスとしてラップするには、専用のフレームワークが必要です。この節では、vLLM、TGI、Ollama の3つの主要な選択肢を紹介します。

vLLM：PagedAttention による高性能推論

vLLM は UC Berkeley チームが開発した推論エンジンで、最大の特徴は高速であることです。

vLLM の核となる革新はPagedAttention——オペレーティングシステムの仮想メモリページングに着想を得た、GPU メモリを効率的に管理する技術です。

従来の推論フレームワークでは、各リクエストの KV Cache（キーバリューキャッシュ）が連続した GPU メモリ空間を占有する必要があります。

リクエストの長さが異なる場合、GPU メモリは断片化し、利用率が非常に低くなります。

PagedAttention は GPU メモリを固定サイズのページに分割し、各リクエストの KV Cache を異なるページに分散して保存し、ページテーブルで位置を記録します。

これにより GPU メモリの利用率が大幅に向上し、より多くのリクエストを同時に処理できます。

例

# vLLM をインストール
pip install vllm

# vLLM で OpenAI 互換の API サービスを起動
# --model でモデルを指定、--host と --port で待受アドレスを指定
# --tensor-parallel-size で並列推論に使う GPU の枚数を指定
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192

サービス起動後、OpenAI SDK で直接呼び出せます：

例

# ファイルパス: test_vllm_client.py
from openai import OpenAI

# ローカル vLLM サービスに接続
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="サンプル-demo-key" # vLLM はデフォルトで実際の API キー不要
)

# チャット補完を呼び出し
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "あなたは親切な AI アシスタントです。"},
{"role": "user", "content": "サンプルチュートリアルを一言で紹介してください"}
],
temperature=0.7,
max_tokens=500,
stream=True # ストリーミング出力
)

print("AI 返信：", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

vLLM は Python コードでカスタムサービスを直接記述することもサポートしています：

例

# ファイルパス: custom_vllm_server.py
from vllm import LLM, SamplingParams

# モデルを初期化
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
gpu_memory_utilization=0.9, # GPU メモリ使用割合
tensor_parallel_size=1, # テンソル並列度（GPU の枚数）
max_model_len=8192, # 最大コンテキスト長
)

# サンプリングパラメータを設定
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=500,
stop=["</s>"],
)

# バッチ推論
prompts = [
"サンプルを紹介してください",
"AI とは何ですか？",
"プログラミングを学ぶには？",
]

# 返信を生成
outputs = llm.generate(prompts, sampling_params)

# 結果を出力
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Prompt: {prompt}")
print(f"Generated: {generated_text}")
print("-" * 50)

TGI：Hugging Face の推論エンジン

TGI（Text Generation Inference）は Hugging Face が提供する推論サービスフレームワークです。

その特徴は、エコシステムが充実しており、Hugging Face Hub と深く統合されていることです。Flash Attention、動的バッチ処理などの最適化をサポートします。

例

# Docker で TGI を実行（最も簡単な方法）
# 注意：Docker をインストールし、GPU サポートを構成する必要があります
docker run -d \
--gpus all \
-p 8080:80 \
-v $PWD/data:/data \
--name tgi-server \
ghcr.io/huggingface/text-generation-inference:latest \
--model-id Qwen/Qwen2.5-7B-Instruct \
--max-input-length 4096 \
--max-total-tokens 8192 \
--max-batch-prefill-tokens 4096

# または Python で直接インストール（開発環境）
# pip install text-generation

TGI サービスを呼び出します：

例

# ファイルパス: test_tgi_client.py
from text_generation import Client

# TGI サービスに接続
client = Client("http://localhost:8080", timeout=60)

# 非ストリーミング呼び出し
response = client.generate(
"サンプルチュートリアルを紹介してください",
max_new_tokens=500,
temperature=0.7,
top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# ストリーミング呼び出し
print("ストリーミング出力：")
for chunk in client.generate_stream(
"Python で Hello World を書いてください",
max_new_tokens=200,
):
if not chunk.token.special:
print(chunk.token.text, end="", flush=True)
print()

Ollama：ローカルデモから本番デプロイまで

Ollama は「ワンクリックでローカルモデルを実行」で知られ、多くの人が開発やデモに使用しています。

しかし実際には、Ollama は小規模な本番環境へのデプロイにも適しています。シンプルで安定しており、リソース消費も制御可能です。

例

# Ollama をインストール（Linux サーバー）
curl -fsSL https://ollama.com/install.sh | sh

# Ollama がすべてのネットワークインターフェースをリッスンするよう設定（本番環境ではファイアウォールに注意）
# systemd サービスファイルを編集
sudo systemctl edit ollama.service

# 以下を追加：
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# サービスを再起動
sudo systemctl restart ollama

# モデルをプルして実行
ollama pull qwen2.5:7b

API 経由で Ollama を呼び出します：

例

# ファイルパス: test_ollama_client.py
import requests
import json

# Ollama API アドレス
base_url = "http://localhost:11434/api"

# チャットインターフェース（ストリーミング）
def chat_stream(model, messages):
url = f"{base_url}/chat"
data = {
"model": model,
"messages": messages,
"stream": True,
"options": {
"temperature": 0.7,
"num_ctx": 8192,
}
}

response = requests.post(url, json=data, stream=True)
full_response = ""

for line in response.iter_lines():
if line:
chunk = json.loads(line)
if "message" in chunk:
content = chunk["message"].get("content", "")
full_response += content
print(content, end="", flush=True)
if chunk.get("done", False):
break

print()
return full_response

# 呼び出し例
messages = [
{"role": "system", "content": "あなたは親切な AI アシスタントです。"},
{"role": "user", "content": サンプルチュートリアルを紹介してください}
]

print(AI応答：, end="")
result = chat_stream("qwen2.5:7b", messages)

フレームワーク選定比較

3つのフレームワークにはそれぞれ特徴があります。どう選ぶ？あなたのニーズ次第です。

フレームワーク	利点	欠点	適用シーン
vLLM	性能最強、スループット最高	設定が比較的複雑	高並行本番環境、性能の追求
TGI	Hugging Face エコシステム、デプロイ簡単	性能はvLLMよりやや低い	すでにHugging Faceツールチェーンを使用中
Ollama	超シンプルなデプロイ、保守コストが低い	超大規模な並行には不向き	中小規模サービス、迅速な立ち上げ

初期段階：まずOllamaを使う。シンプルで信頼性が高い。トラフィックが増えたら：vLLMに移行。性能優先。

推論最適化技術

同じモデルでも、実行方法が異なると速度とコストが数倍変わる可能性がある。

このセクションでは本番環境でよく使われる推論最適化技術を紹介する。

連続バッチ処理（Continuous Batching）

従来のバッチ処理は静的：十分なリクエストが集まったら、まとめて推論し、すべて完了してから次のバッチを処理する。

問題は：リクエストによって短いもの（例えば10文字だけ生成）もあれば、長いもの（500文字生成）もある。

短いリクエストは長いリクエストの完了を待たなければならず、リソースをまとめて解放できないため、GPU利用率が上がらない。

連続バッチ処理は動的：リクエストが完了したら、すぐにキュー内の新しいリクエストを追加する。バッチ全体の完了を待つ必要はない。

これによりGPUはほぼ常にフル稼働し、スループットが大幅に向上する。

vLLMとTGIには連続バッチ処理が組み込まれている。手動実装は不要で、有効にするだけ。

投機的デコーディング（Speculative Decoding）

大規模モデルが1文字生成するには、完全な順伝播を1回呼び出す必要があり、遅い。

投機的サンプリングの考え方は：小さなモデルで後続の数文字を"推測"し、大規模モデルに一括で検証させる。

小さなモデルは推測が速く、必ずしも全問正解ではないが、正解した部分は一括出力できる。

間違った部分は捨てて、再び推測する。

これにより全体の速度は2〜3倍向上でき、出力品質も変わらない。

例

# vLLMで投機的サンプリングを有効にする
# --speculative-model は小さなモデル（ドラフトモデル）を指定する
# --num-speculative-tokens は毎回推測する文字数を指定する
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--speculative-model Qwen/Qwen2.5-0.5B-Instruct \
--num-speculative-tokens 5 \
--host 0.0.0.0 \
--port 8000

量子化推論：AWQ と GPTQ

モデル量子化は、モデルの重みを16ビット浮動小数点数（FP16）から8ビットまたは4ビットの整数に圧縮すること。

VRAM使用量が半分以下になり、推論速度もさらに速くなる。

最も一般的な2つの量子化方式は AWQ と GPTQ です。

量子化方式	特徴	精度損失	推論速度
AWQ	アクティベーション認識重み量子化	極小	快
GPTQ	層ごとの最適化量子化	小	快
GGUF	ローカル実行に適する	許容可能	快

例

# vLLMでAWQ量子化モデルを直接実行する
# 多くの量子化モデルはHugging Faceから直接ダウンロードできる
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Qwen2.5-7B-Instruct-AWQ \
--quantization awq \
--host 0.0.0.0 \
--port 8000

# またはOllamaでGGUF量子化モデルを実行する
ollama run qwen2.5:7b-instruct-q4_0

PythonでAWQ量子化モデルをロードする：

例

# ファイルパス：awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# AWQ量子化モデルをロード
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# 注意：実際に実行するには autoawq ライブラリのインストールが必要
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)

# 推論
messages = [
{"role": "system", "content": "あなたは親切なAIアシスタントです。"},
{"role": "user", "content": "サンプルを紹介"}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.7,
)

response = tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
print(response)

モデル蒸留

蒸留は、大規模モデル（教師）を使って小さなモデル（生徒）を訓練し、小さなモデルに大規模モデルの能力を学習させること。

小さなモデルはより速く、より安価で、効果はやや劣るが、多くのシーンで十分に使える。

例えば、Qwen2.5-72Bを教師として、Qwen2.5-7Bの生徒を訓練すると、効果は72Bに近い可能性があるが、速度とコストは7Bレベル。

蒸留の利点は通常次の通り：モデルサイズが1/10に縮小、推論速度が5〜10倍向上、効果は90%以上保持。

API サービス設計

モデルをラップするのは最初の一歩にすぎない。本番環境のAPIには認証、レート制限、ストリーミング応答、エラー処理など多くの機能が必要。

FastAPIはこれに最適な選択肢です。

FastAPI で LLM サービスをラップ

認証、レート制限、ストリーミング応答を含む完全なAPIサービスを作成しましょう。

例

# ファイルパス：llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. アプリケーションと設定の初期化
# ============================================

app = FastAPI(title="サンプル LLM API", version="1.0.0")

# シンプルなAPI Key認証（本番環境ではより堅牢な方式を推奨）
VALID_API_KEYS = {
"sk-サンプル-123456": {"user": "demo_user", "quota": 1000},
"sk-サンプル-789012": {"user": "pro_user", "quota": 10000},
}

# シンプルなレートリミッター（メモリ実装、本番環境ではRedisを推奨）
class RateLimiter:
def __init__(self):
self.requests = defaultdict(list) # key -> [timestamp, ...]

async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
"""レート制限を超えているかチェック"""
now = time.time()
# 期限切れレコードをクリーンアップ
self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
# 制限超過かどうかをチェック
if len(self.requests[key]) >= max_requests:
return False
# このリクエストを記録
self.requests[key].append(now)
return True

rate_limiter = RateLimiter()

# ============================================
# 2. データモデル
# ============================================

class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"

class Message(BaseModel):
role: Role = Field(..., description="メッセージロール")
content: str = Field(..., description="メッセージ内容")

class ChatCompletionRequest(BaseModel):
model: str = Field(..., description="モデル名")
messages: List[Message] = Field(..., description="会話メッセージリスト")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="サンプリング温度")
top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Top-P サンプリング")
max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="最大生成トークン数")
stream: Optional[bool] = Field(default=False, description="ストリーミング出力かどうか")

class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[dict]
usage: dict

# ============================================
# 3. 認証依存関係
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""API Keyを検証し、ユーザー情報を返す"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key"
)
return VALID_API_KEYS[api_key]

# ============================================
# 4. LLM生成のシミュレーション（実際のプロジェクトでは実モデル呼び出しに置き換え）
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
"""LLM生成プロセスのシミュレーション"""
# ここでは単なるシミュレーション。実際のプロジェクトでは実モデルを呼び出すこと
response_text = (
"こんにちは！サンプルのAIアシスタントです。\n\n"
"サンプルはPython、Java、C++、フロントエンド開発などの豊富なプログラミングチュートリアルを提供しています。\n"
"https://www.example.com にアクセスすると、さらに多くのコンテンツを学べます！\n\n"
"何かお手伝いできることはありますか？"
)
# 生成遅延をシミュレーション
await asyncio.sleep(0.5)
return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
"""LLMストリーミング生成をシミュレーション"""
response_text = await mock_llm_generate(messages, max_tokens)
# 1文字ずつ出力してストリーミング効果をシミュレーション
for char in response_text:
yield char
await asyncio.sleep(0.03) # 各文字30ms遅延

# ============================================
# 5. APIエンドポイント
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
request: ChatCompletionRequest,
user: dict = Depends(get_current_user)
):
"""チャット補完インターフェース、OpenAI形式と互換性あり"""

# レート制限チェック：各ユーザー毎分最大60リクエスト
user_key = user["user"]
if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded"
)

# リクエストIDを生成
request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(datetime.now().timestamp())

if request.stream:
# ストリーミング応答
async def stream_generator():
async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
# SSE形式のチャンクを構築
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": created,
"model": request.model,
"choices": [{
"index": 0,
"delta": {"content": token},
"finish_reason": None
}]
}
yield f"data: {__import__('json').dumps(chunk)}\n\n"
# 終了マーカーを送信
yield "data: [DONE]\n\n"

return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)

else:
# 非ストリーミング応答
response_text = await mock_llm_generate(request.messages, request.max_tokens)

return ChatCompletionResponse(
id=request_id,
created=created,
model=request.model,
choices=[{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}],
usage={
"prompt_tokens": sum(len(m.content) for m in request.messages),
"completion_tokens": len(response_text),
"total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
}
)

@app.get("/health")
async def health_check():
"""ヘルスチェックエンドポイント"""
return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
"""利用可能なモデルを一覧表示"""
return {
"object": "list",
"data": [
{
"id": "qwen2.5-7b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "サンプル"
},
{
"id": "llama-3.1-8b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "サンプル"
}
]
}

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

サービスを起動：

例

# 依存関係をインストール
pip install fastapi uvicorn python-multipart pydantic

# サービスを起動
python llm_api_server.py

# 別のターミナルでテスト
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-サンプル-123456" \
-d '{
"model": "qwen2.5-7b-instruct",
"messages": [
{"role": "user", "content": "こんにちは"}
],
"stream": false
}'

Pythonクライアントで呼び出す：

例

# ファイルパス：client_example.py
from openai import OpenAI

# 自分たちのAPIサービスに接続
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="sk-サンプル-123456"
)

# 非ストリーミングをテスト
print("非ストリーミング呼び出し：")
response = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "サンプルを紹介"}],
stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# ストリーミングをテスト
print("ストリーミング呼び出し：")
stream = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "サンプルを紹介"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

OpenAI 互換インターフェースの利点

APIをOpenAI互換形式にすると、いくつかの明確な利点がある：

ユーザーはコードを変更する必要がなく、base_url と api_key を変更するだけでよい。

豊富なエコシステム——OpenAIをサポートするあらゆるSDK、ツール、アプリケーションがそのままあなたのサービスを利用できる。

独自にAPI形式を設計する必要はなく、OpenAIの設計はすでに十分に考慮されている。

本番環境の認証とレートリミット

前述の例ではシンプルなメモリ内レート制限を使用したが、本番環境ではRedisを使用した分散レート制限を推奨する。

例

# ファイルパス：redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
"""Redisベースの分散レートリミッター"""

def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)

async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int
) -> bool:
"""
スライディングウィンドウレート制限アルゴリズム
key: 限流键（例：user_id や api_key）
max_requests: ウィンドウ内の最大リクエスト数
window_seconds: ウィンドウサイズ（秒）
"""
now = time.time()
window_start = now - window_seconds

# Redis キー
redis_key = f"rate_limit:{key}"

# pipeline アトミック操作を使用
async with self.redis.pipeline() as pipe:
# 1. ウィンドウ外のレコードを削除
pipe.zremrangebyscore(redis_key, 0, window_start)
# 2. 現在のウィンドウ内のリクエスト数を集計
pipe.zcard(redis_key)
# 3. 現在のリクエストを追加
pipe.zadd(redis_key, {str(now): now})
# 4. 有効期限を設定
pipe.expire(redis_key, window_seconds)
# 実行
_, current_count, _, _ = await pipe.execute()

return current_count < max_requests

async def close(self):
await self.redis.close()

# 使用例
async def main():
limiter = RedisRateLimiter("redis://localhost:6379")

# テスト：同一ユーザーは毎分最大10回のリクエスト
user_id = "user_123"
for i in range(15):
allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
print(f"リクエスト {i+1}: {'許可' if allowed else '拒否'}")

await limiter.close()

if __name__ == "__main__":
import asyncio
asyncio.run(main())

認証とレート制限はAPIのインフラ基盤であり、多少複雑になっても構わない。突破されてから後悔するよりはましだ。

負荷分散とスケーリング

ユーザーが増えるにつれて、1台のサーバーでは捌ききれなくなる。マシンを追加し、トラフィックを複数台のマシンに分散させる必要がある。

これが負荷分散と自動スケーリングで解決すべき問題だ。

ステートレスサービス設計

まず覚えておくべきこと：AIサービスはステートレスに設計する。

ステートレスとは：リクエスト間でメモリを共有せず、どのサーバーがどのリクエストを処理してもよいということ。

そうすればスケールアウトは簡単だ：マシンを追加するだけでよく、データ同期を考慮する必要がない。

セッション状態（例：会話履歴）はクライアント側に保存するか（毎回ユーザーに送信してもらう）、Redisに保存すればよい。

Nginx リバースプロキシ

Nginxは最も一般的なロードバランサーで、トラフィックを複数のバックエンドサーバーに分散できる。

例

# ファイルパス：nginx.conf
# Nginx設定：複数台のLLMサービスの負荷分散

upstream llm_backends {
# 最小接続数アルゴリズム：現在の接続数が最も少ないサーバーにリクエストを送信
least_conn;

# バックエンドサーバーリスト
server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

# 複数種類のGPUがある場合、weightでトラフィックの比率を割り当てられる
# weight=2 は、このサーバーが他の2倍のトラフィックを受け持つことを意味する
}

server {
listen 80;
server_name api.example.com;

# APIリクエストの転送
location /v1/ {
# バックエンドに転送
proxy_pass http://llm_backends;

# タイムアウト設定（AI生成は遅い可能性がある）
proxy_connect_timeout 10s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;

# クライアントの実際のIPを渡す
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# ストリーミングレスポンスに必要な設定
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
}

# ヘルスチェックエンドポイント
location /health {
proxy_pass http://llm_backends/health;
proxy_connect_timeout 5s;
proxy_read_timeout 5s;
}

# レート制限：各IPは毎秒最大10リクエスト
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
}

Kubernetes HPA による自動スケーリング

Kubernetesでデプロイしている場合、HPA（Horizontal Pod Autoscaler）を使用して負荷に応じて自動でスケールイン・アウトできる。

例

# ファイルパス：llm-deployment.yaml
# LLMサービスのKubernetesデプロイ設定

# 1. Deployment：LLM Podをデプロイ
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-server
namespace: サンプル
spec:
replicas: 3 # 初期レプリカ数
selector:
matchLabels:
app: llm-server
template:
metadata:
labels:
app: llm-server
spec:
containers:
- name: llm-server
image: サンプル/llm-server:latest
ports:
- containerPort: 8000
resources:
requests:
cpu: "4"
memory: "16Gi"
nvidia.com/gpu: 1 # GPUを1つ要求
limits:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: 1 # GPUを1つに制限
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300 # モデルのロードに時間が必要
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300
periodSeconds: 10
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"

---
# 2. Service：負荷分散Service
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: サンプル
spec:
selector:
app: llm-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

---
# 3. HPA：自動スケーリング
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-hpa
namespace: サンプル
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-server
minReplicas: 3 # 最小3 Pod
maxReplicas: 20 # 最大20 Pod
metrics:
# CPU使用率に基づいてスケールアウト
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# GPU使用率に基づいてスケールアウト（カスタムメトリクスが必要）
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: 80
# キューの長さに基づいてスケールアウト（カスタムメトリクス）
- type: Pods
pods:
metric:
name: pending_requests
target:
type: AverageValue
averageValue: 10
behavior:
scaleUp:
stabilizationWindowSeconds: 60 # スケールアウト前に60秒観察
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # スケールイン前に5分観察
policies:
- type: Percent
value: 20
periodSeconds: 120

GPU リソースプール管理

GPUは高価なので、最大限に活用する必要がある。

一般的な戦略は：モデルタイプごとにグループ化し、同じモデルのPodをまとめて配置する。

例：

リソースプール	GPUモデル	実行モデル	用途
small-pool	RTX 3090 / L4	7Bモデル	通常トラフィック
medium-pool	A10 / L40	13B-70Bモデル	複雑なタスク
large-pool	A100 / H100	70B+モデル	高価値ユーザー
spot-pool	各種マシンタイプ	オフラインのバッチ処理	低価格のSpotインスタンスを使用

AI モデルの A/B テスト

新バージョンのモデルをリリースしたとき、旧バージョンより優れているかどうかをどう判断するか？A/Bテストによる。

A/Bテストの核心は：トラフィックを複数のグループに分け、グループごとに異なるモデルを使い、指標を比較する。

トラフィック分割戦略

一般的なトラフィック分割方法はいくつかある：

例

# ファイルパス：ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
"""A/Bテストの一種"""
name: str # バリアント名
model_name: str # モデル名
weight: int # トラフィックの重み
config: dict # その他の設定

class ABTestRouter:
"""A/Bテストトラフィックルーター"""

def __init__(self, test_name: str, variants: List[Variant]):
self.test_name = test_name
self.variants = variants
# 総重みを計算
total_weight = sum(v.weight for v in variants)
# セグメント範囲を構築
self.ranges: List[Tuple[int, int, Variant]] = []
current = 0
for v in variants:
self.ranges.append((current, current + v.weight, v))
current += v.weight
self.total_weight = total_weight

def get_variant_by_user(self, user_id: str) -> Variant:
"""
ユーザーIDに基づく安定した分割
同じユーザーは常に同じグループに割り当てられ、体験の不一致を避ける
"""
# ハッシュでユーザーのスコアを計算
hash_input = f"{self.test_name}-{user_id}".encode()
hash_value = hashlib.md5(hash_input).hexdigest()
# 0-999の数値に変換
score = int(hash_value[:4], 16) % 1000
# 対応するバリアントを検索
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

def get_variant_random(self) -> Variant:
"""ランダム分割（粘着性が不要なシーン向け）"""
score = random.randint(0, self.total_weight - 1)
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

# 使用例
def main():
# 3つのバリアントを定義
variants = [
Variant(
name="control",
model_name="Qwen2.5-7B-Instruct",
weight=50, # 50% トラフィック
config={"temperature": 0.7}
),
Variant(
name="treatment-v1",
model_name="Qwen2.5-14B-Instruct",
weight=30, # 30% トラフィック
config={"temperature": 0.7}
),
Variant(
name="treatment-v2",
model_name="Llama-3.1-8B-Instruct",
weight=20, # 20% トラフィック
config={"temperature": 0.8}
)
]

router = ABTestRouter("model-comparison-v1", variants)

# ユーザーの分割をテスト
test_users = [f"user_{i}" for i in range(10)]
for user_id in test_users:
variant = router.get_variant_by_user(user_id)
print(f"ユーザー {user_id} → グループ {variant.name} → モデル {variant.model_name}")

# 分布を集計
count: Dict[str, int] = {}
for i in range(10000):
variant = router.get_variant_random()
count[variant.name] = count.get(variant.name, 0) + 1
print("\nランダム分割の統計（10000回）：")
for name, cnt in count.items():
print(f" {name}: {cnt} 回 ({cnt/100}%)")

if __name__ == "__main__":
main()

指標の定義と収集

A/Bテストでは「どのモデルの応答がより賢いか」だけを見るのではなく、定量化可能な指標を定義する必要がある。

指標タイプ	具体的な指標	説明
システム指標	レイテンシ、スループット、エラー率	モデルが速いか、安定しているか
ユーザー行動	ユーザー満足度、会話ターン数、継続率	ユーザーが気に入っているか
ビジネス指標	コンバージョン率、リテンション率、収益	ビジネスに役立つか
品質指標	人手評価、有害率、事実正確性	出力品質が良いか

例

# ファイルパス：metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
"""チャットリクエストの完全な指標"""
# 基本情報
request_id: str
user_id: str
variant_name: str # A/Bテストグループ名
model_name: str

# 時間指標
timestamp: str
ttft: float # Time to First Token
total_time: float # 総所要時間

# 生成指標
prompt_tokens: int
completion_tokens: int
total_tokens: int

# ビジネス指標
user_rating: Optional[int] = None # ユーザー評価（1-5）
is_error: bool = False
error_message: Optional[str] = None

class MetricsCollector:
"""メトリクスコレクター"""

def __init__(self):
# 実際のプロジェクトでは、ここにKafkaやデータベースへ書き込むべき
self.buffer = []

def collect(self, metrics: ChatMetrics):
"""1つのメトリクスを収集"""
self.buffer.append(asdict(metrics))
# 簡単に出力。実際のプロジェクトではストレージに書き込む
print(f"[Metrics] {json.dumps(asdict(metrics), ensure_ascii=False)}")

# バッファが満杯になったらバッチ書き込み
if len(self.buffer) >= 100:
self.flush()

def flush(self):
"""バッチでストレージに書き込む"""
if self.buffer:
# 実際のプロジェクトでは：ClickHouse、BigQueryなどに書き込む
print(f"Flushed {len(self.buffer)} metrics")
self.buffer = []

# 使用例
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
"""1回のチャットリクエストを処理し、メトリクスを記録"""
request_id = str(uuid.uuid4())

# 分割
variant = router.get_variant_by_user(user_id)

# 開始時間を記録
start_time = time.time()
ttft = 0.0
prompt_tokens = len(prompt)
completion_tokens = 0

try:
# ここは実際のモデル呼び出しに置き換える
# response = call_model(variant.model_name, prompt)
response = "これはAIの応答です"
completion_tokens = len(response)

# TTFTをシミュレート（0.2秒後に最初のトークンを受信したと仮定）
ttft = 0.2

# 成功メトリクスを記録
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
is_error=False
)
collector.collect(metrics)
return response, request_id

except Exception as e:
# エラーメトリクスを記録
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=0,
total_tokens=prompt_tokens,
is_error=True,
error_message=str(e)
)
collector.collect(metrics)
raise

# ユーザーフィードバック
def submit_user_feedback(request_id: str, rating: int):
"""ユーザーが評価フィードバックを送信"""
# 実際のプロジェクトでは：データベース内の対応するレコードを更新
print(f"Request {request_id} got rating {rating}")

統計的有意性の判断

データを取得した後、A群52%、B群55%だからといって、単純にB群の方が良いとは言えない。

統計的有意性検定を行い、差異がランダムな変動によるものではないことを確認する必要がある。

例

# ファイルパス：ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
control_conv: float,
treatment_conv: float,
control_total: int,
treatment_total: int
) -> Tuple[float, float]:
"""
Z検定のp値を計算
control_conv: 対照群のコンバージョン率
treatment_conv: 実験群のコンバージョン率
control_total: 対照群のサンプル数
treatment_total: 実験群のサンプル数
"""
# プールされたコンバージョン率
p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
# 標準誤差
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
# Z統計量
z_score = (treatment_conv - control_conv) / se_pool
# p値を計算（両側検定）
p_value = 2 * (1 - normal_cdf(abs(z_score)))
return z_score, p_value

def normal_cdf(x: float) -> float:
"""標準正規分布の累積分布関数（近似）"""
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
control_conversions: int,
control_total: int,
treatment_conversions: int,
treatment_total: int,
alpha: float = 0.05
):
"""A/Bテスト結果の分析"""
control_conv = control_conversions / control_total
treatment_conv = treatment_conversions / treatment_total

z_score, p_value = calculate_z_test(
control_conv, treatment_conv,
control_total, treatment_total
)

relative_lift = (treatment_conv - control_conv) / control_conv

print("A/Bテスト結果の分析")
print("=" * 50)
print(f"対照群: {control_conversions}/{control_total} ({control_conv:.2%})")
print(f"実験群: {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
print(f"相対リフト: {relative_lift:+.2%}")
print("-" * 50)
print(f"Z統計量: {z_score:.4f}")
print(f"p-value: {p_value:.4f}")
print("-" * 50)

if p_value < alpha:
print(f✅ 結果は統計のに有意（p < {alpha}）)
if relative_lift > 0:
print(" 実験群の方が優れています！")
else:
print(" 対照群の方が優れています！")
else:
print(f"❌ 結果は有意ではない（p >= {alpha}）")
print(" どちらのバージョンが良いか判断できないため、より多くのデータが必要です")

# 例：A/Bテスト結果を仮定
if __name__ == "__main__":
# シナリオ 1：有意な差がある
print("シナリオ 1：有意な差がある")
analyze_ab_test(
control_conversions=450, # 対照群 450人が満足
control_total=1000, # 対照群は合計1000人
treatment_conversions=520, # 実験群 520人が満足
treatment_total=1000 # 実験群は合計1000人
)

print("\n" + "="*50 + "\n")

# シナリオ 2：有意な差がない
print("シナリオ 2：有意な差がない")
analyze_ab_test(
control_conversions=48,
control_total=100,
treatment_conversions=52,
treatment_total=100
)

統計的有意性は重要です。検証されていない「改善」は、単なるランダムな変動かもしれません。

MLflow 実験トラッキング

数十のバージョンのモデルを訓練しましたが、どれが最も効果が良いですか？デプロイしたのはどのバージョンですか？

MLflowは実験の管理、モデルの追跡、メトリクスの記録を支援します。

実験管理

例

# ファイルパス：mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# MLflowトラッキングアドレスを設定（ローカルまたはリモートサーバー）
mlflow.set_tracking_uri("http://localhost:5000")

# 実験を作成または設定
experiment_name = "サンプル-llm-finetuning"
try:
experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# LLMファインチューニング実験をシミュレート
def run_finetuning_experiment(
model_name: str,
learning_rate: float,
batch_size: int,
num_epochs: int,
lora_rank: int,
dataset_name: str = "サンプル-demo"
):
"""ファインチューニング実験を1回実行し、MLflowに記録する"""

with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
run_id = run.info.run_id

# パラメータを記録
mlflow.log_param("model_name", model_name)
mlflow.log_param("learning_rate", learning_rate)
mlflow.log_param("batch_size", batch_size)
mlflow.log_param("num_epochs", num_epochs)
mlflow.log_param("lora_rank", lora_rank)
mlflow.log_param("dataset_name", dataset_name)

# トレーニングプロセスをシミュレートし、メトリクスを記録
print(f"訓練実験 {run_id} を開始...")

for epoch in range(num_epochs):
# トレーニング損失をシミュレート（エポックとともに減少）
train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

# メトリクスを記録
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("val_loss", val_loss, step=epoch)
mlflow.log_metric("epoch", epoch, step=epoch)

print(f"Epoch {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
time.sleep(0.5)

# 最終評価メトリクス
final_val_loss = 1.2 + np.random.normal(0, 0.1)
final_accuracy = 0.85 + np.random.normal(0, 0.03)
final_f1 = 0.83 + np.random.normal(0, 0.04)

mlflow.log_metric("final_val_loss", final_val_loss)
mlflow.log_metric("final_accuracy", final_accuracy)
mlflow.log_metric("final_f1", final_f1)

# ファイルを記録（例：トレーニングログ、サンプル出力）
with open("training_log.txt", "w") as f:
f.write(f"Experiment run_id: {run_id}\n")
f.write(f"Model: {model_name}\n")
f.write(f"Final accuracy: {final_accuracy:.4f}\n")

mlflow.log_artifact("training_log.txt")

# モデルを記録（ここではsklearnでシミュレート、実際のプロジェクトでは実際のLLMを記録）
# 実際のプロジェクトでは mlflow.transformers.log_model を使用可能
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

mlflow.sklearn.log_model(
model,
"model",
registered_model_name=f"{model_name.replace('/', '-')}-demo"
)

print(f実験完了！run_id: {run_id})
print(f"最終精度: {final_accuracy:.4f}")

return run_id, final_accuracy

# 数回実験を実行
if __name__ == "__main__":
print("MLflow UIを起動（別のターミナルで実行）：")
print(" mlflow ui --port 5000")
print("\n実験を開始...\n")

# 実験 1：基本設定
run_id1, acc1 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()

# 実験 2：より大きいLoRAランク
run_id2, acc2 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=16
)

print()

# 実験 3：より高い学習率
run_id3, acc3 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=5e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()
print("すべての実験が完了しました！")
print(f"実験 1 (run_id={run_id1}): {acc1:.4f}")
print(f"実験 2 (run_id={run_id2}): {acc2:.4f}")
print(f"実験 3 (run_id={run_id3}): {acc3:.4f}")
print("\n結果を表示: http://localhost:5000")

モデルレジストリ

MLflow Model Registryはモデルのライフサイクルを管理します：Staging（テスト）からProduction（本番）まで。

例

# ファイルパス：mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
model_name: str,
version: int,
description: str = ""
):
"""モデルをProduction段階に昇格する"""
# バージョンの説明を更新
client.update_model_version(
name=model_name,
version=version,
description=description
)

# このバージョンをProductionに設定
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
archive_existing_versions=True # 旧バージョンをアーカイブ
)
print(f"モデル {model_name} のバージョン {version} はProductionに昇格しました")

def get_production_model(model_name: str):
現在のProductionモデルを取得する
versions = client.get_latest_versions(model_name, stages=["Production"])
if versions:
return versions[0]
return None

def list_all_models():
"""登録済みのすべてのモデルを一覧表示する"""
models = client.search_registered_models()
print("登録済みのモデル：")
for model in models:
print(f" - {model.name}")
# そのモデルのすべてのバージョンを一覧表示
versions = client.search_model_versions(f"name='{model.name}'")
for v in versions:
print(f" Version {v.version}: {v.current_stage}")

# 使用例
if __name__ == "__main__":
model_name = "Qwen2.5-7B-Instruct-demo"

print("すべてのモデルを一覧表示：")
list_all_models()

print("\n" + "="*50)
# バージョン1を本番環境にプッシュすると仮定
# promote_model_to_production(
# model_name=model_name,
# version=1,
# description="本番環境モデル：2024年最適化版"
# )

print("\n現在の本番モデルを取得：”)
prod_model = get_production_model(model_name)
if prod_model:
print(f" モデル名: {prod_model.name}")
print(f" バージョン: {prod_model.version}")
print(f" ステージ: {prod_model.current_stage}")
print(f" Run ID: {prod_model.run_id}")
else:
print(" 本番モデルはありません")

監視とアラート

サービスが稼働を開始しましたが、正常に動作しているかどうかはどうやってわかりますか？ユーザーはタイムアウトに遭遇するでしょうか？モデルはどんどん不正確になるでしょうか？

完全な監視システムが必要です。

主要指標：TTFT、TPS、エラー率

AIサービスにはいくつかのコア監視メトリクスがあります：

メトリクス	正式名称	意味	アラートしきい値の例
TTFT	Time to First Token	ユーザーがリクエストを送信してから最初の文字を受信するまでの時間	> 3 秒でアラート
TPOT	Time per Output Token	各出力トークンの平均所要時間	> 100ms でアラート
TPS	Tokens Per Second	1秒あたりに生成されるトークン数	< 100 でアラート
エラー率	Error Rate	失敗したリクエストの割合	> 5% でアラート
GPU使用率	GPU Utilization	GPU使用率	< 30% または > 95% でアラート
GPUメモリ使用量	GPU Memory	GPUメモリ使用率	> 90% でアラート

Prometheus + Grafana

Prometheusがメトリクスを収集し、Grafanaがダッシュボードを表示する、これが監視の黄金のコンビです。

例

# ファイルパス：metrics_prometheus.py
from prometheus_client import (
Counter, Histogram, Gauge,
start_http_server, generate_latest
)
import time
import random

# ============================================
# メトリクスを定義
# ============================================

# Counter：増えるだけで減らないカウンター
REQUEST_COUNT = Counter(
"llm_requests_total",
"Total number of LLM requests",
["model", "variant", "status"] # ラベル
)

# Histogram：統計分布（例：レイテンシ）
REQUEST_LATENCY = Histogram(
"llm_request_latency_seconds",
"LLM request latency",
["model", "variant"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60] # バケット
)

TTFT_HISTOGRAM = Histogram(
"llm_ttft_seconds",
"Time to first token",
["model", "variant"],
buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Gauge：増減できるゲージ
ACTIVE_CONNECTIONS = Gauge(
"llm_active_connections",
"Number of active connections",
["model"]
)

GPU_UTILIZATION = Gauge(
"llm_gpu_utilization_percent",
"GPU utilization",
["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
"llm_gpu_memory_used_bytes",
"GPU memory used",
["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
"llm_tokens_per_second",
"Tokens generated per second",
["model"]
)

# ============================================
# サービスをシミュレート
# ============================================

def process_request(model: str, variant: str):
"""1回のリクエスト処理をシミュレートし、メトリクスを記録する"""
start_time = time.time()

# アクティブ接続数 +1
ACTIVE_CONNECTIONS.labels(model=model).inc()

try:
# TTFT（最初のトークンの時間）をシミュレート
ttft = random.uniform(0.05, 0.3)
TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

# 処理時間をシミュレート
processing_time = random.uniform(0.5, 3.0)
time.sleep(processing_time)

# 成功をシミュレート
REQUEST_COUNT.labels(
model=model, variant=variant, status="success"
).inc()

# レイテンシを記録
latency = time.time() - start_time
REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

# TPSをシミュレート
tps = random.uniform(50, 200)
TOKENS_PER_SECOND.labels(model=model).set(tps)

except Exception:
# エラーをシミュレート
REQUEST_COUNT.labels(
model=model, variant=variant, status="error"
).inc()
raise
finally:
# アクティブ接続数 -1
ACTIVE_CONNECTIONS.labels(model=model).dec()

def update_gpu_metrics():
"""GPUメトリクスの更新をシミュレートする"""
# GPU 0 のメトリクスをシミュレート
gpu0_util = random.uniform(40, 90)
gpu0_mem = random.uniform(10, 20) * 1024**3 # 10-20 GB
GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)

# ============================================
# サービスを起動
# ============================================

if __name__ == "__main__":
# Prometheus metrics エンドポイントを起動
start_http_server(8081)
print("Prometheus metrics server started on port 8081")
print("Metrics available at http://localhost:8081/metrics")

# トラフィックをシミュレート
models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
variants = ["control", "treatment-v1"]

while True:
model = random.choice(models)
variant = random.choice(variants)
process_request(model, variant)
update_gpu_metrics()
time.sleep(random.uniform(0.1, 0.5))

Prometheus設定ファイル：

例

# ファイルパス：prometheus.yml
global:
scrape_interval: 15s

scrape_configs:
# LLMサービスのメトリクスを取得
- job_name: "llm-service"
static_configs:
- targets: ["localhost:8081"]
labels:
service: "llm-api"
environment: "production"

# ノードメトリクスを取得（GPU、CPU、メモリ）
- job_name: "node"
static_configs:
- targets: ["localhost:9100"]

alerting:
alertmanagers:
- static_configs:
- targets: ["localhost:9093"]

# アラートルール
rule_files:
- "alerts.yml"

アラートルール：

例

# ファイルパス：alerts.yml
groups:
- name: llm_alerts
interval: 30s
rules:
# 高エラー率アラート
- alert: HighErrorRate
expr: |
sum(rate(llm_requests_total{status="error"}[5m]))
/
sum(rate(llm_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "高エラー率アラート"
description: "エラー率が5%を超えました。現在値: {{ $value | humanizePercentage }}"

# TTFT 高すぎアラート
- alert: SlowTTFT
expr: |
histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "TTFT 高すぎ"
description: "モデル {{ $labels.model }} の 95パーセンタイル TTFT が 2 秒を超えました"

# GPU 使用率が低すぎる（リソースの無駄）
- alert: LowGPUUtilization
expr: llm_gpu_utilization_percent < 30
for: 30m
labels:
severity: info
annotations:
summary: "GPU 使用率が低すぎる"
description: "GPU {{ $labels.gpu_id }} 使用率 {{ $value }}%、リソースの無駄の可能性があります"

# GPU 使用率が高すぎる（ボトルネックのリスク）
- alert: HighGPUUtilization
expr: llm_gpu_utilization_percent > 95
for: 5m
labels:
severity: warning
annotations:
summary: "GPU 使用率が高すぎる"
description: "GPU {{ $labels.gpu_id }} 使用率 {{ $value }}%、ボトルネックになる可能性があります"

AI 特有の監視：ドリフト検出

AI モデルは「壊れる」だけでなく、「ドリフト」もします。時間の経過とともにデータ分布が変わり、モデルの性能が低下します。

例

# ファイルパス: drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
"""分布統計情報"""
mean: float
std: float
min: float
max: float
percentiles: dict

class DataDriftDetector:
"""データドリフト検出器"""

def __init__(self, baseline_data: np.ndarray):
"""ベースラインデータで初期化する"""
self.baseline_stats = self._compute_stats(baseline_data)
self.baseline_data = baseline_data

def _compute_stats(self, data: np.ndarray) -> DistributionStats:
"""データの統計情報を計算する"""
return DistributionStats(
mean=float(np.mean(data)),
std=float(np.std(data)),
min=float(np.min(data)),
max=float(np.max(data)),
percentiles={
"p10": float(np.percentile(data, 10)),
"p25": float(np.percentile(data, 25)),
"p50": float(np.percentile(data, 50)),
"p75": float(np.percentile(data, 75)),
"p90": float(np.percentile(data, 90))
}
)

def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""
ドリフトを検出
戻り値: ドリフトがあるかどうか、p値、統計情報
"""
# KS 検定: 2つの分布が同じかどうかを比較
ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

# 新しいデータの統計を計算
new_stats = self._compute_stats(new_data)

# 平均値の差（パーセント）を計算
mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

# ドリフトかどうかを判定
is_drift = ks_pvalue < threshold

return {
"is_drift": is_drift,
"ks_statistic": float(ks_statistic),
"ks_pvalue": float(ks_pvalue),
"mean_diff_percent": float(mean_diff_pct),
"baseline_stats": self.baseline_stats,
"new_stats": new_stats,
"threshold": threshold
}

# シミュレーション: ユーザー入力長のドリフト検出
def simulate_input_length_drift():
# ベースラインデータ: ユーザー入力の平均長 50
np.random.seed(42)
baseline = np.random.normal(loc=50, scale=15, size=1000)
baseline = np.clip(baseline, 5, 200)

detector = DataDriftDetector(baseline)

print("データドリフト検出の例")
print("="*50)
print(f"ベースライン: 平均={detector.baseline_stats.mean:.2f}, 標準偏差={detector.baseline_stats.std:.2f}")
print()

# シナリオ 1: 正常データ（ドリフトなし）
print("シナリオ 1: 正常データ")
normal_data = np.random.normal(loc=50, scale=15, size=500)
normal_data = np.clip(normal_data, 5, 200)
result = detector.detect(normal_data)
print(f" 新しいデータの平均: {result['new_stats'].mean:.2f}")
print(f" KS p-value: {result['ks_pvalue']:.4f}")
print(f" ドリフト? {'はい' if result['is_drift'] else 'いいえ'}")
print()

# シナリオ 2: ドリフトあり（入力が長くなった）
print("シナリオ 2: 入力が長くなった")
drift_data = np.random.normal(loc=100, scale=20, size=500)
drift_data = np.clip(drift_data, 5, 300)
result = detector.detect(drift_data)
print(f" 新しいデータの平均: {result['new_stats'].mean:.2f}")
print(f" KS p-value: {result['ks_pvalue']:.4f}")
print(f" ドリフト? {'はい' if result['is_drift'] else 'いいえ'}")
print(f" 平均差: {result['mean_diff_percent']:.1f}%")

if __name__ == "__main__":
simulate_input_length_drift()

コスト最適化戦略

GPU は高価です。コスト最適化は「けち」ではなく「エンジニアリング能力」です。

リクエストキャッシュ：セマンティックキャッシュ

従来のキャッシュは完全一致です。まったく同じ質問には、以前の回答を直接返します。

セマンティックキャッシュはさらに進んで、類似した質問でも以前の回答や中間結果を再利用できます。

例

# ファイルパス: semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
"""シンプルなセマンティックキャッシュ実装"""

def __init__(self, threshold: float = 0.9):
self.threshold = threshold
self.queries = [] # 質問のベクトルを保存
self.answers = [] # 対応する回答を保存
self.timestamps = [] # 時間を保存

def _simulate_embedding(self, text: str) -> np.ndarray:
"""
embedding をシミュレートします。実際のプロジェクトでは OpenAI Embeddings、Sentence-Transformers などを使用します。
"""
# シンプルなハッシュシミュレーション。実際のプロジェクトでは実際の embedding モデルを使用
np.random.seed(hash(text) % (2**32))
return np.random.randn(128) # 128次元ベクトル

def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""コサイン類似度を計算"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get(self, query: str) -> Optional[str]:
"""キャッシュを照会し、類似した質問が見つかったら回答を返す"""
query_vec = self._simulate_embedding(query)

for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
# 有効期限のクリーンアップ（1時間超）
if time.time() - ts > 3600:
continue

similarity = self._cosine_similarity(query_vec, q_vec)
if similarity >= self.threshold:
return answer

return None

def put(self, query: str, answer: str):
"""キャッシュに保存"""
query_vec = self._simulate_embedding(query)
self.queries.append(query_vec)
self.answers.append(answer)
self.timestamps.append(time.time())

# シンプルな容量制御: 最大1000件保存
if len(self.queries) > 1000:
self.queries = self.queries[-1000:]
self.answers = self.answers[-1000:]
self.timestamps = self.timestamps[-1000:]

# 使用例
def main():
cache = SemanticCache(threshold=0.9)

# 最初のリクエスト
query1 = "Python で CSV ファイルを読み取る方法は？"
print(f"クエリ 1: {query1}")
answer1 = cache.get(query1)
if answer1:
print(f"キャッシュヒット: {answer1}")
else:
print("キャッシュミス、モデルを呼び出し...")
answer1 = "pandas.read_csv('file.csv') で CSV ファイルを読み取れます。"
cache.put(query1, answer1)
print(f"キャッシュに保存: {answer1}")

print()

# 2回目: 類似した質問
query2 = "Python で CSV をどう読む？"
print(f"クエリ 2: {query2}")
answer2 = cache.get(query2)
if answer2:
print(f"キャッシュヒット: {answer2}")
else:
print("キャッシュミス、モデルを呼び出し...")

print()

# 3回目: まったく異なる質問
query3 = "機械学習とは何ですか？"
print(f"クエリ 3: {query3}")
answer3 = cache.get(query3)
if answer3:
print(f"キャッシュヒット: {answer3}")
else:
print("キャッシュミス、モデルを呼び出し...")

if __name__ == "__main__":
main()

モデルルーティング：小モデルによるフォールバック

すべてのリクエストに大規模モデルが必要なわけではありません。簡単な質問は小規模モデル、複雑な質問は大規模モデルを使用します。

例

# ファイルパス: model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float
max_tokens: int
capabilities: List[str] # 能力リスト

class SmartModelRouter:
"""スマートモデルルーター"""

def __init__(self):
self.models = {
"small": ModelConfig(
name="Qwen2.5-0.5B-Instruct",
cost_per_1k_tokens=0.001,
max_tokens=4096,
capabilities=["chat", "simple_qa", "summarization"]
),
"medium": ModelConfig(
name="Qwen2.5-7B-Instruct",
cost_per_1k_tokens=0.01,
max_tokens=8192,
capabilities=["chat", "qa", "coding", "reasoning"]
),
"large": ModelConfig(
name="Qwen2.5-72B-Instruct",
cost_per_1k_tokens=0.1,
max_tokens=32768,
capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
)
}

def _classify_query(self, query: str) -> str:
"""
簡単な質問分類（実際のプロジェクトでは分類器を使用可能）
戻り値: small/medium/large
"""
query_lower = query.lower()

# 簡単な質問のキーワード
simple_keywords = [
"こんにちは", "hello", "ありがとう", "hi", "さようなら",
"今日の天気", "何時", "日付",
"翻訳を手伝う", "translate",
"簡単な紹介", "とは"
]

# 複雑な質問のキーワード
complex_keywords = [
"コード", "書く", "実装", "code", "python",
"分析", "理由を説明", "導出", "証明",
"比較", "対比", "違い",
"詳細", "深い", "複雑"
]

# 簡単な質問かどうかを確認
for keyword in simple_keywords:
if keyword in query_lower:
return "small"

# 複雑な質問かどうかを確認
for keyword in complex_keywords:
if keyword in query_lower:
return "large"

# デフォルトは中規模モデル
return "medium"

def route(self, query: str, user_tier: str = "free") -> str:
"""
ルーティング決定
"""
# まず質問を分類
category = self._classify_query(query)

# 有料ユーザーはより良いモデルを使用可能
if user_tier == "pro":
upgrade_map = {
"small": "small",
"medium": "medium",
"large": "large"
}
category = upgrade_map[category]
elif user_tier == "enterprise":
upgrade_map = {
"small": "medium",
"medium": "large",
"large": "large"
}
category = upgrade_map.get(category, category)

# モデル名を返す
return self.models[category].name

def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""コストを見積もる"""
for model in self.models.values():
if model.name == model_name:
return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
return 0.0

# 使用例
def main():
router = SmartModelRouter()

test_queries = [
"こんにちは、知りたいのですが",
"Python とは何ですか？",
"Python のクイックソートを書いてください",
"Transformer アーキテクチャを詳しく分析してください"
]

print("スマートルーティングの例")
print("="*50)

for query in test_queries:
model = router.route(query, user_tier="free")
cost = router.get_cost_estimate(model, 100, 200)
print(f"質問: {query}")
print(f" ルーティング先: {model}")
print(f" 見積もりコスト: ${cost:.4f}")
print()

# 異なるユーザーレベル
print("異なるユーザーレベルの比較")
print("-"*50)
query = "Python のクイックソートを書いてください"
for tier in ["free", "pro", "enterprise"]:
model = router.route(query, user_tier=tier)
print(f"{tier} ユーザー: {model}")

if __name__ == "__main__":
main()

Spot インスタンスとバッチ処理

リアルタイムリクエストには On-Demand インスタンス（安定だが高価）、オフラインバッチ処理には Spot インスタンス（安価だがプリエンプションされる可能性あり）を使用します。

コスト最適化のいくつかの方法のまとめ：

戦略	シナリオ	コスト削減
セマンティックキャッシュ	重複する質問が多い	30-70%
モデルルーティング	簡単な質問の割合が高い	50-80%
モデル量子化	すべてのシナリオ	30-50%
Spot インスタンス	オフラインバッチ処理	60-90%
自動スケーリング	トラフィックの変動が大きい	30-60%

障害処理とデグレード

本番環境に「障害が起きない」はありません。あるのは「障害時にどう優雅に対応するか」だけです。

デグレード戦略

例

# ファイルパス: fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"

@dataclass
class FallbackResponse:
content: str
is_fallback: bool
fallback_reason: Optional[str] = None

class FallbackHandler:
"""フォールバックハンドラー"""

def __init__(self):
self.error_count = 0
self.last_error_time = 0
self.circuit_open = False
self.circuit_open_time = 0

# サーキットブレーカー設定
self.error_threshold = 5 # 連続5回エラーでサーキットブレーカーが作動
self.circuit_timeout = 30 # サーキットブレーカー作動から30秒後に復旧を試みる

def _call_primary_service(self, query: str) -> str:
"""メインサービスの呼び出しをシミュレート"""
# ここを実際のモデル呼び出しに置き換える
return "これはメインサービスの返信です"

def _call_fallback_service(self, query: str) -> str:
"""フォールバックサービスの呼び出しをシミュレート（高速だが品質はやや低い）"""
# より小さいモデル、キャッシュされた応答、さらにはプリセットテンプレートでもよい
return "申し訳ございません。サービスが一時的に混雑しています。これは簡略化された応答です。"

def _call_static_fallback(self, query: str) -> str:
静的フォールバック応答（最後の防衛ライン）
return (
"大変申し訳ございませんが、現在サービスに問題が発生しています。\n"
"しばらくしてからもう一度お試しください。緊急の問題は [email protected] までご連絡ください。"
)

def process_request(self, query: str) -> FallbackResponse:
"""完全なフォールバックロジックを備えたリクエスト処理"""

# 1. サーキットブレーカーが開いているか確認
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_timeout:
# タイムアウトしたため、ハーフオープン状態を試行し、少量のリクエストを許可
print("サーキットブレーカーがハーフオープン、復旧を試行中...")
self.circuit_open = False
else:
# 直接フォールバック
print("サーキットブレーカーがオープン、直接フォールバック")
return FallbackResponse(
content=self._call_fallback_service(query),
is_fallback=True,
fallback_reason="circuit_open"
)

# 2. メインサービスを試行
try:
result = self._call_primary_service(query)
# 成功：エラーカウントをリセット
self.error_count = 0
return FallbackResponse(
content=result,
is_fallback=False
)

except Exception as e:
print(f"メインサービスの呼び出しに失敗: {e}")
self.error_count += 1
self.last_error_time = time.time()

# 3. サーキットブレーカーが必要かどうか確認
if self.error_count >= self.error_threshold:
print("エラー回数が多すぎるため、サーキットブレーカーをオープン")
self.circuit_open = True
self.circuit_open_time = time.time()

# 4. フォールバックサービスを試行
try:
print("フォールバックサービスの呼び出しを試行")
result = self._call_fallback_service(query)
return FallbackResponse(
content=result,
is_fallback=True,
fallback_reason="primary_failed"
)
except Exception as e2:
print(f"フォールバックサービスも失敗: {e2}")

# 5. 最後の静的応答
return FallbackResponse(
content=self._call_static_fallback(query),
is_fallback=True,
fallback_reason="all_failed"
)

# 使用例
def main():
handler = FallbackHandler()

print("フォールバック処理の例")
print("="*50)

# 通常のリクエスト
print("通常のリクエスト:")
resp = handler.process_request("こんにちは")
print(f" 応答: {resp.content}")
print(f" フォールバック? {resp.is_fallback}")
print()

# エラーをシミュレート（実際のプロジェクトでは実際に例外がスローされる）
# ここではロジックを簡潔にデモンストレーション
print("フォールバック戦略のまとめ：")
print(" 1. メインサービスを優先")
print(" 2. メインサービス失敗 → 小モデル/キャッシュを使用")
print(" 3. 連続失敗 → サーキットブレーカーを開き、雪崩を防止")
print(" 4. サーキットブレーカー作動中 → フォールバック応答を迅速に返す")
print(5. 最後の防衛ライン → 静的応答)

カナリアリリースとロールバック

新バージョンは一度に全量リリースせず、まず1%のトラフィックに流し、問題がなければ5%、10%、100%と増やす。

問題が発生したら迅速にロールバックできるようにする。ユーザーから長時間苦情が来てから慌てて設定を変更するのは避けること。

その他の拡張