[trainer] feat: generic remote backend abstraction for RL Training

examples/arctic_rl/run_gsm8k_grpo_arl_zorro_yes.sh

@@ -0,0 +1,87 @@
+#!/bin/bash
+# Canonical GSM8K example for the Arctic RemoteBackend.
+#
+# Demonstrates the generic `verl.remote_backend` abstraction with the
+# Arctic adapter (`trainer.remote_backend=arctic`). Single-GPU, GRPO,
+# Qwen3-0.6B; intended as a quick convergence sanity check.
+
+set -x
+export PYTHONUNBUFFERED=1
+export HYDRA_FULL_ERROR=1
+export RAY_DEDUP_LOGS=0
+export HF_HUB_OFFLINE=1
+export HF_HOME=/checkpoint/huggingface
+export USE_ARCTIC_TRAINING_CLIENT=1
+export VLLM_BATCH_INVARIANT=1
+export VLLM_ENABLE_V1_MULTIPROCESSING=0
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+export PYTHONPATH="$REPO_ROOT:$PYTHONPATH"
+
+gpu_name=$(nvidia-smi --query-gpu=gpu_name --format=csv,noheader -i 0)
+if   [[ $gpu_name == *"H200"* ]]; then flash_attention_v=flash_attention_3
+elif [[ $gpu_name == *"B200"* || $gpu_name == *"B300"* ]]; then flash_attention_v=flash_attention_2
+else flash_attention_v=flash_attention_2
+fi
+
+python3 -m verl.trainer.main_ppo \
+    algorithm.adv_estimator=grpo \
+    data.train_files=/code/shared/gsm8k/train.parquet \
+    data.val_files=/code/shared/gsm8k/test.parquet \
+    data.train_batch_size=16 \
+    data.max_prompt_length=512 \
+    data.max_response_length=1024 \
+    data.filter_overlong_prompts=True \
+    data.truncation='error' \
+    data.shuffle=False \
+    +data.seed=42 \
+    actor_rollout_ref.actor.data_loader_seed=42 \
+    reward.num_workers=1 \
+    actor_rollout_ref.rollout.agent.num_workers=1 \
+    actor_rollout_ref.model.path=Qwen/Qwen3-0.6B \
+    actor_rollout_ref.actor.optim.lr=1e-6 \
+    actor_rollout_ref.model.use_remove_padding=False \
+    actor_rollout_ref.actor.ppo_mini_batch_size=16 \
+    actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=16 \
+    actor_rollout_ref.actor.use_kl_loss=False \
+    actor_rollout_ref.actor.kl_loss_coef=0.001 \
+    actor_rollout_ref.actor.kl_loss_type=low_var_kl \
+    actor_rollout_ref.actor.entropy_coeff=0 \
+    actor_rollout_ref.model.enable_gradient_checkpointing=True \
+    +actor_rollout_ref.model.override_config.attn_implementation=$flash_attention_v \
+    actor_rollout_ref.actor.strategy=fsdp2 \
+    actor_rollout_ref.actor.fsdp_config.param_offload=False \
+    actor_rollout_ref.actor.fsdp_config.optimizer_offload=False \
+    actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=16 \
+    actor_rollout_ref.rollout.tensor_model_parallel_size=1 \
+    actor_rollout_ref.rollout.name=arctic \
+    actor_rollout_ref.rollout.gpu_memory_utilization=0.6 \
+    actor_rollout_ref.rollout.enforce_eager=True \
+    actor_rollout_ref.rollout.n=5 \
+    actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=16 \
+    actor_rollout_ref.ref.fsdp_config.param_offload=False \
+    actor_rollout_ref.ref.strategy=fsdp2 \
+    algorithm.use_kl_in_reward=False \
+    trainer.use_legacy_worker_impl=disable \
+    trainer.remote_backend=arctic \
+    remote_backend.arctic.colocate=False \
+    remote_backend.arctic.training_gpus=1 \
+    remote_backend.arctic.sampling_gpus=1 \
+    remote_backend.arctic.log_prob_gpus=0 \
+    remote_backend.arctic.zero_optimization.stage=2 \
+    remote_backend.arctic.zero_optimization.offload_optimizer.device=none \
+    remote_backend.arctic.zero_optimization.offload_param.device=none \
+    remote_backend.arctic.use_zorro=True \
+    trainer.critic_warmup=0 \
+    trainer.logger="['console']" \
+    trainer.experiment_name=gsm8k_grpo_qwen3_0p6b_ngpu1_gbs16_rolln5_zorroTrue \
+    trainer.project_name=arctic_rl_gsm8k_public \
+    trainer.val_before_train=False \
+    trainer.n_gpus_per_node=1 \
+    trainer.nnodes=1 \
+    trainer.save_freq=-1 \
+    trainer.test_freq=-1 \
+    trainer.total_training_steps=80 \
+    trainer.total_epochs=15 \
+    "$@" 2>&1

verl/experimental/agent_loop/agent_loop.py

@@ -918,13 +918,14 @@ def __init__(
         worker_group: RayWorkerGroup = None,
         rollout_resource_pool: RayResourcePool = None,
         reward_loop_worker_handles: list[ray.actor.ActorHandle] = None,
-    ):
+        **kwargs,
+  ):
         self.config = config
         self.rollout_config, self.model_config = _get_rollout_and_model_config(config)
         self.worker_group = worker_group
         self.rollout_resource_pool = rollout_resource_pool
         self.reward_loop_worker_handles = reward_loop_worker_handles
-
+        self.kwargs = kwargs
         assert worker_group is not None or self.rollout_config.nnodes > 0, "nnodes must be > 0 in standalone mode"
 
         # for recipe to change
@@ -941,9 +942,10 @@ async def create(
         worker_group: RayWorkerGroup = None,
         rollout_resource_pool: RayResourcePool = None,
         reward_loop_worker_handles: list[ray.actor.ActorHandle] = None,
+        **kwargs,
     ):
         """Create agent loop manager."""
-        instance = cls(config, worker_group, rollout_resource_pool, reward_loop_worker_handles)
+        instance = cls(config, worker_group, rollout_resource_pool, reward_loop_worker_handles, **kwargs)
         await instance._initialize_llm_servers()
         await instance._init_global_load_balancer()
         await instance._init_agent_loop_workers()
@@ -968,6 +970,7 @@ async def _initialize_llm_servers(self):
                 config=self.rollout_config,
                 model_config=self.model_config,
                 gpus_per_node=self.rollout_config.n_gpus_per_node,
+                **self.kwargs,
             )
             for replica_rank in range(num_replicas)
         ]

verl/remote_backend/__init__.py

@@ -0,0 +1,28 @@
+"""Generic remote-backend abstraction for verl.
+
+Lets verl drive an out-of-process RL backend (training + rollout +
+log-prob + checkpoint) that owns its own GPUs. Verl talks to a CPU-only
+forwarder worker group; the forwarder forwards every dispatched call to
+a :class:`RemoteBackend` implementation behind a Ray actor (or any other
+RPC the backend prefers).
+
+Pieces:
+
+* :class:`RemoteBackend` (``base.py``) — the all-abstract contract
+  every backend implements.
+* :class:`RemoteBackendRegistry` (``base.py``) — name → class lookup so
+  ``trainer.remote_backend="<name>"`` resolves to a concrete adapter.
+* :class:`RemoteBackendTrainer` (``trainer.py``) — `RayPPOTrainer` subclass
+  that creates the backend on the driver and threads its reconnect handle
+  to every worker.
+* :class:`RemoteBackendActorRolloutRefWorker` (``worker.py``) — the
+  backend-agnostic CPU forwarder.
+* ``worker_utils.py`` — small generic tensor / metric helpers shared by
+  the forwarder and backend adapters.
+
+See :mod:`verl.trainer.ppo.arctic_rl_client` for a reference adapter.
+"""
+
+from verl.remote_backend.base import RemoteBackend, RemoteBackendRegistry
+
+__all__ = ["RemoteBackend", "RemoteBackendRegistry"]