Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions examples/nemo_gym/run_grpo_nemo_gym.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,7 @@
from nemo_rl.algorithms.utils import get_tokenizer
from nemo_rl.data.utils import setup_response_data
from nemo_rl.distributed.virtual_cluster import init_ray
from nemo_rl.environments.nemo_gym import (
setup_nemo_gym_config,
)
from nemo_rl.environments.nemo_gym import setup_nemo_gym_config
from nemo_rl.experience.rollouts import run_nemo_gym_rollout_sync
from nemo_rl.models.generation import configure_generation_config
from nemo_rl.utils.config import (
Expand Down
71 changes: 11 additions & 60 deletions nemo_rl/algorithms/grpo.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import ray
import torch
from pydantic import BaseModel, Field, model_validator
from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
from torchdata.stateful_dataloader import StatefulDataLoader
from transformers import AutoProcessor
from transformers.tokenization_utils_base import PreTrainedTokenizerBase
Expand Down Expand Up @@ -80,12 +79,7 @@
prepare_segment_topology,
)
from nemo_rl.environments.interfaces import EnvironmentInterface
from nemo_rl.environments.nemo_gym import (
NemoGym,
NemoGymConfig,
get_nemo_gym_uv_cache_dir,
get_nemo_gym_venv_dir,
)
from nemo_rl.environments.nemo_gym import spinup_nemo_gym_actor
from nemo_rl.experience.interfaces import (
NEMO_GYM_TASK_INDEX_KEY,
NEXT_NEMO_GYM_TASK_INDEX_KEY,
Expand Down Expand Up @@ -594,67 +588,24 @@ def init_train_dataloader(dataset, suffix: str = ""):
master_config, enable_nemo_gym=enable_nemo_gym
)
nemo_gym_actor = None
if enable_nemo_gym:
nemo_gym_num_nodes = env_configs.get("nemo_gym", {}).get("num_gpu_nodes", 0)
ray_runtime_ctx = ray.get_runtime_context()
ray_cur_node_id = ray_runtime_ctx.get_node_id()
else:
nemo_gym_num_nodes = 0
ray_cur_node_id = None

def _spinup_nemo_gym(base_urls, model_name):
"""Spin up the NeMo Gym actor against the given generation server URLs."""
t0 = time.perf_counter()
nemo_gym_py_exec = get_actor_python_env("nemo_rl.environments.nemo_gym.NemoGym")
if nemo_gym_py_exec.startswith("uv"):
nemo_gym_py_exec = create_local_venv_on_each_node(
nemo_gym_py_exec, "nemo_rl.environments.nemo_gym.NemoGym"
)
nemo_gym_dict = dict(env_configs["nemo_gym"])
# NeMo-RL-side detection knobs are top-level NemoGymConfig fields
# (where the detector reads them), not part of Gym's global config.
invalid_tool_call_patterns = nemo_gym_dict.pop(
"invalid_tool_call_patterns", None
enable_router_replay = router_replay_enabled(policy_config)
routed_experts_dtype = (
resolve_routed_experts_dtype_name_for_model(model_name)
if enable_router_replay
else "int16"
)
thinking_tags = nemo_gym_dict.pop("thinking_tags", None)
# Pass prebuilt cache + venv dirs through the global config so the gym reuses
# image-baked venvs instead of rebuilding them.
uv_cache_dir = get_nemo_gym_uv_cache_dir()
if uv_cache_dir is not None:
nemo_gym_dict.setdefault("uv_cache_dir", uv_cache_dir)
uv_venv_dir = get_nemo_gym_venv_dir()
if uv_venv_dir is not None:
nemo_gym_dict.setdefault("uv_venv_dir", uv_venv_dir)
nemo_gym_cfg = NemoGymConfig(
model_name=model_name,
actor = spinup_nemo_gym_actor(
env_configs=env_configs,
base_urls=base_urls,
invalid_tool_call_patterns=invalid_tool_call_patterns,
thinking_tags=thinking_tags,
require_routed_experts=router_replay_enabled(policy_config),
routed_experts_dtype=(
resolve_routed_experts_dtype_name_for_model(model_name)
if router_replay_enabled(policy_config)
else "int16"
),
model_name=model_name,
enable_router_replay=enable_router_replay,
routed_experts_dtype=routed_experts_dtype,
use_fastokens=bool(policy_config["tokenizer"].get("use_fastokens")),
initial_global_config_dict=nemo_gym_dict,
)
nemo_gym_opts = {}
if nemo_gym_num_nodes:
nemo_gym_opts["scheduling_strategy"] = NodeAffinitySchedulingStrategy(
node_id=ray_cur_node_id,
soft=True,
)
nemo_gym_opts["runtime_env"] = {
"py_executable": nemo_gym_py_exec,
"env_vars": {
**os.environ,
"VIRTUAL_ENV": nemo_gym_py_exec,
"UV_PROJECT_ENVIRONMENT": nemo_gym_py_exec,
},
}
actor = NemoGym.options(**nemo_gym_opts).remote(nemo_gym_cfg)
ray.get(actor._spinup.remote())
return actor, time.perf_counter() - t0

total_nodes = cluster_config["num_nodes"]
Expand Down
16 changes: 12 additions & 4 deletions nemo_rl/algorithms/single_controller_utils/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,7 @@
from transformers.tokenization_utils_base import PreTrainedTokenizerBase

from nemo_rl.algorithms.async_utils.replay_buffer import TQReplayBuffer
from nemo_rl.algorithms.grpo import (
MasterConfig as GrpoMasterConfig,
)
from nemo_rl.algorithms.grpo import MasterConfig as GrpoMasterConfig
from nemo_rl.algorithms.grpo import (
_create_advantage_estimator,
_should_use_nemo_gym,
Expand All @@ -51,6 +49,9 @@
from nemo_rl.environments.interfaces import EnvironmentInterface
from nemo_rl.environments.nemo_gym import spinup_nemo_gym_actor
from nemo_rl.experience.rollout_manager import RolloutManager
from nemo_rl.models.generation.interfaces import (
resolve_routed_experts_dtype_name_for_model,
)
from nemo_rl.models.generation.sglang.config import SGLangConfig
from nemo_rl.models.generation.sglang.sglang_generation import SGLangGeneration
from nemo_rl.models.generation.vllm import VllmGeneration
Expand Down Expand Up @@ -384,12 +385,19 @@ def setup_single_controller(
if use_nemo_gym:
# TODO(#2625): Mirror GRPO's deferred vLLM load so NeMo-Gym spinup
# overlaps model loading instead of running serially afterward.
enable_router_replay = router_replay_enabled(master_config.policy)
enable_router_replay = router_replay_enabled(policy_config)
routed_experts_dtype = (
resolve_routed_experts_dtype_name_for_model(generation_config["model_name"])
if enable_router_replay
else "int16"
)
env_handles["nemo_gym"] = spinup_nemo_gym_actor(
env_configs=master_config.env,
base_urls=generation.dp_openai_server_base_urls,
model_name=generation_config["model_name"],
enable_router_replay=enable_router_replay,
routed_experts_dtype=routed_experts_dtype,
use_fastokens=bool(policy_config["tokenizer"].get("use_fastokens")),
)

# ==========================
Expand Down
29 changes: 18 additions & 11 deletions nemo_rl/environments/nemo_gym.py
Original file line number Diff line number Diff line change
Expand Up @@ -613,26 +613,31 @@ def spinup_nemo_gym_actor(
env_configs: dict[str, Any],
base_urls: list[Optional[str]],
model_name: str,
*,
enable_router_replay: bool,
routed_experts_dtype: str,
use_fastokens: bool,
) -> Any:
Comment thread
yuki-97 marked this conversation as resolved.
"""Spin up the NeMo-Gym actor against the given generation server URLs.

When ``env_configs["nemo_gym"]["num_gpu_nodes"] > 0``, the actor is
scheduled with soft NodeAffinity to the current Ray node so its colocated
GPU resources land where the caller expects.
When env_configs["nemo_gym"]["num_gpu_nodes"] > 0, the actor is scheduled
with soft NodeAffinity to the current Ray node so its colocated GPU
resources land where the caller expects.

Args:
env_configs: The ``master_config.env`` mapping; ``env_configs["nemo_gym"]``
supplies the Gym global config plus NeMo-RL detection knobs
(``invalid_tool_call_patterns``, ``thinking_tags``, ``num_gpu_nodes``).
base_urls: Per-DP-rank OpenAI-compatible server base URLs from the
generation backend.
env_configs: The master_config.env mapping; env_configs["nemo_gym"] supplies
the Gym global config plus NeMo-RL detection knobs (invalid_tool_call_patterns,
thinking_tags, num_gpu_nodes).
base_urls: Per-DP-rank OpenAI-compatible server base URLs from the generation backend.
model_name: Served model name the Gym rollouts should target.
enable_router_replay: Sets ``require_routed_experts`` on the
``NemoGymConfig``.
enable_router_replay: Sets require_routed_experts on the NemoGymConfig.
routed_experts_dtype: Dtype name for R3 routed_experts tensors ("int8"/"int16"/"int32"),
resolved by the caller from the model's expert count.
use_fastokens: Forwarded from policy.tokenizer.use_fastokens so the rollout actor
patches its tokenizer consistently with the driver.

Returns:
The spun-up ``NemoGym`` Ray actor handle (``_spinup`` already awaited).
The spun-up NemoGym Ray actor handle (_spinup already awaited).
"""
nemo_gym_dict = dict(env_configs["nemo_gym"])

Expand All @@ -656,6 +661,8 @@ def spinup_nemo_gym_actor(
invalid_tool_call_patterns=invalid_tool_call_patterns,
thinking_tags=thinking_tags,
require_routed_experts=enable_router_replay,
routed_experts_dtype=routed_experts_dtype,
use_fastokens=use_fastokens,
initial_global_config_dict=nemo_gym_dict,
)

Expand Down
3 changes: 3 additions & 0 deletions tests/unit/single_controller/test_single_controller_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ def _make_master_config(
policy={
"train_global_batch_size": num_prompts_per_step * 2,
"max_total_sequence_length": 32,
"tokenizer": {"use_fastokens": False},
"megatron_cfg": {"enabled": megatron_enabled},
"generation": {
"backend": backend,
Expand Down Expand Up @@ -421,6 +422,8 @@ def test_nemo_gym_wires_env_handle(self, patched_factories):
].return_value.dp_openai_server_base_urls,
model_name="test-model",
enable_router_replay=False,
routed_experts_dtype="int16",
use_fastokens=False,
)
assert actor_args.env_handles["nemo_gym"] is fake_gym_actor

Expand Down
Loading