`nemo_rl.models.generation.dynamo.dynamo_generation`#

Generation and NCCL refit through a driver-owned Dynamo vLLM fleet.

Module Contents#

Classes#

DynamoGeneration

Own a fixed Dynamo service fleet and expose it for NeMo-RL rollouts.

Functions#

`_is_retryable_http_response`	Return whether an internal HTTP error shape represents a transient error.
`_parse_dynamo_completion_response`	Parse the Dynamo OpenAI completion response for direct generation.

Data#

`LOGGER`
`_HTTP_MAX_ATTEMPTS`
`_HTTP_RETRY_DELAY_S`
`_RETRYABLE_HTTP_STATUS_CODES`

API#

nemo_rl.models.generation.dynamo.dynamo_generation.LOGGER#: ‘getLogger(…)’

nemo_rl.models.generation.dynamo.dynamo_generation._HTTP_MAX_ATTEMPTS#: 3

nemo_rl.models.generation.dynamo.dynamo_generation._HTTP_RETRY_DELAY_S#: 1.0

nemo_rl.models.generation.dynamo.dynamo_generation._RETRYABLE_HTTP_STATUS_CODES#: None

nemo_rl.models.generation.dynamo.dynamo_generation._is_retryable_http_response(response: Any) → bool#: Return whether an internal HTTP error shape represents a transient error.

nemo_rl.models.generation.dynamo.dynamo_generation._parse_dynamo_completion_response( response: dict[str, Any], *, request_url: str, ) → tuple[list[int], list[float], bool]#: Parse the Dynamo OpenAI completion response for direct generation.

class nemo_rl.models.generation.dynamo.dynamo_generation.DynamoGeneration( cluster: Optional[nemo_rl.distributed.virtual_cluster.RayVirtualCluster], config: dict[str, Any], tokenizer: Any | None = None, tokenizer_config: Optional[dict[str, Any]] = None, )#

Bases: nemo_rl.models.generation.interfaces.GenerationInterface

Own a fixed Dynamo service fleet and expose it for NeMo-RL rollouts.

Initialization

prepare_for_generation(

*args: Any,

**kwargs: Any,

) → bool#

property frontend_url: str#: Return the internal managed Dynamo OpenAI frontend URL.

finish_generation(*args: Any, **kwargs: Any) → bool#: Invalidate cached rollout state after synchronous generation.

get_logger_metrics() → dict[str, Any]#: Return per-worker Dynamo metric timelines for generation logging.

clear_logger_metrics() → None#: Clear the Dynamo metric timelines for the next logging window.

get_inference_world_size() → int#: Return the number of vLLM ranks across all discovered workers.

get_collective_sender_spec() → nemo_rl.models.generation.interfaces.CollectiveSenderSpec#: Return vLLM’s NCCL protocol and packed-transfer geometry.

shutdown() → bool#: Stop process-local helpers and any driver-owned managed runtime.

__getstate__() → dict[str, Any]#

Serialize only HTTP clients needed by Ray rollout actors.

Driver-owned subprocesses, threads, and Ray worker handles are excluded. The endpoint-only refit channel is retained so AREAL-style cache invalidation still reaches every managed worker after deserialization.

__setstate__(state: dict[str, Any]) → None#: Restore a client-only rollout copy with no service ownership.

_completion_url() → str#

_request_timeout_s() → float#

_merge_stop_strings( batch_stop_strings: Any, ) → Optional[list[str]]#

_prompt_token_ids( data: nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationDatumSpec], sample_idx: int, ) → list[int]#

_build_completion_request( *, prompt_token_ids: list[int], greedy: bool, stop_strings: Optional[list[str]], max_new_tokens: int, ) → dict[str, Any]#

_allowed_new_tokens(input_length: int) → int#: Return the generation budget for a prompt.

_assert_response_within_context( *, input_length: int, generated_length: int, ) → None#

async _post_completion_request( *, prompt_token_ids: list[int], greedy: bool, stop_strings: Optional[list[str]], max_new_tokens: int, ) → tuple[list[int], list[float], bool]#

_single_sample_output( *, input_ids: torch.Tensor, input_length: int, generated_token_ids: list[int], generated_logprobs: list[float], truncated: bool, ) → nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationOutputSpec]#

abstractmethod generate( data: nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationDatumSpec], greedy: bool = False, ) → nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationOutputSpec]#

Reject the unused blocking interface.

Both synchronous and asynchronous GRPO trainers use generate_async for the managed HTTP frontend.

async generate_async( data: nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationDatumSpec], greedy: bool = False, ) → AsyncGenerator[tuple[int, nemo_rl.distributed.batched_data_dict.BatchedDataDict[nemo_rl.models.generation.interfaces.GenerationOutputSpec]], None]#: Generate one token-ID prompt asynchronously through the managed frontend.

init_collective( ip: str, port: int, world_size: int, *, train_world_size: int, ) → list[ray.ObjectRef]#: Initialize native vLLM NCCL transfer on every managed worker.

prepare_refit_info(state_dict_info: dict[str, Any]) → None#: Serialize checkpoint-format tensor metadata for native vLLM refit.

abstractmethod update_weights_via_ipc_zmq() → list[ray.ObjectRef]#

update_weights_from_collective() → list[ray.ObjectRef]#: Receive packed checkpoint-format weights on every Dynamo worker.

invalidate_kv_cache() → bool#: Flush every fixed Dynamo worker’s prefix/KV cache.

nemo_rl.models.generation.dynamo.dynamo_generation#