support for external dynamo deployment

Signed-off-by: jthomson04 <jwillthomson19@gmail.com>

Author: jthomson04

nemo_rl/algorithms/grpo.py

@@ -395,6 +395,10 @@ def init_train_dataloader(dataset, suffix: str = ""):
     # ==========================
     print("\n▶ Setting up compute cluster...", flush=True)
     colocated_inference = generation_config["colocated"]["enabled"]
+    is_external_dynamo = (
+        generation_config.get("backend") == "dynamo"
+        and bool(generation_config.get("dynamo_cfg", {}).get("external_url"))
+    )
 
     env_name_list = extract_necessary_env_names(data_config)
     rm_env_enabled = "reward_model" in env_name_list
@@ -417,7 +421,29 @@ def init_train_dataloader(dataset, suffix: str = ""):
             f"policy_nodes:{policy_nodes} + rm_nodes:{rm_nodes} = total_nodes:{total_nodes}"
         )
 
-    if colocated_inference:
+    if is_external_dynamo:
+        # External Dynamo: no local inference resources — all GPUs go to training.
+        if total_nodes == 1:
+            train_gpus_per_node = cluster_config["gpus_per_node"] - rm_gpus_per_node
+        else:
+            train_gpus_per_node = cluster_config["gpus_per_node"]
+
+        train_cluster = RayVirtualCluster(
+            name="grpo_train_cluster",
+            bundle_ct_per_node_list=[train_gpus_per_node] * policy_nodes,
+            use_gpus=True,
+            num_gpus_per_node=train_gpus_per_node,
+            max_colocated_worker_groups=1,
+        )
+        inference_cluster = None
+        colocated_inference = False
+        print(
+            f"  ✓ External Dynamo mode — all {policy_nodes} node(s) with "
+            f"{train_gpus_per_node} GPUs/node allocated to training",
+            flush=True,
+        )
+
+    elif colocated_inference:
         if total_nodes == 1:
             policy_gpus_per_node = cluster_config["gpus_per_node"] - rm_gpus_per_node
             assert policy_gpus_per_node > 0, (

nemo_rl/models/generation/__init__.py

@@ -73,6 +73,7 @@ def configure_generation_config(
     # dynamo setting — always loads real weights (no refit support)
     elif config["backend"] == "dynamo":
         config = cast(DynamoVllmConfig, config)
-        config["vllm_cfg"]["load_format"] = "auto"
+        if not config.get("dynamo_cfg", {}).get("external_url"):
+            config["vllm_cfg"]["load_format"] = "auto"
 
     return config

nemo_rl/models/generation/dynamo/config.py

@@ -36,6 +36,7 @@ class DynamoSpecificArgs(TypedDict):
 class DynamoCfg(TypedDict, total=False):
     """Dynamo infrastructure configuration."""
 
+    external_url: str  # URL of an external Dynamo deployment (skips all local infra)
     frontend_http_port: int  # 0 = auto-assign
     router_mode: str  # "round-robin", "kv", "random", "least-loaded"
     etcd_port: int  # 0 = auto-assign

nemo_rl/models/generation/dynamo/dynamo_generation.py

@@ -76,15 +76,33 @@ class DynamoVllmGeneration(GenerationInterface):
 
     def __init__(
         self,
-        cluster: RayVirtualCluster,
+        cluster: Optional[RayVirtualCluster],
         config: DynamoVllmConfig,
         name_prefix: str = "dynamo_vllm",
         workers_per_node: Optional[Union[int, list[int]]] = None,
     ):
         self.cfg = config
-        vllm_cfg = config["vllm_cfg"]
         dynamo_cfg = config.get("dynamo_cfg", {})
 
+        # External mode: point at an already-running Dynamo deployment.
+        external_url = dynamo_cfg.get("external_url")
+        if external_url:
+            self._external = True
+            self._etcd_process = None
+            self._nats_process = None
+            self._frontend_process = None
+            self._planner_process = None
+            self._pool = None
+            self.dp_openai_server_base_urls: list[Optional[str]] = [external_url]
+            print(
+                f"  [Dynamo] External mode — using {external_url}",
+                flush=True,
+            )
+            return
+
+        self._external = False
+        vllm_cfg = config["vllm_cfg"]
+
         self.tp_size = vllm_cfg["tensor_parallel_size"]
         assert cluster.world_size() % self.tp_size == 0, (
             f"World size {cluster.world_size()} must be divisible by "
@@ -485,6 +503,8 @@ def _handle_scaling_event(self, event: Any) -> None:
 
     def _check_subprocesses(self) -> None:
         """Raise if any managed subprocess has exited unexpectedly."""
+        if self._external:
+            return
         for attr, name, _ in _SUBPROCESS_REGISTRY:
             proc: Optional[subprocess.Popen] = getattr(self, attr)
             if proc is not None:
@@ -502,6 +522,9 @@ def finish_generation(self, *args: Any, **kwargs: Any) -> bool:
         return True
 
     def shutdown(self) -> bool:
+        if self._external:
+            return True
+
         self._stop_virtual_connector_client()
 
         for attr, name, timeout in _SUBPROCESS_REGISTRY: