Skip to content

vllm.distributed.weight_transfer.clients

Built-in VLLMWeightSyncClient implementations.

These adapt the inference engine's weight-sync control plane to concrete transports. A TrainerWeightTransferEngine takes one of these (or any object with the same four methods — the protocol is structural) and drives the full handshake through it.

Imports of ray / requests are deferred to call time so this module is importable without those packages installed.

Classes:

HTTPVLLMWeightSyncClient

Talks to a vLLM server over the RLHF HTTP routes.

Mirrors vllm/entrypoints/rl/online/api_router.py: /init_weight_transfer_engine, /start_weight_update, /update_weights, /finish_weight_update, /shutdown_weight_transfer_engine.

Source code in vllm/distributed/weight_transfer/clients.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class HTTPVLLMWeightSyncClient:
    """Talks to a vLLM server over the RLHF HTTP routes.

    Mirrors `vllm/entrypoints/rl/online/api_router.py`:
    `/init_weight_transfer_engine`, `/start_weight_update`, `/update_weights`,
    `/finish_weight_update`, `/shutdown_weight_transfer_engine`.
    """

    def __init__(self, base_url: str, timeout: float = 300) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def _post(self, path: str, json: dict[str, Any] | None = None) -> None:
        import requests

        response = requests.post(
            f"{self.base_url}/{path}", json=json, timeout=self.timeout
        )
        response.raise_for_status()

    def init_weight_transfer_engine(self, init_info: dict[str, Any]) -> None:
        self._post("init_weight_transfer_engine", {"init_info": init_info})

    def shutdown_weight_transfer_engine(self) -> None:
        self._post("shutdown_weight_transfer_engine")

    def start_weight_update(self) -> None:
        self._post("start_weight_update")

    def update_weights(self, update_info: WeightTransferUpdatePayload) -> None:
        self._post(
            "update_weights", {"update_info": _json_safe_update_payload(update_info)}
        )

    def finish_weight_update(self, weight_version: str | None = None) -> None:
        json = (
            {"weight_version": weight_version} if weight_version is not None else None
        )
        self._post("finish_weight_update", json)

RayVLLMWeightSyncClient

Talks to one or more vLLM AsyncLLM/LLM Ray actors.

Each call fans out to every handle and blocks on all of them, so a multi-actor (e.g. multi-DP) deployment is driven as one unit.

Source code in vllm/distributed/weight_transfer/clients.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class RayVLLMWeightSyncClient:
    """Talks to one or more vLLM `AsyncLLM`/`LLM` Ray actors.

    Each call fans out to every handle and blocks on all of them, so a
    multi-actor (e.g. multi-DP) deployment is driven as one unit.
    """

    def __init__(self, handle: "ActorHandle | list[ActorHandle]") -> None:
        self.handles = handle if isinstance(handle, list) else [handle]

    def init_weight_transfer_engine(self, init_info: dict[str, Any]) -> None:
        import ray

        request = WeightTransferInitRequest(init_info=init_info)
        ray.get([h.init_weight_transfer_engine.remote(request) for h in self.handles])

    def shutdown_weight_transfer_engine(self) -> None:
        import ray

        ray.get([h.shutdown_weight_transfer_engine.remote() for h in self.handles])

    def start_weight_update(self) -> None:
        import ray

        ray.get([h.start_weight_update.remote() for h in self.handles])

    def update_weights(self, update_info: WeightTransferUpdatePayload) -> None:
        import ray

        request = WeightTransferUpdateRequest(update_info=update_info)
        ray.get([h.update_weights.remote(request) for h in self.handles])

    def finish_weight_update(self, weight_version: str | None = None) -> None:
        import ray

        ray.get([h.finish_weight_update.remote() for h in self.handles])
        if weight_version is not None:
            ray.get(
                [h.update_weight_version.remote(weight_version) for h in self.handles]
            )

_json_safe_update_info(update_info)

Make an update_info dict JSON-serializable for HTTP transport.

CUDA IPC handles (ipc_handles) are tuples of non-JSON-native objects, so over HTTP they are pickled+base64-encoded into ipc_handles_pickled (which the worker auto-deserializes when VLLM_ALLOW_INSECURE_SERIALIZATION=1). Other backends (NCCL) carry only JSON-native metadata and pass through unchanged. Mirrors the old IPC _do_send HTTP branch.

Source code in vllm/distributed/weight_transfer/clients.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def _json_safe_update_info(update_info: dict[str, Any]) -> dict[str, Any]:
    """Make an update_info dict JSON-serializable for HTTP transport.

    CUDA IPC handles (`ipc_handles`) are tuples of non-JSON-native objects, so
    over HTTP they are pickled+base64-encoded into `ipc_handles_pickled` (which
    the worker auto-deserializes when `VLLM_ALLOW_INSECURE_SERIALIZATION=1`).
    Other backends (NCCL) carry only JSON-native metadata and pass through
    unchanged. Mirrors the old IPC `_do_send` HTTP branch.
    """
    ipc_handles = update_info.get("ipc_handles")
    if ipc_handles is None:
        return update_info

    import pickle

    import pybase64 as base64

    out = {k: v for k, v in update_info.items() if k != "ipc_handles"}
    out["ipc_handles_pickled"] = base64.b64encode(pickle.dumps(ipc_handles)).decode(
        "utf-8"
    )
    return out