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
Original file line number Diff line number Diff line change
Expand Up @@ -527,8 +527,8 @@ class NIXLWeightBroadcastConfig(InMemoryWeightBroadcastConfig):

class ConcurrencyConfig(BaseConfig):
"""Adaptive in-flight concurrency control. The orchestrator sizes the
in-flight episode cap from engine KV capacity and learned per-env episode
costs; these fields only bound and seed it."""
in-flight episode cap from engine KV pressure; these fields only bound and
seed it."""

initial_inflight: int | None = Field(None, ge=1)
"""Optional initial in-flight episodes to start from. Set it when a good value is known to skip the initial ramp; otherwise auto-derive a pessimistic bound at runtime."""
Expand Down
7 changes: 0 additions & 7 deletions src/prime_rl/eval/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,10 +34,6 @@
from prime_rl.orchestrator.eval_source import EvalSource
from prime_rl.orchestrator.inference_metrics import InferenceMetricsCollector
from prime_rl.orchestrator.metrics import dispatch_failure_metrics
from prime_rl.orchestrator.patches import (
monkey_patch_chat_completion_logprobs,
monkey_patch_oai_iterable_types,
)
from prime_rl.orchestrator.periodic_logger import PeriodicLogger
from prime_rl.orchestrator.types import DispatchFailure, EvalBatch, GroupCancellation, Policy
from prime_rl.orchestrator.utils import (
Expand All @@ -49,9 +45,6 @@
from prime_rl.utils.logger import format_time, get_logger
from prime_rl.utils.pathing import get_config_dir

monkey_patch_oai_iterable_types()
monkey_patch_chat_completion_logprobs()

# How often ``run_epoch`` re-checks for a superseding checkpoint while it waits for episodes.
POLL_INTERVAL_S = 2.0

Expand Down
4 changes: 0 additions & 4 deletions src/prime_rl/inference/vllm/worker/filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,6 @@
class FileSystemWeightUpdateWorker(Worker):
"""vLLM worker extension for updating weights in-place using shared filesystem."""

def init_broadcaster(self) -> None:
"""Initialize the broadcaster."""
...

def liveness_probe(self) -> None:
"""No-op RPC used by the API server liveness endpoint."""
return None
Expand Down
2 changes: 1 addition & 1 deletion src/prime_rl/orchestrator/concurrency.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,7 @@ def bind(

# ── inbound hooks ────────────────────────────────────────────────────────

def record_episode(self, env_name: str, kind: str, tokens: int, duration: float) -> None:
def record_episode(self, tokens: int) -> None:
"""One completed unit (from the dispatcher). Advances the turnover
clock and, while the last poll was green, grows the cap — so growth
is paced by the pipeline itself at any concurrency scale."""
Expand Down
8 changes: 3 additions & 5 deletions src/prime_rl/orchestrator/dispatcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ def __init__(
max_off_policy_steps: int,
run_id: str,
run_name: str | None,
on_episode_complete: Callable[[str, str, int, float], None] | None = None,
on_episode_complete: Callable[[int], None] | None = None,
) -> None:
self.policy = policy
self.progress = progress
Expand All @@ -157,7 +157,7 @@ def __init__(
self.max_off_policy_steps = max_off_policy_steps
self.run_id = run_id
self.run_name = run_name
# ``(env_name, kind, total_tokens, duration_s)`` per completed episode
# Called with ``total_tokens`` per completed episode
self.on_episode_complete = on_episode_complete

# Starting value of the dynamic cap (the concurrency controller moves
Expand Down Expand Up @@ -694,9 +694,7 @@ async def handle_completed_request(self, task: asyncio.Task) -> None:
if not episode.ok and not episode.traces:
self.metrics.record_error(kind=meta.kind, env_name=meta.env_name)
if self.on_episode_complete is not None and meta.started_at > 0:
self.on_episode_complete(
meta.env_name, meta.kind, episode.num_total_tokens, time.monotonic() - meta.started_at
)
self.on_episode_complete(episode.num_total_tokens)
await self.emit_episode(meta, group, episode)

def complete_group_member(self, meta: InflightEpisode, group: GroupState | None) -> int:
Expand Down
12 changes: 2 additions & 10 deletions src/prime_rl/orchestrator/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,10 +48,6 @@
from prime_rl.orchestrator.inference_metrics import InferenceMetricsCollector
from prime_rl.orchestrator.metrics import TrainEpisodes, dispatch_failure_metrics
from prime_rl.orchestrator.packing import BatchPacker
from prime_rl.orchestrator.patches import (
monkey_patch_chat_completion_logprobs,
monkey_patch_oai_iterable_types,
)
from prime_rl.orchestrator.periodic_logger import PeriodicLogger
from prime_rl.orchestrator.train_sink import TrainSink
from prime_rl.orchestrator.train_source import TrainSource
Expand All @@ -78,12 +74,8 @@
from prime_rl.utils.async_utils import EventLoopLagMonitor, EventLoopLagStats, safe_cancel
from prime_rl.utils.heartbeat import Heartbeat
from prime_rl.utils.logger import format_time, get_logger, setup_logger
from prime_rl.utils.pathing import get_broadcast_dir, get_config_dir
from prime_rl.utils.utils import clean_exit, resolve_latest_ckpt_step

monkey_patch_oai_iterable_types()
monkey_patch_chat_completion_logprobs()

from prime_rl.utils.pathing import get_broadcast_dir, get_config_dir, resolve_latest_ckpt_step
from prime_rl.utils.utils import clean_exit

# Wall-clock budget for post-training cleanup; force-exit if graceful
# shutdown wedges (env-server ZMQ recv, vLLM admin aclose, etc)
Expand Down
115 changes: 0 additions & 115 deletions src/prime_rl/orchestrator/patches.py

This file was deleted.

2 changes: 1 addition & 1 deletion src/prime_rl/trainer/ckpt.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
from prime_rl.trainer.optim import OffloadOptimizer, OptimizerLike
from prime_rl.trainer.world import get_world
from prime_rl.utils.logger import format_time, get_logger
from prime_rl.utils.utils import get_all_ckpt_steps, get_ckpt_dir, get_step_path
from prime_rl.utils.pathing import get_all_ckpt_steps, get_ckpt_dir, get_step_path


@dataclass
Expand Down
2 changes: 0 additions & 2 deletions src/prime_rl/trainer/models/glm4_moe/modeling_glm4_moe.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
from transformers.modeling_outputs import BaseModelOutputWithPast
from transformers.processing_utils import Unpack
from transformers.utils import TransformersKwargs, auto_docstring
from transformers.utils.deprecation import deprecate_kwarg

from prime_rl.trainer.models.base import PreTrainedModelPrimeRL
from prime_rl.trainer.models.glm4_moe.configuration_glm4_moe import Glm4MoeConfig
Expand Down Expand Up @@ -89,7 +88,6 @@ def __init__(self, config: Glm4MoeConfig, layer_idx: int):
self.input_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))
self.post_attention_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))

@deprecate_kwarg("past_key_value", new_name="past_key_values", version="4.58")
def forward(
self,
hidden_states: torch.Tensor,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
from transformers.modeling_outputs import BaseModelOutputWithPast
from transformers.processing_utils import Unpack
from transformers.utils import TransformersKwargs, auto_docstring
from transformers.utils.deprecation import deprecate_kwarg

from prime_rl.trainer.models.base import PreTrainedModelPrimeRL
from prime_rl.trainer.models.glm_moe_dsa.configuration_glm_moe_dsa import GlmMoeDsaConfig, _index_cache_skip_topk
Expand Down Expand Up @@ -87,7 +86,6 @@ def __init__(self, config: GlmMoeDsaConfig, layer_idx: int):
self.input_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))
self.post_attention_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))

@deprecate_kwarg("past_key_value", new_name="past_key_values", version="4.58")
def forward(
self,
hidden_states: torch.Tensor,
Expand Down
2 changes: 0 additions & 2 deletions src/prime_rl/trainer/models/llama/modeling_llama.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
from transformers.models.llama.configuration_llama import LlamaConfig
from transformers.processing_utils import Unpack
from transformers.utils import TransformersKwargs, auto_docstring, can_return_tuple, logging
from transformers.utils.deprecation import deprecate_kwarg

from prime_rl.trainer.models.base import PreTrainedModelPrimeRL
from prime_rl.trainer.models.layers.attn import ATTN_IMPL2CLASS, AttentionConfig
Expand Down Expand Up @@ -71,7 +70,6 @@ def __init__(self, config: LlamaConfig, layer_idx: int):
self.input_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))
self.post_attention_layernorm = RMSNorm(RMSNormConfig(hidden_size=config.hidden_size, eps=config.rms_norm_eps))

@deprecate_kwarg("past_key_value", new_name="past_key_values", version="4.58")
def forward(
self,
hidden_states: torch.Tensor,
Expand Down
3 changes: 2 additions & 1 deletion src/prime_rl/trainer/rl/train.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,8 @@
from prime_rl import monitors
from prime_rl.utils.config import cli
from prime_rl.utils.process import set_proc_title
from prime_rl.utils.utils import clean_exit, resolve_latest_ckpt_step
from prime_rl.utils.pathing import resolve_latest_ckpt_step
from prime_rl.utils.utils import clean_exit


@clean_exit
Expand Down
19 changes: 0 additions & 19 deletions src/prime_rl/utils/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,33 +2,14 @@
import functools
import importlib
import math
import os
import sys
from collections import defaultdict
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Callable

import wandb

from prime_rl.utils.logger import get_logger

# TODO: Change all imports to use utils.pathing
# ruff: noqa: F401
from prime_rl.utils.pathing import (
get_all_ckpt_steps,
get_batch_dir,
get_broadcast_dir,
get_ckpt_dir,
get_eval_dir,
get_file_monitor_dir,
get_log_dir,
get_step_path,
resolve_latest_ckpt_step,
sync_wait_for_path,
wait_for_path,
)


def import_object(dotted_path: str) -> Any:
"""Import an object from a dotted path like 'my_module.submodule.MyClass'."""
Expand Down
1 change: 0 additions & 1 deletion tests/unit/inference/test_serving_chat_with_tokens.py

This file was deleted.

Loading