ray

Ray Agent Skill

Agent-optimized skill for Ray distributed computing framework.

Quick Reference

Ray Core - Tasks and Actors

import ray
ray.init()

# Task (remote function)
@ray.remote
def square(x):
    return x ** 2

future = square.remote(4)
result = ray.get(future)  # 16

# Parallel tasks
futures = [square.remote(i) for i in range(100)]
results = ray.get(futures)

# Actor (remote class)
@ray.remote
class Counter:
    def __init__(self):
        self.count = 0
    def increment(self):
        self.count += 1
        return self.count

counter = Counter.remote()
result = ray.get(counter.increment.remote())

Ray Data - Distributed Data Processing

import ray

# Read data
ds = ray.data.read_parquet("s3://bucket/data/*.parquet")
ds = ray.data.read_csv("data/*.csv")
ds = ray.data.read_json("data/*.json")

# Transform data
def preprocess(batch):
    batch["new_col"] = batch["col1"] * 2
    return batch

ds = ds.map_batches(preprocess, batch_format="pandas")

# Filter
ds = ds.filter(lambda row: row["value"] > 10)

# Write results
ds.write_parquet("output/")
ds.write_csv("output/")

# Batch inference
def predict(batch):
    # Model inference
    return {"predictions": model.predict(batch["features"])}

predictions = ds.map_batches(predict, batch_size=32)

Ray Train - Distributed Training

import ray
from ray import train
from ray.train import ScalingConfig
from ray.train.torch import TorchTrainer

def train_func(config):
    # Your training code
    model = create_model()
    train_dataset = train.get_dataset_shard("train")

    for epoch in range(config["num_epochs"]):
        # Training loop
        loss = train_epoch(model, train_dataset)
        train.report({"loss": loss})

# Configure distributed training
trainer = TorchTrainer(
    train_func,
    train_loop_config={"num_epochs": 10, "lr": 0.001},
    scaling_config=ScalingConfig(
        num_workers=4,
        use_gpu=True,
        resources_per_worker={"CPU": 2, "GPU": 1}
    ),
    datasets={"train": train_dataset}
)

result = trainer.fit()