Ray 分布式 AI 框架 2026 版：超大模型推理与训练

from ray import Actor

@Actor
class Counter:
    def __init__(self):
        self.value = 0
    
    def increment(self):
        self.value += 1
        return self.value

import ray
ray.init()

@ray.remote
def process_large_batch(data):
    return expensive_computation(data)

futures = [process_large_batch(d) for d in batches]
results = ray.get(futures)

@ray.remote(num_cpus=4, num_gpus=2)
class ModelServer:
    def __init__(self):
        self.model = load_model()

from ray.train.torch import TorchTrainer
from ray.train import TrainingConfig

trainer = TorchTrainer(
    train_loop_per_worker=train_fn,
    scaling_config=ScalingConfig(num_workers=8),
)
result = trainer.fit()

from ray import serve

@serve.deployment(num_replicas=3, ray_actor_options={"num_gpus": 1})
class LLMPredictor:
    def __init__(self):
        self.model = load_llm_model()
    
    async def __call__(self, http_request):
        prompt = await http_request.json()
        return {"response": self.model.generate(prompt)}

LLMPredictor.deploy()

from ray import tune, rllib

from ray.rllib.agents.ppo import PPOTrainer

config = {
    "env": "CartPole-v1",
    "num_workers": 4,
    "framework": "torch"
}

trainer = PPOTrainer(config=config)

from ray import data

ds = data.read_parquet("s3://bucket/large_dataset/")
ds = ds.filter(lambda x: x["score"] > 0.5)
ds = ds.random_shuffle()
train_ds, test_ds = ds.train_test_split(0.8)

ray start --head --num-gpus=4

ray start --head --address='10.0.0.1:6379'
ray start --address='10.0.0.1:6379'  # Worker

apiVersion: ray.io/v1
kind: RayCluster
metadata:
  name: ray-cluster
spec:
  headGroupSpec:
    serviceType: ClusterIP
    replicas: 1
    rayStarterOptions: {}