Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions docs/algorithms.md
Original file line number Diff line number Diff line change
Expand Up @@ -157,9 +157,9 @@ Algorithms operate on native verifier artifacts and annotate their message graph
- `async score_episode(episode)` — rollout-local scoring as one episode arrives.
- `async score_group(episodes)` — group-relative scoring after the cohort completes.

The pipeline drives these through `finalize_episode` and `finalize_group`. Advantages, reference logprobs, and named loss weights stay on verifier nodes through admission. Only admitted traces are flattened into `TrainingSample`s.
The pipeline calls `score_group` directly and `score_episode` through `finalize_episode`, which skips episodes with no trainable sampled tokens. Advantages, reference logprobs, and named loss weights stay on verifier nodes through admission. Only admitted traces are flattened into `TrainingSample`s.

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 clients (`self.clients`). 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.train.algo.advantage`.
The algorithm config states which loss component its action tokens feed (`action_loss_type`, a class variable on each config class). Every class is constructed with its algorithm config plus the one host-owned resource it can't rebuild — the live policy clients (`self.clients`). 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). 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.train.algo.advantage`.

## Async / Off-Policy Training

Expand Down
1 change: 0 additions & 1 deletion src/prime_rl/orchestrator/algo/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,6 @@ class owns its scoring hooks

def build_algorithm(config: AlgoConfig, clients: InferenceClient) -> Algorithm:
cls = ALGORITHM_CLASSES[config.type]
assert cls.action_loss_type == config.action_loss_type # config and runtime declare in two places
# The Algorithm is the runtime of the algorithm config's training signal
# (its sibling GenerationSource interprets the sampling half). Every algorithm is
# handed the live policy pool — opsd self-distills against it, others may
Expand Down
11 changes: 3 additions & 8 deletions src/prime_rl/orchestrator/algo/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@

from __future__ import annotations

from typing import TYPE_CHECKING, ClassVar
from typing import TYPE_CHECKING

import verifiers.v1 as vf

from prime_rl.configs.algorithm import ActionLossType, AlgoConfig, FrozenModelConfig
from prime_rl.configs.algorithm import AlgoConfig, FrozenModelConfig
from prime_rl.utils.logger import get_logger

if TYPE_CHECKING:
Expand Down Expand Up @@ -50,9 +50,8 @@ class Algorithm:
annotated traces into transport samples only after admission.
"""

action_loss_type: ClassVar[ActionLossType] = "rl"

def __init__(self, config: AlgoConfig, clients: InferenceClient):
self.action_loss_type = config.action_loss_type
self.clients = clients
self.connected: InferenceClient | None = None

Expand All @@ -74,7 +73,3 @@ async def finalize_episode(self, episode: vf.Episode) -> None:
"""Run rollout-local scoring when the episode has trainable traces."""
if any(True for _ in iter_trainable_traces([episode])):
await self.score_episode(episode)

async def finalize_group(self, episodes: list[vf.Episode]) -> None:
"""Run group-relative scoring over the native episodes."""
await self.score_group(episodes)
2 changes: 0 additions & 2 deletions src/prime_rl/orchestrator/algo/opd.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,6 @@ class OPDAlgorithm(Algorithm):
credit is assigned — rollouts keep ``advantages=None`` and samples ship no
advantage stream; ``group_size`` only fans out sampling."""

action_loss_type = "ref_kl"

def __init__(self, config: OPDAlgoConfig, clients: InferenceClient):
super().__init__(config, clients)
self.teacher = config.teacher
Expand Down
2 changes: 0 additions & 2 deletions src/prime_rl/orchestrator/algo/opsd.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,6 @@ class OPSDAlgorithm(Algorithm):
multimodal prompts and any number of turns. No scalar advantage is
assigned."""

action_loss_type = "ref_kl"

def __init__(self, config: OPSDAlgoConfig, clients: InferenceClient):
super().__init__(config, clients)
self.demo_key = config.demo_key
Expand Down
2 changes: 0 additions & 2 deletions src/prime_rl/orchestrator/algo/sft.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,3 @@ class SFTDistillAlgorithm(Algorithm):
Assigns no advantage — the ``ce`` loss ignores credit, and SFT trains on
every sampled token. A curriculum can reject results using reward or any
other finalized rollout data."""

action_loss_type = "ce"
2 changes: 1 addition & 1 deletion src/prime_rl/orchestrator/train_sink.py
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ async def process_group(self, group_id: str) -> None:

survivors = [trace for _, trace in iter_trainable_traces(group)]
if survivors:
await env.algorithm.finalize_group(group)
await env.algorithm.score_group(group)
admitted = self._admit(group) if group else False
if not survivors or not admitted:
self.pending_episodes.extend(group, admitted=admitted)
Expand Down
Loading