From 41abd29b9355b164769165288b7949d15bbce764 Mon Sep 17 00:00:00 2001 From: youhengfei <474029121@qq.com> Date: Thu, 22 Oct 2026 20:20:34 +0800 Subject: [PATCH 1/2] [Bugfix][M9] Restore model-neutral PD coordinator metadata PDEngine rebuilt every worker ready payload as OuroConfig, so any non-Ouro family failed at coordinator startup (Nanbeige TypeError num_loops, Huginn TypeError n_embd). The coordinator only validates vocab_size, total_ut_steps and max_position_embeddings in add_request. Replace the OuroConfig reconstruction with a coordinator-scoped PDModelInfo adopted from the ready report, ignoring family-specific fields and failing fast at startup when required limits are missing or mistyped. Drop the OuroConfig import. Add CPU regression tests covering the helper, the real PDEngine startup path on a Nanbeige payload, and request validation against Nanbeige limits. Signed-off-by: youhengfei <474029121@qq.com> --- tests/test_pd.py | 81 +++++++++++++++++++++++++++++++++++++++++++ vllm_rlt/pd/engine.py | 26 ++++++++++++-- 2 files changed, 105 insertions(+), 2 deletions(-) diff --git a/tests/test_pd.py b/tests/test_pd.py index 496ed968..117d6c35 100644 --- a/tests/test_pd.py +++ b/tests/test_pd.py @@ -105,6 +105,87 @@ def test_pd_configuration_validation(options): PDConfig(**options) +def test_pd_coordinator_model_metadata_is_model_neutral(): + from dataclasses import asdict + + from tests.helpers import tiny_nanbeige_config + from vllm_rlt.models import HuginnConfig + from vllm_rlt.pd.engine import PDModelInfo + + def ready(config): + return dict(stage_id="stage-0", model=asdict(config)) + + for config in (tiny_ouro_config(), tiny_nanbeige_config()): + info = PDModelInfo.from_ready(ready(config)) + assert info == PDModelInfo( + config.vocab_size, config.total_ut_steps, config.max_position_embeddings + ) + # Huginn derives loop/context limits through properties, which asdict omits + # from the ready payload: fail at startup, not at request validation. + with pytest.raises(ValueError, match="model metadata lacks"): + PDModelInfo.from_ready(ready(HuginnConfig())) + payload = ready(tiny_ouro_config()) + payload["model"]["vocab_size"] = True + with pytest.raises(ValueError, match="model metadata lacks"): + PDModelInfo.from_ready(payload) + + +def test_pd_nanbeige_startup_uses_worker_limits_for_request_validation(monkeypatch): + from dataclasses import asdict + from unittest.mock import Mock + + from tests.helpers import tiny_nanbeige_config + from vllm_rlt.pd.engine import PDEngine + + config = tiny_nanbeige_config() + + def ready_client(model, *, stage_config, **options): + # Replace only the process boundary. The coordinator still constructs, + # adopts worker metadata, connects peers and validates real requests. + assert model is config + return Mock( + ready=True, + connected=True, + stopped=True, + info=dict( + kind="ready", + stage_id=stage_config.stage_id, + role=stage_config.role, + model=asdict(model), + model_type=model.to_dict()["model_type"], + fingerprint="same-model-and-cache", + info={"agent": stage_config.stage_id}, + num_blocks=64, + block_size=16, + depths=model.total_ut_steps, + ), + ) + + clients = Mock(side_effect=ready_client) + monkeypatch.setattr("vllm_rlt.pd.engine.StageClient", clients) + with PDEngine(config, attention_backend="triton") as engine: + assert clients.call_count == 2 + # Accept the upper vocabulary/context boundaries at Nanbeige's depth. + params = SamplingParams( + max_tokens=config.max_position_embeddings, max_loops=config.total_ut_steps + ) + engine.add_request("valid", [config.vocab_size - 1], params) + assert engine.has_request("valid") + with pytest.raises(ValueError, match="valid model token IDs"): + engine.add_request("bad_token", [config.vocab_size], params) + with pytest.raises(ValueError, match="loop bounds exceed model depth"): + engine.add_request( + "bad_depth", [1], SamplingParams(max_loops=config.total_ut_steps + 1) + ) + with pytest.raises(ValueError, match="exceeds context capacity"): + engine.add_request( + "bad_context", [1], SamplingParams(max_tokens=config.max_position_embeddings + 1) + ) + assert engine.request_ids() == ("valid",) + assert engine.closed + assert not engine.has_unfinished_requests() + + def drain(engine): outputs = {} deadline = time.monotonic() + 90 diff --git a/vllm_rlt/pd/engine.py b/vllm_rlt/pd/engine.py index ce2b3b7d..30d6e330 100644 --- a/vllm_rlt/pd/engine.py +++ b/vllm_rlt/pd/engine.py @@ -14,7 +14,7 @@ from vllm_rlt.config import CacheConfig, ExecutionConfig, ExitConfig, SchedulerConfig from vllm_rlt.engine.llm_engine import StageClient from vllm_rlt.engine.stage_config import StageConfig -from vllm_rlt.models import OuroConfig, resolve_model_config +from vllm_rlt.models import resolve_model_config from vllm_rlt.profiling import ProfileConfig from vllm_rlt.profiling_artifacts import _write_json_atomic, profile_timestamp from vllm_rlt.request import FinishReason, Request, RequestOutput, RequestPhase @@ -24,6 +24,28 @@ from .connector import PDStageConnector +@dataclass(frozen=True) +class PDModelInfo: + """Model metadata required by the CPU coordinator, not a model config.""" + + vocab_size: int + total_ut_steps: int + max_position_embeddings: int + + @classmethod + def from_ready(cls, info): + """Adopt the coordinator-consumed limits from a stage's ready report.""" + model = info["model"] + names = ("vocab_size", "total_ut_steps", "max_position_embeddings") + invalid = [name for name in names if type(model.get(name)) is not int] + if invalid: + raise ValueError( + f"stage {info['stage_id']} model metadata lacks {invalid}; " + "PD coordinator cannot validate requests for this model family" + ) + return cls(**{name: model[name] for name in names}) + + @dataclass class Peer: name: str @@ -142,7 +164,7 @@ def __init__( fingerprints = {p.client.info["fingerprint"] for p in self.peers.values()} if len(fingerprints) != 1: raise ValueError("P/D model weights or KV configuration do not match") - self.model_config = OuroConfig(**next(iter(self.peers.values())).client.info["model"]) + self.model_config = PDModelInfo.from_ready(next(iter(self.peers.values())).client.info) for peer in self.peers.values(): self._send( peer.name, From af665110067fc47776207593d78d107a1bbe17a9 Mon Sep 17 00:00:00 2001 From: youhengfei <474029121@qq.com> Date: Sun, 25 Oct 2026 14:02:19 +0800 Subject: [PATCH 2/2] [Bugfix][M9] Initialize shared model configuration before worker startup Restore Nanbeige PD startup by resolving ModelConfig at deployment entry points and passing the concrete model configuration to clients and workers. - Preserve family-specific configuration and derived request limits. - Reuse supplied configuration when loading checkpoint weights. - Remove PDModelInfo and ready-message configuration reconstruction. - Keep runtime capacity, transport metadata and fingerprint checks in ready. - Reject unsupported PD model families before spawning workers. - Cover configuration propagation, checkpoint loading and request validation. Validation: - CPU: 834 passed, 329 skipped. - PD GPU: 38 passed, 1 skipped on two H100s. - GSM8K-100: Ouro 69/100, Nanbeige 63/100; native and PD token sequences match on all 100 questions for each model. Signed-off-by: youhengfei <474029121@qq.com> --- docs/unified_execution.md | 14 ++++++ tests/test_auto_model.py | 73 +++++++++++++++++++++++++++++++ tests/test_pd.py | 42 +++++++----------- tests/test_stage_engine.py | 6 +++ vllm_rlt/engine/llm_engine.py | 17 +++++--- vllm_rlt/engine/worker.py | 17 +++++--- vllm_rlt/entrypoints/llm.py | 16 ++++--- vllm_rlt/entrypoints/pd_serve.py | 3 +- vllm_rlt/entrypoints/serve.py | 2 + vllm_rlt/models/__init__.py | 23 ++++++++-- vllm_rlt/models/config.py | 74 ++++++++++++++++++++++++++++++++ vllm_rlt/models/huginn.py | 6 ++- vllm_rlt/models/nanbeige.py | 6 ++- vllm_rlt/models/ouro.py | 6 ++- vllm_rlt/pd/engine.py | 34 ++++----------- 15 files changed, 261 insertions(+), 78 deletions(-) create mode 100644 vllm_rlt/models/config.py diff --git a/docs/unified_execution.md b/docs/unified_execution.md index 30ff96d4..ad098cb7 100644 --- a/docs/unified_execution.md +++ b/docs/unified_execution.md @@ -157,6 +157,20 @@ and optional stream overlap remain execution choices behind this interface. | Preemption snapshots | Engine-owned PreemptionManager, invoked through scheduler callbacks | | Transport staging and remote-access lifetime | Connector; local Request/KV changes are coordinated through StageEngine | +**Model configuration.** Deployment entry points construct a shared `ModelConfig` +from the resolved checkpoint JSON, an existing concrete config, or `module.config` +before spawning workers. It retains the concrete model config and delegates common +request limits, including derived properties. StageClient and the PD coordinator +retain it; spawned workers receive copies. Checkpoint workers load weights using +the concrete config without reparsing JSON; config-only inputs construct models, +and module inputs retain their supplied weights. Models and StageEngine continue +to use the concrete config. Configuration is treated as read-only after initialization. +Worker `ready` messages report runtime capacity and transport information, not a +serialized model config. PD still verifies worker-computed weight/KV fingerprints +before connecting peers; static configuration does not prove runtime compatibility. +PD validates model-family support before spawning workers and explicitly rejects +unsupported families. + **Connector and deployment.** Worker calls `connector.progress()` independently of engine progress. Connector readiness/backpressure reaches Scheduler through the engine's readiness predicate. Import/export and safe release also coordinate diff --git a/tests/test_auto_model.py b/tests/test_auto_model.py index c6cd4992..64ea1fe8 100644 --- a/tests/test_auto_model.py +++ b/tests/test_auto_model.py @@ -12,12 +12,85 @@ from tests.helpers import tiny_nanbeige_config, tiny_ouro_config from vllm_rlt.models import ( AutoModelForCausalLM, + ModelConfig, NanbeigeForCausalLM, OuroForCausalLM, + prepare_model, resolve_local_model_path, ) +@pytest.mark.parametrize("family", ["ouro", "nanbeige", "huginn"]) +def test_entry_config_preserves_properties_and_checkpoint_weights(tmp_path, monkeypatch, family): + from tests.test_huginn import tiny_huginn_config + from vllm_rlt.models import HuginnForCausalLM + + config_factory, model_cls = { + "ouro": (tiny_ouro_config, OuroForCausalLM), + "nanbeige": (tiny_nanbeige_config, NanbeigeForCausalLM), + "huginn": (tiny_huginn_config, HuginnForCausalLM), + }[family] + concrete = config_factory() + original = model_cls(concrete) + (tmp_path / "config.json").write_text(json.dumps(concrete.to_dict())) + save_file( + {key: value.clone() for key, value in original.state_dict().items()}, + tmp_path / "model.safetensors", + ) + source, revision, config = prepare_model(tmp_path) + assert config.total_ut_steps == concrete.total_ut_steps + assert config.max_position_embeddings == concrete.max_position_embeddings + assert config.vocab_size == concrete.vocab_size + assert ModelConfig(concrete).model_specific_config is concrete + assert prepare_model(original)[2].model_specific_config is original.config + with pytest.raises(ValueError, match="do not match"): + prepare_model(original, model_config=ModelConfig(tiny_ouro_config(vocab_size=99))) + + # Neither the Auto factory nor the family loader may reconstruct configuration. + (tmp_path / "config.json").unlink() + resolver = Mock(side_effect=AssertionError("configuration resolved twice")) + monkeypatch.setattr("vllm_rlt.models.resolve_model_config", resolver) + loaded = AutoModelForCausalLM.from_pretrained( + source, revision=revision, model_config=config, dtype=torch.float32 + ) + assert loaded.config is config.model_specific_config + for key, value in original.state_dict().items(): + torch.testing.assert_close(loaded.state_dict()[key], value, rtol=0, atol=0) + resolver.assert_not_called() + + +@pytest.mark.parametrize( + "model_cls,config_factory", + [ + (OuroForCausalLM, tiny_ouro_config), + (NanbeigeForCausalLM, tiny_nanbeige_config), + ], +) +def test_stage_checkpoint_uses_entry_config_after_spawn(tmp_path, model_cls, config_factory): + from vllm_rlt.config import CacheConfig + from vllm_rlt.engine.llm_engine import StageClient + + concrete = config_factory() + model = model_cls(concrete) + save_file(model.state_dict(), tmp_path / "model.safetensors") + # No config.json exists: the child must use the supplied, pickled configuration. + config = ModelConfig(concrete) + client = StageClient( + tmp_path, + model_config=config, + dtype="float32", + attention_backend="torch", + cache_config=CacheConfig(64, 2), + ) + try: + assert client.model_config is config + assert client.ready + assert "model" not in client.info and "model_type" not in client.info + assert client.stats()["requests"] == 0 + finally: + client.close() + + @pytest.fixture def confirm_download(monkeypatch): monkeypatch.setattr("sys.stdin.isatty", lambda: True) diff --git a/tests/test_pd.py b/tests/test_pd.py index 117d6c35..f4108989 100644 --- a/tests/test_pd.py +++ b/tests/test_pd.py @@ -105,33 +105,20 @@ def test_pd_configuration_validation(options): PDConfig(**options) -def test_pd_coordinator_model_metadata_is_model_neutral(): - from dataclasses import asdict +def test_pd_rejects_huginn_before_spawning(monkeypatch): + from unittest.mock import Mock - from tests.helpers import tiny_nanbeige_config from vllm_rlt.models import HuginnConfig - from vllm_rlt.pd.engine import PDModelInfo + from vllm_rlt.pd.engine import PDEngine - def ready(config): - return dict(stage_id="stage-0", model=asdict(config)) + clients = Mock() + monkeypatch.setattr("vllm_rlt.pd.engine.StageClient", clients) + with pytest.raises(ValueError, match="PD does not support Huginn"): + PDEngine(HuginnConfig()) + clients.assert_not_called() - for config in (tiny_ouro_config(), tiny_nanbeige_config()): - info = PDModelInfo.from_ready(ready(config)) - assert info == PDModelInfo( - config.vocab_size, config.total_ut_steps, config.max_position_embeddings - ) - # Huginn derives loop/context limits through properties, which asdict omits - # from the ready payload: fail at startup, not at request validation. - with pytest.raises(ValueError, match="model metadata lacks"): - PDModelInfo.from_ready(ready(HuginnConfig())) - payload = ready(tiny_ouro_config()) - payload["model"]["vocab_size"] = True - with pytest.raises(ValueError, match="model metadata lacks"): - PDModelInfo.from_ready(payload) - - -def test_pd_nanbeige_startup_uses_worker_limits_for_request_validation(monkeypatch): - from dataclasses import asdict + +def test_pd_nanbeige_startup_uses_entry_config_for_request_validation(monkeypatch): from unittest.mock import Mock from tests.helpers import tiny_nanbeige_config @@ -141,8 +128,9 @@ def test_pd_nanbeige_startup_uses_worker_limits_for_request_validation(monkeypat def ready_client(model, *, stage_config, **options): # Replace only the process boundary. The coordinator still constructs, - # adopts worker metadata, connects peers and validates real requests. + # connects peers and validates requests without model metadata in ready. assert model is config + assert options["model_config"].model_specific_config is config return Mock( ready=True, connected=True, @@ -151,8 +139,6 @@ def ready_client(model, *, stage_config, **options): kind="ready", stage_id=stage_config.stage_id, role=stage_config.role, - model=asdict(model), - model_type=model.to_dict()["model_type"], fingerprint="same-model-and-cache", info={"agent": stage_config.stage_id}, num_blocks=64, @@ -165,6 +151,10 @@ def ready_client(model, *, stage_config, **options): monkeypatch.setattr("vllm_rlt.pd.engine.StageClient", clients) with PDEngine(config, attention_backend="triton") as engine: assert clients.call_count == 2 + assert engine.model_config.model_specific_config is config + assert all( + call.kwargs["model_config"] is engine.model_config for call in clients.call_args_list + ) # Accept the upper vocabulary/context boundaries at Nanbeige's depth. params = SamplingParams( max_tokens=config.max_position_embeddings, max_loops=config.total_ut_steps diff --git a/tests/test_stage_engine.py b/tests/test_stage_engine.py index f3c82045..0eaf45d5 100644 --- a/tests/test_stage_engine.py +++ b/tests/test_stage_engine.py @@ -254,12 +254,14 @@ def test_spawned_mixed_worker_matches_local_without_connector(asynchronous): import multiprocessing from vllm_rlt.engine.worker import worker_main + from vllm_rlt.models import ModelConfig local = make_stage_engine(mixed=True, asynchronous=asynchronous) options = dict( seed=123, dtype="float32", revision=None, + model_config=ModelConfig(tiny_ouro_config()), engine=dict( cache_config=CacheConfig(128, 2), scheduler_config=SchedulerConfig( @@ -296,6 +298,7 @@ def receive(): ready = receive() assert ready["kind"] == "ready" assert ready["role"] == "prefill_decode" + assert "model" not in ready and "model_type" not in ready assert "info" not in ready # No registered transport or PD handshake. params = SamplingParams(max_tokens=3, ignore_eos=True) for request_id, tokens in (("a", [1, 2, 3, 4, 5]), ("b", [2, 3])): @@ -561,6 +564,9 @@ def test_module_weight_source_preserves_explicit_deployment( _, device, weights, settings = launch.call_args.args assert device == expected_device assert settings["dtype"] == expected_dtype + assert not client.ready + assert client.model_config.model_specific_config is model.config + assert settings["model_config"] is client.model_config assert weights is not model assert next(weights.parameters()).device.type == "cpu" client.closed = True diff --git a/vllm_rlt/engine/llm_engine.py b/vllm_rlt/engine/llm_engine.py index 766c9ade..8402b24a 100644 --- a/vllm_rlt/engine/llm_engine.py +++ b/vllm_rlt/engine/llm_engine.py @@ -4,7 +4,6 @@ import time from collections import deque from dataclasses import replace -from types import SimpleNamespace from uuid import uuid4 import torch @@ -12,6 +11,7 @@ from vllm_rlt.engine.stage_config import StageConfig from vllm_rlt.engine.stage_engine import StageEngine from vllm_rlt.engine.worker import start_stage_worker +from vllm_rlt.models import prepare_model LLMEngine = StageEngine @@ -30,6 +30,7 @@ def __init__( device=None, dtype=None, revision=None, + model_config=None, seed=0, startup_timeout=120, shutdown_timeout=30, @@ -39,6 +40,9 @@ def __init__( wait_for_ready=True, **engine_options, ): + model, revision, self.model_config = prepare_model( + model, revision=revision, model_config=model_config + ) # An explicitly supplied model is a weight source, never a second runner. # Checkpoint paths load directly inside the worker without a host copy. if isinstance(model, torch.nn.Module): @@ -69,7 +73,13 @@ def __init__( self.stage_config, device, model, - dict(seed=seed, dtype=dtype, revision=revision, engine=engine_options), + dict( + seed=seed, + dtype=dtype, + revision=revision, + engine=engine_options, + model_config=self.model_config, + ), connector_factory=connector_factory, max_control_messages=max_control_messages, ) @@ -131,9 +141,6 @@ def _receive(self, *, timeout=0): raise error(f"stage worker failed:\n{self.failure}") if kind == "ready": self.info, self.ready = message, True - self.model_config = SimpleNamespace( - **message["model"], model_type=message["model_type"] - ) elif kind == "connected": self.connected = True elif kind == "reply": diff --git a/vllm_rlt/engine/worker.py b/vllm_rlt/engine/worker.py index ce15bf9d..9a2b2805 100644 --- a/vllm_rlt/engine/worker.py +++ b/vllm_rlt/engine/worker.py @@ -9,7 +9,7 @@ import time import traceback from collections.abc import Callable -from dataclasses import asdict, is_dataclass +from dataclasses import is_dataclass import torch @@ -43,14 +43,21 @@ def __init__( self.channel = channel self.max_control_messages = max_control_messages dtype = getattr(torch, options["dtype"]) + model_config = options["model_config"] if isinstance(model_source, torch.nn.Module): self.model = model_source.to(device=device, dtype=dtype) elif is_dataclass(model_source): - model_cls = MODEL_MAPPING[model_source.to_dict()["model_type"]] - self.model = model_cls(model_source).to(device=device, dtype=dtype) + model_cls = MODEL_MAPPING[model_config.model_type] + self.model = model_cls(model_config.model_specific_config).to( + device=device, dtype=dtype + ) else: self.model = AutoModelForCausalLM.from_pretrained( - model_source, revision=options["revision"], device=str(device), dtype=dtype + model_source, + revision=options["revision"], + device=str(device), + dtype=dtype, + model_config=model_config, ) self.engine = StageEngine(self.model, stage_config=stage_config, **options["engine"]) self.connector = ( @@ -66,8 +73,6 @@ def __init__( info = dict( stage_id=stage_config.stage_id, role=stage_config.role, - model=asdict(self.model.config), - model_type=self.model.config.to_dict()["model_type"], num_blocks=cache.num_blocks, block_size=cache.block_size, depths=cache.storage_depths, diff --git a/vllm_rlt/entrypoints/llm.py b/vllm_rlt/entrypoints/llm.py index b3554417..55e46104 100644 --- a/vllm_rlt/entrypoints/llm.py +++ b/vllm_rlt/entrypoints/llm.py @@ -5,7 +5,7 @@ from vllm_rlt.engine.llm_engine import StageClient from vllm_rlt.models import ( load_tokenizer, - resolve_model_config, + prepare_model, resolve_model_source, ) from vllm_rlt.sampling_params import SamplingParams @@ -29,15 +29,17 @@ def __init__( execution_config=None, speculative_config=None, ): - self._tokenizer_source = None - if isinstance(model, str): - tokenizer_source = resolve_model_source(model) - model, revision, _ = resolve_model_config(model, revision=revision) - if tokenizer is None: - self._tokenizer_source = (tokenizer_source, revision) + tokenizer_source = resolve_model_source(model) if isinstance(model, str) else None + model, revision, model_config = prepare_model(model, revision=revision) + self._tokenizer_source = ( + (tokenizer_source, revision) + if tokenizer is None and tokenizer_source is not None + else None + ) self.tokenizer = tokenizer self.engine = StageClient( model, + model_config=model_config, device=device, dtype=dtype, revision=revision, diff --git a/vllm_rlt/entrypoints/pd_serve.py b/vllm_rlt/entrypoints/pd_serve.py index 5730c17b..39ac160c 100644 --- a/vllm_rlt/entrypoints/pd_serve.py +++ b/vllm_rlt/entrypoints/pd_serve.py @@ -11,7 +11,7 @@ profile_config_from_args, runtime_configs, ) -from vllm_rlt.models import load_tokenizer, resolve_model_config, resolve_model_source +from vllm_rlt.models import ModelConfig, load_tokenizer, resolve_model_config, resolve_model_source from vllm_rlt.pd.config import PDConfig from vllm_rlt.pd.engine import PDEngine @@ -37,6 +37,7 @@ def load_engine(args): cache = options.pop("cache_config") engine = PDEngine( source, + model_config=ModelConfig.from_dict(config), revision=revision, dtype=args.dtype, pd_config=PDConfig( diff --git a/vllm_rlt/entrypoints/serve.py b/vllm_rlt/entrypoints/serve.py index 0c011228..5b11b3c3 100644 --- a/vllm_rlt/entrypoints/serve.py +++ b/vllm_rlt/entrypoints/serve.py @@ -15,6 +15,7 @@ runtime_configs, ) from vllm_rlt.models import ( + ModelConfig, load_tokenizer, resolve_model_config, resolve_model_source, @@ -42,6 +43,7 @@ def load_engine(args): raise ValueError("serving requires the Ouro byte-level tokenizer") engine = StageClient( source, + model_config=ModelConfig.from_dict(config), revision=revision, device=args.device, dtype=args.dtype, diff --git a/vllm_rlt/models/__init__.py b/vllm_rlt/models/__init__.py index 6d02f2b2..3da1d193 100644 --- a/vllm_rlt/models/__init__.py +++ b/vllm_rlt/models/__init__.py @@ -8,6 +8,7 @@ import torch +from .config import ModelConfig, prepare_model from .huginn import HuginnConfig, HuginnForCausalLM from .nanbeige import NanbeigeConfig, NanbeigeForCausalLM from .ouro import OuroConfig, OuroForCausalLM @@ -136,14 +137,19 @@ class AutoModelForCausalLM: @classmethod def _dispatch_from_config( cls, - config_data: dict[str, Any], + config_data: dict[str, Any] | ModelConfig, path_or_repo: str | Path, *, revision: str | None = None, device: torch.device | str = "cpu", dtype: torch.dtype = torch.bfloat16, ): - model_type = config_data.get("model_type", "").lower() + config = ( + config_data + if isinstance(config_data, ModelConfig) + else ModelConfig.from_dict(config_data) + ) + model_type = config.model_type model_cls = MODEL_MAPPING.get(model_type) if model_cls is None: supported = sorted(MODEL_MAPPING.keys()) @@ -152,7 +158,11 @@ def _dispatch_from_config( f"Supported types: {supported}" ) return model_cls.from_pretrained( - path_or_repo, revision=revision, device=device, dtype=dtype + path_or_repo, + revision=revision, + device=device, + dtype=dtype, + config=config.model_specific_config, ) @classmethod @@ -160,17 +170,22 @@ def from_pretrained( cls, path_or_repo: str | Path, *, + model_config: ModelConfig | None = None, revision: str | None = None, device: torch.device | str = "cpu", dtype: torch.dtype = torch.bfloat16, ): - source, resolved_revision, config = resolve_model_config(path_or_repo, revision=revision) + source, resolved_revision, config = prepare_model( + path_or_repo, revision=revision, model_config=model_config + ) return cls._dispatch_from_config( config, source, revision=resolved_revision, device=device, dtype=dtype ) __all__ = [ + "ModelConfig", + "prepare_model", "OuroConfig", "OuroForCausalLM", "NanbeigeConfig", diff --git a/vllm_rlt/models/config.py b/vllm_rlt/models/config.py new file mode 100644 index 00000000..10339b7d --- /dev/null +++ b/vllm_rlt/models/config.py @@ -0,0 +1,74 @@ +"""Entry-owned model configuration shared by deployment clients and workers.""" + +from dataclasses import dataclass +from pathlib import Path + +from torch import nn + +from .huginn import HuginnConfig +from .nanbeige import NanbeigeConfig +from .ouro import OuroConfig + +_CONFIG_TYPES = { + "ouro": OuroConfig, + "nanbeige": NanbeigeConfig, + "huginn_raven": HuginnConfig, +} + + +@dataclass(frozen=True) +class ModelConfig: + """A concrete model config, treated as read-only after initialization. + + Spawned workers receive copies, not shared mutable configuration. Derived + limits remain properties of the concrete config rather than duplicated fields. + """ + + model_specific_config: OuroConfig | NanbeigeConfig | HuginnConfig + + def __post_init__(self): + if type(self.model_specific_config) not in _CONFIG_TYPES.values(): + raise ValueError("Unsupported model configuration type") + + @classmethod + def from_dict(cls, values): + model_type = values.get("model_type", "").lower() + if model_type not in _CONFIG_TYPES: + raise ValueError(f"Unsupported model_type {model_type!r}") + return cls(_CONFIG_TYPES[model_type].from_dict(values)) + + @property + def model_type(self): + return self.model_specific_config.to_dict()["model_type"] + + @property + def vocab_size(self): + return self.model_specific_config.vocab_size + + @property + def total_ut_steps(self): + return self.model_specific_config.total_ut_steps + + @property + def max_position_embeddings(self): + return self.model_specific_config.max_position_embeddings + + +def prepare_model(model, *, revision=None, model_config=None): + """Resolve a source once, before spawning; preserve its weight-source semantics.""" + if model_config is not None: + if not isinstance(model_config, ModelConfig): + raise TypeError("model_config must be a ModelConfig") + concrete = model.config if isinstance(model, nn.Module) else model + if not isinstance(model, (str, Path)) and concrete != model_config.model_specific_config: + raise ValueError("Model source and model_config do not match") + return model, revision, model_config + if isinstance(model, (str, Path)): + # The resolver also owns download approval and snapshot pinning. + from . import resolve_model_config + + model, revision, values = resolve_model_config(model, revision=revision) + model_config = ModelConfig.from_dict(values) + else: + model_config = ModelConfig(model.config if isinstance(model, nn.Module) else model) + return model, revision, model_config diff --git a/vllm_rlt/models/huginn.py b/vllm_rlt/models/huginn.py index 9312df67..b7c4c117 100644 --- a/vllm_rlt/models/huginn.py +++ b/vllm_rlt/models/huginn.py @@ -310,6 +310,7 @@ def from_pretrained( revision: str | None = None, device: torch.device | str = "cpu", dtype: torch.dtype = torch.bfloat16, + config: HuginnConfig | None = None, ) -> "HuginnForCausalLM": """Strictly stream safetensors into a meta model without Hub code execution.""" from safetensors import safe_open @@ -322,7 +323,10 @@ def from_pretrained( source, _, _ = resolve_model_config(path_or_repo, revision=revision) folder = Path(source) - config = HuginnConfig.from_dict(json.loads((folder / "config.json").read_text())) + if config is None: + config = HuginnConfig.from_dict(json.loads((folder / "config.json").read_text())) + elif not isinstance(config, HuginnConfig): + raise TypeError("config must be a HuginnConfig") index_path = folder / "model.safetensors.index.json" index = json.loads(index_path.read_text())["weight_map"] if index_path.is_file() else None if index is None: diff --git a/vllm_rlt/models/nanbeige.py b/vllm_rlt/models/nanbeige.py index 1b573fa4..a94aee5b 100644 --- a/vllm_rlt/models/nanbeige.py +++ b/vllm_rlt/models/nanbeige.py @@ -273,6 +273,7 @@ def from_pretrained( revision: str | None = None, device: torch.device | str = "cpu", dtype: torch.dtype = torch.bfloat16, + config: NanbeigeConfig | None = None, ) -> "NanbeigeForCausalLM": """Stream strictly checked safetensors into a meta model; never execute Hub code.""" from safetensors import safe_open @@ -285,7 +286,10 @@ def from_pretrained( source, revision, _ = resolve_model_config(path_or_repo, revision=revision) folder = Path(source) - config = NanbeigeConfig.from_dict(json.loads((folder / "config.json").read_text())) + if config is None: + config = NanbeigeConfig.from_dict(json.loads((folder / "config.json").read_text())) + elif not isinstance(config, NanbeigeConfig): + raise TypeError("config must be a NanbeigeConfig") index_path = folder / "model.safetensors.index.json" index = json.loads(index_path.read_text())["weight_map"] if index_path.is_file() else None if index is None: diff --git a/vllm_rlt/models/ouro.py b/vllm_rlt/models/ouro.py index eae29724..4af3ff44 100644 --- a/vllm_rlt/models/ouro.py +++ b/vllm_rlt/models/ouro.py @@ -301,6 +301,7 @@ def from_pretrained( revision: str | None = None, device: torch.device | str = "cpu", dtype: torch.dtype = torch.bfloat16, + config: OuroConfig | None = None, ) -> "OuroForCausalLM": """Stream strictly checked safetensors into a meta model; never execute Hub code. @@ -316,7 +317,10 @@ def from_pretrained( source, revision, _ = resolve_model_config(path_or_repo, revision=revision) folder = Path(source) - config = OuroConfig.from_dict(json.loads((folder / "config.json").read_text())) + if config is None: + config = OuroConfig.from_dict(json.loads((folder / "config.json").read_text())) + elif not isinstance(config, OuroConfig): + raise TypeError("config must be an OuroConfig") index_path = folder / "model.safetensors.index.json" index = json.loads(index_path.read_text())["weight_map"] if index_path.is_file() else None if index is None: diff --git a/vllm_rlt/pd/engine.py b/vllm_rlt/pd/engine.py index 30d6e330..47f58ad1 100644 --- a/vllm_rlt/pd/engine.py +++ b/vllm_rlt/pd/engine.py @@ -14,7 +14,7 @@ from vllm_rlt.config import CacheConfig, ExecutionConfig, ExitConfig, SchedulerConfig from vllm_rlt.engine.llm_engine import StageClient from vllm_rlt.engine.stage_config import StageConfig -from vllm_rlt.models import resolve_model_config +from vllm_rlt.models import prepare_model from vllm_rlt.profiling import ProfileConfig from vllm_rlt.profiling_artifacts import _write_json_atomic, profile_timestamp from vllm_rlt.request import FinishReason, Request, RequestOutput, RequestPhase @@ -24,28 +24,6 @@ from .connector import PDStageConnector -@dataclass(frozen=True) -class PDModelInfo: - """Model metadata required by the CPU coordinator, not a model config.""" - - vocab_size: int - total_ut_steps: int - max_position_embeddings: int - - @classmethod - def from_ready(cls, info): - """Adopt the coordinator-consumed limits from a stage's ready report.""" - model = info["model"] - names = ("vocab_size", "total_ut_steps", "max_position_embeddings") - invalid = [name for name in names if type(model.get(name)) is not int] - if invalid: - raise ValueError( - f"stage {info['stage_id']} model metadata lacks {invalid}; " - "PD coordinator cannot validate requests for this model family" - ) - return cls(**{name: model[name] for name in names}) - - @dataclass class Peer: name: str @@ -88,12 +66,16 @@ def __init__( attention_backend="flash_attn", dtype="bfloat16", revision=None, + model_config=None, seed=0, speculative_config=None, ): self.speculative_config = speculative_config - if isinstance(model, (str, Path)): - model, revision, _ = resolve_model_config(model, revision=revision) + model, revision, self.model_config = prepare_model( + model, revision=revision, model_config=model_config + ) + if self.model_config.model_type == "huginn_raven": + raise ValueError("PD does not support Huginn") self.config = pd_config or PDConfig() self.exit_config = exit_config or ExitConfig( "ouro" if speculative_config else "ouro_delayed" @@ -135,6 +117,7 @@ def __init__( name = f"{epoch}-{role}-{device}" client = StageClient( model, + model_config=self.model_config, device=device, stage_config=StageConfig(stage_id=name, role=role), connector_factory=partial(PDStageConnector.create, config=self.config), @@ -164,7 +147,6 @@ def __init__( fingerprints = {p.client.info["fingerprint"] for p in self.peers.values()} if len(fingerprints) != 1: raise ValueError("P/D model weights or KV configuration do not match") - self.model_config = PDModelInfo.from_ready(next(iter(self.peers.values())).client.info) for peer in self.peers.values(): self._send( peer.name,