nemo_gym.rollout_collection

Module Contents

Classes

Name	Description
`E2ERolloutCollectionConfig`	Spin up all necessary servers and perform a batch of rollout collection using each dataset inside the provided configs.
`RolloutAggregationConfig`	Aggregate metrics across rollout shards produced by `gym eval run --no-serve +disable_aggregation=true`.
`RolloutAggregationHelper`	-
`RolloutCollectionConfig`	Perform a batch of rollout collection.
`RolloutCollectionHelper`	-
`SharedRolloutCollectionConfig`	-
`_CompletedRollout`	A finished `/run` dispatch, with timing carried alongside (not inside) the raw result.

Functions

Name	Description
`_agent_request_failure_row`	One sidecar row for a `/run` call that came back without a result.
`_attach_ng_perf`	-
`_attach_trajectory_record`	-
`_build_ng_perf`	Assemble the per-rollout `ng_perf` summary from `ng_trajectory`.
`_build_trajectory_record`	-
`_coverage_report`	State how much of the input the score covers, for the runs where it is not all of it.
`_expand_input_glob`	Expand a glob-or-comma-separated-globs string into a sorted, deduplicated list of paths.
`_failure_rows_counted_as_zero`	Sidecar rows the caller opted to count in the metrics denominator.
`_get_max_rollout_attempts`	Read `NEMO_GYM_MAX_ROLLOUT_ATTEMPTS` (positive int) or default to 3.
`_has_observation_gap`	-
`_latest_failure_rows`	The last attempt recorded for each rollout across the failures sidecars.
`_masking_step_metrics`	In-progress view of what a run is losing to its environment rather than its policy.
`_nonnegative_int`	-
`_normalize_health_check_ignored_checks`	-
`_rollout_for_export`	Return an exporter view without the complete trajectory or raw capture payloads.
`_rollout_request_debug_summary`	-
`_strip_capture_payloads`	-
`_trajectory_identity`	-
`_truncated_body`	Decode at most the kept prefix, so a huge error page is never decoded in full.
`loads_jsonl_line`	Parse one JSONL line, raising a clean `ConfigError` (naming file + line) on malformed JSON.

Data

AGENT_REQUEST_FAILED_FAILURE_CLASS

AGENT_RUN_ERROR_FAILURE_CLASS

NG_FAILURE_CLASS_KEY

_AGENT_DID_NOT_RUN_STATUSES

_DEFAULT_MAX_ROLLOUT_ATTEMPTS

_MAX_FAILURE_BODY_CHARS

_MODEL_CALL_PAYLOAD_KEYS

_NO_RESULT_FAILURE_CLASSES

_RUN_FAILURE_ERRORS

__getattr__

_failures_path_for

logger

API

class nemo_gym.rollout_collection.E2ERolloutCollectionConfig()

Bases: SharedRolloutCollectionConfig

Spin up all necessary servers and perform a batch of rollout collection using each dataset inside the provided configs.

Examples:

gym eval run         +output_jsonl_fpath=weather_rollouts.jsonl         +num_samples_in_parallel=10

reuse_existing_data_preparation

bool = False

split

Union[Literal['train'], Literal['validation'], Literal['benchmark']]

nemo_gym.rollout_collection.E2ERolloutCollectionConfig._reject_example_split(
    data
)

classmethod

nemo_gym.rollout_collection.E2ERolloutCollectionConfig._reject_input_jsonl_fpath(
    data
)

classmethod

class nemo_gym.rollout_collection.RolloutAggregationConfig()

Bases: BaseNeMoGymCLIConfig

Aggregate metrics across rollout shards produced by gym eval run --no-serve +disable_aggregation=true.

Reads every JSONL file matching input_glob, computes aggregate metrics by POSTing to each agent server’s /aggregate_metrics endpoint over the global union of records, and writes a single <output_jsonl_fpath stem>_aggregate_metrics.json next to the rollouts. By default also concatenates all shards into output_jsonl_fpath.

Examples:

gym eval aggregate         "+config_paths=[benchmarks/aime24/config.yaml,responses_api_models/vllm_model/configs/vllm_model.yaml]"         +input_glob='results/rollouts-rs*-chunk*.jsonl'         +output_jsonl_fpath=results/rollouts.jsonl

count_failure_classes_as_zero

List[str]

disable_health_check

bool

health_check_ignored_checks

List[str]

health_check_workers

Optional[int]

input_glob

str

merge_shards

bool

output_jsonl_fpath

str

nemo_gym.rollout_collection.RolloutAggregationConfig._validate_health_check_ignored_checks(
    value
)

classmethod

class nemo_gym.rollout_collection.RolloutAggregationHelper()

Bases: BaseModel

nemo_gym.rollout_collection.RolloutAggregationHelper.run_from_config(
    config: nemo_gym.rollout_collection.RolloutAggregationConfig
) -> typing.Optional[pathlib.Path]

async

class nemo_gym.rollout_collection.RolloutCollectionConfig()

Bases: SharedRolloutCollectionConfig

Perform a batch of rollout collection.

Examples:

gym eval run --no-serve         +agent_name=example_single_tool_call_simple_agent         +input_jsonl_fpath=weather_query.jsonl         +output_jsonl_fpath=weather_rollouts.jsonl         +limit=100         +num_repeats=4         +num_samples_in_parallel=10

agent_map

Optional[Dict[str, str]]

agent_name

Optional[str]

fan_out

Optional[Dict[str, List[str]]]

input_jsonl_fpath

str

limit

Optional[int]

materialized_jsonl_fpath

Path

num_repeats

Union[int, Dict[str, int]]

num_repeats_add_seed

bool

prompt_config

Optional[str]

resume_from_cache

bool

skills

Optional[SkillsConfig]

nemo_gym.rollout_collection.RolloutCollectionConfig._coerce_null_num_repeats(
    v
)

classmethod

nemo_gym.rollout_collection.RolloutCollectionConfig._fold_agent_name_into_agent_map() -> nemo_gym.rollout_collection.RolloutCollectionConfig

nemo_gym.rollout_collection.RolloutCollectionConfig._validate_fan_out() -> nemo_gym.rollout_collection.RolloutCollectionConfig

nemo_gym.rollout_collection.RolloutCollectionConfig._validate_num_repeats() -> nemo_gym.rollout_collection.RolloutCollectionConfig

class nemo_gym.rollout_collection.RolloutCollectionHelper()

Bases: BaseModel

nemo_gym.rollout_collection.RolloutCollectionHelper._call_aggregate_metrics(
    results: typing.List[typing.Dict],
    rows: typing.List[typing.Dict],
    output_fpath: pathlib.Path
) -> typing.Optional[pathlib.Path]

async

Call /aggregate_metrics on each agent server after rollouts complete.

Writes a single _aggregate_metrics.json with one entry per agent (same shape as the old _agent_metrics.json). Returns the file path.

nemo_gym.rollout_collection.RolloutCollectionHelper._load_from_cache(
    config: nemo_gym.rollout_collection.RolloutCollectionConfig
) -> typing.Tuple[typing.List[typing.Dict], typing.List[typing.Dict], typing.List[typing.Dict], typing.List[typing.List[str]]]

nemo_gym.rollout_collection.RolloutCollectionHelper._preprocess_raw_rows(
    raw_rows: typing.List[typing.Tuple[int, str, typing.Dict]],
    config: nemo_gym.rollout_collection.RolloutCollectionConfig
) -> typing.List[typing.Dict]

staticmethod

nemo_gym.rollout_collection.RolloutCollectionHelper._preprocess_rows_from_config(
    config: nemo_gym.rollout_collection.RolloutCollectionConfig
) -> typing.List[typing.Dict]

nemo_gym.rollout_collection.RolloutCollectionHelper._run_examples_with_metadata(
    examples: typing.List[typing.Dict],
    head_server_config: typing.Optional[nemo_gym.config_types.BaseServerConfig] = None,
    semaphore: typing.Optional[asyncio.Semaphore] = None,
    route_failures_to_sidecar: bool = False
) -> typing.Iterator[asyncio.Future]

Internal dispatch shared by run_examples and Gym’s own collection paths.

Identical contract to run_examples, but each future resolves to a _CompletedRollout that carries rollout_latency_ms alongside the raw /run result instead of inside it, so internal-only timing never has to be smuggled through (and stripped back out of) a dict that a direct caller of run_examples could also observe.

nemo_gym.rollout_collection.RolloutCollectionHelper._run_from_config(
    config: nemo_gym.rollout_collection.RolloutCollectionConfig
) -> typing.Tuple[typing.List[typing.Dict]]

async

nemo_gym.rollout_collection.RolloutCollectionHelper._validate_agent_names(
    examples: typing.List[typing.Dict],
    global_config_dict: omegaconf.DictConfig
) -> None

staticmethod

Fail before any dispatch when a row names an agent absent from the running config.

Without this, the first bad row dies mid-collection with a raw omegaconf ConfigKeyError after valid rows have already been dispatched.

nemo_gym.rollout_collection.RolloutCollectionHelper._validate_agent_pairings(
    examples: typing.List[typing.Dict],
    global_config_dict: omegaconf.DictConfig
) -> None

staticmethod

Fail before dispatch when a row points to agent incompatible with the resources server it runs on.

nemo_gym.rollout_collection.RolloutCollectionHelper.preprocess_examples(
    examples: typing.List[typing.Dict],
    agent_map: typing.Optional[typing.Dict[str, str]] = None,
    fan_out: typing.Optional[typing.Dict[str, typing.List[str]]] = None,
    num_repeats: typing.Union[int, typing.Dict[str, int]] = 1,
    num_repeats_add_seed: bool = False,
    global_config_dict: typing.Optional[omegaconf.DictConfig] = None
) -> typing.List[typing.Dict]

Apply run-level routing and repetition to caller-held rows.

Public entry point for direct run_examples callers (e.g. trainer integrations that drive dispatch themselves): run_examples resolves task_sources and validates agent names, but agent_map, fan_out and num_repeats are applied only during preprocessing. Call this first, then pass the returned rows to run_examples.

Pass global_config_dict (the merged config) to also resolve task_source-only rows to their agents here; leave it None to defer that to run_examples, which does it against the head server’s config. Input rows are not mutated; the expanded, stamped copies are returned.

nemo_gym.rollout_collection.RolloutCollectionHelper.resolve_task_sources(
    examples: typing.List[typing.Dict],
    global_config_dict: omegaconf.DictConfig
) -> None

staticmethod

Stamp an agent_ref onto every row that carries only a task_source.

task_source names the config instance that declared the row’s dataset. Resolution is :func:~nemo_gym.global_config.resolve_dataset_agent — the same rules benchmark discovery uses, so dispatch can never disagree with the listing. Conflicting agent: pins across one instance’s datasets are a hard error (rows carry only the instance name), as are unknown/non-routable instances; +agent_map is the disambiguator.

Rows that already have an agent_ref are left untouched, so this is a no-op on legacy datasets and on already-resolved (materialized) rows. Runs before any dispatch.

nemo_gym.rollout_collection.RolloutCollectionHelper.run_examples(
    examples: typing.List[typing.Dict],
    head_server_config: typing.Optional[nemo_gym.config_types.BaseServerConfig] = None,
    semaphore: typing.Optional[asyncio.Semaphore] = None,
    route_failures_to_sidecar: bool = False
) -> typing.Iterator[asyncio.Future]

We provide this function as a lower level interface for running rollout collection.

Rows are dispatched as given: task_sources are resolved and agent names validated here, but run-level knobs (agent_map, fan_out, num_repeats) are NOT applied — call preprocess_examples first if you need them.

route_failures_to_sidecar makes a failed /run a failure row instead of an exception that ends every rollout still in flight. It defaults off because those rollouts then leave the score.

Every future resolves to exactly the (row, result) pair Gym’s own /run endpoint returned — no Gym-private fields are ever added to result.

nemo_gym.rollout_collection.RolloutCollectionHelper.run_from_config(
    config: nemo_gym.rollout_collection.RolloutCollectionConfig
) -> typing.Tuple[typing.List[typing.Dict]]

async

Collect rollouts for a whole config. Wrapped in the run-scoped job span.

This is the driver side of an evaluation run and the outermost span Gym produces, so every rollout it dispatches is a descendant of it. job is in the default preset but deliberately not in per_rollout, where each rollout is meant to be its own bounded root trace.

nemo_gym.rollout_collection.RolloutCollectionHelper.setup_server_client(
    head_server_config: typing.Optional[nemo_gym.config_types.BaseServerConfig] = None
) -> nemo_gym.server_utils.ServerClient

class nemo_gym.rollout_collection.SharedRolloutCollectionConfig()

Bases: UploadRolloutsConfigMixin, BaseNeMoGymCLIConfig

count_failure_classes_as_zero

List[str]

disable_aggregation

bool

disable_health_check

bool

health_check_ignored_checks

List[str]

health_check_workers

Optional[int]

num_samples_in_parallel

Optional[int]

output_jsonl_fpath

str

responses_create_params

Dict[str, Any]

rollout_collection_driver

Optional[str]

route_failures_to_sidecar

bool

nemo_gym.rollout_collection.SharedRolloutCollectionConfig._validate_health_check_ignored_checks(
    value
)

classmethod

class nemo_gym.rollout_collection._CompletedRollout(
    row: typing.Dict[str, typing.Any],
    result: typing.Dict[str, typing.Any],
    rollout_latency_ms: typing.Optional[float]
)

Dataclass

A finished /run dispatch, with timing carried alongside (not inside) the raw result.

result

Dict[str, Any]

rollout_latency_ms

Optional[float]

row

Dict[str, Any]

nemo_gym.rollout_collection._agent_request_failure_row(
    exc: BaseException,
    status: typing.Optional[int]
) -> typing.Dict[str, typing.Any]

One sidecar row for a /run call that came back without a result.

No reward and no response: an infrastructure failure is not a verifier score of zero, and a placeholder would read as real generation data to token capture, aggregation and trainers. The class says whether the rollout ran. A NeMo Gym agent answers 500 when its own handler raises, so any status it answered with means the agent ran and broke, which is also how a model server rejecting the model’s own output arrives here. A gateway status, or no reply to take a status from, says nothing about the rollout. Neither class carries a reward; an evaluation that wants the first counted names it in count_failure_classes_as_zero.

nemo_gym.rollout_collection._attach_ng_perf(
    result: dict[str, typing.Any],
    observability_enabled: bool,
    rollout_latency_ms: typing.Optional[float] = None
) -> None

nemo_gym.rollout_collection._attach_trajectory_record(
    row: dict[str, typing.Any],
    result: dict[str, typing.Any]
) -> None

nemo_gym.rollout_collection._build_ng_perf(
    result: dict[str, typing.Any],
    rollout_latency_ms: typing.Optional[float]
) -> typing.Optional[dict[str, typing.Any]]

Assemble the per-rollout ng_perf summary from ng_trajectory.

Returns None (ng_perf stays absent) unless at least one reasoning turn was observed: per-turn evidence is needed rather than just raw model-call capture, so a rollout collected with observability disabled produces no ng_perf at all.

Token fields are summed over every model call referenced by a reasoning-turn AgentInvocation. This includes compaction calls whenever the harness also lists them in AgentInvocation.model_calls.

num_turns counts reasoning turns summed across all invocations (an AgentInvocation is one root-agent or subagent conversation that may span many turns). Each invocation contributes its explicit TrajectoryTurn count when the harness emits turn records, falling back to its owned model-call count (one assistant response per turn), then to 1 (an invocation that ran had at least one turn) — so hybrid trajectories where only some invocations report turns still count every conversation.

token_observability_coverage reports what fraction of those turns actually resolved to a captured call: a turn whose ModelCallRef was unmatched or ambiguous silently loses its tokens from the sums below, and this is the only signal that it happened.

nemo_gym.rollout_collection._build_trajectory_record(
    row: dict[str, typing.Any],
    result: dict[str, typing.Any]
) -> nemo_gym.rollout_observability.TrajectoryRecord

nemo_gym.rollout_collection._coverage_report(
    expected: int,
    scored: int,
    failure_counts: collections.Counter,
    failures_hint: typing.Any
) -> str

State how much of the input the score covers, for the runs where it is not all of it.

Silence here is what makes a partial run look complete, so this reports against the materialized input rather than the rollouts one hop happened to dispatch.

nemo_gym.rollout_collection._expand_input_glob(
    input_glob: str
) -> typing.List[str]

Expand a glob-or-comma-separated-globs string into a sorted, deduplicated list of paths.

Examples:

'results/rollouts.jsonl' -> ['results/rollouts.jsonl'] (if it exists)
'a/*.jsonl, b/*.jsonl'   -> matches of both patterns, deduplicated

nemo_gym.rollout_collection._failure_rows_counted_as_zero(
    failures_fpaths: typing.List[pathlib.Path],
    failure_classes: typing.List[str],
    scored_keys: set
) -> typing.List[typing.Dict[str, typing.Any]]

Sidecar rows the caller opted to count in the metrics denominator.

The last attempt of a rollout is the one that stands, so it is selected across every failure class before the wanted classes are picked out. Selecting the other way round would let a stale attempt be counted after a later one landed in a class the caller did not ask for.

A row that already carries a reward is counted as it stands. A row that carries none records that no rollout happened, so it is counted as a zero here and only here: the score enters the metric input, never the sidecar or the rollouts jsonl, which keeps the artifacts free of a verdict no verifier gave. A rollout that also succeeded is never counted.

nemo_gym.rollout_collection._get_max_rollout_attempts() -> int

Read NEMO_GYM_MAX_ROLLOUT_ATTEMPTS (positive int) or default to 3.

nemo_gym.rollout_collection._has_observation_gap(
    result: dict[str, typing.Any],
    code: str
) -> bool

nemo_gym.rollout_collection._latest_failure_rows(
    failures_fpaths: typing.List[pathlib.Path]
) -> typing.Dict[typing.Tuple[typing.Any, typing.Any], typing.Dict[str, typing.Any]]

The last attempt recorded for each rollout across the failures sidecars.

nemo_gym.rollout_collection._masking_step_metrics(
    agent_name: str,
    scored: collections.Counter,
    dropped: collections.Counter
) -> typing.Dict[str, float]

In-progress view of what a run is losing to its environment rather than its policy.

scored covers persisted rollouts only, split into the unmasked ones (count, reward) and the masked ones; dropped counts what never reached the main output at all. reward_unmasked averages over the unmasked rollouts alone, so the gap against the existing reward series is the score lost to infrastructure. Failed and omitted attempts are reported as counts, never folded into a quality average.

Empty until something is actually masked or dropped, so a healthy run exports exactly what it exported before. The final numbers come from /aggregate_metrics; this is the progress view while the run is still going.

nemo_gym.rollout_collection._nonnegative_int(
    value: typing.Any
) -> typing.Optional[int]

nemo_gym.rollout_collection._normalize_health_check_ignored_checks(
    value
) -> typing.List[str]

nemo_gym.rollout_collection._rollout_for_export(
    result: dict[str, typing.Any]
) -> dict[str, typing.Any]

Return an exporter view without the complete trajectory or raw capture payloads.

nemo_gym.rollout_collection._rollout_request_debug_summary(
    row: typing.Dict[str, typing.Any]
) -> typing.Dict[str, typing.Any]

nemo_gym.rollout_collection._strip_capture_payloads(
    result: dict[str, typing.Any]
) -> None

nemo_gym.rollout_collection._trajectory_identity(
    row: dict[str, typing.Any]
) -> tuple[str, str]

nemo_gym.rollout_collection._truncated_body(
    body: typing.Optional[bytes]
) -> typing.Optional[str]

Decode at most the kept prefix, so a huge error page is never decoded in full.

nemo_gym.rollout_collection.loads_jsonl_line(
    raw,
    fpath,
    line_no: int
)

Parse one JSONL line, raising a clean ConfigError (naming file + line) on malformed JSON.

nemo_gym.rollout_collection.AGENT_REQUEST_FAILED_FAILURE_CLASS = 'agent_request_failed'

nemo_gym.rollout_collection.AGENT_RUN_ERROR_FAILURE_CLASS = 'agent_run_error'

nemo_gym.rollout_collection.NG_FAILURE_CLASS_KEY = '_ng_failure_class'

nemo_gym.rollout_collection.NG_NO_PERSIST_KEY = '_ng_no_persist'

nemo_gym.rollout_collection.NG_PERF_KEY = 'ng_perf'

nemo_gym.rollout_collection.NG_TERMINAL_KEY = '_ng_failure_terminal'

nemo_gym.rollout_collection.NG_TRAJECTORY_KEY = 'ng_trajectory'

nemo_gym.rollout_collection._AGENT_DID_NOT_RUN_STATUSES = frozenset({429, 502, 503, 504})

nemo_gym.rollout_collection._DEFAULT_MAX_ROLLOUT_ATTEMPTS = 3

nemo_gym.rollout_collection._MAX_FAILURE_BODY_CHARS = 2000

nemo_gym.rollout_collection._MODEL_CALL_PAYLOAD_KEYS = ('request', 'response', 'request_raw', 'response_raw')

nemo_gym.rollout_collection._NO_RESULT_FAILURE_CLASSES = frozenset({AGENT_REQUEST_FAILED_FAILURE_CLASS, AGENT_RUN_ERROR_FAILURE_CLASS})

nemo_gym.rollout_collection._RUN_FAILURE_ERRORS = (ClientError, orjson.JSONDecodeError, TimeoutError)

nemo_gym.rollout_collection.__getattr__ = moved_attr_getter(__name__, {'collect_rollouts': 'nemo_gym.cli.eval', 'aggregate...

nemo_gym.rollout_collection._failures_path_for = failures_path_for

nemo_gym.rollout_collection.logger = logging.getLogger(__name__)