diff --git a/deps/verifiers b/deps/verifiers index d30a3f48e5..3a38df548e 160000 --- a/deps/verifiers +++ b/deps/verifiers @@ -1 +1 @@ -Subproject commit d30a3f48e5f14b06b3081b2102ec32cc3149b849 +Subproject commit 3a38df548e27ef58c3a2e7b151fc7962a64f8d8a diff --git a/docs/algorithms.md b/docs/algorithms.md index de5b3594ed..a2bcd1e7db 100644 --- a/docs/algorithms.md +++ b/docs/algorithms.md @@ -166,12 +166,12 @@ At runtime, each env's resolved config builds two objects: a `Sampler` (`prime_r | `opsd` | `OPSDAlgorithm` | `score_rollout`: demo-conditioned prefill under the live policy | | `sft` | `SFTDistillAlgorithm` | `score_group`: group-norm credit (feeds filters) | -Each class owns its hooks outright — reading one top to bottom reads the algorithm, and everything on the class is an override point. The two hooks are one scope-and-timing ladder — the wider scope is unlocked by a later barrier, so the two axes coincide. Each is handed the `Rollout` directly — the env's typed trace (`reward`, `nodes`, `num_turns`, ...) with `samples` attached, plus `assign_advantages` to write credit: +Each class owns its hooks outright — reading one top to bottom reads the algorithm, and everything on the class is an override point. The two hooks are one scope-and-timing ladder — the wider scope is unlocked by a later barrier, so the two axes coincide. Each is handed the env's own data directly — a `Rollout` (the typed trace: `reward`, `nodes`, `num_turns`, ... with `samples` attached) on arrival, the group's `Episode`s at group time — plus `assign_advantages` to write credit: -- `async score_rollout(rollout)` — one rollout, **on arrival** (as it's tokenized, before its group is complete): rollout-local credit (`rollout.assign_advantages(...)`, scalar broadcast or per-token), observation ce weights, **or** model I/O — query a reference pool (e.g. `self.teacher_pool`, connected in `setup()` via `self.connect(...)`, or the live `self.policy_pool` for opsd) and attach per-token results (e.g. teacher logprobs) with bounded concurrency. No siblings. `echo` weights observation tokens here, identifying env-provided observation nodes by their non-sampled status and source step role attribution, applying the optional user filter, and writing the `ce_weights` stream. Model I/O runs *before* the pre-batch filters, so it pays compute on rollouts that may then be filtered out. -- `score_group(group)` — the cohort, **before filtering** (filters read the streams), synchronous: group-relative credit (GRPO/MaxRL baselines). `group` is a list of `Rollout`. +- `async score_rollout(rollout)` — one rollout, **on arrival** (as it's tokenized, before its group is complete): rollout-local credit (`rollout.assign_advantages(...)`), observation ce weights, **or** model I/O — query a reference pool (e.g. `self.teacher_pool`, connected in `setup()` via `self.connect(...)`, or the live `self.policy_pool` for opsd) and attach per-token results (e.g. teacher logprobs) with bounded concurrency. No siblings. `echo` weights observation tokens here, identifying env-provided observation nodes by their non-sampled status and source step role attribution, applying the optional user filter, and writing the `ce_weights` stream. Model I/O runs *before* the pre-batch filters, so it pays compute on rollouts that may then be filtered out. +- `score_group(group)` — the cohort, **before filtering** (filters read the streams), synchronous: group-relative credit (GRPO/MaxRL baselines). `group` is a list of `TrainEpisode`, so an algorithm can compare within an episode as well as across them (`hierarchical_grpo` does); `group_rollouts(group)` flattens it for the ones that only compare across. -The pipeline drives the hooks through two non-virtual methods it never looks inside: `algorithm.finalize_rollout(rollout)` per arrival (rollout-local scoring + reference I/O) and `algorithm.finalize_group(rollouts)` per group (scoring + wire stamping; after this the records are frozen — groups die at stamping). Sample construction (interleaving) is pure pipeline — observation-token provenance is available through structural attribution (`node.sampled`, `node.is_content`) for any algorithm that trains on env-provided tokens. +The pipeline drives the hooks through two non-virtual methods it never looks inside: `algorithm.finalize_rollout(rollout)` per arrival (rollout-local scoring + reference I/O) and `algorithm.finalize_group(episodes)` per group (scoring + wire stamping; after this the records are frozen — groups die at stamping). Sample construction (interleaving) is pure pipeline — observation-token provenance is available through structural attribution (`node.sampled`, `node.is_content`) for any algorithm that trains on env-provided tokens. Class-level declarations state what the algorithm needs: which loss component its action tokens feed (`action_loss_type`). Every class is constructed with its algorithm config plus the one host-owned resource it can't rebuild — the live policy pool (`self.policy_pool`). Everything else an algorithm needs it builds from its own config in `setup()`: `opd` connects its frozen `teacher`; `opsd` builds the renderer for its demonstration hint (tokenizer is always the live policy's — self-distillation has no separate model). The pipeline only ever calls the two `finalize_*` methods — writing your own algorithm is subclassing `Algorithm` and overriding the hooks its signal needs (see [Authoring an Algorithm](#authoring-an-algorithm)). Shared math (efficiency shaping, prefill alignment) lives as plain functions in `prime_rl.orchestrator.algo.advantage`. @@ -420,26 +420,28 @@ Both of `kuhn-poker-v1`'s agents late-bind to the run's own model — shared-pol ### Authoring an Algorithm -There is no config hook that points at user code — a new credit-assignment scheme is a new named algorithm in the repo. Subclass `Algorithm`, assign credit in the scoring hook whose timing fits your signal, and register the class. The hook receives the group's `Rollout`s (each the env's typed `verifiers.Trace` — turns, tool calls, metadata in `info` — with `samples` attached) and writes credit via `assign_advantages`: +There is no config hook that points at user code — a new credit-assignment scheme is a new named algorithm in the repo. Subclass `Algorithm`, assign credit in the scoring hook whose timing fits your signal, and register the class. The hook receives the group's `TrainEpisode`s, each holding the env's typed `verifiers.Trace`s — turns, tool calls, metadata in `info` — with `samples` attached, and writes credit via `assign_advantages`: ```python # src/prime_rl/orchestrator/algo/my_algo.py import torch from prime_rl.orchestrator.algo.base import Algorithm +from prime_rl.orchestrator.types import group_rollouts class MyAlgorithm(Algorithm): async def score_group(self, group): - rewards = torch.tensor([rollout.reward for rollout in group], dtype=torch.float32) + rollouts = group_rollouts(group) + rewards = torch.tensor([rollout.reward for rollout in rollouts], dtype=torch.float32) advantages = ... # one value per rollout - for rollout, advantage in zip(group, advantages.tolist(), strict=True): + for rollout, advantage in zip(rollouts, advantages.tolist(), strict=True): rollout.assign_advantages(advantage) ``` -Add a typed `MyAlgoConfig` to `prime_rl.configs.algorithm` and its discriminated union, then register `"my_algo": MyAlgorithm` in `ALGORITHM_CLASSES`. Pick the hook by *when* your signal is ready: `score_rollout` for per-arrival credit or credit that needs a model call (it's `async`), `score_group` for group-relative credit (GRPO/MaxRL). `assign_advantages` takes a scalar (broadcast over the rollout's trainable tokens — the common case) or a full-length per-token list aligned to the concatenated sample token_ids (process rewards, step-level credit; `0.0` off-mask). +Add a typed `MyAlgoConfig` to `prime_rl.configs.algorithm` and its discriminated union, then register `"my_algo": MyAlgorithm` in `ALGORITHM_CLASSES`. Pick the hook by *when* your signal is ready: `score_rollout` for per-arrival credit or credit that needs a model call (it's `async`), `score_group` for group-relative credit (GRPO/MaxRL). `assign_advantages` takes a scalar and broadcasts it over the rollout's trainable tokens, writing it onto the graph's nodes so branches sharing a node cannot disagree about its credit. -Each per-token list must match the rollout's completion-token count exactly — validated loudly when the view writes it. Advantage-based filters and metrics derive from the streams (the zero-advantage filter checks for all-zero streams; logged distributions use per-rollout means). Signals that depend on the live policy's weights (like OPD's reverse KL) cannot be precomputed here; those are reference-scoring algorithms, evaluated in the trainer. +Advantage-based filters and metrics derive from the streams (the zero-advantage filter checks for all-zero streams; logged distributions use per-rollout means). Signals that depend on the live policy's weights (like OPD's reverse KL) cannot be precomputed here; those are reference-scoring algorithms, evaluated in the trainer. ### Reference Scoring diff --git a/pyproject.toml b/pyproject.toml index 2274aa3aa0..2b75aa872f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,7 +33,7 @@ dependencies = [ "setproctitle>=1.3.0", "uvloop>=0.21.0", "torchtitan", - "verifiers[harbor]>=0.2.2.dev76", + "verifiers[harbor]>=0.2.2.dev81", "renderers", "dion", "tilelang>=0.1.8", diff --git a/src/prime_rl/orchestrator/algo/__init__.py b/src/prime_rl/orchestrator/algo/__init__.py index 745a8ebc79..5e5f657696 100644 --- a/src/prime_rl/orchestrator/algo/__init__.py +++ b/src/prime_rl/orchestrator/algo/__init__.py @@ -17,10 +17,9 @@ ``finalize_rollout`` / ``finalize_group`` methods the pipeline drives. Advantages are per-token everywhere they are stored or shipped — there is no scalar advantage in the pipeline. An algorithm assigns credit in its scoring - hook via ``Rollout.assign_advantages``: a scalar that is *broadcast* over the - rollout's completion tokens (uniform credit, the common case), or an explicit - full-length-N per-token list aligned to the concatenated sample token_ids - (0.0 off-mask). + hook via ``TrainRollout.assign_advantages``, which broadcasts one value over + the rollout's trainable tokens and writes it onto the graph's nodes, where + the tokens themselves live. - ``routing`` — wire-field stamping: per-token component weight streams (rl / ce / ref_kl) and the per-token advantage stream. """ @@ -39,7 +38,7 @@ from prime_rl.orchestrator.algo.rae import RAEAlgorithm from prime_rl.orchestrator.algo.routing import stamp_advantages, stamp_loss_routing from prime_rl.orchestrator.algo.sft import SFTDistillAlgorithm -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.types import TrainRollout if TYPE_CHECKING: from prime_rl.configs.algorithm import AlgoConfig @@ -79,7 +78,7 @@ def build_algorithm(config: AlgoConfig, policy_pool: InferencePool) -> Algorithm "OPDAlgorithm", "OPSDAlgorithm", "RAEAlgorithm", - "Rollout", + "TrainRollout", "SFTDistillAlgorithm", "build_algorithm", "connect_frozen_pool", diff --git a/src/prime_rl/orchestrator/algo/base.py b/src/prime_rl/orchestrator/algo/base.py index b12d17df62..a7962c0327 100644 --- a/src/prime_rl/orchestrator/algo/base.py +++ b/src/prime_rl/orchestrator/algo/base.py @@ -19,7 +19,7 @@ I/O against another model — an inference pool the algorithm connected in ``setup()`` (a frozen teacher) or the live policy (opsd's self-distillation), queried with bounded concurrency. No siblings. -- ``score_group(group)`` — the cohort, on group completion, *before* filtering +- ``score_group(group)`` — the cohort of episodes, on group completion, *before* filtering (filters read the streams): group-relative credit (GRPO/MaxRL baselines). How rollouts are *produced* is not the algorithm's concern: that is the env's @@ -44,12 +44,13 @@ from prime_rl.configs.algorithm import ActionLossType, AlgoConfig, FrozenModelConfig from prime_rl.orchestrator.algo.routing import stamp_advantages, stamp_loss_routing +from prime_rl.orchestrator.types import group_rollouts from prime_rl.utils.logger import get_logger if TYPE_CHECKING: from renderers import RendererConfig - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode, TrainRollout from prime_rl.utils.client import InferencePool @@ -91,9 +92,10 @@ class Algorithm: (``action_loss_type``); - lifecycle — :meth:`setup` connects client pools to the frozen models the algorithm declares, resolving each reference via :meth:`connect`; - - the two scoring hooks, each ``async`` and given the :class:`Rollout` - directly — read the trace, write credit via - :meth:`Rollout.assign_advantages`. They are + - the two scoring hooks, each ``async`` and given the env's own data + directly — a :class:`TrainRollout` on arrival, the group's + :class:`Episode`\ s at group time — so a hook reads the trace and + writes credit via :meth:`TrainRollout.assign_advantages`. They are async so either stage may do I/O — e.g. a process-reward model or a teacher at arrival, or a judge at group time whose signal a pre-batch filter then reads; a hook that only does advantage math simply never @@ -103,9 +105,9 @@ class Algorithm: observation ce weights, or per-token results from a model the algorithm connected in :meth:`setup` (e.g. teacher reference logprobs). Default: nothing. - - :meth:`score_group` — the cohort, *before* filtering (filters read the - streams): group-relative credit. Default: nothing — rollouts keep - ``advantages=None``, so advantage-based filters skip them. + - :meth:`score_group` — the cohort of episodes, *before* filtering + (filters read the streams): group-relative credit. Default: nothing — + rollouts keep ``advantages=None``, so advantage-based filters skip them. Model I/O lives in :meth:`score_rollout`: it runs at arrival, *before* the pre-batch filters, so it pays compute on rollouts that may then be filtered @@ -137,29 +139,31 @@ async def connect(self, reference: FrozenModelConfig) -> InferencePool: self.connected_pools.append(pool) return pool - async def score_rollout(self, rollout: Rollout) -> None: + async def score_rollout(self, rollout: TrainRollout) -> None: """Arrival phase, one rollout, before its group is complete: write rollout-local credit (``rollout.assign_advantages``), observation ce weights (echo), or per-token results from a model — an inference pool connected in :meth:`setup`, or the live policy (opsd). No siblings, no group stats.""" - async def score_group(self, group: list[Rollout]) -> None: + async def score_group(self, group: list[Episode]) -> None: """Group phase, the finalized cohort, before filtering: write - group-relative credit.""" + group-relative credit. The cohort arrives as episodes, so an algorithm + can compare within one episode as well as across them; ``group_rollouts`` + flattens it for the algorithms that only compare across.""" - async def finalize_rollout(self, rollout: Rollout) -> None: + async def finalize_rollout(self, rollout: TrainRollout) -> None: """Arrival phase (non-virtual): rollout-local scoring as each rollout is tokenized.""" if rollout.samples: await self.score_rollout(rollout) - async def finalize_group(self, rollouts: list[Rollout]) -> None: + async def finalize_group(self, episodes: list[Episode]) -> None: """Group phase (non-virtual): group-relative scoring, then stamp each sample's wire fields (the advantage stream + loss routing). After this the records are frozen — groups die at stamping.""" - await self.score_group(rollouts) - for rollout in rollouts: + await self.score_group(episodes) + for rollout in group_rollouts(episodes): stamp_advantages(rollout) for sample in rollout.samples: stamp_loss_routing(sample, self.action_loss_type) diff --git a/src/prime_rl/orchestrator/algo/echo.py b/src/prime_rl/orchestrator/algo/echo.py index d4ecf74fa3..c7c5e15dda 100644 --- a/src/prime_rl/orchestrator/algo/echo.py +++ b/src/prime_rl/orchestrator/algo/echo.py @@ -11,7 +11,7 @@ if TYPE_CHECKING: import verifiers.v1 as vf - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import TrainRollout from prime_rl.utils.client import InferencePool @@ -34,12 +34,12 @@ def __init__(self, config: EchoAlgoConfig, policy_pool: InferencePool): if config.filter is not None: self.filter_fn = partial(import_object(config.filter.import_path), **config.filter.kwargs) - async def score_rollout(self, rollout: Rollout) -> None: + async def score_rollout(self, rollout: TrainRollout) -> None: # Observation weighting is rollout-local; the group-relative GRPO # baseline is inherited unchanged as ``score_group``. self._weight_observations(rollout) - def _weight_observations(self, rollout: Rollout) -> None: + def _weight_observations(self, rollout: TrainRollout) -> None: """Write each sample's ``ce_weights`` stream over the env-provided observation tokens of later turns. Provenance is structural under v1: within a branch, the non-sampled nodes that follow the first model diff --git a/src/prime_rl/orchestrator/algo/grpo.py b/src/prime_rl/orchestrator/algo/grpo.py index 0b87fa7cee..1ca2024320 100644 --- a/src/prime_rl/orchestrator/algo/grpo.py +++ b/src/prime_rl/orchestrator/algo/grpo.py @@ -6,9 +6,10 @@ from prime_rl.configs.algorithm import GRPOAlgoConfig from prime_rl.orchestrator.algo.base import Algorithm +from prime_rl.orchestrator.types import group_rollouts if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode from prime_rl.utils.client import InferencePool @@ -21,15 +22,16 @@ def __init__(self, config: GRPOAlgoConfig, policy_pool: InferencePool): super().__init__(config, policy_pool) self.length_penalty = config.length_penalty - async def score_group(self, group: list[Rollout]) -> None: - rewards = torch.tensor([rollout.reward for rollout in group], dtype=torch.float32) + async def score_group(self, group: list[Episode]) -> None: + rollouts = group_rollouts(group) + rewards = torch.tensor([rollout.reward for rollout in rollouts], dtype=torch.float32) length_penalty = self.length_penalty if length_penalty is None: advantages = rewards - rewards.mean() else: - output = torch.tensor([rollout.num_output_tokens for rollout in group], dtype=rewards.dtype) - total = torch.tensor([rollout.num_total_tokens for rollout in group], dtype=rewards.dtype) - turns = torch.tensor([rollout.num_turns for rollout in group], dtype=rewards.dtype) + output = torch.tensor([rollout.num_output_tokens for rollout in rollouts], dtype=rewards.dtype) + total = torch.tensor([rollout.num_total_tokens for rollout in rollouts], dtype=rewards.dtype) + turns = torch.tensor([rollout.num_turns for rollout in rollouts], dtype=rewards.dtype) input = total - output penalty_frac = ( length_penalty.num_output_tokens_weight * (output / output.max().clamp(min=1)) @@ -39,5 +41,5 @@ async def score_group(self, group: list[Rollout]) -> None: penalty = rewards.mean() * penalty_frac shaped_rewards = rewards - penalty advantages = shaped_rewards - shaped_rewards.mean() - for rollout, advantage in zip(group, advantages.tolist(), strict=True): + for rollout, advantage in zip(rollouts, advantages.tolist(), strict=True): rollout.assign_advantages(advantage) diff --git a/src/prime_rl/orchestrator/algo/hierarchical_grpo.py b/src/prime_rl/orchestrator/algo/hierarchical_grpo.py index 2945e341de..f9a2b6eb54 100644 --- a/src/prime_rl/orchestrator/algo/hierarchical_grpo.py +++ b/src/prime_rl/orchestrator/algo/hierarchical_grpo.py @@ -5,9 +5,10 @@ from prime_rl.configs.algorithm import HierarchicalGRPOAlgoConfig from prime_rl.orchestrator.algo.base import Algorithm +from prime_rl.orchestrator.types import rollouts_of if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode, TrainRollout from prime_rl.utils.client import InferencePool @@ -21,19 +22,19 @@ class HierarchicalGRPOAlgorithm(Algorithm): interchangeable. ``episode_agents`` lists the roles, normally ``solver``, that are compared - within one episode. Other roles are compared across the full rollout group. - A comparison group with one trace produces zero advantage.""" + within one episode. Other roles are compared across the whole group. A + comparison group with one trace produces zero advantage.""" def __init__(self, config: HierarchicalGRPOAlgoConfig, policy_pool: InferencePool): super().__init__(config, policy_pool) self.episode_agents = set(config.episode_agents) - async def score_group(self, group: list[Rollout]) -> None: - peers: dict[tuple[str, str | None], list[Rollout]] = defaultdict(list) - for rollout in group: - episode_scoped = rollout.agent.name in self.episode_agents - key = (rollout.agent.name, rollout.episode_id if episode_scoped else None) - peers[key].append(rollout) + async def score_group(self, group: list[Episode]) -> None: + peers: dict[tuple[str, str | None], list[TrainRollout]] = defaultdict(list) + for episode in group: + for rollout in rollouts_of(episode): + episode_scoped = rollout.agent.name in self.episode_agents + peers[(rollout.agent.name, episode.id if episode_scoped else None)].append(rollout) for members in peers.values(): baseline = sum(rollout.reward for rollout in members) / len(members) for rollout in members: diff --git a/src/prime_rl/orchestrator/algo/max_rl.py b/src/prime_rl/orchestrator/algo/max_rl.py index 9a3978108d..f013b24df2 100644 --- a/src/prime_rl/orchestrator/algo/max_rl.py +++ b/src/prime_rl/orchestrator/algo/max_rl.py @@ -5,9 +5,10 @@ import torch from prime_rl.orchestrator.algo.base import Algorithm +from prime_rl.orchestrator.types import group_rollouts if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode class MaxRLAlgorithm(Algorithm): @@ -23,9 +24,10 @@ class MaxRLAlgorithm(Algorithm): <= 0 carries no signal and gets zero advantages (the zero-advantage filter drops it, matching the paper's no-success convention).""" - async def score_group(self, group: list[Rollout]) -> None: - rewards = torch.tensor([rollout.reward for rollout in group], dtype=torch.float32) + async def score_group(self, group: list[Episode]) -> None: + rollouts = group_rollouts(group) + rewards = torch.tensor([rollout.reward for rollout in rollouts], dtype=torch.float32) mean = rewards.mean() advantages = torch.zeros_like(rewards) if mean <= 0 else (rewards - mean) / mean - for rollout, advantage in zip(group, advantages.tolist(), strict=True): + for rollout, advantage in zip(rollouts, advantages.tolist(), strict=True): rollout.assign_advantages(advantage) diff --git a/src/prime_rl/orchestrator/algo/opd.py b/src/prime_rl/orchestrator/algo/opd.py index 3135f2a9b2..96ea134738 100644 --- a/src/prime_rl/orchestrator/algo/opd.py +++ b/src/prime_rl/orchestrator/algo/opd.py @@ -8,7 +8,7 @@ from prime_rl.utils.client import StaticInferencePool if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import TrainRollout from prime_rl.transport import TrainingSample from prime_rl.utils.client import InferencePool @@ -37,7 +37,7 @@ async def setup(self) -> None: raise TypeError("opd teacher must be a static endpoint — prefill scoring needs fixed endpoints") self.teacher_pool = pool - async def score_rollout(self, rollout: Rollout) -> None: + async def score_rollout(self, rollout: TrainRollout) -> None: pool = self.teacher_pool assert pool is not None, "teacher pool not connected — Algorithm.setup() must run first" diff --git a/src/prime_rl/orchestrator/algo/opsd.py b/src/prime_rl/orchestrator/algo/opsd.py index 737666bea9..888a2b7391 100644 --- a/src/prime_rl/orchestrator/algo/opsd.py +++ b/src/prime_rl/orchestrator/algo/opsd.py @@ -9,7 +9,7 @@ if TYPE_CHECKING: from renderers.base import Renderer - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import TrainRollout from prime_rl.transport import TrainingSample from prime_rl.utils.client import InferencePool @@ -50,7 +50,7 @@ async def setup(self) -> None: self.renderer = create_renderer(load_tokenizer(self.policy_pool.model_name), self.renderer_config) - def _demonstration(self, rollout: Rollout) -> str: + def _demonstration(self, rollout: TrainRollout) -> str: demonstration = rollout.info.get(self.demo_key) if demonstration is None: demonstration = getattr(rollout.task.data, self.demo_key, None) @@ -61,7 +61,7 @@ def _demonstration(self, rollout: Rollout) -> str: ) return demonstration - async def score_rollout(self, rollout: Rollout) -> None: + async def score_rollout(self, rollout: TrainRollout) -> None: pool = self.teacher_pool renderer = self.renderer assert renderer is not None, "renderer not built — Algorithm.setup() must run first" diff --git a/src/prime_rl/orchestrator/algo/rae.py b/src/prime_rl/orchestrator/algo/rae.py index 264ae48f84..33abfdc1e5 100644 --- a/src/prime_rl/orchestrator/algo/rae.py +++ b/src/prime_rl/orchestrator/algo/rae.py @@ -5,9 +5,10 @@ from prime_rl.configs.algorithm import RAEAlgoConfig from prime_rl.orchestrator.algo.base import Algorithm +from prime_rl.orchestrator.types import group_rollouts if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode from prime_rl.utils.client import InferencePool @@ -34,8 +35,8 @@ def __init__(self, config: RAEAlgoConfig, policy_pool: InferencePool): self.decay = config.decay self.baselines: dict[str, float] = defaultdict(float) - async def score_group(self, group: list[Rollout]) -> None: - for rollout in group: + async def score_group(self, group: list[Episode]) -> None: + for rollout in group_rollouts(group): baseline = self.baselines[rollout.agent.name] rollout.assign_advantages(rollout.reward - baseline) self.baselines[rollout.agent.name] = self.decay * baseline + (1.0 - self.decay) * rollout.reward diff --git a/src/prime_rl/orchestrator/algo/routing.py b/src/prime_rl/orchestrator/algo/routing.py index 0337fa696c..aaac58f132 100644 --- a/src/prime_rl/orchestrator/algo/routing.py +++ b/src/prime_rl/orchestrator/algo/routing.py @@ -14,10 +14,11 @@ from typing import TYPE_CHECKING from prime_rl.configs.algorithm import ActionLossType +from prime_rl.orchestrator.trajectories import iter_trainable_branches from prime_rl.transport import TrainingSample if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import TrainRollout def stamp_loss_routing(sample: TrainingSample, action_loss_type: ActionLossType) -> None: @@ -51,24 +52,13 @@ def stamp_loss_routing(sample: TrainingSample, action_loss_type: ActionLossType) sample.ref_kl_weights = action_weights -def stamp_advantages(rollout: Rollout) -> None: - """Stamp the rollout's per-token advantage stream onto its samples' wire - fields. The stream is full-length-N — aligned to the samples' ``token_ids`` - concatenated in order, 0.0 on non-trainable positions — and sliced across - them. Rollouts with no credit assigned (``advantages=None``, e.g. opd/opsd) - ship no advantage stream. - """ - advantages = rollout.advantages - if advantages is None: - return - total = sum(len(sample.token_ids) for sample in rollout.samples) - if len(advantages) != total: - raise ValueError( - f"advantage stream must align with the rollout's tokens: " - f"got {len(advantages)}, expected {total} (env '{rollout.env_name}')." - ) - offset = 0 - for sample in rollout.samples: - num_tokens = len(sample.token_ids) - sample.advantages = list(advantages[offset : offset + num_tokens]) - offset += num_tokens +def stamp_advantages(rollout: TrainRollout) -> None: + """Copy each trainable branch's per-token credit onto the sample built from it, zeroed where + the sample does not train. The branch spreads its nodes' values across its own tokens, so the + two align by construction, but a node shared with an earlier branch is credited there and is + only context here. A rollout that was never scored (opd/opsd) ships no advantage stream.""" + for sample, (branch, _) in zip(rollout.samples, iter_trainable_branches(rollout), strict=True): + advantages = branch.advantages + if advantages is None: + continue + sample.advantages = [a if trains else 0.0 for a, trains in zip(advantages, sample.mask, strict=True)] diff --git a/src/prime_rl/orchestrator/dispatcher.py b/src/prime_rl/orchestrator/dispatcher.py index 5fee725f6a..a65cd62f51 100644 --- a/src/prime_rl/orchestrator/dispatcher.py +++ b/src/prime_rl/orchestrator/dispatcher.py @@ -5,17 +5,18 @@ N rollouts in one call reserves N permits (each bridged v0 rollout is its own single-agent episode). - Optional rate limiting via ``AsyncLimiter(tasks_per_minute, 60)``. -- Emit-everything invariant: every dispatched episode eventually reaches - ``out_q`` exactly once, as a ``list[Rollout]``. Failures - (env error, empty trajectory, task exception, off-policy cancel) carry - ``trace.last_error`` set; sinks decide drop / partial-train policy. +- Emit-everything invariant: every dispatched episode eventually reaches ``out_q`` exactly once, + as an ``Episode``. An env-side failure (env error, empty trajectory) rides the trace that hit it + (``trace.last_error``); a failure with no trace to ride — the env produced none, the task raised, + the group was cancelled — is an episode with no traces and the reason on ``episode.errors``, + never a stand-in rollout. Sinks decide drop / partial-train policy. - ``DispatcherMode.PREFER_TRAIN`` / ``PREFER_EVAL`` controls which kind to schedule next. Transitions are level-triggered (driven by the eval source's emptiness), so in-flight rollouts of the opposite kind drain naturally on either side of an eval boundary. - ``on_version_pending`` (called by the watcher before the engines pause for - the weight update) bumps ``off_policy_steps`` on in-flight train rollouts and - drops groups past ``max_off_policy_steps``. + the weight update) drops in-flight train groups whose generating policy would + fall further behind than ``max_off_policy_steps``. Eval rollouts are measurements for the policy version they started with, so they are allowed to finish even if training advances. Train rollouts sampled from a frozen model never age — their sampler doesn't change @@ -40,10 +41,10 @@ from prime_rl.orchestrator.eval_source import EvalSource from prime_rl.orchestrator.train_source import TrainSource from prime_rl.orchestrator.types import ( + Episode, GroupState, - InflightRollout, + InflightEpisode, Policy, - Rollout, RolloutKind, ) from prime_rl.utils.async_utils import safe_cancel, safe_cancel_all @@ -131,11 +132,13 @@ def __init__( eval_source: EvalSource | None, policy_pool: InferencePool, policy: Policy, + run_id: str, max_inflight_episodes: int, tasks_per_minute: float | None, max_off_policy_steps: int, ) -> None: self.policy = policy + self.run_id = run_id self.train_envs = train_envs self.eval_envs = eval_envs # Train rollouts go to the env sampler's pool; eval always @@ -151,12 +154,12 @@ def __init__( AsyncLimiter(tasks_per_minute, time_period=60) if tasks_per_minute else None ) - self.inflight: dict[asyncio.Task, InflightRollout] = {} + self.inflight: dict[asyncio.Task, InflightEpisode] = {} self.groups: dict[uuid.UUID, GroupState] = {} # Bounded so the dispatcher backpressures on a slow sink. One entry per # episode — the sinks count episodes, never loose traces. - self.out_q: asyncio.Queue[list[Rollout]] = asyncio.Queue(maxsize=max(8, self.max_inflight)) + self.out_q: asyncio.Queue[Episode] = asyncio.Queue(maxsize=max(8, self.max_inflight)) self.mode: DispatcherMode = DispatcherMode.PREFER_TRAIN # Set by the orchestrator after the final train step; pipeline then @@ -183,11 +186,11 @@ def _train_pool_for(self, env_name: str) -> tuple[InferencePool, str, bool]: @property def inflight_train_count(self) -> int: - return sum(m.rollout_count for m in self.inflight.values() if m.kind == "train") + return sum(m.episodes_owed for m in self.inflight.values() if m.kind == "train") @property def inflight_eval_count(self) -> int: - return sum(m.rollout_count for m in self.inflight.values() if m.kind == "eval") + return sum(m.episodes_owed for m in self.inflight.values() if m.kind == "eval") @property def available_permits(self) -> int: @@ -197,7 +200,7 @@ def available_permits(self) -> int: def inflight_by_env(self) -> dict[tuple[RolloutKind, str], int]: counts: dict[tuple[RolloutKind, str], int] = defaultdict(int) for meta in self.inflight.values(): - counts[(meta.kind, meta.env_name)] += meta.rollout_count + counts[(meta.kind, meta.env_name)] += meta.episodes_owed return dict(counts) @property @@ -225,15 +228,19 @@ def disable_train_scheduling(self) -> None: triggered eval drain naturally.""" self.train_scheduling_disabled = True + def _inflight_lag(self) -> list[int]: + """How far behind the live policy each in-flight train dispatch has fallen.""" + return [self.policy.version - m.policy_version for m in self.inflight.values() if m.kind == "train"] + @property def max_off_policy_level(self) -> int: - steps = [m.off_policy_steps for m in self.inflight.values() if m.kind == "train"] - return max(steps) if steps else 0 + lag = self._inflight_lag() + return max(lag) if lag else 0 @property def mean_off_policy_level(self) -> float: - steps = [m.off_policy_steps for m in self.inflight.values() if m.kind == "train"] - return sum(steps) / len(steps) if steps else 0.0 + lag = self._inflight_lag() + return sum(lag) / len(lag) if lag else 0.0 # ── lifecycle ────────────────────────────────────────────────────────── @@ -269,8 +276,13 @@ async def stop(self) -> None: await safe_cancel(self.task) self.task = None + def samples_from_live_policy(self, meta: InflightEpisode) -> bool: + """Whether this dispatch follows the live policy at all. A frozen sampler does not, so it + never ages and has no version to be behind.""" + return meta.kind == "eval" or self.train_envs.get(meta.env_name).sampler.samples_from_live_policy + async def on_version_pending(self, step: int) -> None: - """Bump off-policy counters and drop groups past + """Drop groups whose generating policy is now further behind than ``max_off_policy_steps`` (drop_group emits ``Cancelled`` markers so the sink still finalizes the partial group). Eval rollouts are not aged because they are tied to their start-time policy version. @@ -286,10 +298,11 @@ async def on_version_pending(self, step: int) -> None: continue # Frozen-sourced rollouts never go stale — their sampler doesn't # change with policy updates. - if not self.train_envs.get(meta.env_name).sampler.samples_from_live_policy: + if not self.samples_from_live_policy(meta): continue - meta.off_policy_steps += 1 - if meta.off_policy_steps > self.max_off_policy_steps: + # The live version is about to become ``step``'s, so this dispatch will be one + # further behind than it is now. + if (self.policy.version + 1) - meta.policy_version > self.max_off_policy_steps: stale_groups.add(meta.group_id) for gid in stale_groups: @@ -472,12 +485,12 @@ async def schedule_group_rollout(self, group_id: uuid.UUID, group: GroupState) - ) ) - self.inflight[task] = InflightRollout( + self.inflight[task] = InflightEpisode( kind=group.kind, env_name=group.env_name, group_id=group_id, policy_version=group.policy_version_at_start, - rollout_count=permits, + episodes_owed=permits, client_config=client, eval_step=group.eval_step, ) @@ -496,8 +509,8 @@ def release(self, n: int) -> None: async def handle_completed_rollout(self, task: asyncio.Task) -> None: """Emit every dispatched episode exactly once to ``out_q``: a ``run`` - result as one episode, a legacy ``run_group`` result as ``rollout_count`` - single-trace episodes. Task exceptions synthesize ``rollout_count`` + result as one episode, a legacy ``run_group`` result as ``episodes_owed`` + single-trace episodes. Task exceptions synthesize ``episodes_owed`` error-marker episodes so the sink's count-to-``group_size`` finalization still triggers. Cancelled tasks (popped by ``drop_group``) raise ``CancelledError`` and are discarded — ``drop_group`` already emitted @@ -506,32 +519,20 @@ async def handle_completed_rollout(self, task: asyncio.Task) -> None: meta = self.inflight.pop(task, None) if meta is None: return # already handled by drop_group / cancel_inflight_rollouts - self.release(meta.rollout_count) + self.release(meta.episodes_owed) group = self.groups.get(meta.group_id) - is_synth_exception = False try: result = task.result() - rollouts: list[Rollout] = result if isinstance(result, list) else [result] - if not rollouts: - raise RuntimeError("env run returned an empty episode (no traces)") + episodes: list[vf.WireEpisode] = result if isinstance(result, list) else [result] except asyncio.CancelledError: return except Exception as exc: get_logger().warning(f"Rollout task failed in group {meta.group_id} ({meta.env_name}): {exc!r}") - task_idx = group.task_idx if group is not None else -1 - rollouts = [ - Rollout( - task=vf.TraceTask(type="Task", data=vf.TaskData(idx=task_idx, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), - ) - for _ in range(meta.rollout_count) - ] - for r in rollouts: - r.record_error(exc) - is_synth_exception = True + await self.emit_failed_episodes(meta, group, vf.Error(type="TaskFailed", message=repr(exc))) + return - for r in rollouts: + for r in (r for e in episodes for r in e.traces): if not r.has_error and r.num_turns == 0: # Empty trajectory: promote to an explicit error so the sink # treats it like any other failure (``has_error`` reads ``ok``) @@ -540,22 +541,29 @@ async def handle_completed_rollout(self, task: asyncio.Task) -> None: get_logger().warning(f"Empty trajectory in group {meta.group_id} ({meta.env_name})") if r.has_error: self.metrics.record_error(kind=meta.kind, env_name=meta.env_name) - if not is_synth_exception and r.last_error is not None: + if r.last_error is not None: get_logger().warning( f"Rollout failed in group {meta.group_id} ({meta.env_name}) — {r.last_error.type}: {r.last_error.message}" ) - if meta.rollout_count == 1: - # A ``run`` task: the whole result is one episode. - await self.emit_episode(meta, group, rollouts) - else: - # A legacy ``run_group`` task: one single-trace episode per trace. - for r in rollouts: - await self.emit_episode(meta, group, [r]) - - async def emit_episode(self, meta: InflightRollout, group: GroupState | None, rollouts: list[Rollout]) -> None: - """Stamp prime-rl metadata onto one completed episode and put it on - ``out_q``. Pops the group from ``self.groups`` once every owed episode - has been emitted.""" + # A ``run`` task answers one episode; a legacy ``run_group`` task one per rollout. + for episode in episodes: + if not episode.traces: + # No trace to carry the failure, so the episode's own error is all there is. + error = episode.last_error + detail = f"{error.type}: {error.message}" if error is not None else "no error recorded" + get_logger().warning(f"Env returned no traces in group {meta.group_id} ({meta.env_name}) — {detail}") + self.metrics.record_error(kind=meta.kind, env_name=meta.env_name) + await self.emit_episode(meta, group, episode) + + async def emit_episode( + self, + meta: InflightEpisode, + group: GroupState | None, + episode: vf.WireEpisode, + ) -> None: + """Put one completed episode on ``out_q``, stamped with the facts of the dispatch it came + from. Pops the group from ``self.groups`` once every owed episode has been emitted. The + same facts also ride each trace, where the algorithms and the saved records read them.""" eval_step = meta.eval_step policy_version = meta.policy_version if group is not None: @@ -565,16 +573,24 @@ async def emit_episode(self, meta: InflightRollout, group: GroupState | None, ro if group.emitted >= group.target_rollouts: self.groups.pop(meta.group_id, None) - for rollout in rollouts: - rollout.kind = meta.kind + for rollout in episode.traces: rollout.env_name = meta.env_name - rollout.group_id = meta.group_id - rollout.policy_version = policy_version - rollout.off_policy_steps = meta.off_policy_steps - if meta.kind == "eval": - assert eval_step is not None, "eval rollout missing eval_step" - rollout.eval_step = eval_step - await self.out_q.put(rollouts) + # Generation is over, so the span closes at whatever version is live now: an episode that + # outlived an update spans more than one. + policy = ( + vf.PolicySpan(start=policy_version, end=self.policy.version) + if self.samples_from_live_policy(meta) + else None + ) + await self.out_q.put(meta.stamp(episode, run_id=self.run_id, policy=policy, eval_step=eval_step)) + + async def emit_failed_episodes(self, meta: InflightEpisode, group: GroupState | None, error: vf.Error) -> None: + """Emit one traceless episode per rollout the task owed, so the sink still counts its way to + ``group_size``. This is the shape vf already gives a run that produced nothing — the reason + on ``errors`` — so prime-rl's own failures need no type of their own.""" + for _ in range(meta.episodes_owed): + self.metrics.record_error(kind=meta.kind, env_name=meta.env_name) + await self.emit_episode(meta, group, Episode.model_construct(errors=[error])) async def drop_group(self, group_id: uuid.UUID) -> int: """Cancel remaining in-flight tasks for this group and emit a @@ -582,34 +598,27 @@ async def drop_group(self, group_id: uuid.UUID) -> int: (both in-flight and not-yet-scheduled). Returns the count for off-policy metrics.""" group = self.groups.pop(group_id, None) - task_idx = group.task_idx if group is not None else -1 # Sync claim phase: pop matching tasks from ``self.inflight`` and # release their permits in one non-yielding sweep. After this loop # the dropped tasks are no longer reachable from ``self.inflight``, # so ``handle_completed_rollout``'s existing None-guard makes the # subsequent async emit phase race-free. - claimed: list[tuple[asyncio.Task, InflightRollout]] = [] + claimed: list[tuple[asyncio.Task, InflightEpisode]] = [] for task, meta in list(self.inflight.items()): if meta.group_id != group_id: continue del self.inflight[task] - self.release(meta.rollout_count) + self.release(meta.episodes_owed) claimed.append((task, meta)) tasks_to_cancel = [task for task, _ in claimed] - inflight_cancelled = sum(meta.rollout_count for _, meta in claimed) - last_meta: InflightRollout | None = claimed[-1][1] if claimed else None + inflight_cancelled = sum(meta.episodes_owed for _, meta in claimed) + last_meta: InflightEpisode | None = claimed[-1][1] if claimed else None + cancel = vf.Error(type="Cancelled", message="Off-policy cancel") for _, meta in claimed: - for _ in range(meta.rollout_count): - trace = Rollout( - task=vf.TraceTask(type="Task", data=vf.TaskData(idx=task_idx, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), - ok=False, - errors=[vf.Error(type="Cancelled", message="Off-policy cancel")], - stop_condition="error", - ) - await self.emit_episode(meta, group, [trace]) + for _ in range(meta.episodes_owed): + await self.emit_episode(meta, group, vf.WireEpisode.model_construct(errors=[cancel])) # For non-group-scoring envs, the group may have rollouts that # were never dispatched (``rollouts_to_schedule > 0``). Emit @@ -620,34 +629,27 @@ async def drop_group(self, group_id: uuid.UUID) -> int: # and us reaching it — synthesize a stand-in from the group state unscheduled_cancelled = 0 if group is not None and group.rollouts_to_schedule > 0: - fallback_meta = last_meta or InflightRollout( + fallback_meta = last_meta or InflightEpisode( kind=group.kind, env_name=group.env_name, group_id=group_id, policy_version=group.policy_version_at_start, - rollout_count=1, + episodes_owed=1, eval_step=group.eval_step, ) unscheduled_cancelled = group.rollouts_to_schedule for _ in range(unscheduled_cancelled): - trace = Rollout( - task=vf.TraceTask(type="Task", data=vf.TaskData(idx=task_idx, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), - ok=False, - errors=[vf.Error(type="Cancelled", message="Off-policy cancel")], - stop_condition="error", - ) - await self.emit_episode(fallback_meta, group, [trace]) + await self.emit_episode(fallback_meta, group, vf.WireEpisode.model_construct(errors=[cancel])) cancelled = inflight_cancelled + unscheduled_cancelled if cancelled > 0: meta_for_log = last_meta or ( - InflightRollout( + InflightEpisode( kind=group.kind, env_name=group.env_name, group_id=group_id, policy_version=group.policy_version_at_start if group else 0, - rollout_count=1, + episodes_owed=1, eval_step=group.eval_step, ) if group is not None @@ -668,8 +670,8 @@ async def cancel_inflight_rollouts(self) -> None: """Cancel all in-flight rollouts. Used on shutdown — doesn't emit markers since the sinks are being torn down anyway.""" for meta in self.inflight.values(): - self.metrics.record_cancellation(kind=meta.kind, env_name=meta.env_name, n=meta.rollout_count) - self.release(meta.rollout_count) + self.metrics.record_cancellation(kind=meta.kind, env_name=meta.env_name, n=meta.episodes_owed) + self.release(meta.episodes_owed) tasks = list(self.inflight.keys()) self.inflight.clear() self.groups.clear() @@ -687,9 +689,9 @@ async def cancel_inflight_train_rollouts(self) -> int: if meta.kind != "train": continue self.inflight.pop(task, None) - self.release(meta.rollout_count) - self.metrics.record_cancellation(kind="train", env_name=meta.env_name, n=meta.rollout_count) - cancelled += meta.rollout_count + self.release(meta.episodes_owed) + self.metrics.record_cancellation(kind="train", env_name=meta.env_name, n=meta.episodes_owed) + cancelled += meta.episodes_owed train_tasks.append(task) train_group_ids.add(meta.group_id) for gid in train_group_ids: diff --git a/src/prime_rl/orchestrator/envs.py b/src/prime_rl/orchestrator/envs.py index a0c86e1908..88ec2c9cfb 100644 --- a/src/prime_rl/orchestrator/envs.py +++ b/src/prime_rl/orchestrator/envs.py @@ -32,13 +32,16 @@ from prime_rl.configs.orchestrator import EnvConfig, EvalSourceConfig, TrainSourceConfig from prime_rl.orchestrator.algo import Algorithm, build_algorithm from prime_rl.orchestrator.sampler import Sampler -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.types import TrainRollout from prime_rl.utils.logger import get_logger # Every wire trace validates into this type. WireTaskData (extra="allow") keeps the env's task # fields without importing the env package — the orchestrator never reads them typed (only # task.idx + task.model_dump). -ROLLOUT_TYPE = Rollout[vf.WireTaskData] +ROLLOUT_TYPE = TrainRollout[vf.WireTaskData] +# The env server answers a wire episode; we keep that envelope and only re-type its traces. The +# dispatcher then mints the Train/Eval episode that carries the dispatch's own facts. +EPISODE_TYPE = vf.WireEpisode # Max wait for the env server to answer health. Generous because the launcher spawns # servers concurrently with the orchestrator, and a legacy server loads its dataset @@ -112,12 +115,15 @@ async def run( cache_salt: str | None, task_data: dict | None = None, task_idx: int | None = None, - ) -> list[Rollout]: - """Run one episode; return its typed Traces. A v1 env takes the task itself - (``task_data``); the legacy bridge is addressed by dataset row (``task_idx``). - A zero-trace episode raises (the dispatcher synthesizes the error marker); a - not-``ok`` episode marks its clean traces failed so partial episodes never - train.""" + ) -> EPISODE_TYPE: + """Run one episode; return it with its traces re-typed as ``Rollout``. A v1 env takes the + task itself (``task_data``); the legacy bridge is addressed by dataset row (``task_idx``). + A zero-trace episode comes back as-is — vf already recorded on it why it produced nothing; + a not-``ok`` episode marks its clean traces failed so partial episodes never train. + + The episode itself is kept rather than flattened away: it is what groups an env's seats, + and downstream reads its aggregates (``by_agent``, ``num_turns``) instead of rebuilding + them from loose traces.""" episode = await self.env_client.run( task_data=task_data, task_idx=task_idx, @@ -125,24 +131,19 @@ async def run( model=model_name, sampling=self._sampling(cache_salt), ) - if not episode.traces: - error = episode.last_error - detail = f"{error.type}: {error.message}" if error is not None else "no traces and no error recorded" - raise RuntimeError(f"episode failed before any trace was produced — {detail}") rollouts = [ROLLOUT_TYPE.model_construct(**dict(wire)) for wire in episode.traces] for rollout in rollouts: - rollout.episode_id = episode.id if not episode.ok and rollout.ok: error = episode.last_error or vf.Error( type="EpisodeFailed", message="A sibling trace in this episode failed" ) rollout.errors = [*rollout.errors, error] rollout.ok = False - return rollouts + return EPISODE_TYPE.model_construct(**{**dict(episode), "traces": rollouts}) async def run_group( self, client: vf.ClientConfig, task_idx: int, model_name: str, group_size: int, cache_salt: str | None - ) -> list[Rollout]: + ) -> list[TrainRollout]: """Run a group of rollouts for ``task_idx`` (group-scoring envs); return typed Traces.""" wires = await self.env_client.run_group( task_idx=task_idx, @@ -151,7 +152,7 @@ async def run_group( model=model_name, sampling=self._sampling(cache_salt), ) - return [ROLLOUT_TYPE.model_construct(**dict(wire)) for wire in wires] + return [EPISODE_TYPE.model_construct(traces=[ROLLOUT_TYPE.model_construct(**dict(wire))]) for wire in wires] class TrainEnv(Env): diff --git a/src/prime_rl/orchestrator/eval_sink.py b/src/prime_rl/orchestrator/eval_sink.py index c02ebab7b9..f1bc00eeea 100644 --- a/src/prime_rl/orchestrator/eval_sink.py +++ b/src/prime_rl/orchestrator/eval_sink.py @@ -8,41 +8,60 @@ 3. ``process_batch`` — at ``num_examples × group_size`` episodes, return an ``EvalBatch`` with the full returned cohort (metrics are computed downstream). -``add()`` takes one episode (``list[Rollout]``) and returns ``EvalBatch | None``; +``add()`` takes one ``Episode`` and returns ``EvalBatch | None``; all accounting counts episodes, never loose traces. """ from __future__ import annotations -import uuid from collections import defaultdict +import verifiers.v1 as vf + from prime_rl.orchestrator.envs import EvalEnvs from prime_rl.orchestrator.metrics import EvalRollouts -from prime_rl.orchestrator.types import EvalBatch, Rollout +from prime_rl.orchestrator.types import ( + Episode, + EvalBatch, + Rollout, + env_name_of, + group_id_of, + rollouts_of, + run_of, +) from prime_rl.utils.logger import get_logger +def eval_step_of(episode: Episode) -> int: + """The eval epoch an episode belongs to, off the run the dispatcher recorded when it landed. + An online eval belongs to the training run, and its metadata requires the step, unlike an + episode to train on, whose step is the window it lands in.""" + metadata = run_of(episode).metadata + assert isinstance(metadata, vf.EvalMetadata), "not an eval episode" + return metadata.step + + class EvalSink: """Constructed only when eval is configured.""" def __init__(self, *, eval_envs: EvalEnvs) -> None: self.eval_envs = eval_envs - self.pending_groups: dict[uuid.UUID, list[Rollout]] = defaultdict(list) + self.pending_groups: dict[str, list[Episode]] = defaultdict(list) # Episodes arrived per group / per batch bucket — the finalization counts. - self.pending_group_episodes: dict[uuid.UUID, int] = defaultdict(int) - self.pending_batches: dict[tuple[str, int], list[Rollout]] = defaultdict(list) + self.pending_group_episodes: dict[str, int] = defaultdict(int) + self.pending_batches: dict[tuple[str, int], list[Episode]] = defaultdict(list) self.pending_batch_episodes: dict[tuple[str, int], int] = defaultdict(int) - def add(self, episode: list[Rollout]) -> EvalBatch | None: + def add(self, episode: Episode) -> EvalBatch | None: """Process one episode arrival; finalize the group on the ``group_size``-th - episode and the per-env epoch on the ``num_examples × group_size``-th.""" - env_name = episode[0].env_name - group_id = episode[0].group_id - for rollout in episode: + episode and the per-env epoch on the ``num_examples × group_size``-th. A failed + episode brings no rollouts but still counts toward both.""" + env_name = env_name_of(episode) + group_id = group_id_of(episode) + for rollout in rollouts_of(episode): self.process_rollout(rollout) - bkey = (env_name, episode[0].eval_step) - self.pending_groups[group_id].extend(episode) + bkey = (env_name, eval_step_of(episode)) + self.pending_groups[group_id].append(episode) self.pending_group_episodes[group_id] += 1 if self.pending_group_episodes[group_id] >= self.group_size_for(env_name): self.process_group(group_id) @@ -66,13 +85,13 @@ def batch_progress(self) -> list[tuple[str, int, int, int, int]]: ``buffered`` is partial-group episode arrivals from non-group-scoring envs.""" batch_counts: dict[tuple[str, int], int] = dict(self.pending_batch_episodes) buffered: dict[tuple[str, int], int] = {} - for group_id, rollouts in self.pending_groups.items(): - if not rollouts: + for group_id, episodes in self.pending_groups.items(): + if not episodes: continue - env_name = rollouts[0].env_name + env_name = env_name_of(episodes[0]) if self.eval_envs.get(env_name).requires_group_scoring: continue - bkey = (env_name, rollouts[0].eval_step) + bkey = (env_name, eval_step_of(episodes[0])) buffered[bkey] = buffered.get(bkey, 0) + self.pending_group_episodes.get(group_id, 0) return [ ( @@ -96,16 +115,19 @@ def process_rollout(self, rollout: Rollout) -> None: # ── level 2: per-group (move into batch bucket) ─────────────────────── - def process_group(self, group_id: uuid.UUID) -> None: - group = self.pending_groups.pop(group_id, []) + def process_group(self, group_id: str) -> None: + finished = self.pending_groups.pop(group_id, []) episodes = self.pending_group_episodes.pop(group_id, 0) - if not group: + if not finished: return - env_name = group[0].env_name - task_idx = group[0].task.data.idx - eval_step = group[0].eval_step + # Read the group's facts off an episode, not a trace: every episode in it may have + # produced none (a whole group cancelled off-policy). + env_name = env_name_of(finished[0]) + eval_step = eval_step_of(finished[0]) + group = [t for e in finished for t in rollouts_of(e)] + task_idx = group[0].task.data.idx if group else -1 bucket = self.pending_batches[(env_name, eval_step)] - bucket.extend(group) + bucket.extend(finished) self.pending_batch_episodes[(env_name, eval_step)] += episodes survivors = [r for r in group if not r.has_error] @@ -123,6 +145,6 @@ def process_batch(self, key: tuple[str, int]) -> EvalBatch: downstream via ``EvalBatch.rollouts.metrics`` over the all/effective subsets, so the sink does no aggregation.""" env_name, step = key - rollouts = self.pending_batches.pop(key, []) + episodes = self.pending_batches.pop(key, []) self.pending_batch_episodes.pop(key, None) - return EvalBatch(env_name=env_name, step=step, rollouts=EvalRollouts(rollouts)) + return EvalBatch(env_name=env_name, step=step, rollouts=EvalRollouts(episodes)) diff --git a/src/prime_rl/orchestrator/metrics.py b/src/prime_rl/orchestrator/metrics.py index 4e67190dc3..c4f1eca34c 100644 --- a/src/prime_rl/orchestrator/metrics.py +++ b/src/prime_rl/orchestrator/metrics.py @@ -23,10 +23,11 @@ from typing import TYPE_CHECKING, Any, Callable, Iterator, Literal +from prime_rl.orchestrator.types import env_name_of, group_id_of, narrow, rollouts_of from prime_rl.orchestrator.utils import compute_pass_metrics if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode, Rollout, TrainRollout Subset = Literal["all", "effective"] @@ -172,7 +173,14 @@ class TraceMetrics(StatGroup): """Trace-level metrics for one agent, every one of them flat over that agent's traces: one sample is one trace, so a fan-out (n same-agent traces in one episode, e.g. n solvers) simply contributes n samples. Weighing whole episodes against each other is the episode level's job; - inside a seat the trace is the unit, which is also what the advantage computation samples.""" + inside a seat the trace is the unit, which is also what the advantage computation samples. + + Built from the episodes narrowed to that agent, not a loose trace list — the episode is the + atomic unit, so anything episode-scoped (which example a trace answered) is still in reach.""" + + def __init__(self, episodes: list[Episode]) -> None: + self.episodes = episodes + super().__init__([t for e in episodes for t in rollouts_of(e)]) DISTRIBUTIONS = ("reward", "num_total_tokens", "num_input_tokens", "num_output_tokens", "num_turns", "num_branches") RATES = ("is_truncated", "is_completed") @@ -227,8 +235,8 @@ def solve_rates(self) -> dict[str, float]: (every trace scored 1.0), and ``solved_some`` (the mixed remainder — the GRPO-signal groups).""" groups: dict = {} - for r in self.rollouts: - groups.setdefault(r.group_id, []).append(r) + for e in self.episodes: + groups.setdefault(group_id_of(e), []).extend(rollouts_of(e)) n_groups = len(groups) solved_none = sum(1 for g in groups.values() if sum(r.reward for r in g) == 0) solved_all = sum(1 for g in groups.values() if all(r.reward == 1.0 for r in g)) @@ -259,52 +267,46 @@ def to_dict(self, prefix: str, *, subset: Subset) -> dict[str, float]: class EpisodeMetrics: - """Metrics shared by train and eval over a rollout list. The count metrics (tokens/turns/ - branches) are episode-level — one value per episode, summing its traces — and are the only - per-metric keys ``to_wandb`` emits at the env level; every trace-level metric is emitted per - agent via ``by_agent()``. The boolean ``Stat`` properties (0/1 distributions, ``.mean()`` is - the rate) serve the console log lines. ``TrainMetrics`` / ``EvalMetrics`` extend ``to_wandb`` - with the train pipeline rates and the eval scores.""" - - def __init__(self, rollouts: list[Rollout]) -> None: - self.rollouts = rollouts + """Metrics shared by train and eval over a list of episodes. The count metrics (tokens/turns/ + branches) are read straight off ``vf.Episode``, which already sums an episode's traces, and are + the only per-metric keys ``to_wandb`` emits at the env level; every trace-level metric is + emitted per agent via ``by_agent()``. The boolean ``Stat`` properties (0/1 distributions, + ``.mean()`` is the rate) serve the console log lines. ``TrainMetrics`` / ``EvalMetrics`` extend + ``to_wandb`` with the train pipeline rates and the eval scores.""" - def episodes(self) -> list[list[Rollout]]: - """The subset's rollouts grouped into their episodes. A rollout without an - ``episode_id`` (legacy envs, synthesized error markers) is its own episode.""" - grouped: dict[str, list[Rollout]] = {} - for r in self.rollouts: - grouped.setdefault(r.episode_id or r.id, []).append(r) - return list(grouped.values()) + def __init__(self, episodes: list[Episode]) -> None: + self.episodes = episodes + self.rollouts: list[Rollout] = [t for e in episodes for t in e.traces] def by_agent(self) -> dict[str, TraceMetrics]: - """Per-agent metric views (``vf.Episode.by_agent`` over the subset's rollouts).""" - per_agent: dict[str, list[Rollout]] = {} - for r in self.rollouts: - per_agent.setdefault(r.agent.name, []).append(r) - return {name: TraceMetrics(rollouts) for name, rollouts in sorted(per_agent.items())} - - # Episode-level count metrics: one value per episode, summing its traces — the same - # aggregation as ``vf.Episode.num_turns`` / ``num_*_tokens``. + """Per-agent metric views: the episodes narrowed to one agent's traces, keyed by the agent + names each episode's own ``vf.Episode.by_agent`` reports.""" + names = sorted({name for e in self.episodes for name in e.by_agent}) + return { + name: TraceMetrics([n for e in self.episodes if (n := narrow(e, lambda r: r.agent.name == name))]) + for name in names + } + + # Episode-level count metrics, one value per episode — ``vf.Episode``'s own aggregates. @property def num_total_tokens(self) -> Stat: - return Stat([float(sum(r.num_total_tokens for r in episode)) for episode in self.episodes()]) + return Stat([float(e.num_total_tokens) for e in self.episodes]) @property def num_input_tokens(self) -> Stat: - return Stat([float(sum(r.num_input_tokens for r in episode)) for episode in self.episodes()]) + return Stat([float(e.num_input_tokens) for e in self.episodes]) @property def num_output_tokens(self) -> Stat: - return Stat([float(sum(r.num_output_tokens for r in episode)) for episode in self.episodes()]) + return Stat([float(e.num_output_tokens) for e in self.episodes]) @property def num_turns(self) -> Stat: - return Stat([float(sum(r.num_turns for r in episode)) for episode in self.episodes()]) + return Stat([float(e.num_turns) for e in self.episodes]) @property def num_branches(self) -> Stat: - return Stat([float(sum(r.num_branches for r in episode)) for episode in self.episodes()]) + return Stat([float(sum(t.num_branches for t in e.traces)) for e in self.episodes]) # Boolean rate metrics for the console log lines (0/1 distributions — ``.mean()`` is the # rate); to_wandb emits their per-agent counterparts instead. @@ -357,14 +359,14 @@ def to_wandb(self, *, prefix: str, subset: Subset) -> dict[str, float]: return out -def pass_at_k(rollouts: list[Rollout]) -> dict[str, float]: +def pass_at_k(episodes: list[Episode]) -> dict[str, float]: """pass@k / pass^k averaged over examples; ``{}`` for non-binary rewards.""" - rewards = [r.reward for r in rollouts] + rewards = [r.reward for e in episodes for r in rollouts_of(e)] if not set(rewards).issubset({0.0, 1.0}): return {} by_example: dict = {} - for r in rollouts: - by_example.setdefault(r.group_id, []).append(r.reward) + for e in episodes: + by_example.setdefault(group_id_of(e), []).extend(r.reward for r in rollouts_of(e)) per_example = [compute_pass_metrics(rs) for rs in by_example.values()] keys = sorted({k for d in per_example for k in d}) return {k: sum(d[k] for d in per_example if k in d) / sum(1 for d in per_example if k in d) for k in keys} @@ -377,8 +379,8 @@ class EvalMetrics(EpisodeMetrics): per example) is supplied by the container so the ``all`` and ``effective`` subsets — and every agent — share one stable key.""" - def __init__(self, rollouts: list[Rollout], group_size: int) -> None: - super().__init__(rollouts) + def __init__(self, episodes: list[Episode], group_size: int) -> None: + super().__init__(episodes) self.group_size = group_size @property @@ -391,54 +393,64 @@ def to_wandb(self, *, prefix: str, subset: Subset) -> dict[str, float]: p = f"{prefix}/{subset}/{agent}" out[f"{p}/avg@{self.group_size}"] = traces.stats()["reward"].mean() if subset == "effective": - out |= {f"{p}/{k}": v for k, v in pass_at_k(traces.rollouts).items()} + out |= {f"{p}/{k}": v for k, v in pass_at_k(traces.episodes).items()} return out class TrainRollouts: - """A list of train rollouts (everything that came back, errored + filtered + untrainable - included). ``effective`` is the clean trainable subset (a view of the same traces); - ``metrics`` builds ``TrainMetrics`` over them.""" + """The train episodes of one window (everything that came back, errored + filtered + + untrainable included). ``effective`` is the clean trainable subset — the same episodes, each + narrowed to its surviving traces; ``metrics`` builds ``TrainMetrics`` over them. Sized and + iterated by *rollout*, since that is the unit almost every consumer wants.""" - def __init__(self, rollouts: list[Rollout] | None = None) -> None: - self.rollouts = rollouts if rollouts is not None else [] + def __init__(self, episodes: list[Episode] | None = None) -> None: + self.episodes = episodes if episodes is not None else [] - def append(self, rollout: Rollout) -> None: - self.rollouts.append(rollout) + def append(self, episode: Episode) -> None: + self.episodes.append(episode) + + @property + def rollouts(self) -> list[TrainRollout]: + return [t for e in self.episodes for t in rollouts_of(e)] def __len__(self) -> int: - return len(self.rollouts) + return sum(len(e.traces) for e in self.episodes) - def __iter__(self) -> Iterator[Rollout]: + def __iter__(self) -> Iterator[TrainRollout]: return iter(self.rollouts) @property def effective(self) -> TrainRollouts: - return TrainRollouts([r for r in self.rollouts if not r.has_error and not r.is_filtered and r.agent.trainable]) + kept = (narrow(e, lambda r: not r.has_error and not r.is_filtered and r.agent.trainable) for e in self.episodes) + return TrainRollouts([e for e in kept if e is not None]) def by_env(self) -> dict[str, TrainRollouts]: - grouped: dict[str, list[Rollout]] = {} - for r in self.rollouts: - grouped.setdefault(r.env_name, []).append(r) - return {env: TrainRollouts(rs) for env, rs in grouped.items()} + grouped: dict[str, list[Episode]] = {} + for episode in self.episodes: + grouped.setdefault(env_name_of(episode), []).append(episode) + return {env: TrainRollouts(episodes) for env, episodes in grouped.items()} @property def metrics(self) -> TrainMetrics: - return TrainMetrics(self.rollouts) + return TrainMetrics(self.episodes) class EvalRollouts: - """A list of eval rollouts (errored + untrainable included). ``effective`` is the - non-errored trainable subset (a view). + """The eval episodes of one epoch (errored + untrainable included). ``effective`` is the + non-errored trainable subset — the same episodes, each narrowed to its surviving traces. ``group_size`` (rollouts per example, the ``avg@k`` k) is derived from the full epoch and carried onto ``effective`` so both subsets share one stable key; ``metrics`` builds ``EvalMetrics``.""" - def __init__(self, rollouts: list[Rollout] | None = None, group_size: int | None = None) -> None: - self.rollouts = rollouts if rollouts is not None else [] + def __init__(self, episodes: list[Episode] | None = None, group_size: int | None = None) -> None: + self.episodes = episodes if episodes is not None else [] self._group_size = group_size + @property + def rollouts(self) -> list[Rollout]: + return [t for e in self.episodes for t in e.traces] + def __len__(self) -> int: - return len(self.rollouts) + return sum(len(e.traces) for e in self.episodes) def __iter__(self) -> Iterator[Rollout]: return iter(self.rollouts) @@ -452,17 +464,16 @@ def group_size(self) -> int: if self._group_size is not None: return self._group_size counts: dict = {} - for r in self.rollouts: - if r.agent.trainable: - counts[r.group_id] = counts.get(r.group_id, 0) + 1 + for e in self.episodes: + trainable = sum(1 for r in rollouts_of(e) if r.agent.trainable) + counts[group_id_of(e)] = counts.get(group_id_of(e), 0) + trainable return max(counts.values(), default=0) @property def effective(self) -> EvalRollouts: - return EvalRollouts( - [r for r in self.rollouts if not r.has_error and r.agent.trainable], group_size=self.group_size - ) + kept = (narrow(e, lambda r: not r.has_error and r.agent.trainable) for e in self.episodes) + return EvalRollouts([e for e in kept if e is not None], group_size=self.group_size) @property def metrics(self) -> EvalMetrics: - return EvalMetrics(self.rollouts, self.group_size) + return EvalMetrics(self.episodes, self.group_size) diff --git a/src/prime_rl/orchestrator/orchestrator.py b/src/prime_rl/orchestrator/orchestrator.py index 7d7b095985..6db9e0853f 100644 --- a/src/prime_rl/orchestrator/orchestrator.py +++ b/src/prime_rl/orchestrator/orchestrator.py @@ -3,8 +3,8 @@ ``Orchestrator`` owns the shared state (policy, progress, ckpt, monitor) and drives the pipeline. Components are single-purpose: -- ``RolloutDispatcher`` schedules rollouts; emits ``Rollout`` (train/eval - discriminated by ``kind``) on its queue. +- ``RolloutDispatcher`` schedules rollouts; emits ``Episode`` (train/eval discriminated by + ``kind``) on its queue — the env's own episode, or a traceless one saying why there is none. - ``TrainSink`` ingests train rollouts (tokenize → advantages → filters) and returns a ``TrainBatch`` when the threshold is met. - ``EvalSink`` ingests eval rollouts and returns an ``EvalBatch`` (the full @@ -28,7 +28,6 @@ from typing import TYPE_CHECKING import tomli_w -import verifiers.v1 as vf from modelexpress import p2p_pb2 from modelexpress.client import MxClient @@ -40,6 +39,8 @@ from prime_rl.transport.base import TrainingBatchSender from prime_rl.utils.client import InferencePool from prime_rl.utils.monitor.base import Monitor +import verifiers.v1 as vf + import prime_rl._compat # noqa: F401 — patch ring_flash_attn compat before transitive imports from prime_rl.configs.orchestrator import OrchestratorConfig from prime_rl.orchestrator.ckpt import setup_ckpt_manager @@ -57,16 +58,19 @@ from prime_rl.orchestrator.train_sink import TrainSink from prime_rl.orchestrator.train_source import TrainSource from prime_rl.orchestrator.types import ( + Episode, EvalBatch, Policy, Progress, - Rollout, TrainBatch, + group_id_of, + run_of, + to_record, ) from prime_rl.orchestrator.utils import ( get_weight_dir, intercept_vf_logging, - save_rollouts, + save_episodes, set_default_executor, setup_policy_inference_pool, trim_process_memory, @@ -401,6 +405,7 @@ async def setup(self) -> None: eval_source=self.eval_source, policy_pool=self.policy_inference, policy=self.policy, + run_id=self.run_id, max_inflight_episodes=config.max_inflight_episodes, tasks_per_minute=config.tasks_per_minute, max_off_policy_steps=config.max_off_policy_steps, @@ -505,7 +510,7 @@ async def start(self) -> None: trim_process_memory() async def main_loop(self) -> None: - """Consume episodes (``list[Rollout]``) from the dispatcher and route them + """Consume ``Episode``\\ s from the dispatcher and route them to the train / eval sink. Both sinks return a finalized batch (or ``None``) from ``add()``; we just dispatch on the result.""" while not self.stopped.is_set(): @@ -515,37 +520,27 @@ async def main_loop(self) -> None: break try: - episode: list[Rollout] = await asyncio.wait_for(self.dispatcher.out_q.get(), timeout=0.5) + episode: Episode = await asyncio.wait_for(self.dispatcher.out_q.get(), timeout=0.5) except asyncio.TimeoutError: continue - # Every completed rollout — errored, filtered, or never batched — lands in the - # ``all`` trace file the moment it arrives, so it survives crashes and drains. - # Train rollouts belong to the batch window currently collecting (``progress.step``), - # eval rollouts to the step whose eval triggered them. - kind = episode[0].kind - step = episode[0].eval_step if kind == "eval" else self.progress.step + # Every completed episode — errored, filtered, or never batched — lands in the + # ``all`` trace file the moment it arrives, so it survives crashes and drains. One + # episode per line, so an episode that produced no traces still records why. + # An eval episode already knows its step (the eval that triggered it); a train one + # belongs to the batch window collecting right now, which only this loop knows. + run = run_of(episode) + if isinstance(run.metadata, vf.TrainMetadata): + run.metadata.step = self.progress.step + step = run.metadata.step assert step is not None - run: vf.RunInfo = ( - vf.EvalRunInfo(id=self.run_id, step=step) - if kind == "eval" - else vf.TrainRunInfo(id=self.run_id, step=step) - ) - for rollout in episode: - rollout.record_run( - run, - env_name=rollout.env_name, - group_id=str(rollout.group_id), - episode_id=rollout.episode_id, - policy_version=rollout.policy_version, - ) await asyncio.to_thread( - save_rollouts, - [rollout.to_record() for rollout in episode], - get_trace_path(self.config.output_dir, step, kind, "all"), + save_episodes, + [to_record(episode)], + get_trace_path(self.config.output_dir, step, run.metadata.type, "all"), ) - if kind == "eval": + if run.metadata.type == "eval": assert self.eval_sink is not None # eval rollouts only emitted when eval is configured eval_batch = self.eval_sink.add(episode) if eval_batch is not None: @@ -560,7 +555,7 @@ async def main_loop(self) -> None: async def finalize_train_batch(self, batch: TrainBatch) -> None: """Ship one ``TrainBatch`` out to the trainer and handle the I/O - side-effects (ckpt, save_rollouts, reference scoring, sender.send, + side-effects (ckpt, save_episodes, reference scoring, sender.send, metrics, heartbeat, progress, eval trigger). The sink has already done all data-transformation work.""" config = self.config @@ -625,21 +620,12 @@ async def finalize_train_batch(self, batch: TrainBatch) -> None: await self.version_advanced.wait() self.wait_for_policy_time += time.perf_counter() - hold_start - # Stamp each rollout's true staleness: batch ``step`` trains on policy - # v{step-1}, so a rollout generated from v{k} is (step-1)-k versions - # off-policy — queue time included, unlike the dispatcher's in-flight - # counter, which only sees weight updates during generation. Frozen- - # sourced rollouts stay 0 (their sampler doesn't follow the policy). - for r in batch.rollouts: - if self.train_envs.get(r.env_name).sampler.samples_from_live_policy: - r.off_policy_steps = (step - 1) - r.policy_version - # The effective (clean, trained-on) subset lands in the per-step ``effective`` trace file # at ship time; the full arrival window already streamed into ``all`` on arrival. # to_record drops the per-node training tensors — they're for training, not the rollout # record, and can't round-trip json (raw numpy bytes). - records = [r.to_record() for r in effective] - await asyncio.to_thread(save_rollouts, records, get_trace_path(config.output_dir, step, "train", "effective")) + records = [to_record(e) for e in effective.episodes] + await asyncio.to_thread(save_episodes, records, get_trace_path(config.output_dir, step, "train", "effective")) await self.sender.send(TrainingBatch(examples=batch.samples, step=step)) self.progress.step += 1 @@ -664,7 +650,7 @@ async def finalize_train_batch(self, batch: TrainBatch) -> None: num_input = sum(r.num_input_tokens for r in effective) num_output = sum(r.num_output_tokens for r in effective) num_rollouts = len(batch.rollouts) - num_unique_examples = len({r.group_id for r in batch.rollouts}) + num_unique_examples = len({group_id_of(e) for e in batch.rollouts.episodes}) metrics |= { "progress/tokens": num_tokens, "progress/input_tokens": num_input, @@ -689,7 +675,7 @@ async def finalize_train_batch(self, batch: TrainBatch) -> None: metrics[f"pre_filters/all/{name}/rate"] = count / self.train_sink.pre_filter_seen self.monitor.log(metrics, step=step) self.wait_for_policy_time = 0.0 - self.monitor.log_samples(effective.rollouts, step=step) + self.monitor.log_samples(effective.episodes, step=step) self.monitor.log_distributions( distributions={ "rewards": [r.reward for r in effective], @@ -813,7 +799,7 @@ def log_train_batch(self, batch: TrainBatch, *, step: int, step_time: float) -> n_effective = len(effective) n_trainable = sum(1 for r in effective if r.is_trainable) trainable_rate = (n_trainable / n_effective) if n_effective else 0.0 - max_off_policy = max((r.off_policy_steps for r in effective), default=0) + max_off_policy = max((run_of(e).metadata.off_policy_steps or 0 for e in effective.episodes), default=0) head = ( f"Step {step} | {format_time(step_time):>7} | Reward {eff.reward.mean():.4f} | " @@ -837,13 +823,13 @@ def log_train_batch(self, batch: TrainBatch, *, step: int, step_time: float) -> lines.append( f"╰─ {env_name:<{name_width}} | Ratio {ratio:.1%} | Reward {env_eff.reward.mean():.4f} | " f"Turns {env_eff.num_turns.mean():.1f} | Branches {env_eff.num_branches.mean():.1f} | " - f"Max Off-Policy {max((r.off_policy_steps for r in env_eff_pool), default=0)} | " + f"Max Off-Policy {max((run_of(e).metadata.off_policy_steps or 0 for e in env_eff_pool.episodes), default=0)} | " f"Error {pool.metrics.has_error.mean():.1%} | Truncation {env_eff.is_truncated.mean():.1%}" ) get_logger().success("\n\t\t ".join(lines)) async def finalize_eval_batch(self, batch: EvalBatch) -> None: - """Persist + log one completed eval epoch (save_rollouts, + """Persist + log one completed eval epoch (save_episodes, monitor.log_eval_samples, monitor.log).""" if not batch.rollouts: get_logger().warning(f"Eval @ step={batch.step} env={batch.env_name}: no rollouts returned, skipping log") @@ -853,13 +839,13 @@ async def finalize_eval_batch(self, batch: EvalBatch) -> None: # completion (multiple eval envs share the step file — each epoch appends its cohort # once, and every record carries ``env_name``); the full returned cohort already # streamed into ``all`` on arrival. - records = [r.to_record() for r in batch.rollouts.effective] + records = [to_record(e) for e in batch.rollouts.effective.episodes] await asyncio.to_thread( - save_rollouts, records, get_trace_path(self.config.output_dir, batch.step, "eval", "effective") + save_episodes, records, get_trace_path(self.config.output_dir, batch.step, "eval", "effective") ) - self.monitor.log_eval_samples(batch.rollouts, env_name=batch.env_name, step=batch.step) - policy_versions = {r.policy_version for r in batch.rollouts} - policy_version = min(policy_versions) + self.monitor.log_eval_samples(batch.rollouts.episodes, env_name=batch.env_name, step=batch.step) + policy_versions = {m.policy.start for e in batch.rollouts.episodes if (m := run_of(e).metadata).policy} + policy_version = min(policy_versions, default=0) if len(policy_versions) > 1: get_logger().warning( f"Eval {batch.env_name} step {batch.step} had mixed policy versions: {sorted(policy_versions)}" diff --git a/src/prime_rl/orchestrator/train_sink.py b/src/prime_rl/orchestrator/train_sink.py index fb0dc6abc1..b340404a7b 100644 --- a/src/prime_rl/orchestrator/train_sink.py +++ b/src/prime_rl/orchestrator/train_sink.py @@ -2,24 +2,23 @@ 1. ``process_rollout`` — eager per-rollout tokenization (overlaps with dispatcher producing more rollouts), then the env algorithm's - ``finalize_rollout`` (rollout-local scoring + any reference I/O). Errored - and untrainable rollouts skip this. -2. ``process_group`` — filters errored rollouts, hands the trainable - survivors to the env algorithm's ``finalize_group`` (advantages + - per-sample wire stamping), runs the pre-batch filter pass. -3. ``process_batch`` — applies post-batch filter annotations and assembles - the trainer-bound ``TrainingSample`` list. Returns a ``TrainBatch``. + ``finalize_rollout`` (rollout-local scoring + any reference + I/O). Errored and untrainable rollouts skip this. +2. ``process_group`` — filters errored rollouts, hands the episodes narrowed + to their trainable survivors to the env algorithm's ``finalize_group`` + (advantages + per-sample wire stamping), runs the pre-batch filter pass. +3. ``process_batch`` — applies post-batch filter annotations and assembles the + trainer-bound ``TrainingSample`` list. Returns a ``TrainBatch``. -``add()`` takes one episode (``list[Rollout]``) and returns +``add()`` takes one ``Episode`` and returns ``TrainBatch | None``; group accounting counts episodes, never loose traces. -I/O concerns (ship to trainer, save_rollouts, monitor.log) live on the +I/O concerns (ship to trainer, save_episodes, monitor.log) live on the orchestrator. """ from __future__ import annotations import asyncio -import uuid from collections import defaultdict from prime_rl.configs.orchestrator import OrchestratorConfig @@ -27,12 +26,21 @@ from prime_rl.orchestrator.filters import RolloutFilter, apply_filters from prime_rl.orchestrator.metrics import TrainRollouts from prime_rl.orchestrator.trajectories import trace_to_samples -from prime_rl.orchestrator.types import Rollout, TrainBatch +from prime_rl.orchestrator.types import ( + Episode, + TrainBatch, + TrainRollout, + env_name_of, + group_id_of, + group_rollouts, + narrow, + rollouts_of, +) from prime_rl.transport import TrainingSample from prime_rl.utils.logger import get_logger -def payload_tokens(rollout: Rollout) -> int: +def payload_tokens(rollout: TrainRollout) -> int: """Token cost of the rollout's trainer-bound payload — the samples built by ``process_rollout``. This is what actually ships: forked traces can drop branches with no trainable tokens, so ``Trace.num_total_tokens`` (which sums @@ -79,11 +87,11 @@ def __init__( # Keyed by the dispatcher's group UUID. ``(env_name, task_idx)`` # isn't unique — the same task can be re-sampled while an # earlier group is still in flight - self.pending_groups: dict[uuid.UUID, list[Rollout]] = defaultdict(list) + self.pending_groups: dict[str, list[Episode]] = defaultdict(list) # Episodes arrived per group — the finalization count (an episode may # add several traces to ``pending_groups`` but counts once here). - self.pending_group_episodes: dict[uuid.UUID, int] = defaultdict(int) - self.pending_batch: list[Rollout] = [] + self.pending_group_episodes: dict[str, int] = defaultdict(int) + self.pending_batch: list[TrainRollout] = [] # Running payload-token total of ``pending_batch`` (token-batched # runs), kept in sync on append/pop so the readiness check never # re-sums per arrival. @@ -112,8 +120,8 @@ def buffered_count(self) -> int: (non-group-scoring envs) — buffered in the sink ahead of the batch.""" return sum( self.pending_group_episodes.get(group_id, 0) - for group_id, rollouts in self.pending_groups.items() - if rollouts and not self.train_envs.get(rollouts[0].env_name).requires_group_scoring + for group_id, episodes in self.pending_groups.items() + if episodes and not self.train_envs.get(env_name_of(episodes[0])).requires_group_scoring ) def pending_batch_by_env(self) -> dict[str, int]: @@ -124,16 +132,17 @@ def pending_batch_by_env(self) -> dict[str, int]: counts[r.env_name] += 1 return dict(counts) - async def add(self, episode: list[Rollout]) -> TrainBatch | None: + async def add(self, episode: Episode) -> TrainBatch | None: """Process one episode arrival; finalize the group on the ``group_size``-th episode; return a ``TrainBatch`` if the finalization pushed (or left) the batch over its threshold. Arrivals into - still-incomplete groups never ship a batch.""" - group_id = episode[0].group_id - env_name = episode[0].env_name - for rollout in episode: + still-incomplete groups never ship a batch. A failed episode brings no + rollouts, but still counts toward the group so finalization triggers.""" + group_id = group_id_of(episode) + env_name = env_name_of(episode) + for rollout in rollouts_of(episode): await self.process_rollout(rollout) - self.pending_groups[group_id].extend(episode) + self.pending_groups[group_id].append(episode) self.pending_group_episodes[group_id] += 1 if self.pending_group_episodes[group_id] < self.group_size_for(env_name): return None @@ -150,7 +159,7 @@ async def add(self, episode: list[Rollout]) -> TrainBatch | None: return self.process_batch() return None - async def process_rollout(self, rollout: Rollout) -> None: + async def process_rollout(self, rollout: TrainRollout) -> None: """Build training samples from the rollout's Trace (one per branch), walking the message graph. Training is renderer-only across all modes (RL/OPD student, SFT teacher), so every node already carries its tokens. Errored rollouts are dropped at the group @@ -169,24 +178,26 @@ async def process_rollout(self, rollout: Rollout) -> None: # tokenized — before its group is complete. await self.train_envs.get(rollout.env_name).algorithm.finalize_rollout(rollout) - async def process_group(self, group_id: uuid.UUID) -> None: + async def process_group(self, group_id: str) -> None: """Finalize one GRPO group: drop errored rollouts (the whole group when ``requires_group_scoring`` and any failed), assign advantages, run pre-batch filters, append survivors to ``pending_batch``.""" - group = self.pending_groups.pop(group_id, []) + episodes = self.pending_groups.pop(group_id, []) self.pending_group_episodes.pop(group_id, None) - if not group: + if not episodes: return + # Read the group's facts off an episode, not a trace: every episode in it may have + # produced none (a whole group cancelled off-policy). + env_name = env_name_of(episodes[0]) + group = [t for e in episodes for t in rollouts_of(e)] # Window membership follows group finalization, not arrival: a rollout # only becomes observable (metrics / persistence) once its whole group # is finalized, so a batch's window never claims rollouts of a group # that ships later. Dropped groups still land here — they were observed. - for r in group: - self.pending_rollouts.append(r) - env_name = group[0].env_name - task_idx = group[0].task.data.idx - survivors = [r for r in group if not r.has_error] - num_errored = len(group) - len(survivors) + for episode in episodes: + self.pending_rollouts.append(episode) + task_idx = group[0].task.data.idx if group else -1 + num_errored = sum(r.has_error for r in group) # Group-scoring envs: any failure makes survivors' rewards unsafe # (computed relative to the missing ones) @@ -197,8 +208,10 @@ async def process_group(self, group_id: uuid.UUID) -> None: f"rollouts={len(group)} (errored={num_errored}) | dropped: group-scored partial" ) return - # Untrainable traces carry no samples and must not skew the group baseline. - survivors = [r for r in survivors if r.agent.trainable] + # Untrainable traces carry no samples and must not skew the group baseline. The cohort + # stays episodes so the algorithm can still see which attempts shared one. + cohort = [n for e in episodes if (n := narrow(e, lambda r: not r.has_error and r.agent.trainable))] + survivors = group_rollouts(cohort) if not survivors: get_logger().debug( f"Finished group | env={env_name} task_idx={task_idx} | " @@ -209,7 +222,7 @@ async def process_group(self, group_id: uuid.UUID) -> None: # Advantages + per-sample wire stamping (advantage stream, loss # routing) are the algorithm's job (finalize_group); the sink only # owns the grouping mechanics. - await env.algorithm.finalize_group(survivors) + await env.algorithm.finalize_group(cohort) # The env has a single sampling temperature; fan it out per token # (context tokens are masked out, so their temperature is don't-care). @@ -251,11 +264,9 @@ async def process_group(self, group_id: uuid.UUID) -> None: ) def process_batch(self) -> TrainBatch: - """Pop a cohort off ``pending_batch`` (by rollout count when - ``batch_size`` is set, by token count when ``token_batch_size`` is - set), apply post-batch filter annotations, and assemble the - trainer-bound ``TrainingSample`` list. Overflow stays for the next - batch.""" + """Pop a cohort off ``pending_batch`` (by rollout count when ``batch_size`` is set, by + token count when ``token_batch_size`` is set) and flatten it into the trainer-bound + ``TrainingSample`` list. Overflow stays for the next batch.""" if self.batch_size is not None: cohort = self.pending_batch[: self.batch_size] self.pending_batch = self.pending_batch[self.batch_size :] diff --git a/src/prime_rl/orchestrator/types.py b/src/prime_rl/orchestrator/types.py index 65c56d3181..eadf05fe3f 100644 --- a/src/prime_rl/orchestrator/types.py +++ b/src/prime_rl/orchestrator/types.py @@ -3,12 +3,16 @@ from __future__ import annotations import uuid +from collections.abc import Callable, Iterable from dataclasses import dataclass -from typing import TYPE_CHECKING, Generic, Literal, Protocol +from typing import TYPE_CHECKING, Any, Generic, Literal, Protocol, cast import verifiers.v1 as vf from pydantic import ConfigDict, Field +from verifiers.v1.configs.agent import WireAgentConfig +from verifiers.v1.state import State from verifiers.v1.task import DataT +from verifiers.v1.trace import EXCLUDE_FIELDS from prime_rl.transport import TrainingSample @@ -38,21 +42,6 @@ class Progress: RolloutKind = Literal["train", "eval"] -@dataclass -class InflightRollout: - """Per-task scheduling state in the dispatcher; one entry per in-flight - ``run`` / ``run_group`` task.""" - - kind: RolloutKind - env_name: str - group_id: uuid.UUID - policy_version: int - rollout_count: int - client_config: vf.ClientConfig | None = None - off_policy_steps: int = 0 - eval_step: int | None = None - - @dataclass class GroupState: """Per-group dispatcher state: what's left to schedule + the pinned @@ -72,72 +61,154 @@ class GroupState: policy_version_at_start: int = 0 -class Rollout(vf.Trace[DataT], Generic[DataT]): - """A completed rollout: the env's typed ``vf.Trace`` *is* the rollout — prime-rl's - orchestration metadata lives on it directly (set by the dispatcher once the rollout - returns), so there's no wrapper. Train vs eval is the ``kind`` discriminator. All metadata - fields are ``exclude=True``, so dumping a Rollout yields a plain trace on the wire; the - orchestrator mirrors them onto the trace's own fields via ``vf.Trace.record_run`` (``kind`` as - ``run.type``, the run id, the step, and ``env_name``/``group_id``/``policy_version`` into - ``info``) when a rollout arrives, so the on-disk records stay fully placeable. +Rollout = vf.WireTrace +"""The env's own trace, unextended — what an eval rollout is, and the read side of a training one. +Where a trace sits (its episode, its group, its env) is the episode's to say.""" + + +class TrainRollout(vf.Trace[DataT, State, WireAgentConfig], Generic[DataT]): + """A rollout on the training path — the one place prime-rl extends a verifiers type, because + trainer-bound state has nowhere else to live: the samples built from its branches and the + filter verdicts over them. All of it is ``exclude=True``, so dumping one yields a plain trace + on the wire. - It is also the single currency the scoring hooks receive: a hook reads the trace - directly (``rollout.reward``, ``rollout.nodes``, ``rollout.num_turns``) and writes - credit through :meth:`assign_advantages` (scalar broadcast or per-token), which - spreads over the samples' trainable (mask-True) tokens.""" + ``env_name`` rides along because a sample is routed by it (the trainer's per-env loss config) + and the sink's pending batch is a trace list. Eval rollouts carry the fields unset.""" model_config = ConfigDict(arbitrary_types_allowed=True) # ``samples`` holds msgspec structs - kind: RolloutKind = Field(default="train", exclude=True) env_name: str = Field(default="", exclude=True) - group_id: uuid.UUID = Field(default_factory=uuid.uuid4, exclude=True) - # Links the traces of one episode; stamped into ``info`` on arrival so - # saved records keep their grouping. - episode_id: str = Field(default="", exclude=True) - policy_version: int = Field(default=0, exclude=True) - off_policy_steps: int = Field(default=0, exclude=True) samples: list[TrainingSample] = Field(default_factory=list, exclude=True) - # Per-token rl advantage stream, full-length-N (= len(token_ids)) per - # sample, concatenated across the rollout's samples in order; 0.0 on - # non-trainable positions. None = no credit assigned (advantage-based - # filters skip it; the wire ships no advantage stream). - advantages: list[float] | None = Field(default=None, exclude=True) is_filtered: bool = Field(default=False, exclude=True) filter_results: dict[str, bool] = Field(default_factory=dict, exclude=True) - eval_step: int | None = Field(default=None, exclude=True) - - def assign_advantages(self, values: float | list[float]) -> None: - """Write the rl advantage stream: a scalar broadcast over the - rollout's trainable (mask-True) tokens (0.0 elsewhere), or a per-token - list already aligned full-length to the samples' concatenated - ``token_ids``. A rollout never assigned ships no advantage stream.""" - total = sum(len(sample.token_ids) for sample in self.samples) - if isinstance(values, (int, float)): - self.advantages = [ - float(values) if trainable else 0.0 for sample in self.samples for trainable in sample.mask - ] - return - if len(values) != total: - raise ValueError( - f"per-token advantages must align with the rollout's tokens: " - f"got {len(values)}, expected {total} (env '{self.env_name}')." - ) - self.advantages = [float(v) for v in values] + + def assign_advantages(self, value: float) -> None: + """Write ``value`` as the credit for every trainable token, node by node. Credit lives on + the nodes so branches sharing one cannot disagree about it, and so the trainer reads it + aligned to the tokens (``Branch.advantages``) rather than re-sliced by offset.""" + for node in self.nodes: + trainable = sum(node.mask) + node.advantages = [value] * trainable if trainable else None + + @property + def advantages(self) -> list[float] | None: + """Every assigned credit on this trace, or ``None`` if it was never scored — which a + trace assigned all zeros is not.""" + if all(node.advantages is None for node in self.nodes): + return None + return [a for node in self.nodes for a in (node.advantages or [])] def scalar_advantage(self) -> float | None: - """Scalar view of the per-token advantage stream for monitoring: the - mean over assigned (non-zero) positions — exact for the uniform GRPO - case, 0.0 for a zero-advantage group, None when no credit was assigned.""" - if not self.advantages: + """Scalar view of the credit for monitoring: the mean over assigned (non-zero) positions — + exact for the uniform GRPO case, 0.0 for a zero-advantage group, None when unscored.""" + advantages = self.advantages + if not advantages: return None - nonzero = [a for a in self.advantages if a != 0.0] + nonzero = [a for a in advantages if a != 0.0] return sum(nonzero) / len(nonzero) if nonzero else 0.0 @property def is_trainable(self) -> bool: """Whether the rollout carries a training signal — a nonzero advantage on some token. A uniform-reward GRPO group (all-zero advantages) or an unscored rollout has no gradient.""" - return bool(self.advantages) and any(a != 0.0 for a in self.advantages) + advantages = self.advantages + return bool(advantages) and any(a != 0.0 for a in advantages) + + +Episode = vf.WireEpisode +"""The env's own episode, unextended: everything prime-rl needs to say about a dispatch has a +place on it already — the env it ran (``env.name``), the group it was planned in (``group``), and +the run it belongs to (``run``, which on the training path carries the policy version and how +stale it got).""" + + +def narrow(episode: Episode, keep: Callable[[Rollout], bool]) -> Episode | None: + """The episode with only the traces that pass ``keep``, or ``None`` if none do. A subset stays + a list of episodes rather than a flat trace list, so the episode-level aggregates keep + describing what survived. The kept traces are the same objects, not copies.""" + traces = [t for t in episode.traces if keep(t)] + return episode.model_copy(update={"traces": traces}) if traces else None + + +def rollouts_of(episode: Episode) -> list[TrainRollout]: + """An episode's traces, typed as the rollouts prime-rl works with. Every trace is built as a + ``TrainRollout`` (``Env.run`` is shared), so the training fields are always reachable — on the + eval path they simply stay empty.""" + return cast(list[TrainRollout], episode.traces) + + +def to_record(episode: Episode) -> dict[str, Any]: + """JSON record without the per-node training tensors — the episode form of + ``Trace.to_record``, and the unit ``traces.jsonl`` stores: one episode per line. The tensors + are the trainer's, not the record's, and raw numpy bytes don't round-trip through json.""" + return episode.model_dump(mode="json", exclude={"traces": {"__all__": EXCLUDE_FIELDS}}) + + +GROUP_ID = "group_id" +"""``Episode.info`` key for the comparison group. Verifiers has no notion of one — a group is the +consumer's: the episodes it planned from one task and scores against each other.""" + + +def group_id_of(episode: Episode) -> str: + """The group an episode was planned in. The dispatcher plans every episode into one, so this + is always set by the time anything downstream asks.""" + group_id = episode.info.get(GROUP_ID) + assert isinstance(group_id, str), "the dispatcher plans every episode into a group" + return group_id + + +def env_name_of(episode: Episode) -> str: + """The env as prime-rl names it — its config key, which is not vf's ``env.id``.""" + return episode.env.name or "" + + +def run_of(episode: Episode) -> vf.TrainRunInfo: + """An episode's run record. Every episode the orchestrator produces belongs to the training + run — the ones it trains on and the ones it evaluates along the way — so this is always a + ``TrainRunInfo``, and ``kind`` says which of the two it is.""" + assert isinstance(episode.run, vf.TrainRunInfo), "the dispatcher records the run on arrival" + return episode.run + + +def group_rollouts(episodes: Iterable[Episode]) -> list[TrainRollout]: + """Every trace of a group, flat — the view an algorithm comparing across the whole group + wants, where the episode an attempt came from does not matter.""" + return [r for e in episodes for r in rollouts_of(e)] + + +@dataclass +class InflightEpisode: + """One episode in flight, and the facts of the dispatch that will be stamped onto it when it + lands. The pair is the whole lifecycle: an ``InflightEpisode`` going out, an ``Episode`` coming + back — so nothing downstream has to know how a rollout was scheduled.""" + + kind: RolloutKind + env_name: str + group_id: uuid.UUID + policy_version: int + episodes_owed: int + """How many episodes this dispatch owes the sink — one, except on the legacy group path.""" + client_config: vf.ClientConfig | None = None + eval_step: int | None = None + + def stamp(self, episode: Episode, *, run_id: str, policy: vf.PolicySpan | None, eval_step: int | None) -> Episode: + """Write the dispatch's facts onto the landed episode, in the places the episode already + has for them. The group's values win over this dispatch's when it is still alive, so they + are passed in rather than read off ``self``. + + The run's metadata says what the episode is to the run, which is all the rest of the + orchestrator needs to route it. An eval knows its step here; an episode to train on does + not — it belongs to whichever batch window is collecting when it lands, so the main loop + fills that in.""" + episode.env.name = self.env_name + episode.info[GROUP_ID] = str(self.group_id) + if self.kind == "eval": + assert eval_step is not None, "eval episode missing its step" + metadata: vf.EpisodeMetadata = vf.EvalMetadata(step=eval_step, policy=policy) + else: + metadata = vf.TrainMetadata(policy=policy) + episode.run = vf.TrainRunInfo(id=run_id, metadata=metadata) + return episode @dataclass diff --git a/src/prime_rl/orchestrator/utils.py b/src/prime_rl/orchestrator/utils.py index 374a1070f0..62ca24b872 100644 --- a/src/prime_rl/orchestrator/utils.py +++ b/src/prime_rl/orchestrator/utils.py @@ -49,15 +49,16 @@ async def setup_policy_inference_pool(*, config: OrchestratorConfig, tokenizer): return renderer, inference_pool -def save_rollouts(rollouts: list[dict], path: Path) -> None: - """Append rollouts (Trace record dicts, already JSON-serializable) to a JSONL file. - The trace streams are append-only: ``all`` grows one rollout at a time as they - complete, ``effective`` one batch at a time on finalize.""" +def save_episodes(episodes: list[dict], path: Path) -> None: + """Append episodes (Episode record dicts, already JSON-serializable) to a JSONL file — one + episode per line, the shape verifiers writes and its ``read_episodes`` reads. The streams are + append-only: ``all`` grows one episode at a time as they complete, ``effective`` one batch at a + time on finalize.""" path.parent.mkdir(parents=True, exist_ok=True) opts = orjson.OPT_APPEND_NEWLINE | orjson.OPT_SERIALIZE_NUMPY with open(path, "ab") as f: - for rollout in rollouts: - f.write(orjson.dumps(rollout, default=str, option=opts)) + for episode in episodes: + f.write(orjson.dumps(episode, default=str, option=opts)) def intercept_vf_logging(logger: str = "verifiers", level: str = "DEBUG", prefix: str | None = None): diff --git a/src/prime_rl/utils/monitor/base.py b/src/prime_rl/utils/monitor/base.py index 7a0a51a6e4..cbe63df6ff 100644 --- a/src/prime_rl/utils/monitor/base.py +++ b/src/prime_rl/utils/monitor/base.py @@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode _DROPPED_JSON_VALUE = object() @@ -85,11 +85,11 @@ def log(self, metrics: dict[str, Any], step: int) -> None: pass @abstractmethod - def log_samples(self, rollouts: list[Rollout], step: int) -> None: + def log_samples(self, episodes: list[Episode], step: int) -> None: pass @abstractmethod - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: pass @abstractmethod @@ -118,10 +118,10 @@ def log(self, metrics: dict[str, Any], step: int) -> None: else: self.history = [metrics] - def log_samples(self, rollouts: list[Rollout], step: int) -> None: + def log_samples(self, episodes: list[Episode], step: int) -> None: pass - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: pass def save_final_summary(self, filename: str = "final_summary.json") -> None: diff --git a/src/prime_rl/utils/monitor/file.py b/src/prime_rl/utils/monitor/file.py index 08a5240bd9..a44e7a9e0a 100644 --- a/src/prime_rl/utils/monitor/file.py +++ b/src/prime_rl/utils/monitor/file.py @@ -12,7 +12,7 @@ from prime_rl.utils.monitor.base import Monitor, drop_non_finite_json_values if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode class FileMonitor(Monitor): @@ -79,10 +79,10 @@ def log(self, metrics: dict[str, Any], step: int) -> None: self._file.write(json.dumps(row)) self._file.write("\n") - def log_samples(self, rollouts: list[Rollout], step: int) -> None: + def log_samples(self, episodes: list[Episode], step: int) -> None: pass - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: pass def log_distributions(self, distributions: dict[str, list[float]], step: int) -> None: diff --git a/src/prime_rl/utils/monitor/multi.py b/src/prime_rl/utils/monitor/multi.py index 043953546d..11c64e1e84 100644 --- a/src/prime_rl/utils/monitor/multi.py +++ b/src/prime_rl/utils/monitor/multi.py @@ -7,7 +7,7 @@ from prime_rl.utils.monitor.prime import PrimeMonitor if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode class MultiMonitor(Monitor): @@ -33,17 +33,17 @@ def log(self, metrics: dict[str, Any], step: int) -> None: except Exception as e: self.logger.warning(f"Failed to log metrics to {monitor.__class__.__name__}: {e}") - def log_samples(self, rollouts: list[Rollout], step: int) -> None: + def log_samples(self, episodes: list[Episode], step: int) -> None: for monitor in self.monitors: try: - monitor.log_samples(rollouts=rollouts, step=step) + monitor.log_samples(episodes=episodes, step=step) except Exception as e: self.logger.warning(f"Failed to log samples to {monitor.__class__.__name__}: {e}") - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: for monitor in self.monitors: try: - monitor.log_eval_samples(rollouts=rollouts, env_name=env_name, step=step) + monitor.log_eval_samples(episodes=episodes, env_name=env_name, step=step) except Exception as e: self.logger.warning(f"Failed to log eval samples to {monitor.__class__.__name__}: {e}") diff --git a/src/prime_rl/utils/monitor/prime.py b/src/prime_rl/utils/monitor/prime.py index eee454a9aa..468a63653d 100644 --- a/src/prime_rl/utils/monitor/prime.py +++ b/src/prime_rl/utils/monitor/prime.py @@ -23,7 +23,7 @@ from prime_rl.utils.monitor.base import Monitor, drop_non_finite_json_values, sample_items_for_logging if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode _SAMPLE_SCHEMA = pa.schema( @@ -248,7 +248,7 @@ def log(self, metrics: dict[str, Any], step: int) -> None: }, ) - def log_samples(self, rollouts: list[Rollout], step: int) -> None: + def log_samples(self, episodes: list[Episode], step: int) -> None: """Logs rollouts to Prime Intellect API using presigned URLs for direct R2 upload.""" if not self.is_master: return @@ -262,21 +262,21 @@ def log_samples(self, rollouts: list[Rollout], step: int) -> None: ): return - rollouts = sample_items_for_logging( - rollouts, + episodes = sample_items_for_logging( + episodes, self.config.log_extras.sample_ratio, ) - if not rollouts: + if not episodes: return assert self.last_log_samples_step <= step, "Step must be greater than last logged step" assert step not in self._pending_sample_steps, f"Step {step} upload already in progress" assert self.logger is not None, "Logger is required for sample logging" - self.logger.info(f"Logging {len(rollouts)} samples to Prime Intellect API at step {step}") + self.logger.info(f"Logging {len(episodes)} episodes to Prime Intellect API at step {step}") start_time = time.perf_counter() - parquet_bytes = self._rollouts_to_parquet_bytes(rollouts, step) + parquet_bytes = self._rollouts_to_parquet_bytes(episodes, step) if not parquet_bytes: self.logger.warning(f"No samples to log at step {step}") @@ -291,8 +291,9 @@ def log_samples(self, rollouts: list[Rollout], step: int) -> None: f"Initiated samples upload at step {step} to Prime Intellect API in {time.perf_counter() - start_time:.2f}s" ) - def _rollouts_to_parquet_bytes(self, rollouts: list[Rollout], step: int) -> bytes | None: - """Convert rollouts to Parquet bytes for upload. One row per rollout. The conversation + def _rollouts_to_parquet_bytes(self, episodes: list[Episode], step: int) -> bytes | None: + """Convert episodes to Parquet bytes for upload. One row per trace, carrying the episode + it belongs to. The conversation is the unit (no prompt/completion split — meaningless mid-branch): `completion` is the last branch's messages and `trajectory` is one message list per branch. Shares `verifiers.v1.utils.platform.trace_to_sample` with verifiers' eval `--push`, so a training-run @@ -301,8 +302,10 @@ def _rollouts_to_parquet_bytes(self, rollouts: list[Rollout], step: int) -> byte now = datetime.now(timezone.utc) rows = [] - for sample_id, rollout in enumerate(rollouts): - sample = trace_to_sample(rollout, rollout_number=sample_id + 1, episode_id=rollout.episode_id or None) + for sample_id, (episode, rollout) in enumerate( + (episode, trace) for episode in episodes for trace in episode.traces + ): + sample = trace_to_sample(rollout, rollout_number=sample_id + 1, episode_id=episode.id) trajectory = sample["trajectory"] if not trajectory: # no branches (e.g. a rollout that errored before any message) continue @@ -326,7 +329,7 @@ def _rollouts_to_parquet_bytes(self, rollouts: list[Rollout], step: int) -> byte "completion": json.dumps(sample["completion"]), "trajectory": json.dumps(trajectory), "answer": "", - "env_name": rollout.env_name, + "env_name": episode.env.name or "", "task": json.dumps(sample["task"]), "info": json.dumps(rollout.info), "reward": sample["reward"], @@ -447,7 +450,7 @@ async def _confirm_samples_upload(self, step: int, s3_key: str, max_retries: int await asyncio.sleep(delay) return False - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: pass def log_distributions(self, distributions: dict[str, list[float]], step: int) -> None: diff --git a/src/prime_rl/utils/monitor/wandb.py b/src/prime_rl/utils/monitor/wandb.py index 7340b3c6de..c87e1274af 100644 --- a/src/prime_rl/utils/monitor/wandb.py +++ b/src/prime_rl/utils/monitor/wandb.py @@ -23,7 +23,7 @@ from prime_rl.utils.monitor.base import Monitor, sample_items_for_logging if TYPE_CHECKING: - from prime_rl.orchestrator.types import Rollout + from prime_rl.orchestrator.types import Episode def _loggable_task(task) -> str: @@ -159,13 +159,32 @@ def init_wandb(max_retries: int): if config is not None and isinstance(config, WandbWithExtrasConfig) and config.log_extras: if config.log_extras.samples: self.last_log_samples_step = -1 - self.samples_cols = ["step", "env_name", "task", "task_idx", "messages", "input_ids", "reward"] + self.samples_cols = [ + "step", + "env_name", + "task", + "task_idx", + "agent", + "branch_idx", + "messages", + "input_ids", + "reward", + ] self.samples_table = wandb.Table( columns=self.samples_cols, log_mode="INCREMENTAL", ) self.tokenizer = tokenizer - self.eval_samples_cols = ["step", "env", "task", "task_idx", "completion", "reward"] + self.eval_samples_cols = [ + "step", + "env", + "task", + "task_idx", + "agent", + "branch_idx", + "completion", + "reward", + ] self.eval_samples_table = wandb.Table( columns=self.eval_samples_cols, log_mode="INCREMENTAL", @@ -189,8 +208,10 @@ def log(self, metrics: dict[str, Any], step: int) -> None: return wandb.log({**metrics, "step": step}) - def log_samples(self, rollouts: list[Rollout], step: int) -> None: - """Logs rollouts to W&B table.""" + def log_samples(self, episodes: list[Episode], step: int) -> None: + """Log a sample of episodes to the W&B table, one row per branch — the agent that ran it + and which of its branches the row is are columns, so a multi-agent episode reads as its + seats rather than as one blurred cell.""" if not self.is_master: return if ( @@ -203,45 +224,47 @@ def log_samples(self, rollouts: list[Rollout], step: int) -> None: # Do not log samples if not enabled or not log interval step return - rollouts = sample_items_for_logging( - rollouts, + episodes = sample_items_for_logging( + episodes, self.config.log_extras.sample_ratio, ) - if not rollouts: + if not episodes: return assert self.tokenizer is not None, "Tokenizer is required for sample logging" assert self.last_log_samples_step <= step, "Step must be greater than last logged step" assert self.logger is not None, "Logger is required for sample logging" - self.logger.info(f"Logging {len(rollouts)} samples to W&B table at step {step}") + self.logger.info(f"Logging {len(episodes)} episodes to W&B table at step {step}") start_time = time.perf_counter() - for rollout in rollouts: - trace = rollout - for branch in trace.branches: - token_ids = branch.token_ids - if not token_ids: - continue - sample = { - "step": step, - "env_name": rollout.env_name, - "task": _loggable_task(trace.task.data), - "task_idx": trace.task.data.idx, - "messages": self.tokenizer.decode(token_ids), - "input_ids": str(token_ids), - "reward": trace.reward, - } - assert list(sample.keys()) == self.samples_cols, ( - "Order of columns in the table must be the same as order of the keys here" - ) - self.samples_table.add_data(*sample.values()) + for episode in episodes: + for trace in episode.traces: + for branch_idx, branch in enumerate(trace.branches): + token_ids = branch.token_ids + if not token_ids: + continue + sample = { + "step": step, + "env_name": episode.env.name or "", + "task": _loggable_task(trace.task.data), + "task_idx": trace.task.data.idx, + "agent": trace.agent.name, + "branch_idx": branch_idx, + "messages": self.tokenizer.decode(token_ids), + "input_ids": str(token_ids), + "reward": trace.reward, + } + assert list(sample.keys()) == self.samples_cols, ( + "Order of columns in the table must be the same as order of the keys here" + ) + self.samples_table.add_data(*sample.values()) wandb.log({"samples": self.samples_table, "step": step}) self.last_log_samples_step = step self.logger.debug(f"Logged samples at step {step} to W&B table in {time.perf_counter() - start_time:.2f}s") - def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> None: + def log_eval_samples(self, episodes: list[Episode], env_name: str, step: int) -> None: """Logs eval rollouts to a separate W&B table.""" if not self.is_master: return @@ -253,23 +276,25 @@ def log_eval_samples(self, rollouts: list[Rollout], env_name: str, step: int) -> ): return - for rollout in rollouts: - trace = rollout - for branch in trace.branches: - # Eval runs the openai client (no token ids), so show the assistant message - # content rather than decoded tokens. - completion = "".join(m.content or "" for m in branch.messages if m.role == "assistant") - if not completion: - continue - sample = { - "step": step, - "env": env_name, - "task": _loggable_task(trace.task.data), - "task_idx": trace.task.data.idx, - "completion": completion, - "reward": trace.reward, - } - self.eval_samples_table.add_data(*sample.values()) + for episode in episodes: + for trace in episode.traces: + for branch_idx, branch in enumerate(trace.branches): + # Eval runs the openai client (no token ids), so show the assistant message + # content rather than decoded tokens. + completion = "".join(m.content or "" for m in branch.messages if m.role == "assistant") + if not completion: + continue + sample = { + "step": step, + "env": env_name, + "task": _loggable_task(trace.task.data), + "task_idx": trace.task.data.idx, + "agent": trace.agent.name, + "branch_idx": branch_idx, + "completion": completion, + "reward": trace.reward, + } + self.eval_samples_table.add_data(*sample.values()) wandb.log({"eval/samples": self.eval_samples_table, "step": step}) diff --git a/tests/unit/orchestrator/test_advantage.py b/tests/unit/orchestrator/test_advantage.py index 9da1e56e21..6e7cb269d8 100644 --- a/tests/unit/orchestrator/test_advantage.py +++ b/tests/unit/orchestrator/test_advantage.py @@ -2,6 +2,7 @@ import pytest import verifiers.v1 as vf +from verifiers.v1.configs.agent import WireAgentConfig from prime_rl.configs.algorithm import ( GRPOAlgoConfig, @@ -10,8 +11,9 @@ ) from prime_rl.orchestrator.algo.grpo import GRPOAlgorithm from prime_rl.orchestrator.algo.max_rl import MaxRLAlgorithm -from prime_rl.orchestrator.trajectories import trace_to_samples -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.algo.routing import stamp_advantages +from prime_rl.orchestrator.trajectories import iter_trainable_branches, trace_to_samples +from prime_rl.orchestrator.types import Episode, TrainRollout def _build_rollout( @@ -21,8 +23,8 @@ def _build_rollout( obs_lengths: list[int] | None = None, env_name: str = "test", metrics: dict | None = None, -) -> Rollout: - """Build a ``Rollout`` (a ``vf.Trace``) as an alternating message graph. +) -> TrainRollout: + """Build a ``TrainRollout`` (a ``vf.Trace``) as an alternating message graph. ``sampled_lengths`` gives the token count of each model turn (a sampled ``AssistantMessage`` node); ``obs_lengths`` (one shorter, if given) gives the @@ -97,9 +99,9 @@ def _take(n: int) -> list[int]: ) parent = len(nodes) - 1 - rollout = Rollout[vf.TaskData]( + rollout = TrainRollout[vf.TaskData]( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=nodes, calls=calls, rewards={"reward": vf.Reward(score=reward)}, @@ -116,8 +118,8 @@ def _make_rollout( num_turns: int = 1, env_name: str = "test", metrics: dict | None = None, -) -> Rollout: - """Build a ``Rollout`` carrying ``completion_len`` model-sampled tokens split +) -> TrainRollout: + """Build a ``TrainRollout`` carrying ``completion_len`` model-sampled tokens split across ``num_turns`` sampled turns. Always carries at least one trainable token so credit broadcasts somewhere.""" num_turns = max(num_turns, 1) @@ -127,8 +129,8 @@ def _make_rollout( return _build_rollout(reward, sampled_lengths=sampled_lengths, env_name=env_name, metrics=metrics) -def _make_group(rewards, completion_lengths=None, num_turns=None) -> list[Rollout]: - """Build one group of ``Rollout``\\ s from 1D arrays of rewards/lengths/turns — +def _make_group(rewards, completion_lengths=None, num_turns=None) -> list[TrainRollout]: + """Build one group of ``TrainRollout``\\ s from 1D arrays of rewards/lengths/turns — exactly what ``score_group`` sees.""" rollouts = [] for i, reward in enumerate(rewards): @@ -138,24 +140,30 @@ def _make_group(rewards, completion_lengths=None, num_turns=None) -> list[Rollou return rollouts -def _scalar(rollout: Rollout) -> float: - """The per-rollout advantage scalar an algorithm assigned — broadcast over - the rollout's trainable (mask-True) tokens, so any trainable position holds it.""" - mask = [m for sample in rollout.samples for m in sample.mask] - return rollout.advantages[mask.index(True)] +def _as_episodes(group: list[TrainRollout]) -> list[Episode]: + """One episode per rollout — the shape a single-agent env produces, and what the + algorithms are handed.""" + return [Episode.model_construct(id=f"e{i}", traces=[rollout]) for i, rollout in enumerate(group)] -def _grpo(group: list[Rollout], length_penalty=None) -> list[float]: +def _scalar(rollout: TrainRollout) -> float: + """The per-rollout advantage scalar an algorithm assigned — broadcast over the rollout's + trainable tokens, so every assigned position holds it.""" + assert rollout.advantages is not None + return rollout.advantages[0] + + +def _grpo(group: list[TrainRollout], length_penalty=None) -> list[float]: """Drive ``GRPOAlgorithm.score_group`` and read back each per-rollout scalar.""" algo = GRPOAlgorithm(GRPOAlgoConfig(length_penalty=length_penalty), policy_pool=None) - asyncio.run(algo.score_group(group)) + asyncio.run(algo.score_group(_as_episodes(group))) return [_scalar(rollout) for rollout in group] -def _max_rl(group: list[Rollout]) -> list[float]: +def _max_rl(group: list[TrainRollout]) -> list[float]: """Drive ``MaxRLAlgorithm.score_group`` and read back each per-rollout scalar.""" algo = MaxRLAlgorithm(MaxRLAlgoConfig(), policy_pool=None) - asyncio.run(algo.score_group(group)) + asyncio.run(algo.score_group(_as_episodes(group))) return [_scalar(rollout) for rollout in group] @@ -217,7 +225,7 @@ def test_linear_context_term_penalizes_more_context(): _build_rollout(1.0, sampled_lengths=[10], obs_lengths=[]), _build_rollout(1.0, sampled_lengths=[10], obs_lengths=[100]), ] - asyncio.run(GRPOAlgorithm(GRPOAlgoConfig(length_penalty=cfg), policy_pool=None).score_group(group)) + asyncio.run(GRPOAlgorithm(GRPOAlgoConfig(length_penalty=cfg), policy_pool=None).score_group(_as_episodes(group))) advs = [_scalar(rollout) for rollout in group] assert advs[0] > advs[1] assert sum(advs) == pytest.approx(0.0, abs=1e-6) @@ -240,23 +248,66 @@ def test_linear_turns_term_penalizes_more_turns(): def test_assign_advantages_broadcasts_scalar(): - """A scalar broadcasts uniformly over the rollout's trainable (mask-True) tokens.""" - rollout = _build_rollout(0.0, sampled_lengths=[2]) - # one user prompt token (masked) + 2 sampled tokens (trainable) + """A scalar broadcasts over the trainable tokens, and only those — the credit is stored per + node against `mask`, so untrainable positions hold nothing rather than a zero.""" + rollout = _build_rollout(0.0, sampled_lengths=[2]) # 1 masked prompt token + 2 trainable rollout.assign_advantages(0.7) - assert rollout.advantages == [0.0, 0.7, 0.7] + assert rollout.advantages == [0.7, 0.7] + # The branch view widens it back to the token sequence the trainer indexes. + (branch, _), *_ = iter_trainable_branches(rollout) + assert branch.advantages == [0.0, 0.7, 0.7] def test_assign_advantages_zeros_non_trainable(): - """Non-trainable (mask=False) positions stay 0.0 under scalar broadcast.""" + """A non-trainable (mask=False) position reads 0.0 in the branch view.""" # prompt(1, masked) + sampled(1) + obs(1, masked): mask is [F, T, F] rollout = _build_rollout(0.0, sampled_lengths=[1], obs_lengths=[1]) rollout.assign_advantages(0.7) - assert rollout.advantages == [0.0, 0.7, 0.0] - - -def test_assign_advantages_rejects_misaligned(): - rollout = _build_rollout(0.0, sampled_lengths=[2]) - # full length is 3 (prompt + 2 sampled); a 1-element list must be rejected - with pytest.raises(ValueError, match="align"): - rollout.assign_advantages([0.5]) + (branch, _), *_ = iter_trainable_branches(rollout) + assert branch.advantages == [0.0, 0.7, 0.0] + + +def test_unassigned_credit_is_not_zero_credit(): + """An unscored rollout is distinguishable from one scored zero, all the way to the sample.""" + unscored = _build_rollout(0.0, sampled_lengths=[2]) + assert unscored.advantages is None and not unscored.is_trainable + scored = _build_rollout(0.0, sampled_lengths=[2]) + scored.assign_advantages(0.0) + assert scored.advantages == [0.0, 0.0] and not scored.is_trainable # scored, but no gradient + + +def test_stamp_advantages_zeros_a_shared_node_in_the_later_branch(): + """A forked node is trained in the first branch containing it; in the later branch it is + context, so its credit must not ride along on that branch's sample.""" + root = vf.MessageNode( + message=vf.AssistantMessage(role="assistant", content=""), + sampled=True, + token_ids=[1, 2], + mask=[True, True], + logprobs=[-0.1, -0.2], + ) + leaves = [ + vf.MessageNode( + parent=0, + message=vf.AssistantMessage(role="assistant", content=""), + sampled=True, + token_ids=[3], + mask=[True], + logprobs=[-0.3], + ) + for _ in range(2) + ] + rollout = TrainRollout[vf.TaskData]( + task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt=None)), + agent=vf.AgentInfo(config=WireAgentConfig()), + nodes=[root, *leaves], + rewards={"reward": vf.Reward(score=0.0)}, + ) + rollout.env_name = "test" + rollout.samples = trace_to_samples(rollout, env_name="test") + rollout.assign_advantages(0.5) + stamp_advantages(rollout) + + first, second = rollout.samples + assert first.advantages == [0.5, 0.5, 0.5] + assert second.advantages == [0.0, 0.0, 0.5] diff --git a/tests/unit/orchestrator/test_algorithms.py b/tests/unit/orchestrator/test_algorithms.py index b2dbb01d05..aa5a2da772 100644 --- a/tests/unit/orchestrator/test_algorithms.py +++ b/tests/unit/orchestrator/test_algorithms.py @@ -4,13 +4,14 @@ import pydantic import pytest import verifiers.v1 as vf +from verifiers.v1.configs.agent import WireAgentConfig from verifiers.v1.graph import MessageNode from verifiers.v1.types import AssistantMessage, ToolMessage, UserMessage from prime_rl.configs.algorithm import AlgoConfig, FrozenModelConfig from prime_rl.orchestrator.algo import EchoAlgorithm, stamp_advantages, stamp_loss_routing -from prime_rl.orchestrator.trajectories import trace_to_samples -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.trajectories import iter_trainable_branches, trace_to_samples +from prime_rl.orchestrator.types import TrainRollout from prime_rl.transport.types import TrainingSample FROZEN = {"name": "org/ref-model", "base_url": ["http://ref:8001/v1"]} @@ -157,36 +158,39 @@ def test_stamp_loss_routing_merges_action_weights_into_ce_stream(): assert sample.ref_kl_weights is None -def _make_rollout( - samples: list[TrainingSample], - advantages: list[float] | None = None, -) -> Rollout: - rollout = Rollout( +def _make_rollout(samples: list[TrainingSample]) -> TrainRollout: + """A rollout whose graph mirrors ``_make_sample``: one node per sample, carrying that + sample's tokens and mask, so branches and samples line up as they do in a real trace.""" + nodes = [ + MessageNode( + parent=None, + message=AssistantMessage(role="assistant", content=""), + sampled=True, + token_ids=list(sample.token_ids), + mask=list(sample.mask), + logprobs=[lp for lp, m in zip(sample.logprobs, sample.mask, strict=True) if m], + ) + for sample in samples + ] + rollout = TrainRollout( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), - nodes=[], + agent=vf.AgentInfo(config=WireAgentConfig()), + nodes=nodes, rewards={}, env_name="test-env", ) rollout.samples = samples - rollout.advantages = advantages return rollout -def test_stamp_advantages_full_length_stream(): - # The advantage stream is full-length-N: 0.0 on prompt + non-trainable - # positions, the rl credit on trainable (mask True) tokens. - rollout = _make_rollout([_make_sample()], advantages=[0.0, 0.0, 0.5, -0.5, 0.0, 1.0]) - stamp_advantages(rollout) - assert rollout.samples[0].advantages == [0.0, 0.0, 0.5, -0.5, 0.0, 1.0] - - -def test_stamp_advantages_slices_across_samples(): - samples = [_make_sample(), _make_sample()] - rollout = _make_rollout(samples, advantages=[1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0]) +def test_stamp_advantages_copies_the_branch_view(): + """Each sample takes its branch's spread credit — no offsets, so the two align by + construction. Covered end to end in test_advantage.py, which builds real nodes.""" + rollout = _make_rollout([_make_sample()]) + rollout.assign_advantages(1.0) stamp_advantages(rollout) - assert rollout.samples[0].advantages == [1.0, 2.0, 3.0, 4.0, 5.0, 6.0] - assert rollout.samples[1].advantages == [7.0, 8.0, 9.0, 10.0, 11.0, 12.0] + (branch, _), *_ = iter_trainable_branches(rollout) + assert rollout.samples[0].advantages == branch.advantages def test_stamp_advantages_no_credit_ships_none(): @@ -195,22 +199,10 @@ def test_stamp_advantages_no_credit_ships_none(): assert rollout.samples[0].advantages is None -def test_stamp_advantages_rejects_misaligned(): - rollout = _make_rollout([_make_sample()], advantages=[0.5]) - with pytest.raises(ValueError, match="align"): - stamp_advantages(rollout) - - def test_assign_advantages_scalar_broadcasts_over_mask(): rollout = _make_rollout([_make_sample()]) rollout.assign_advantages(1.0) - assert rollout.advantages == [0.0, 0.0, 1.0, 1.0, 0.0, 1.0] - - -def test_assign_advantages_list_rejects_misaligned(): - rollout = _make_rollout([_make_sample()]) - with pytest.raises(ValueError, match="align"): - rollout.assign_advantages([0.5]) + assert rollout.advantages == [1.0, 1.0, 1.0] # one per trainable token # -------------------------------------------------------------------------- @@ -244,7 +236,7 @@ def _node(message, *, parent, sampled, token_ids, logprobs=None, is_content=None ) -def _two_turn_rollout(observation_role: str = "tool") -> Rollout: +def _two_turn_rollout(observation_role: str = "tool") -> TrainRollout: """A single linear branch: user prompt, an assistant response, an env-provided observation (tool output / user feedback), then a second assistant response. Tokens: prompt [1,2], action [3,4], observation @@ -259,9 +251,9 @@ def _two_turn_rollout(observation_role: str = "tool") -> Rollout: _node(obs_message, parent=1, sampled=False, token_ids=[5, 6]), _node(AssistantMessage(content="A2"), parent=2, sampled=True, token_ids=[7, 8], logprobs=[-0.3, -0.4]), ] - rollout = Rollout( + rollout = TrainRollout( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=nodes, rewards={"r": vf.Reward(score=1.0)}, env_name="test-env", @@ -311,9 +303,9 @@ def test_echo_weights_only_content_tokens_when_is_content_present(): ), _node(AssistantMessage(content="A2"), parent=2, sampled=True, token_ids=[7, 8], logprobs=[-0.3, -0.4]), ] - rollout = Rollout( + rollout = TrainRollout( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt=None)), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=nodes, rewards={"r": vf.Reward(score=1.0)}, env_name="test-env", diff --git a/tests/unit/orchestrator/test_batch.py b/tests/unit/orchestrator/test_batch.py index 8b4b31f7ed..6bf6a22aaa 100644 --- a/tests/unit/orchestrator/test_batch.py +++ b/tests/unit/orchestrator/test_batch.py @@ -93,18 +93,18 @@ def test_training_sample_requires_env_name(): @pytest.mark.parametrize( - ("rollout_count", "num_train_workers", "expected_batches_per_worker"), [(4, 2, 2), (5, 2, 3), (7, 1, 7), (11, 4, 3)] + ("episodes_owed", "num_train_workers", "expected_batches_per_worker"), [(4, 2, 2), (5, 2, 3), (7, 1, 7), (11, 4, 3)] ) def test_prepare_batch_balances_micro_batches_across_workers( - make_training_example, rollout_count, num_train_workers, expected_batches_per_worker + make_training_example, episodes_owed, num_train_workers, expected_batches_per_worker ): - examples = [make_training_example() for i in range(rollout_count)] + examples = [make_training_example() for i in range(episodes_owed)] batches_per_gpu = prepare_batch( rollouts=examples, seq_len=4, num_train_workers=num_train_workers, - idxs=[0] * rollout_count, + idxs=[0] * episodes_owed, num_loras=1, bin_cost=build_bin_cost(None), ) diff --git a/tests/unit/orchestrator/test_filters.py b/tests/unit/orchestrator/test_filters.py index 69ce76d029..914d753793 100644 --- a/tests/unit/orchestrator/test_filters.py +++ b/tests/unit/orchestrator/test_filters.py @@ -1,7 +1,7 @@ import math -import uuid import verifiers.v1 as vf +from verifiers.v1.configs.agent import WireAgentConfig from prime_rl.configs.orchestrator import GibberishFilterConfig, RepetitionFilterConfig from prime_rl.orchestrator.filters import ( @@ -11,7 +11,7 @@ setup_filter, setup_filters, ) -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.types import TrainRollout def _assistant_node(token_ids: list[int], logprobs: list[float]) -> vf.MessageNode: @@ -46,8 +46,8 @@ def _make_rollout( *, reward: float = 1.0, multi_step: bool = False, -) -> Rollout: - """Build a ``Rollout`` (a message-graph trace) carrying the completion tokens — enough for +) -> TrainRollout: + """Build a ``TrainRollout`` (a message-graph trace) carrying the completion tokens — enough for the filters to inspect each node's sampled tokens / logprobs.""" if multi_step: mid = len(completion_ids) // 2 @@ -57,14 +57,13 @@ def _make_rollout( ] else: nodes = [_assistant_node(completion_ids, completion_logprobs)] - rollout = Rollout[vf.TaskData]( + rollout = TrainRollout[vf.TaskData]( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="")), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=nodes, rewards={"reward": vf.Reward(score=reward)}, ) rollout.env_name = "test" - rollout.group_id = uuid.uuid4() return rollout @@ -140,9 +139,9 @@ def test_gibberish_aligns_logprobs_under_generation_prompt_scaffold(): it; reading the aligned branch streams detects it.""" gibberish_filter = _make_gibberish_filter() - rollout = Rollout[vf.TaskData]( + rollout = TrainRollout[vf.TaskData]( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="")), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=[_scaffold_assistant_node([50, 80, 120_000], [-1.0, -0.5, gibberish_filter.logprob_threshold - 1.0])], rewards={"reward": vf.Reward(score=1.0)}, ) diff --git a/tests/unit/orchestrator/test_metrics.py b/tests/unit/orchestrator/test_metrics.py index 546e594d9a..6b8ba82db6 100644 --- a/tests/unit/orchestrator/test_metrics.py +++ b/tests/unit/orchestrator/test_metrics.py @@ -1,11 +1,24 @@ import math +from dataclasses import replace from itertools import count from types import SimpleNamespace +from uuid import uuid4 import pytest import verifiers.v1 as vf +from prime_rl.orchestrator.eval_sink import eval_step_of from prime_rl.orchestrator.metrics import EvalRollouts, Stat, TrainRollouts +from prime_rl.orchestrator.types import ( + GROUP_ID, + Episode, + InflightEpisode, + Rollout, + TrainRollout, + group_id_of, + rollouts_of, + run_of, +) from prime_rl.orchestrator.utils import compute_pass_metrics _ids = count() @@ -29,7 +42,6 @@ def mk( metrics: dict | None = None, rewards: dict | None = None, env_name: str = "env", - group_id: str = "g0", trainable: bool = True, is_trainable: bool = True, is_filtered: bool = False, @@ -61,7 +73,6 @@ def mk( stop_condition=stop_condition, metrics=metrics or {}, env_name=env_name, - group_id=group_id, agent=SimpleNamespace(trainable=trainable, name=agent_name), is_trainable=is_trainable, is_filtered=is_filtered, @@ -79,8 +90,28 @@ def mk( ) -def train_wandb(rollouts, subset: str = "all") -> dict: - return TrainRollouts(rollouts).metrics.to_wandb(prefix="train/agg", subset=subset) +def ep(*rollouts, env_name: str = "env", errors=(), group_id="g0", cls=Episode): + """One episode over these traces, named and grouped the way the dispatcher stamps it. + ``model_construct`` skips validation so the duck-typed stand-ins above can stand in for real + ones.""" + return cls.model_construct( + id=f"e{next(_ids)}", + traces=list(rollouts), + env=vf.EnvInfo(name=env_name), + info={GROUP_ID: group_id}, + errors=list(errors), + ) + + +def solo(rollouts, group_ids=None, cls=Episode): + """Each rollout as its own single-trace episode — the single-agent shape. ``group_ids`` gives + the example each answers; by default they all answer the same one.""" + ids = group_ids or ["g0"] * len(rollouts) + return [ep(r, group_id=g, cls=cls) for r, g in zip(rollouts, ids, strict=True)] + + +def train_wandb(rollouts, subset: str = "all", group_ids=None) -> dict: + return TrainRollouts(solo(rollouts, group_ids)).metrics.to_wandb(prefix="train/agg", subset=subset) def test_stat(): @@ -93,7 +124,12 @@ def test_stat(): def test_container_effective_by_env_and_listlike(): rc = TrainRollouts( - [mk(env_name="a"), mk(env_name="a", has_error=True), mk(env_name="b", is_filtered=True), mk(env_name="b")] + [ + ep(mk(env_name="a"), env_name="a"), + ep(mk(env_name="a", has_error=True), env_name="a"), + ep(mk(env_name="b", is_filtered=True), env_name="b"), + ep(mk(env_name="b"), env_name="b"), + ] ) assert len(rc) == 4 and [r.env_name for r in rc] == ["a", "a", "b", "b"] # sized + iterable eff = rc.effective @@ -101,16 +137,18 @@ def test_container_effective_by_env_and_listlike(): assert all(not r.has_error and not r.is_filtered and r in rc.rollouts for r in eff) # view of references by_env = rc.by_env() assert set(by_env) == {"a", "b"} and len(by_env["a"]) == 2 and isinstance(by_env["a"], TrainRollouts) - rc.append(mk()) - assert len(rc) == 5 + rc.append(ep(mk())) + assert len(rc) == 5 # sized by rollout, appended by episode def test_to_wandb_distributions(): m = TrainRollouts( - [ - mk(reward=1.0, num_total_tokens=10, num_input_tokens=4), - mk(reward=0.0, num_total_tokens=20, num_input_tokens=6), - ] + solo( + [ + mk(reward=1.0, num_total_tokens=10, num_input_tokens=4), + mk(reward=0.0, num_total_tokens=20, num_input_tokens=6), + ] + ) ).metrics assert m.num_input_tokens.mean() == 5.0 # fluent Stat access out = m.to_wandb(prefix="train/agg", subset="all") @@ -124,15 +162,20 @@ def test_to_wandb_distributions(): def test_episode_and_agent_levels(): # Two proposer-solver episodes: one proposer + two solvers each (the solver fan-out). - rollouts = [ - mk(reward=1.0, num_turns=1, agent_name="proposer", episode_id="e1"), - mk(reward=0.0, num_turns=2, agent_name="solver", episode_id="e1"), - mk(reward=1.0, num_turns=4, agent_name="solver", episode_id="e1"), - mk(reward=0.0, num_turns=3, agent_name="proposer", episode_id="e2"), - mk(reward=1.0, num_turns=6, agent_name="solver", episode_id="e2"), - mk(reward=0.0, num_turns=8, agent_name="solver", episode_id="e2"), - ] - m = TrainRollouts(rollouts).metrics + m = TrainRollouts( + [ + ep( + mk(reward=1.0, num_turns=1, agent_name="proposer"), + mk(reward=0.0, num_turns=2, agent_name="solver"), + mk(reward=1.0, num_turns=4, agent_name="solver"), + ), + ep( + mk(reward=0.0, num_turns=3, agent_name="proposer"), + mk(reward=1.0, num_turns=6, agent_name="solver"), + mk(reward=0.0, num_turns=8, agent_name="solver"), + ), + ] + ).metrics assert m.num_turns.mean() == 12.0 # episode-level sums: 1+2+4 and 3+6+8 assert m.num_total_tokens.values == [30.0, 30.0] # summed across the episode's traces out = m.to_wandb(prefix="train/agg", subset="all") @@ -150,11 +193,11 @@ def test_agent_metrics_are_flat_over_traces(): """Inside a seat the trace is the unit of aggregation, so an uneven fan-out (one solver trace from this episode, three from that) never reweights anything: every agent-level metric is the plain figure over that agent's rollouts.""" - rollouts = [ - mk(agent_name="solver", episode_id="e1", is_truncated=True, reward=1.0), - *[mk(agent_name="solver", episode_id="e2", is_truncated=False, reward=0.0) for _ in range(3)], + episodes = [ + ep(mk(agent_name="solver", is_truncated=True, reward=1.0)), + ep(*[mk(agent_name="solver", is_truncated=False, reward=0.0) for _ in range(3)]), ] - out = TrainRollouts(rollouts).metrics.to_wandb(prefix="train/agg", subset="all") + out = TrainRollouts(episodes).metrics.to_wandb(prefix="train/agg", subset="all") assert out["train/agg/all/solver/is_truncated/mean"] == 0.25 # 1 of 4 traces, not (1.0 + 0.0) / 2 assert out["train/agg/all/solver/is_completed/mean"] == 1.0 assert out["train/agg/all/solver/is_trainable/mean"] == 1.0 # sibling rates agree @@ -162,7 +205,9 @@ def test_agent_metrics_are_flat_over_traces(): def test_boolean_rates_and_error_breakdown_all_only(): - rc = TrainRollouts([mk(is_truncated=True), mk(has_error=True, error_type="ProviderError"), mk(is_filtered=True)]) + rc = TrainRollouts( + solo([mk(is_truncated=True), mk(has_error=True, error_type="ProviderError"), mk(is_filtered=True)]) + ) out = rc.metrics.to_wandb(prefix="train/agg", subset="all") assert out["train/agg/all/agent/is_truncated/mean"] == 1 / 3 assert out["train/agg/all/agent/is_completed/mean"] == 1.0 @@ -176,7 +221,9 @@ def test_boolean_rates_and_error_breakdown_all_only(): def test_solve_rates(): groups = {"A": [1.0, 1.0], "B": [0.0, 0.0], "C": [1.0, 0.0], "D": [1.0, 0.0]} # all / none / some / some - out = train_wandb([mk(reward=r, group_id=g) for g, rs in groups.items() for r in rs]) + out = train_wandb( + [mk(reward=r) for _, rs in groups.items() for r in rs], group_ids=[g for g, rs in groups.items() for _ in rs] + ) rates = ( out["train/agg/all/agent/solved_all"], out["train/agg/all/agent/solved_none"], @@ -200,7 +247,7 @@ def test_nested_metrics_and_rewards(): # scoring failed after seeding: unscored (None) entries count as 0.0 on `all` mk(has_error=True, metrics={"acc": None}, rewards={"correct": None, "format": None}), ] - rc = TrainRollouts(rollouts) + rc = TrainRollouts(solo(rollouts)) m = rc.metrics agent = m.by_agent()["agent"] assert agent.metrics["acc"].mean() == pytest.approx(4 / 3) # nested group access @@ -215,13 +262,17 @@ def test_nested_metrics_and_rewards(): assert eff["train/agg/effective/agent/rewards/format/mean"] == 0.5 # cross-env agg: another env's unscored trace carries different keys, so it can't dilute these other = mk(env_name="other", has_error=True, rewards={"solved": None}) - agg = TrainRollouts(rollouts + [other]).metrics.to_wandb(prefix="train/agg", subset="all") + agg = TrainRollouts(solo(rollouts) + [ep(other, env_name="other")]).metrics.to_wandb( + prefix="train/agg", subset="all" + ) assert agg["train/agg/all/agent/rewards/format/mean"] == pytest.approx(1 / 3) assert agg["train/agg/all/agent/rewards/solved/mean"] == 0.0 def test_nested_timing(): - m = TrainRollouts([mk(setup=1.0, agent=2.0, agent_model=1.5, agent_harness=0.5, finalize=0.5, scoring=0.5)]).metrics + m = TrainRollouts( + solo([mk(setup=1.0, agent=2.0, agent_model=1.5, agent_harness=0.5, finalize=0.5, scoring=0.5)]) + ).metrics timing = m.by_agent()["agent"].timing assert timing.setup.mean() == 1.0 and timing.total.mean() == 4.0 # total sums all four phases assert timing.agent_model.mean() == 1.5 and timing.agent_harness.mean() == 0.5 @@ -242,12 +293,12 @@ def test_train_only_metrics_absent_from_eval(): assert out["train/agg/all/agent/is_filtered/mean"] == 0.5 assert out["train/agg/all/agent/filters/gibberish/mean"] == 0.5 assert "train/agg/all/is_trainable/mean" not in out # pipeline verdicts are per-trace - eval_out = EvalRollouts(rollouts).metrics.to_wandb(prefix="eval/x", subset="all") + eval_out = EvalRollouts(solo(rollouts)).metrics.to_wandb(prefix="eval/x", subset="all") assert not any("is_trainable" in k or "is_filtered" in k or "/filters/" in k for k in eval_out) def test_eval_avg_at_k_and_pass_k(): - binary = EvalRollouts([mk(reward=1.0, group_id="g0"), mk(reward=0.0, group_id="g0")]) + binary = EvalRollouts(solo([mk(reward=1.0), mk(reward=0.0)])) eff = binary.effective.metrics.to_wandb(prefix="eval/x", subset="effective") assert eff["eval/x/effective/agent/avg@2"] == 0.5 # mean reward under avg@ (k from the groups) assert "eval/x/effective/avg@2" not in eff # scores are per-agent, never pooled @@ -255,7 +306,7 @@ def test_eval_avg_at_k_and_pass_k(): all_out = binary.metrics.to_wandb(prefix="eval/x", subset="all") assert all_out["eval/x/all/agent/avg@2"] == 0.5 assert not any("pass@" in k or "pass^" in k for k in all_out) # pass@k effective-only - non_binary = EvalRollouts([mk(reward=0.5, group_id="g0"), mk(reward=1.0, group_id="g0")]) + non_binary = EvalRollouts(solo([mk(reward=0.5), mk(reward=1.0)])) assert not any("pass@" in k for k in non_binary.effective.metrics.to_wandb(prefix="eval/x", subset="effective")) @@ -265,3 +316,83 @@ def test_compute_pass_metrics_matches_closed_form(): assert out["pass@2"] == 1.0 - math.comb(2, 2) / math.comb(4, 2) assert out["pass^2"] == math.comb(2, 2) / math.comb(4, 2) assert set(out) == {"pass@1", "pass@2", "pass@4", "pass^1", "pass^2", "pass^4"} + + +def test_traceless_episode_keeps_its_reason(): + """The failure needs no type of its own: ``failed`` is just "no traces", and the reason is the + error vf (or the dispatcher) already put on the episode.""" + episode = ep(errors=[vf.Error(type="Cancelled", message="Off-policy cancel")]) + assert not episode.traces and rollouts_of(episode) == [] + assert episode.last_error is not None and episode.last_error.type == "Cancelled" + assert ep(mk()).traces + # A traceless episode still counts as an episode, but contributes no rollouts. + pool = TrainRollouts([ep(mk()), episode]) + assert len(pool) == 1 and len(pool.episodes) == 2 + assert len(pool.effective.episodes) == 1 # it survives nothing, so the subset drops it + + +def test_training_state_is_train_only(): + """Only a train rollout carries trainer-bound state; an eval trace has no field for it. + Credit is not among them — it lives on the graph's nodes, which every trace has.""" + assert {"samples", "is_filtered", "filter_results"} <= set(TrainRollout.model_fields) + assert "advantages" not in TrainRollout.model_fields # derived from the nodes + # An eval trace is the env's own, unextended — where a trace sits is the episode's to say, + # and it is the same wire specialization the episode's ``traces`` hold. + assert Rollout is vf.WireTrace + assert not {"samples", "is_filtered", "filter_results", "group_id", "episode_id"} & set(Rollout.model_fields) + + +def test_inflight_episode_stamps_what_lands(): + """The dispatch and the landed episode are one pair: `stamp` is the only place the facts of a + dispatch become facts of an episode, and the run it writes is what tells the rest of the + orchestrator which path the episode is on.""" + wire = vf.WireEpisode.model_construct(id="e", traces=[]) + inflight = InflightEpisode(kind="train", env_name="rt", group_id=uuid4(), policy_version=3, episodes_owed=1) + span = vf.PolicySpan(start=3, end=4) # an update landed while it was generating + train = inflight.stamp(wire, run_id="r", policy=span, eval_step=None) + run = run_of(train) + assert isinstance(run.metadata, vf.TrainMetadata) and run.metadata.policy == span + assert run.metadata.step is None # the batch window it lands in is not known yet + assert run.metadata.off_policy_steps is None # so there is nothing to be behind yet + assert train.env.name == "rt" and group_id_of(train) is not None + + run.metadata.step = 6 # the window it landed in, which step 6 trains v5 from + assert run.metadata.off_policy_steps == 2 and run.metadata.policy.drift == 1 + + evaluation = replace(inflight, kind="eval").stamp( + vf.WireEpisode.model_construct(id="e", traces=[]), run_id="r", policy=span, eval_step=12 + ) + # An online eval belongs to the same training run — same id, told apart by its metadata. Its + # off-policy reading is what drifted under it, not its distance from a step it was dispatched + # at: nothing trains on an eval, and a slow one would otherwise look on-policy forever. + eval_meta = run_of(evaluation).metadata + assert isinstance(eval_meta, vf.EvalMetadata) and eval_meta.step == 12 + assert run_of(evaluation).id == run.id and eval_meta.off_policy_steps == span.drift == 1 + with pytest.raises(AssertionError): # an eval episode without its step is not representable + replace(inflight, kind="eval").stamp(wire, run_id="r", policy=span, eval_step=None) + + +def test_eval_sink_reads_the_epoch_a_stamped_episode_landed_in(): + """The eval sink places an episode by the step on its run. Regression: online eval moved onto + the training run (kind="eval"), and nothing unit-tested the sink, so the sink kept asserting on + the old record and only a live run caught it.""" + inflight = InflightEpisode( + kind="eval", env_name="rt", group_id=uuid4(), policy_version=3, episodes_owed=1, eval_step=12 + ) + landed = inflight.stamp(vf.WireEpisode.model_construct(id="e", traces=[]), run_id="r", policy=None, eval_step=12) + assert eval_step_of(landed) == 12 + + trained_on = replace(inflight, kind="train").stamp( + vf.WireEpisode.model_construct(id="e2", traces=[]), run_id="r", policy=None, eval_step=None + ) + with pytest.raises(AssertionError): # an episode to train on is not placed by an eval epoch + eval_step_of(trained_on) + + +def test_empty_is_not_the_same_as_failed(): + """An episode can error and still have traces — that failure rides the traces and must not be + counted as an episode that produced nothing.""" + errored = ep(mk(has_error=True), errors=[vf.Error(type="EnvError", message="boom")]) + assert errored.traces and not errored.ok # failed, but something came back + # the failure rides the trace, where the seat's rate sees it + assert rollouts_of(errored)[0].has_error diff --git a/tests/unit/utils/test_prime_monitor.py b/tests/unit/utils/test_prime_monitor.py index 798e041885..e83ef19800 100644 --- a/tests/unit/utils/test_prime_monitor.py +++ b/tests/unit/utils/test_prime_monitor.py @@ -4,8 +4,9 @@ import pyarrow.parquet as pq import verifiers.v1 as vf +from verifiers.v1.configs.agent import WireAgentConfig -from prime_rl.orchestrator.types import Rollout +from prime_rl.orchestrator.types import TrainRollout from prime_rl.utils.monitor.prime import PrimeMonitor @@ -15,8 +16,8 @@ def _new_monitor() -> PrimeMonitor: return monitor -def _build_rollout(*, example_id: int, reward: float, task: str) -> Rollout: - """Build a v1 ``Rollout`` (message-graph trace). The user node carries the prompt and the +def _build_rollout(*, example_id: int, reward: float, task: str) -> TrainRollout: + """Build a v1 ``TrainRollout`` (message-graph trace). The user node carries the prompt and the assistant node the completion; ``_rollouts_to_parquet_bytes`` reads the conversation off the branches (its ``completion`` column is the last branch's messages, ``trajectory`` is one message list per branch).""" @@ -35,27 +36,30 @@ def _build_rollout(*, example_id: int, reward: float, task: str) -> Rollout: sampled=True, ), ] - rollout = Rollout[vf.TaskData]( + rollout = TrainRollout[vf.TaskData]( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=example_id, prompt=f"prompt-{example_id}")), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), nodes=nodes, rewards={"reward": vf.Reward(score=reward)}, ) rollout.env_name = task - # Per-token advantage stream (full-length-N): 0.0 on the 3 prompt tokens, - # reward/2 on the 2 completion (mask-True) tokens. - rollout.advantages = [0.0, 0.0, 0.0, reward / 2, reward / 2] + rollout.assign_advantages(reward / 2) # over the 2 completion (mask-True) tokens return rollout +def _episode(*rollouts: TrainRollout, env_name: str = "task-a") -> vf.WireEpisode: + """The unit the monitor uploads: one episode, whose traces become the rows.""" + return vf.WireEpisode.model_construct(traces=list(rollouts), env=vf.EnvInfo(name=env_name)) + + def test_rollouts_to_parquet_bytes_preserves_all_rollouts_and_ids(): monitor = _new_monitor() monitor.run_id = "run-123" parquet_bytes = monitor._rollouts_to_parquet_bytes( [ - _build_rollout(example_id=101, reward=1.0, task="task-a"), - _build_rollout(example_id=202, reward=0.0, task="task-b"), + _episode(_build_rollout(example_id=101, reward=1.0, task="task-a")), + _episode(_build_rollout(example_id=202, reward=0.0, task="task-b"), env_name="task-b"), ], step=7, ) @@ -81,14 +85,14 @@ def test_rollouts_to_parquet_bytes_skips_rollouts_without_trajectory(): monitor.run_id = "run-456" rollout_with_branches = _build_rollout(example_id=1, reward=1.0, task="task-a") - rollout_without_branches = Rollout[vf.TaskData]( + rollout_without_branches = TrainRollout[vf.TaskData]( task=vf.TraceTask(type="Task", data=vf.TaskData(idx=2, prompt="missing-trajectory")), - agent=vf.AgentInfo(config=vf.AgentConfig()), + agent=vf.AgentInfo(config=WireAgentConfig()), ) assert rollout_without_branches.branches == [] parquet_bytes = monitor._rollouts_to_parquet_bytes( - [rollout_with_branches, rollout_without_branches], + [_episode(rollout_with_branches, rollout_without_branches)], step=3, )