diff --git a/src/agentcore_rl_toolkit/backends/slime/SETUP.md b/src/agentcore_rl_toolkit/backends/slime/SETUP.md index 8d75e36..3ae0c3d 100644 --- a/src/agentcore_rl_toolkit/backends/slime/SETUP.md +++ b/src/agentcore_rl_toolkit/backends/slime/SETUP.md @@ -149,21 +149,76 @@ cp .wandb.env.example .wandb.env # optional; skip to disable wandb ### 3.4 Run training -train.sh defaults target **8 × H100** (NUM_GPUS=8, TP_SIZE=2, -ROLLOUT_GPUS_PER_ENGINE=2). For smaller clusters override via env -(e.g. `NUM_GPUS=1 TP_SIZE=1 ROLLOUT_GPUS_PER_ENGINE=1` for a single -GPU). Defaults also set `NUM_ROLLOUT=1` for smoke testing — bump to -`NUM_ROLLOUT=100` (slime's production value) for a real run. - -`SLIME_DIR` / -`MEGATRON_DIR` need to point at the slime + Megatron-LM source trees -(inside the `slimerl/slime:latest` container these are `/root/slime` -and `/root/Megatron-LM`). +Two entry points, same job under the hood: + +- **`SlimeRunner` (Python)** — recommended. Write a short `train.py`; + the class handles Ray/SGLang plumbing and slime CLI flags. +- **`train.sh` (bash)** — escape hatch. Use when you need to override + something the class doesn't surface, or to debug the raw slime CLI. + +#### Python (`SlimeRunner`) + +Defaults target **8 × H100** (`num_gpus=8`, `tp_size=2`, +`rollout_gpus_per_engine=2`). Override kwargs for other cluster sizes. +`.train(num_rollout=…)` defaults to 1 rollout for smoke testing — bump +to 100 for a real run. + +`slime_dir` / `megatron_dir` default to the in-container paths +(`/root/slime` and `/root/Megatron-LM`); override for bare-metal +installs. + +```python +# train.py — minimal 3B smoke test +from agentcore_rl_toolkit.backends.slime import SlimeRunner + +SlimeRunner( + exp_id="gsm8k-3b-smoke", + agent_runtime_arn="arn:aws:bedrock-agentcore:...", + s3_bucket="your-bucket", + model_dir="/path/to/Qwen2.5-3B-Instruct", + data_path="/path/to/gsm8k_tiny.jsonl", + model_type="qwen2.5-3B", +).train(num_rollout=1) +``` ```bash cd /path/to/agentcore-rl-toolkit +python src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.py +``` + +32B on 8 GPUs: + +```python +SlimeRunner( + exp_id="gsm8k-32b-run", + agent_runtime_arn="arn:aws:bedrock-agentcore:...", + s3_bucket="your-bucket", + model_dir="/path/to/Qwen2.5-32B-Instruct", + data_path="/path/to/gsm8k_tiny.jsonl", + model_type="qwen2.5-32B", + tp_size=8, + rollout_gpus_per_engine=8, +).train(num_rollout=5) +``` + +Any slime/Megatron-LM/SGLang CLI flag that isn't surfaced as a named +kwarg can be passed through `extra_flags`: + +```python +SlimeRunner(..., extra_flags=["--num-epoch", "3"]).train(num_rollout=50) +``` -# Qwen2.5-3B, 8 GPUs, 1 rollout (smoke test — train.sh defaults) +If you prefer a YAML config, `SlimeRunner.from_yaml("config.yaml").train()` +accepts the same keys. + +#### Bash (`train.sh`) — escape hatch + +`train.sh` takes the same knobs via env vars. It's kept as the low-level +reference for what the Python class replicates, and as a debugging path +for slime flag experiments. + +```bash +# 3B smoke test export SLIME_DIR=/root/slime \ MEGATRON_DIR=/root/Megatron-LM \ MODEL_DIR=/path/to/Qwen2.5-3B-Instruct \ @@ -171,16 +226,7 @@ export SLIME_DIR=/root/slime \ bash src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.sh ``` -For 32B on 8 GPUs: - -```bash -export MODEL_DIR=/path/to/Qwen2.5-32B-Instruct \ - MODEL_TYPE=qwen2.5-32B \ - TP_SIZE=8 \ - ROLLOUT_GPUS_PER_ENGINE=8 \ - NUM_ROLLOUT=5 -bash src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.sh -``` +For 32B on 8 GPUs, add `MODEL_TYPE=qwen2.5-32B TP_SIZE=8 ROLLOUT_GPUS_PER_ENGINE=8 NUM_ROLLOUT=5`. ### 3.5 Run evaluation diff --git a/src/agentcore_rl_toolkit/backends/slime/__init__.py b/src/agentcore_rl_toolkit/backends/slime/__init__.py index e69de29..f6797bc 100644 --- a/src/agentcore_rl_toolkit/backends/slime/__init__.py +++ b/src/agentcore_rl_toolkit/backends/slime/__init__.py @@ -0,0 +1,12 @@ +"""Slime training backend for agentcore-rl-toolkit. + +Primary entry point is :class:`SlimeRunner`. The ``integration/`` subpackage +and ``patches/`` module are implementation detail — users shouldn't import +them directly, but they are load-bearing (slime loads +``agentcore_rl_toolkit.backends.slime.integration.rollout.generate_rollout`` +via ``--rollout-function-path`` at job-submit time). +""" + +from .runner import SlimeRunner + +__all__ = ["SlimeRunner"] diff --git a/src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.py b/src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.py new file mode 100644 index 0000000..0be6601 --- /dev/null +++ b/src/agentcore_rl_toolkit/backends/slime/examples/math_agent/train.py @@ -0,0 +1,22 @@ +"""Train the strands math agent on GSM8K via slime — Python entry point. + +Prerequisites (see ``SETUP.md`` for full instructions): + - Inside a working slime environment (``slimerl/slime:latest`` container or equivalent). + - Agent deployed to ACR; runtime ARN + S3 bucket handy. + - Model checkpoint and training JSONL downloaded locally. + +Run: + python train.py +""" + +from agentcore_rl_toolkit.backends.slime import SlimeRunner + +if __name__ == "__main__": + SlimeRunner( + exp_id="gsm8k-3b-smoke", + agent_runtime_arn="arn:aws:bedrock-agentcore:::runtime/", + s3_bucket="your-bucket-name", + model_dir="/workspace/slime_workdir/models/Qwen2.5-3B-Instruct", + data_path="/workspace/slime_workdir/data/gsm8k_tiny.jsonl", + model_type="qwen2.5-3B", + ).train(num_rollout=1) diff --git a/src/agentcore_rl_toolkit/backends/slime/runner.py b/src/agentcore_rl_toolkit/backends/slime/runner.py new file mode 100644 index 0000000..556ee5c --- /dev/null +++ b/src/agentcore_rl_toolkit/backends/slime/runner.py @@ -0,0 +1,314 @@ +"""SlimeRunner — one Python entry point for slime-backed training. + +Users instantiate ``SlimeRunner`` with a handful of per-experiment fields +and call ``.train()``; the runner reproduces what ``train.sh`` does today +(stop stale processes, start a Ray head, source the slime model script, +submit the slime training job) via subprocess. + +``train.sh`` stays in the repo as the low-level escape hatch; this class +is the primary entry point. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import tempfile +from dataclasses import dataclass, field +from pathlib import Path + +# Keys on SlimeRunner that flow to slime's `args` namespace via the +# --custom-config-path YAML (slime's `utils/arguments.py` setattrs each key +# onto `args` at parse time). Consumed by SlimeArtConfig.from_args(). +_TOOLKIT_CONFIG_KEYS = ( + "agent_runtime_arn", + "s3_bucket", + "exp_id", + "gateway_port", + "acr_timeout", + "model_id", + "acr_tps_limit", + "max_concurrent", + "reward_postprocessing", +) + + +@dataclass +class SlimeRunner: + """One Python entry point for slime-backed training. + + Example: + >>> SlimeRunner( + ... exp_id="gsm8k-3b-smoke", + ... agent_runtime_arn="arn:aws:bedrock-agentcore:...", + ... s3_bucket="my-bucket", + ... model_dir="/path/to/Qwen2.5-3B-Instruct", + ... data_path="/path/to/gsm8k_tiny.jsonl", + ... model_type="qwen2.5-3B", + ... ).train(num_rollout=1) + """ + + # --- Required: per-experiment --- + exp_id: str + agent_runtime_arn: str + s3_bucket: str + model_dir: str + data_path: str + model_type: str + + # --- Optional: cluster --- + num_gpus: int = 8 + tp_size: int = 2 + rollout_gpus_per_engine: int = 2 + slime_dir: str = "/root/slime" + megatron_dir: str = "/root/Megatron-LM" + + # --- Optional: ACR / toolkit (forwarded to slime via custom-config yaml) --- + model_id: str = "default" + acr_timeout: int = 900 + acr_tps_limit: int = 25 + max_concurrent: int = 100 + gateway_port: int = 9090 + reward_postprocessing: str = "grpo" + + # --- Optional: training hyperparameters --- + rollout_batch_size: int = 32 + n_samples_per_prompt: int = 8 + rollout_max_response_len: int = 1024 + rollout_temperature: float = 1.0 + lr: float = 1e-6 + eps_clip: float = 0.2 + eps_clip_high: float = 0.28 + weight_decay: float = 0.1 + adam_beta2: float = 0.98 + sglang_mem_fraction_static: float = 0.7 + max_tokens_per_gpu: int = 9216 + + # --- Wandb (opt-in; no defaults injected if unset) --- + wandb_project: str | None = None + wandb_group: str | None = None + + # --- Escape hatch --- + extra_flags: list[str] = field(default_factory=list) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + @classmethod + def from_yaml(cls, path: str | os.PathLike) -> "SlimeRunner": + """Load kwargs from a YAML file (convenience for config-file workflows).""" + import yaml + + with open(path) as f: + data = yaml.safe_load(f) or {} + return cls(**data) + + def train(self, num_rollout: int = 1) -> None: + """Run the training job. Blocks until the slime job exits. + + Mirrors ``train.sh`` step-by-step: stop stale sglang/ray, start a Ray + head, source the slime model script, submit the slime training job + via ``ray job submit``. Streams stdout/stderr to the parent process. + """ + self._stop_stale_processes() + self._start_ray() + model_args = self._source_model_script() + runtime_env = self._build_runtime_env() + with self._write_toolkit_config() as config_path: + self._submit_ray_job(num_rollout, model_args, runtime_env, config_path) + + # ------------------------------------------------------------------ + # Internals — one per train.sh step, no magic + # ------------------------------------------------------------------ + + def _stop_stale_processes(self) -> None: + subprocess.run(["pkill", "-9", "sglang"], check=False) + subprocess.run(["ray", "stop", "--force"], check=False) + subprocess.run(["sleep", "3"], check=True) + + def _start_ray(self) -> None: + subprocess.run( + ["ray", "start", "--head", "--num-gpus", str(self.num_gpus), "--disable-usage-stats"], + check=True, + ) + + def _source_model_script(self) -> list[str]: + """Source slime's scripts/models/.sh and return MODEL_ARGS. + + slime ships per-model arg files (e.g. qwen2.5-3B.sh) that export + MODEL_ARGS as a bash array. We invoke bash to source the script and + print the array one element per null byte, then split in Python. + """ + script = Path(self.slime_dir) / "scripts" / "models" / f"{self.model_type}.sh" + if not script.exists(): + raise FileNotFoundError( + f"slime model script not found: {script}. " + f"Check slime_dir={self.slime_dir!r} and model_type={self.model_type!r}." + ) + cmd = f'source "{script}"; printf "%s\\0" "${{MODEL_ARGS[@]}}"' + out = subprocess.check_output(["bash", "-c", cmd]) + items = out.split(b"\0") + return [x.decode() for x in items if x] + + def _build_runtime_env(self) -> dict: + """Runtime env forwarded to every Ray worker. + + Mirrors train.sh's inline python snippet: PYTHONPATH for Megatron, + CUDA_DEVICE_MAX_CONNECTIONS for Megatron TP, plus wandb keys when + set in the parent environment. + """ + env_vars: dict[str, str] = { + "PYTHONPATH": self.megatron_dir, + "CUDA_DEVICE_MAX_CONNECTIONS": "1", + } + for key in ("WANDB_API_KEY", "WANDB_ENTITY"): + val = os.environ.get(key) + if val: + env_vars[key] = val + return {"env_vars": env_vars} + + def _write_toolkit_config(self): + """Write a temp YAML of toolkit fields for slime --custom-config-path. + + slime's argparse loads this YAML and setattr's each key onto args, + where our rollout integration reads them via SlimeArtConfig.from_args. + Returned as a context manager so the temp file lives for the job's + duration and is cleaned up after. + """ + import contextlib + + import yaml + + data = {k: getattr(self, k) for k in _TOOLKIT_CONFIG_KEYS} + + @contextlib.contextmanager + def _ctx(): + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", prefix="slime-runner-", delete=False) as f: + yaml.safe_dump(data, f) + path = f.name + try: + yield path + finally: + try: + os.unlink(path) + except OSError: + pass + + return _ctx() + + def _submit_ray_job( + self, + num_rollout: int, + model_args: list[str], + runtime_env: dict, + config_path: str, + ) -> None: + flags = self._build_slime_flags(num_rollout, model_args, config_path) + cmd = [ + "ray", + "job", + "submit", + "--address=http://127.0.0.1:8265", + f"--runtime-env-json={json.dumps(runtime_env)}", + "--", + "python3", + str(Path(self.slime_dir) / "train.py"), + *flags, + ] + subprocess.run(cmd, check=True) + + def _build_slime_flags( + self, + num_rollout: int, + model_args: list[str], + config_path: str, + ) -> list[str]: + """Flags passed to ``python3 slime/train.py`` — mirrors train.sh 1:1.""" + flags: list[str] = [ + *model_args, + "--hf-checkpoint", + self.model_dir, + "--ref-load", + self.model_dir, + "--prompt-data", + self.data_path, + "--num-rollout", + str(num_rollout), + "--tensor-model-parallel-size", + str(self.tp_size), + "--rollout-num-gpus-per-engine", + str(self.rollout_gpus_per_engine), + "--input-key", + "prompt", + "--rollout-batch-size", + str(self.rollout_batch_size), + "--n-samples-per-prompt", + str(self.n_samples_per_prompt), + "--rollout-max-response-len", + str(self.rollout_max_response_len), + "--rollout-temperature", + str(self.rollout_temperature), + "--advantage-estimator", + "grpo", + "--use-kl-loss", + "--kl-loss-type", + "low_var_kl", + "--eps-clip", + str(self.eps_clip), + "--eps-clip-high", + str(self.eps_clip_high), + "--lr", + str(self.lr), + "--lr-decay-style", + "constant", + "--weight-decay", + str(self.weight_decay), + "--adam-beta2", + str(self.adam_beta2), + "--optimizer-cpu-offload", + "--overlap-cpu-optimizer-d2h-h2d", + "--use-precision-aware-optimizer", + "--sequence-parallel", + "--sglang-mem-fraction-static", + str(self.sglang_mem_fraction_static), + "--sglang-cuda-graph-max-bs", + "32", + "--sglang-tool-call-parser", + "qwen25", + "--attention-dropout", + "0.0", + "--hidden-dropout", + "0.0", + "--accumulate-allreduce-grads-in-fp32", + "--attention-softmax-in-fp32", + "--attention-backend", + "flash", + "--actor-num-gpus-per-node", + str(self.num_gpus), + "--colocate", + "--megatron-to-hf-mode", + "bridge", + "--rollout-function-path", + "agentcore_rl_toolkit.backends.slime.integration.rollout.generate_rollout", + "--custom-reward-post-process-path", + "agentcore_rl_toolkit.backends.slime.integration.rewards.normalize_episode_rewards", + "--custom-config-path", + config_path, + "--use-dynamic-global-batch-size", + "--use-dynamic-batch-size", + "--max-tokens-per-gpu", + str(self.max_tokens_per_gpu), + ] + + # Wandb opt-in: only emit --use-wandb if the user (or env) supplied an API key. + if os.environ.get("WANDB_API_KEY"): + flags.append("--use-wandb") + if self.wandb_project: + flags.extend(["--wandb-project", self.wandb_project]) + if self.wandb_group: + flags.extend(["--wandb-group", self.wandb_group]) + + flags.extend(self.extra_flags) + return flags diff --git a/tests/test_slime_runner.py b/tests/test_slime_runner.py new file mode 100644 index 0000000..a84fc68 --- /dev/null +++ b/tests/test_slime_runner.py @@ -0,0 +1,90 @@ +"""Tests for SlimeRunner — the Python entry point around train.sh.""" + +from pathlib import Path +from unittest.mock import patch + +import yaml + +from agentcore_rl_toolkit.backends.slime import SlimeRunner + +REQUIRED_KWARGS = dict( + exp_id="exp-1", + agent_runtime_arn="arn:aws:bedrock-agentcore:us-west-2:111122223333:runtime/foo", + s3_bucket="my-bucket", + model_dir="/models/Qwen2.5-3B-Instruct", + data_path="/data/gsm8k.jsonl", + model_type="qwen2.5-3B", +) + + +def test_build_runtime_env_required_keys_and_wandb_opt_in(): + """PYTHONPATH + CUDA_DEVICE_MAX_CONNECTIONS are always present; wandb keys + are opt-in, forwarded only when set in the parent env.""" + runner = SlimeRunner(**REQUIRED_KWARGS, megatron_dir="/opt/megatron") + + with patch.dict("os.environ", {}, clear=True): + env_no_wandb = runner._build_runtime_env() + assert env_no_wandb == { + "env_vars": { + "PYTHONPATH": "/opt/megatron", + "CUDA_DEVICE_MAX_CONNECTIONS": "1", + } + } + + with patch.dict("os.environ", {"WANDB_API_KEY": "abc", "WANDB_ENTITY": "me"}, clear=True): + env_with_wandb = runner._build_runtime_env() + assert env_with_wandb["env_vars"]["WANDB_API_KEY"] == "abc" + assert env_with_wandb["env_vars"]["WANDB_ENTITY"] == "me" + + +def test_from_yaml_round_trips(tmp_path: Path): + """from_yaml should accept the same keys the dataclass does.""" + config = tmp_path / "config.yaml" + config.write_text(yaml.safe_dump({**REQUIRED_KWARGS, "num_gpus": 4, "lr": 5e-7})) + + runner = SlimeRunner.from_yaml(config) + + assert runner.exp_id == "exp-1" + assert runner.num_gpus == 4 + assert runner.lr == 5e-7 + + +def test_slime_flags_pass_through_key_kwargs(): + """Flags the user cares about most (num_rollout, tp_size, lr, rollout_batch_size, + rollout-function-path) must reach the slime CLI verbatim.""" + runner = SlimeRunner(**REQUIRED_KWARGS, tp_size=4, lr=5e-7, rollout_batch_size=16) + flags = runner._build_slime_flags(num_rollout=10, model_args=["--fake-model-arg"], config_path="/tmp/cfg.yaml") + + # --fake-model-arg comes from the (mocked) model script and must be first + assert flags[0] == "--fake-model-arg" + # Key kwargs flow through as --k v pairs + assert "--num-rollout" in flags and flags[flags.index("--num-rollout") + 1] == "10" + assert "--tensor-model-parallel-size" in flags and flags[flags.index("--tensor-model-parallel-size") + 1] == "4" + assert "--lr" in flags and flags[flags.index("--lr") + 1] == "5e-07" + assert "--rollout-batch-size" in flags and flags[flags.index("--rollout-batch-size") + 1] == "16" + # Integration hooks (load-bearing dotted paths) + assert "agentcore_rl_toolkit.backends.slime.integration.rollout.generate_rollout" in flags + assert "agentcore_rl_toolkit.backends.slime.integration.rewards.normalize_episode_rewards" in flags + assert "/tmp/cfg.yaml" in flags + + +def test_extra_flags_are_appended_to_slime_cli(): + """The escape hatch: extra_flags end up at the end of the CLI unmodified.""" + runner = SlimeRunner(**REQUIRED_KWARGS, extra_flags=["--num-epoch", "3", "--use-rollout-routing-replay"]) + flags = runner._build_slime_flags(num_rollout=1, model_args=[], config_path="/tmp/cfg.yaml") + + assert flags[-3:] == ["--num-epoch", "3", "--use-rollout-routing-replay"] + + +def test_toolkit_config_yaml_includes_acr_pointers(): + """The temp yaml written for --custom-config-path must carry the ACR fields + the rollout integration reads via SlimeArtConfig.from_args.""" + runner = SlimeRunner(**REQUIRED_KWARGS, model_id="qwen-served") + + with runner._write_toolkit_config() as path: + data = yaml.safe_load(Path(path).read_text()) + + assert data["agent_runtime_arn"] == REQUIRED_KWARGS["agent_runtime_arn"] + assert data["s3_bucket"] == REQUIRED_KWARGS["s3_bucket"] + assert data["exp_id"] == REQUIRED_KWARGS["exp_id"] + assert data["model_id"] == "qwen-served"