G1Joshi/Agent-Skills

Ray

Ray es la capa de cómputo para IA. Impulsa el entrenamiento de ChatGPT y cargas de trabajo a gran escala. La versión 3.0 (2025) mejora la eficiencia y añade un Servidor MCP para agentes.

Cuándo Usarlo

• Aplicaciones Python Distribuidas: Escalar tareas Python que requieren mucho cómputo sin problemas desde una sola laptop hasta grandes clústeres en la nube.
• Entrenamiento ML Distribuido con Ray Train: Orquestar trabajos de entrenamiento PyTorch, XGBoost y LightGBM en múltiples nodos.
• Servicio de Modelos en Producción con Ray Serve: Servir LLMs y canalizaciones de microservicios multimodelo con escalado automático dinámico.
• Optimización de Hiperparámetros con Ray Tune: Ejecutar barridos de optimización bayesiana, Optuna y basada en población en cientos de trabajadores.

Inicio Rápido

import ray

ray.init()

# Tarea distribuida remota
@ray.remote
def square(x):
    return x * x

# Lanzar tareas en paralelo a través del clúster
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Cuadrados calculados:", results)

Conceptos Clave

#Tareas Distribuidas y Actores con Estado

Escalar funciones y clases entre los trabajadores del clúster:

import ray
import time

# Inicializar localmente o conectar a un clúster Ray remoto
ray.init(ignore_reinit_error=True)

# 1. Función Remota sin Estado (Tarea Distribuida)
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
    time.sleep(0.5) # Simula cómputo intensivo de CPU/IO
    return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}

# Lanzar tareas concurrentemente a través del clúster
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # Recolectar resultados
print("Fragmentos procesados:", results)

# 2. Clase Remota con Estado (Actor Distribuido)
@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("Conteo final del actor:", ray.get(inc_futures[-1]))

#Servicio de Modelos en Producción con Ray Serve

Desplegar puntos finales REST escalables con agrupación dinámica de solicitudes:

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)

#Optimización Distribuida de Hiperparámetros con Ray Tune

Ejecutar pruebas de hiperparámetros en paralelo:

from ray import tune

def training_function(config):
    # Entrenamiento 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("Mejores hiperparámetros:", results.get_best_result().config)

Patrones Comunes

Actores Distribuidos con Estado para Servicio de Modelos

Problema: Recargar grandes pesos de modelos en memoria en cada invocación de función.

Solución: Usar Actores Ray para mantener el estado en memoria entre solicitudes:

@ray.remote(num_gpus=1)
class ModelServer:
    def __init__(self):
        # Cargar pesos una vez en la VRAM de la 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))

Mejores Prácticas (2026)

• Hacer verificar el Panel de Ray (puerto 8265) para monitorear la utilización de CPU/GPU de los nodos, la ubicación de los actores y la memoria del almacén de objetos.
• Hacer pasar grandes conjuntos de datos de solo lectura a través del almacén de objetos plasma de Ray (ray.put(data)) para evitar serializaciones repetitivas.
• Hacer especificar explícitamente los requisitos de recursos en los decoradores (@ray.remote(num_cpus=2, num_gpus=1)).
• Hacer usar Ray Serve para microservicios multimodelo en producción con agrupación automática incorporada.
• No hacer llamar a ray.get() dentro de tareas remotas; pasar ObjectRefs directamente a otras tareas remotas para preservar la canalización.
• No hacer crear millones de microtareas; agrupar tareas de grano fino para minimizar la sobrecarga del planificador.
• No hacer pasar objetos grandes con estado (como conexiones de base de datos o archivos abiertos) dentro de argumentos remotos.

Solución de Problemas

Error	Causa	Solución
RaySystemError: Object store full	Las tareas producen más datos que la memoria compartida asignada al almacén plasma.	Usar ray.get() para consumir resultados, o almacenar arrays grandes en almacenamiento en disco.
Actor died unexpectedly	El proceso trabajador se bloqueó debido a una excepción no manejada o falta de memoria.	Revisar los registros en el panel de Ray (http://localhost:8265) y aumentar la RAM del trabajador.
Task failed to schedule: Unsatisfiable resource request	Se solicitaron más CPUs/GPUs (num_gpus=2) de los disponibles en el clúster.	Verificar los recursos disponibles mediante ray.cluster_resources().

Referencias

• Documentación de Ray