Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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 @@ -200,6 +200,12 @@ class RemoteInferenceClient:
active_lora_name: Optional[str] = None
"""Name of the active LoRA adapter. If set, generation requests use this adapter instead of the base model."""

uses_lora_weight_sync: bool = False

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need a separate flag here? Can't we use just active_lora_name is not None as the condition?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"""True when the trainer syncs LoRA adapters (rather than full/merged weights). When True,
`sleep()` is forced to level=1: level=2 discards the base model from VRAM with no CPU backup,
and LoRA-only broadcasts cannot repopulate it. Must be kept in sync with the same gate vLLM
uses for `enable_lora` (see `_uses_lora_weight_sync` in inference_servers/utils.py)."""

tokenizer: Optional[Any] = None
"""Optional HF tokenizer for local tokenize/detokenize (avoids HTTP round-trips)."""

Expand Down Expand Up @@ -924,6 +930,15 @@ async def sleep(self, level: int = 2, tags: Optional[List[str]] = None) -> Dict[
Returns:
Dict mapping server_url to response.
"""
# Mirror BaseVLLMInferenceEngine.sleep: when the trainer syncs LoRA adapters
# only, force level=1 so the base model survives via CPU backup. level=2
# discards weights with no source to restore from on wake_up(["weights"]).
if self.uses_lora_weight_sync and level != 1:
logger.info(
"Forcing sleep level=1 (uses_lora_weight_sync=True); requested level=%d would discard the base model.",
level,
)
level = 1
params: Dict[str, Any] = {"level": str(level)}
if tags:
params["tags"] = tags
Expand Down Expand Up @@ -1085,44 +1100,51 @@ async def update_lora_from_disk(
lora_path: str,
) -> Dict[str, Any]:
"""
Update LoRA adapter weights by loading from disk on all backend servers via /v1/load_lora_adapter.

Always loads under self.active_lora_name so the same slot is reused across
weight syncs.
Update LoRA adapter weights by loading from disk on all backend servers.

After loading, generation requests will automatically use the LoRA adapter
by setting the model name to the LoRA adapter name.
Implementation: unload then load (with load_inplace=False), rather than a single
load with load_inplace=True. Reason: vLLM's /v1/load_lora_adapter caches the
LoRARequest object server-side keyed by lora_name (see OpenAIServingModels.lora_requests).
If load_inplace=True is set on that cached request, every subsequent generate request
causes LRUCacheWorkerLoRAManager.add_adapter to re-read the adapter from disk on every
model step (line 299-307 of vllm/lora/worker_manager.py checks `load_inplace` per-step).
The cleanest fix is upstream (clear load_inplace before caching); until that lands,
unloading first lets us reuse the same lora_name without ever needing load_inplace=True.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: Docstring stylke is not consistent with the rest of the codebase. Can you clean this up and also convert this into a TODO:

The cleanest fix is upstream (clear load_inplace before caching); until that lands,
unloading first lets us reuse the same lora_name without ever needing load_inplace=True.


Args:
lora_path: Path to the LoRA adapter on disk (must be accessible from servers).

Returns:
Dict mapping server_url to response.
Dict mapping server_url to load response.
"""
if self.active_lora_name is None:
raise ValueError("active_lora_name must be set on RemoteInferenceClient before loading a LoRA adapter.")

lora_name = self.active_lora_name
payload = {
"lora_name": lora_name,
"lora_path": lora_path,
"load_inplace": True,
}

# Call /v1/load_lora_adapter on all servers directly.
# This endpoint returns a plain text response (not JSON), so we use a
# custom call instead of _call_all_servers which expects JSON.
# Both endpoints return plain text on success or JSON ErrorResponse on failure,
# so we use a session.post directly rather than _call_all_servers (which expects JSON).
session = await self._get_session()

async def _unload_on_server(server_url: str) -> None:
"""Remove the cached LoRARequest server-side. 404 on the first sync is expected."""
url = f"{server_url}/v1/unload_lora_adapter"
async with session.post(url, json={"lora_name": lora_name}) as resp:
if resp.status == 404:
return
if resp.status >= 400:
body = await resp.json()
raise_for_status(resp, body)
Comment on lines +1134 to +1136

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The call to resp.json() is risky here. If the server returns a non-JSON error response (e.g., a 502 Bad Gateway from a proxy or a 500 Internal Server Error with a plain text stack trace), await resp.json() will raise an aiohttp.ContentTypeError. This masks the actual HTTP status code and error message, making debugging harder. It is safer to wrap the JSON parsing in a try-except block or check the content type before parsing, especially since raise_for_status is already designed to handle a None body by falling back to the standard resp.raise_for_status().

Suggested change
if resp.status >= 400:
body = await resp.json()
raise_for_status(resp, body)
if resp.status >= 400:
try:
body = await resp.json()
except Exception:
body = None
raise_for_status(resp, body)


async def _load_on_server(server_url: str):
url = f"{server_url}/v1/load_lora_adapter"
payload = {"lora_name": lora_name, "lora_path": lora_path}
async with session.post(url, json=payload) as resp:
# vLLM returns 200 with text body on success, or JSON ErrorResponse on failure
if resp.status >= 400:
body = await resp.json()
raise_for_status(resp, body)
Comment on lines 1142 to 1144

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to the _unload_on_server implementation, calling resp.json() directly on a non-2xx response can lead to a ContentTypeError if the error body is not valid JSON. Wrapping this in a try-except block ensures that the actual HTTP error is surfaced via raise_for_status even if the error response body is malformed or plain text.

Suggested change
if resp.status >= 400:
body = await resp.json()
raise_for_status(resp, body)
if resp.status >= 400:
try:
body = await resp.json()
except Exception:
body = None
raise_for_status(resp, body)

return server_url, {"status": resp.status, "body": await resp.text()}

await asyncio.gather(*[_unload_on_server(url) for url in self.server_urls])
results = await asyncio.gather(*[_load_on_server(url) for url in self.server_urls])

logger.info(f"Loaded LoRA adapter '{lora_name}' from {lora_path}")
Expand Down
8 changes: 7 additions & 1 deletion skyrl/backends/skyrl_train/inference_servers/setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,12 @@

from .remote_inference_client import RemoteInferenceClient
from .server_group import ServerGroup
from .utils import build_router_args, build_vllm_cli_args, get_pd_cli_args
from .utils import (
_uses_lora_weight_sync,
build_router_args,
build_vllm_cli_args,
get_pd_cli_args,
)
from .vllm_router import VLLMRouter

NIXL_SIDE_CHANNEL_BASE_PORT = 5600
Expand Down Expand Up @@ -284,6 +289,7 @@ def build_new_inference_client(
model_name=cfg.trainer.policy.model.path,
enable_return_routed_experts=ie_cfg.enable_return_routed_experts,
active_lora_name=active_lora_name,
uses_lora_weight_sync=_uses_lora_weight_sync(cfg),
data_parallel_size=ie_cfg.data_parallel_size,
tokenizer=tokenizer,
)
Expand Down
2 changes: 2 additions & 0 deletions tests/backends/skyrl_train/gpu/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
from skyrl.backends.skyrl_train.inference_servers.server_group import ServerGroup
from skyrl.backends.skyrl_train.inference_servers.setup import create_inference_servers
from skyrl.backends.skyrl_train.inference_servers.utils import (
_uses_lora_weight_sync,
build_vllm_cli_args,
)
from skyrl.backends.skyrl_train.inference_servers.vllm_router import VLLMRouter
Expand Down Expand Up @@ -620,6 +621,7 @@ def create(
model_name=served_model_name if served_model_name else cfg.trainer.policy.model.path,
enable_return_routed_experts=ie_cfg.enable_return_routed_experts,
active_lora_name=active_lora_name,
uses_lora_weight_sync=_uses_lora_weight_sync(cfg),
data_parallel_size=ie_cfg.data_parallel_size,
tokenizer=get_tokenizer(cfg.trainer.policy.model.path),
)
Expand Down
Loading