G1Joshi/Agent-Skills: Especialista em Ray para Computação Distr…

Ray

Ray é a camada de computação para IA. Ele alimenta o treinamento do ChatGPT e cargas de trabalho em escala massiva. A v3.0 (2025) melhora a eficiência e adiciona um Servidor MCP para agentes.

Quando Usar

• Aplicações Python Distribuídas: Escalando tarefas Python com uso intenso de computação de forma transparente, de um único laptop a grandes clusters na nuvem.
• Treinamento de ML Distribuído com Ray Train: Orquestrando trabalhos de treinamento multi-nó com PyTorch, XGBoost e LightGBM.
• Implantação de Modelos em Produção com Ray Serve: Servindo LLMs e pipelines de microsserviços multi-modelo com escalonamento automático dinâmico.
• Otimização de Hiperparâmetros com Ray Tune: Executando varreduras com métodos Bayesianos, Optuna e treinamento baseado em população por centenas de workers.

Início Rápido

import ray

ray.init()

# Tarefa distribuída remota
@ray.remote
def square(x):
    return x * x

# Lançar tarefas em paralelo no cluster
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Quadrados computados:", results)

Conceitos Principais

#Tarefas Distribuídas e Atores com Estado

Escalando funções e classes entre workers do cluster:

import ray
import time

# Inicializar localmente ou conectar a um cluster Ray remoto
ray.init(ignore_reinit_error=True)

# 1. Função Remota sem Estado (Tarefa Distribuída)
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
    time.sleep(0.5) # Simulando computação com uso intenso de CPU/IO
    return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}

# Lançar tarefas concorrentemente no cluster
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # Coletar resultados
print("Chunks Processados:", results)

# 2. Classe Remota com Estado (Ator Distribuído)
@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("Contagem final do ator:", ray.get(inc_futures[-1]))

#Implantação de Modelos em Produção com Ray Serve

Implantando endpoints REST escaláveis com agrupamento dinâmico de requisições:

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)

#Otimização Distribuída de Hiperparâmetros com Ray Tune

Executando tentativas de hiperparâmetros em paralelo:

from ray import tune

def training_function(config):
    # Treinamento de modelo simulado
    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("Melhores hiperparâmetros:", results.get_best_result().config)

Padrões Comuns

Atores Distribuídos com Estado para Servir Modelos

Problema: Recarregar grandes pesos de modelo na memória a cada invocação de função.

Solução: Use Atores Ray para manter estado em memória entre requisições:

@ray.remote(num_gpus=1)
class ModelServer:
    def __init__(self):
        # Carregar pesos uma vez na VRAM da GPU
        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))

Práticas Recomendadas (2026)

• Faça verifique o Painel Ray (porta 8265) para monitorar a utilização de CPU/GPU dos nós, colocação de atores e memória do armazenamento de objetos.
• Faça passe grandes conjuntos de dados somente leitura através do armazenamento de objetos plasma do Ray (ray.put(data)) para evitar serializações repetitivas.
• Faça especifique explicitamente os requisitos de recursos nos decoradores (@ray.remote(num_cpus=2, num_gpus=1)).
• Faça use Ray Serve para microsserviços multi-modelo em produção com agrupamento automático de requisições integrado.
• Não faça chame ray.get() dentro de tarefas remotas; passe ObjectRefs diretamente para outras tarefas remotas para preservar o pipelining.
• Não faça crie milhões de microtarefas; agrupe tarefas granulares para minimizar a sobrecarga do escalonador.
• Não faça passe objetos grandes com estado (como conexões de banco de dados ou arquivos abertos) dentro de argumentos remotos.

Solução de Problemas

Erro	Causa	Solução
RaySystemError: Object store full	Tarefas produzindo mais dados do que a memória compartilhada alocada no armazenamento plasma.	Use ray.get() para consumir resultados, ou armazene arrays grandes em disco.
Actor died unexpectedly	Processo worker travou devido a exceção não tratada ou OOM.	Verifique os logs no painel Ray (http://localhost:8265) e aumente a RAM do worker.
Task failed to schedule: Unsatisfiable resource request	Solicitou mais CPUs/GPUs (num_gpus=2) do que o disponível no cluster.	Verifique os recursos disponíveis via ray.cluster_resources().

Referências

• Documentação do Ray