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
63 changes: 39 additions & 24 deletions nemo_skills/pipeline/nemo_rl/grpo.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
get_timeout,
resolve_mount_paths,
run_exp,
temporary_env_update,
get_nsight_cmd,
)
from nemo_skills.utils import get_logger_name, setup_logging

Expand All @@ -57,6 +59,7 @@ class NemoRLTask:
log_dir: str
env_variables: dict
backend: str
profile_step_range: str
extra_arguments: str = ""

def format_train_args(self):
Expand Down Expand Up @@ -94,10 +97,11 @@ def format_wandb_args(self):

def get_cmd(self):
self.logging_params = self.format_wandb_args()

nsight_cmd = get_nsight_cmd(self.profile_step_range)
cmd = (
f"export PYTHONPATH=$PYTHONPATH:/nemo_run/code:/opt/NeMo-RL && "
f"export UV_PROJECT=/opt/NeMo-RL && "
f"{nsight_cmd}"
f"echo 'Starting training' && "
f"uv run --active python /nemo_run/code/nemo_skills/training/nemo_rl/start_grpo.py "
f" {self.format_train_args()} "
Expand Down Expand Up @@ -125,6 +129,7 @@ def get_training_cmd(
log_dir,
env_variables,
backend,
profile_step_range,
):
timeout = get_timeout(cluster_config, partition)

Expand All @@ -144,6 +149,7 @@ def get_training_cmd(
log_dir=log_dir,
env_variables=env_variables,
backend=backend,
profile_step_range=profile_step_range,
)

return task.get_cmd()
Expand Down Expand Up @@ -197,6 +203,12 @@ def grpo_nemo_rl(
wandb_project: str = typer.Option("nemo-skills", help="Weights & Biases project name"),
wandb_group: str = typer.Option(None, help="Weights & Biases group name."),
disable_wandb: bool = typer.Option(False, help="Disable wandb logging"),
profile_step_range: str = typer.Option(
None,
help="Controls which training steps the nsys profiler captures. "
"Format: START:STOP (1-indexed, STOP exclusive, same as slice syntax arr[start:stop]). "
"Example: '3:5' profiles steps 3 and 4 only. NOTE: START must be ≥ 1, so '0:10' is invalid."
),
partition: str = typer.Option(
None, help="Can specify if need interactive jobs or a specific non-default partition"
),
Expand Down Expand Up @@ -302,34 +314,37 @@ def grpo_nemo_rl(
log_dir=f"{log_dir}/training-logs",
env_variables=env_variables,
backend=backend,
profile_step_range=profile_step_range,
)

server_config = None
env_update = {"RAY_LOG_SYNC_FREQUENCY": 20} if profile_step_range else {}
with get_exp(expname, cluster_config, _reuse_exp) as exp:
prev_task = _task_dependencies
for job_id in range(num_training_jobs):
prev_task = add_task(
exp,
cmd=train_cmd,
task_name=f'{expname}-grpo-{job_id}',
log_dir=f"{log_dir}/training-logs",
container=cluster_config["containers"]["nemo-rl"],
num_gpus=num_gpus,
num_nodes=num_nodes,
cluster_config=cluster_config,
server_config=server_config,
partition=partition,
time_min=time_min,
run_after=run_after,
reuse_code=reuse_code,
reuse_code_exp=reuse_code_exp,
task_dependencies=[prev_task] if prev_task is not None else None,
slurm_kwargs={"exclusive": exclusive} if exclusive else None,
heterogeneous=True if server_config is not None else False,
with_sandbox=with_sandbox,
with_ray=True,
installation_command=installation_command,
)
with temporary_env_update(cluster_config, env_update):
for job_id in range(num_training_jobs):
prev_task = add_task(
exp,
cmd=train_cmd,
task_name=f'{expname}-grpo-{job_id}',
log_dir=f"{log_dir}/training-logs",
container=cluster_config["containers"]["nemo-rl"],
num_gpus=num_gpus,
num_nodes=num_nodes,
cluster_config=cluster_config,
server_config=server_config,
partition=partition,
time_min=time_min,
run_after=run_after,
reuse_code=reuse_code,
reuse_code_exp=reuse_code_exp,
task_dependencies=[prev_task] if prev_task is not None else None,
slurm_kwargs={"exclusive": exclusive} if exclusive else None,
heterogeneous=True if server_config is not None else False,
with_sandbox=with_sandbox,
with_ray=True,
installation_command=installation_command,
)

prev_task = add_task(
exp,
Expand Down
80 changes: 48 additions & 32 deletions nemo_skills/pipeline/nemo_rl/sft.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
get_timeout,
resolve_mount_paths,
run_exp,
temporary_env_update,
get_nsight_cmd,
)
from nemo_skills.utils import get_logger_name, setup_logging

Expand Down Expand Up @@ -59,6 +61,7 @@ class NemoRLTask:
log_dir: str
env_variables: dict
backend: str
profile_step_range: str
extra_arguments: str = ""

def format_train_args(self):
Expand Down Expand Up @@ -94,20 +97,22 @@ def format_wandb_args(self):

def get_cmd(self):
self.logging_params = self.format_wandb_args()

nsight_cmd = get_nsight_cmd(self.profile_step_range)
cmd = (
f"export PYTHONPATH=$PYTHONPATH:/nemo_run/code:/opt/NeMo-RL && "
f"export UV_PROJECT=/opt/NeMo-RL && "
f"echo 'Starting training' && "
f"NRL_FORCE_REBUILD_VENVS=true uv run --active python /nemo_run/code/nemo_skills/training/nemo_rl/start_sft.py "
f" {self.format_train_args()} "
f" {self.format_data_args()} "
f" {self.logging_params} "
f" {self.extra_arguments} "
"export PYTHONPATH=$PYTHONPATH:/nemo_run/code:/opt/NeMo-RL && "
"export UV_PROJECT=/opt/NeMo-RL && "
f"{nsight_cmd}"
"echo 'Starting training' && "
"NRL_FORCE_REBUILD_VENVS=true uv run --active "
"python /nemo_run/code/nemo_skills/training/nemo_rl/start_sft.py "
f"{self.format_train_args()} {self.format_data_args()} "
f"{self.logging_params} {self.extra_arguments}"
)

return cmd



def get_training_cmd(
cluster_config,
partition,
Expand All @@ -125,6 +130,7 @@ def get_training_cmd(
log_dir,
env_variables,
backend,
profile_step_range,
):
timeout = get_timeout(cluster_config, partition)

Expand All @@ -144,6 +150,7 @@ def get_training_cmd(
log_dir=log_dir,
env_variables=env_variables,
backend=backend,
profile_step_range=profile_step_range,
)

return task.get_cmd()
Expand Down Expand Up @@ -197,6 +204,12 @@ def sft_nemo_rl(
wandb_project: str = typer.Option("nemo-skills", help="Weights & Biases project name"),
wandb_group: str = typer.Option(None, help="Weights & Biases group name."),
disable_wandb: bool = typer.Option(False, help="Disable wandb logging"),
profile_step_range: str = typer.Option(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same comments here as in the grpo file

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

None,
help="Controls which training steps the nsys profiler captures. "
"Format: START:STOP (1-indexed, STOP exclusive, same as slice syntax arr[start:stop]). "
"Example: '3:5' profiles steps 3 and 4 only. NOTE: START must be ≥ 1, so '0:10' is invalid."
),
partition: str = typer.Option(
None, help="Can specify if need interactive jobs or a specific non-default partition"
),
Expand Down Expand Up @@ -301,34 +314,37 @@ def sft_nemo_rl(
log_dir=f"{log_dir}/training-logs",
env_variables=env_variables,
backend=backend,
profile_step_range=profile_step_range,
)

server_config = None
env_update = {"RAY_LOG_SYNC_FREQUENCY": 20} if profile_step_range else {}
with get_exp(expname, cluster_config, _reuse_exp) as exp:
prev_task = _task_dependencies
for job_id in range(num_training_jobs):
prev_task = add_task(
exp,
cmd=train_cmd,
task_name=f'{expname}-sft-{job_id}',
log_dir=f"{log_dir}/training-logs",
container=cluster_config["containers"]["nemo-rl"],
num_gpus=num_gpus,
num_nodes=num_nodes,
cluster_config=cluster_config,
server_config=server_config,
partition=partition,
time_min=time_min,
run_after=run_after,
reuse_code=reuse_code,
reuse_code_exp=reuse_code_exp,
task_dependencies=[prev_task] if prev_task is not None else None,
slurm_kwargs={"exclusive": exclusive} if exclusive else None,
heterogeneous=True if server_config is not None else False,
with_sandbox=False,
with_ray=True,
installation_command=installation_command,
)
with temporary_env_update(cluster_config, env_update):
for job_id in range(num_training_jobs):
prev_task = add_task(
exp,
cmd=train_cmd,
task_name=f'{expname}-sft-{job_id}',
log_dir=f"{log_dir}/training-logs",
container=cluster_config["containers"]["nemo-rl"],
num_gpus=num_gpus,
num_nodes=num_nodes,
cluster_config=cluster_config,
server_config=server_config,
partition=partition,
time_min=time_min,
run_after=run_after,
reuse_code=reuse_code,
reuse_code_exp=reuse_code_exp,
task_dependencies=[prev_task] if prev_task is not None else None,
slurm_kwargs={"exclusive": exclusive} if exclusive else None,
heterogeneous=True if server_config is not None else False,
with_sandbox=False,
with_ray=True,
installation_command=installation_command,
)

prev_task = add_task(
exp,
Expand Down
1 change: 1 addition & 0 deletions nemo_skills/pipeline/utils/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
get_exp_handles,
get_sandbox_command,
run_exp,
get_nsight_cmd,
)
from nemo_skills.pipeline.utils.generation import (
configure_client,
Expand Down
12 changes: 12 additions & 0 deletions nemo_skills/pipeline/utils/exp.py
Original file line number Diff line number Diff line change
Expand Up @@ -633,3 +633,15 @@ def get_exp(expname, cluster_config, _reuse_exp=None):
if cluster_config['executor'] == 'local':
return run.Experiment(expname, clean_mode=True)
return run.Experiment(expname, clean_mode=True, log_level="WARN")


def get_nsight_cmd(profile_step_range):
cmd = ''
if profile_step_range is not None:
cmd = (
f'export LD_LIBRARY_PATH="/usr/local/cuda/lib64:/usr/local/cuda/lib:/usr/local/nvidia/lib64:/usr/local/nvidia/lib:/usr/lib/x86_64-linux-gnu" && '
f"export NRL_NSYS_PROFILE_STEP_RANGE={profile_step_range} && "
'export NRL_NSYS_WORKER_PATTERNS="*policy*,*vllm*" && '

)
return cmd