Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/jstzwj/ai-infra-plugins --skill ray
The command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Comprehensive reference documentation and skill for Ray - a unified framework for scaling AI and Python applications. Covers Ray Core (tasks, actors, objects, scheduling, placement groups, namespaces, runtime environment, fault tolerance, compiled graphs, direct transport), Ray Data (datasets, transformations, datasources, preprocessors, execution engine, streaming), Ray Serve (model serving, deployments, HTTP handling, autoscaling, model composition, multi-app, multiplexing, monitoring, architecture), Ray Train (distributed training with PyTorch, TensorFlow, HuggingFace, XGBoost, LightGBM, Horovod, DeepSpeed, JAX; scaling config, checkpointing, training iterators, collective operations), Ray Tune (hyperparameter tuning, search algorithms, schedulers, analysis, logging, stoppers, trainables, CLI, experiment execution), Ray RLlib (reinforcement learning algorithms, RL modules, learners, environments, connectors, replay buffers, callbacks, multi-agent, offline training, fault tolerance), Ray Cluster (setup, configuration, autoscaling, cloud providers AWS/GCP/Azure, KubeRay, job submission, runtime environments, observability, dashboard, security, governance), Ray DAG (directed acyclic graphs, compiled graphs, DAG execution), Ray AIR (AI runtime, batch prediction, checkpoints, integrations), Ray Workflow (workflow orchestration, durable execution), Ray LLM (LLM serving integration), Ray Client (remote cluster connection), Ray utility modules (placement groups, scheduling strategies, state API, tracing, collective operations), and Ray internal architecture (GCS, raylet, worker, object store, plasma, memory management). Based on Ray source code analysis.
version
2.47.0
Ray - Unified Framework for Scaling AI and Python Applications
Overview
Ray is an open-source unified framework for scaling AI and Python applications. It provides a simple, universal API for building distributed applications, enabling parallel processing of compute-heavy workloads across clusters of machines. Ray powers some of the most complex and demanding AI workloads in production.
Key Capabilities:
Ray Core: Distributed computing primitives - tasks, actors, objects, and scheduling
Ray Data: Scalable data loading, transformation, and processing
Ray Train: Distributed model training with framework-agnostic APIs
Ray Tune: Scalable hyperparameter tuning with state-of-the-art algorithms
Ray Serve: Scalable model serving with composition and autoscaling
Ray RLlib: Industry-grade reinforcement learning library
Ray Cluster: Multi-node cluster management with autoscaling
Ray Workflow: Long-running, durable workflow execution
Ray DAG: Directed acyclic graph execution and compiled graphs
Ray AIR: Unified AI runtime for end-to-end ML workflows
Ray Client: Remote cluster connection and execution
Ray LLM: LLM serving and deployment integration
Supported Languages: Python, Java, C++ (cross-language support)
Ray Version: 2.47.0 | Python: 3.9+ | License: Apache 2.0
Architecture Overview
+------------------------------------------------------------------+
| Application Layer |
| Ray Train | Ray Tune | Ray Serve | Ray RLlib | Ray Data |
+------------------------------------------------------------------+
| Ray AIR (AI Runtime) |
| Checkpoints | Batch Prediction | Integrations | Metrics |
+------------------------------------------------------------------+
| Ray Core |
| Tasks | Actors | Objects | Scheduling | Placement Groups |
| Namespaces | Runtime Env | Fault Tolerance | DAGs |
+------------------------------------------------------------------+
| Cluster Management |
| GCS (Global Control Service) | Autoscaler | Job Submission |
| Ray Dashboard | KubeRay | Ray Client | Runtime Env |
+------------------------------------------------------------------+
| Distributed Runtime |
| Raylet (per-node) | Worker Processes | Object Store (Plasma) |
| gRPC Communication | Memory Management | Resource Isolation |
+------------------------------------------------------------------+
| Infrastructure |
| AWS | GCP | Azure | Kubernetes | On-Premise | Local |
+------------------------------------------------------------------+
Quick Reference
Initialization & Shutdown
import ray
# Initialize Ray
ray.init() # Local cluster
ray.init(address="auto") # Connect to existing cluster
ray.init(address="ray://cluster:10001") # Ray Client
ray.init(num_cpus=8, num_gpus=2, # Resource specification
object_store_memory=10**9,
dashboard_host="0.0.0.0",
dashboard_port=8265,
namespace="my_app",
runtime_env={"pip": ["requests"]})
# Shutdown
ray.shutdown()
ray.is_initialized() # Check if initialized
Tasks (Remote Functions)
@ray.remote
def my_function(x):
return x * 2
# Execute remote
result_ref = my_function.remote(42)
result = ray.get(result_ref) # Retrieve result
# Multiple returns
@ray.remote(num_returns=3)
def return_three():
return 1, 2, 3
refs = return_three.remote()
# Options
result = my_function.options(
num_cpus=2, num_gpus=1,
resources={"TPU": 4},
memory=2**31,
max_retries=3,
retry_exceptions=True,
scheduling_strategy="SPREAD",
name="my_task",
runtime_env={"pip": ["numpy"]}
).remote(42)
# Batch execution
results = ray.get([my_function.remote(i) for i in range(100)])
# Streaming generators
@ray.remote(num_returns="streaming")
def generate_data(n):
for i in range(n):
yield i
gen = generate_data.remote(10)
for ref in gen:
print(ray.get(ref))
# Put objects in object store
ref = ray.put(42)
value = ray.get(ref)
# Get multiple
values = ray.get([ref1, ref2, ref3])
# Get with timeout
value = ray.get(ref, timeout=5.0)
# Wait for objects
ready, remaining = ray.wait(
[ref1, ref2, ref3],
num_returns=2,
timeout=10.0
)
from ray.tune.search import (
BasicVariantGenerator, # Default grid/random
)
from ray.tune.search.optuna import OptunaSearch
from ray.tune.search.hyperopt import HyperOptSearch
from ray.tune.search.bayesopt import BayesOptSearch
from ray.tune.search.flaml import CFO, BlendSearch
from ray.tune.search.bohb import TuneBOHB
from ray.tune.search.nevergrad import NevergradSearch
from ray.tune.search.zoopt import ZOOptSearch
from ray.tune.search.sigopt import SigOptSearch
from ray.tune.search.hebo import HEBOSearch
tune_config = tune.TuneConfig(
search_alg=OptunaSearch(),
# or
search_alg=HyperOptSearch(metric="score", mode="max"),
)
# Head node
ray start --head --port=6379 --dashboard-host=0.0.0.0 --dashboard-port=8265 \
--num-cpus=8 --num-gpus=4 --object-store-memory=1000000000
# Worker node
ray start --address=<head-ip>:6379 --num-cpus=8 --num-gpus=4
# Stop
ray stop
# Status
ray status
from ray.autoscaler.sdk import (
request_cluster_resources,
get_cluster_resources,
)
Job Submission
from ray.job_submission import JobSubmissionClient
client = JobSubmissionClient("http://<head-ip>:8265")
job_id = client.submit_job(
entrypoint="python train.py --epochs 10",
runtime_env={
"pip": ["torch", "transformers"],
"working_dir": "./",
},
submission_id="my-job-1",
)
# Monitor
job_status = client.get_job_status(job_id)
job_logs = client.get_job_logs(job_id)
# List jobs
jobs = client.list_jobs()
CLI Job Submission
ray job submit --address=http://<head-ip>:8265 \
--runtime-env-json='{"pip": ["torch"]}' \
-- python train.py
ray job status <job_id>
ray job logs <job_id>
ray job stop <job_id>
ray job list
ray job delete <job_id>
Ray Dashboard
Accessible at http://<head-ip>:8265:
Overview: Cluster state, resource usage, active jobs
Jobs: Running/completed jobs with logs
Actors: Actor lifecycle and state
Tasks: Task execution timeline and metrics
Objects: Object store usage
Nodes: Node health and resources
Logs: Centralized log viewer
Metrics: Prometheus metrics dashboard
Serve: Serve deployment status
Data: Dataset statistics
Ray Workflow
import ray
from ray import workflow
@workflow.step
def step1(x):
return x * 2
@workflow.step
def step2(x):
return x + 1
@workflow.step
def combine(*args):
return sum(args)
# Define workflow
dag = combine.step(step1.step(1), step2.step(2))
# Execute
result = dag.run()
# With checkpointing
result = dag.run(workflow_id="my_workflow")
# Resume after failure
result = workflow.resume(workflow_id="my_workflow")
Ray DAG & Compiled Graphs
import ray
from ray.dag import InputNode
@ray.remote
def process(x):
return x * 2
@ray.remote
class Model:
def predict(self, x):
return x + 1
# Build DAG
with InputNode() as inp:
a = process.bind(inp)
model = Model.bind()
b = model.predict.bind(a)
dag = b
# Execute DAG
result = ray.get(dag.execute(42))
# Compiled Graph (optimized execution)
compiled_graph = dag.experimental_compile()
result = ray.get(compiled_graph.execute(42))
Ray AIR
from ray.air import session, RunConfig
from ray.air.config import ScalingConfig, CheckpointConfig, FailureConfig
# Inside training function
def train_func(config):
for epoch in range(10):
loss = train_one_epoch(config)
session.report(
{"loss": loss, "epoch": epoch},
checkpoint=ray.train.Checkpoint.from_directory(f"/tmp/ckpt_{epoch}"),
)
# Preprocessors
from ray.data.preprocessors import StandardScaler
preprocessor = StandardScaler(columns=["feature1"])
# Batch prediction
from ray.train.batch_predictor import BatchPredictor
predictor = BatchPredictor.from_checkpoint(
checkpoint,
MyPredictorClass,
)
predictions = predictor.predict(test_dataset)
Utility Modules
State API
from ray.util.state import (
list_tasks, list_actors, list_objects,
list_nodes, list_jobs, list_placement_groups,
get_task, get_actor, get_object, get_node,
)
# List resources
tasks = list_tasks(detail=True, filters=[("name", "=", "train")])
actors = list_actors(detail=True, filters=[("state", "=", "ALIVE")])
nodes = list_nodes()
# Get specific resource
task = get_task(task_id)
actor = get_actor(actor_id)
Collective Operations
from ray.experimental.collective import (
create_collective_group,
allreduce, allgather, broadcast, reduce, sendsend,
)
# Create collective group
create_collective_group([actor1, actor2, actor3], backend="nccl")
GPU Utilities
ray.get_gpu_ids() # Get GPU IDs for this worker
ray.available_resources() # Available resources
ray.cluster_resources() # Total cluster resources
