G1Joshi/Agent-Skills

Ray

Ray est la couche de calcul pour l'IA. Il alimente l'entraînement de ChatGPT et des charges de travail massives. La version 3.0 (2025) améliore l'efficacité et ajoute un serveur MCP pour les agents.

Quand l'utiliser

• Applications Python distribuées : Passage à l'échelle de tâches Python lourdes de manière transparente d'un seul ordinateur portable à de grands clusters cloud.
• Entraînement ML distribué avec Ray Train : Orchestration d'entraînements multi-nœuds avec PyTorch, XGBoost et LightGBM.
• Déploiement de modèles en production avec Ray Serve : Service de LLMs et pipelines de microservices multi-modèles avec mise à l'échelle automatique dynamique.
• Optimisation des hyperparamètres avec Ray Tune : Exécution de campagnes bayésiennes, Optuna et basées sur la population sur des centaines de workers.

Démarrage rapide

import ray

ray.init()

# Tâche distribuée à distance
@ray.remote
def square(x):
    return x * x

# Lancer des tâches en parallèle sur le cluster
futures = [square.remote(i) for i in range(10)]
results = ray.get(futures)
print("Carrés calculés :", results)

Concepts clés

#Tâches distribuées et acteurs avec état

Passage à l'échelle de fonctions et de classes sur les workers du cluster :

import ray
import time

# Initialiser localement ou se connecter à un cluster Ray distant
ray.init(ignore_reinit_error=True)

# 1. Fonction distante sans état (tâche distribuée)
@ray.remote
def process_data_chunk(chunk_id: int) -> dict:
    time.sleep(0.5) # Simulation d'une tâche CPU/IO
    return {"chunk_id": chunk_id, "processed_items": 1000 * chunk_id}

# Lancer les tâches simultanément sur le cluster
futures = [process_data_chunk.remote(i) for i in range(8)]
results = ray.get(futures) # Récupération des résultats
print("Tranches traitées :", results)

# 2. Classe distante avec état (acteur distribué)
@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("Valeur finale du compteur :", ray.get(inc_futures[-1]))

#Déploiement de modèles en production avec Ray Serve

Déploiement de points de terminaison REST scalables avec regroupement dynamique de requêtes :

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)

#Optimisation distribuée des hyperparamètres avec Ray Tune

Exécution d'essais d'hyperparamètres en parallèle :

from ray import tune

def training_function(config):
    # Entraînement de modèle simulé
    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("Meilleurs hyperparamètres :", results.get_best_result().config)

Patterns courants

Acteurs distribués avec état pour le service de modèles

Problème : Recharger à chaque appel de fonction de lourds poids de modèle en mémoire.

Solution : Utiliser les acteurs Ray pour maintenir l'état en mémoire entre les requêtes :

@ray.remote(num_gpus=1)
class ModelServer:
    def __init__(self):
        # Charger les poids une fois dans la VRAM 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))

Bonnes pratiques (2026)

• À faire : Vérifier le tableau de bord Ray (port 8265) pour surveiller l'utilisation CPU/GPU des nœuds, le placement des acteurs et la mémoire du store d'objets.
• À faire : Passer les grands ensembles de données en lecture seule via le store d'objets plasma (ray.put(data)) pour éviter les sérialisations répétitives.
• À faire : Spécifier explicitement les besoins en ressources dans les décorateurs (@ray.remote(num_cpus=2, num_gpus=1)).
• À faire : Utiliser Ray Serve pour les microservices multi-modèles en production avec regroupement automatique intégré.
• À ne pas faire : Appeler ray.get() à l'intérieur de tâches distantes ; passer les ObjectRefs directement à d'autres tâches distantes pour préserver le pipeline.
• À ne pas faire : Créer des millions de micro-tâches ; regrouper les tâches fines pour minimiser la surcharge de l'ordonnanceur.
• À ne pas faire : Passer des objets volumineux avec état (comme des connexions de base de données ou des fichiers ouverts) dans les arguments distants.

Résolution des problèmes

Erreur	Cause	Solution
RaySystemError: Object store full	Des tâches produisent plus de données que la mémoire partagée allouée.	Utiliser ray.get() pour consommer les résultats, ou stocker les gros tableaux en disque.
Actor died unexpectedly	Le processus worker a planté à cause d'une exception non gérée ou d'un OOM.	Vérifier les logs dans le tableau de bord Ray (http://localhost:8265) et augmenter la RAM worker.
Task failed to schedule: Unsatisfiable resource request	Demande plus de CPU/GPU (num_gpus=2) qu'il n'en existe sur le cluster.	Vérifier les ressources disponibles via ray.cluster_resources().

Références

• Documentation Ray