Ray
RayはAIのためのコンピューティングレイヤーです。ChatGPTのトレーニングや大規模ワークロードを支えています。v3.0（2025年）では、効率性が向上し、エージェント向けのMCPサーバーが追加されました。
使用するタイミング
- 分散Pythonアプリケーション: 計算負荷の高いPythonタスクを、単一のラップトップから大規模なクラウドクラスタまでシームレスにスケーリング。
- Ray Trainによる分散MLトレーニング: マルチノードでのPyTorch、XGBoost、LightGBMのトレーニングジョブをオーケストレーション。
- Ray Serveによる本番モデルサービング: 動的オートスケーリングを備えたLLMやマルチモデルマイクロサービスのパイプラインを提供。
- Ray Tuneによるハイパーパラメータ最適化: ベイズ最適化、Optuna、集団ベースのトレーニングスイープを数百のワーカーで実行。
クイックスタート
import ray
ray.init()
# リモート分散タスク
@ray.remote
def square(x):
return x * x
# クラスタ全体で並列にタスクを起動
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Computed squares:", results)
コアコンセプト
#分散タスクとステートフルアクター
クラスタワーカー間で関数とクラスをスケーリング:
import ray
import time
# ローカルで初期化するか、リモートのRayクラスタに接続
ray.init(ignore_reinit_error=True)
# 1. ステートレスリモート関数（分散タスク）
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
time.sleep(0.5) # CPU/IOバウンドな計算をシミュレート
return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}
# クラスタ全体でタスクを同時に起動
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # 結果を収集
print("Processed Chunks:", results)
# 2. ステートフルリモートクラス（分散アクター）
@ray.remote
class GlobalCounter:
def __init__(self):
self.count = 0
def increment(self) -> int:
self.count += 1
return self.count
counter_actor = GlobalCounter.remote()
inc_futures = [counter_actor.increment.remote() for _ in range(5)]
print("Final actor count:", ray.get(inc_futures[-1]))
#Ray Serveによる本番モデルサービング
動的リクエストバッチ処理を備えたスケーラブルなRESTエンドポイントのデプロイ:
from ray import serve
import torch
from transformers import pipeline
@serve.deployment(num_replicas=2, ray_actor_options={"num_cpus": 1, "num_gpus": 0})
class SentimentClassifier:
def __init__(self):
self.classifier = pipeline("sentiment-analysis")
async def __call__(self, request):
json_data = await request.json()
text = json_data.get("text", "")
return self.classifier(text)
app = SentimentClassifier.bind()
# serve.run(app)
#Ray Tuneによる分散ハイパーパラメータチューニング
並行ハイパーパラメータトライアルの実行:
from ray import tune
def training_function(config):
# シミュレートされたモデルトレーニング
for step in range(10):
intermediate_score = config["alpha"] * step + config["beta"]
tune.report(loss=1.0 / (intermediate_score + 0.1))
search_space = {
"alpha": tune.uniform(0.1, 1.0),
"beta": tune.choice([1, 2, 5])
}
tuner = tune.Tuner(
training_function,
param_space=search_space,
tune_config=tune.TuneConfig(num_samples=10, metric="loss", mode="min")
)
results = tuner.fit()
print("Best hyperparameters:", results.get_best_result().config)
一般的なパターン
モデルサービングのためのステートフル分散アクター
問題: 関数呼び出しのたびに大規模なモデル重みをメモリに再ロードする。
解決策: Rayアクターを使用して、リクエスト間でインメモリ状態を保持する:
@ray.remote(num_gpus=1)
class ModelServer:
def __init__(self):
# 重みを一度GPU VRAMにロード
self.model = load_large_model()
def predict(self, sample):
return self.model(sample)
server_actor = ModelServer.remote()
prediction = ray.get(server_actor.predict.remote(sample_data))
ベストプラクティス（2026年）
- 推奨: Rayダッシュボード（ポート8265）を確認して、ノードのCPU/GPU使用率、アクターの配置、オブジェクトストアメモリを監視する。
- 推奨: 大きな読み取り専用データセットは、Rayプラズマオブジェクトストア（
ray.put(data)）を介して渡し、繰り返しのシリアライゼーションを防ぐ。 - 推奨: デコレータでリソース要件を明示的に指定する（
@ray.remote(num_cpus=2, num_gpus=1)）。 - 推奨: 本番環境でのマルチモデルマイクロサービスには、組み込みの自動バッチ処理を備えたRay Serveを使用する。
- 非推奨:
ray.get()をリモートタスク内で呼び出さない。ObjectRefを他のリモートタスクに直接渡してパイプライン処理を維持する。 - 非推奨: 数百万のマイクロタスクを作成しない。細かいタスクをまとめて、スケジューラのオーバーヘッドを最小限に抑える。
- 非推奨: リモート引数内に、データベース接続や開いているファイルなどの大きなステートフルオブジェクトを渡さない。
トラブルシューティング
| エラー | 原因 | 解決策 |
|---|---|---|
RaySystemError: Object store full | タスクが割り当てられた共有メモリプラズマストアよりも多くのデータを生成している。 | ray.get() を使用して結果を消費するか、大きな配列をディスクストレージに保存する。 |
Actor died unexpectedly | ワーカープロセスが未処理の例外またはOOMでクラッシュした。 | Rayダッシュボード（http://localhost:8265）でログを確認し、ワーカーRAMを増やす。 |
Task failed to schedule: Unsatisfiable resource request | クラスタで利用可能な数よりも多くのCPU/GPU（num_gpus=2）を要求した。 | ray.cluster_resources() で利用可能なリソースを確認する。 |