modelexpress-client 0.4.0

Client library for Model Express gRPC server
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""ModelStreamer loading strategy: stream safetensors via runai-model-streamer.

Supports object storage (S3, GCS, Azure Blob) and local filesystem paths.
File resolution and tensor iteration are delegated to the engine adapter so
native loader integrations stay engine-specific.
"""

from __future__ import annotations

import importlib.util
import logging
import os
from typing import Iterator

import torch

from ..adapter import EngineAdapter, StrategyFailed
from .base import LoadContext, LoadStrategy, _as_load_result, register_tensors
from .context import LoadResult

logger = logging.getLogger("modelexpress.strategy_model_streamer")


class ModelStreamerStrategy(LoadStrategy):
    """Load weights by streaming safetensors via runai-model-streamer.

    Activated by setting MX_MODEL_URI (gate only). The actual URI used to
    stream weights is taken from `model_config.model_weights` if set
    (object-storage URIs: s3://, gs://, az://) and falls back to
    `model_config.model` otherwise (local paths, HF-resolved snapshots).
    Engine adapters provide the concrete ModelStreamer iterator.
    """

    name = "model_streamer"
    requires = (
        EngineAdapter.apply_weight_iter,
        EngineAdapter.build_model_streamer_weight_iter,
    )

    def is_available(self, ctx: LoadContext) -> bool:
        if not super().is_available(ctx):
            return False
        if importlib.util.find_spec("runai_model_streamer") is None:
            logger.info(
                f"[Worker {ctx.global_rank}] runai_model_streamer not installed, skipping"
            )
            return False

        model_uri = os.environ.get("MX_MODEL_URI", "")
        if not model_uri:
            logger.info(
                f"[Worker {ctx.global_rank}] MX_MODEL_URI not set, skipping model streamer"
            )
            return False
        return True

    def load(self, result: LoadResult, ctx: LoadContext) -> LoadResult:
        result = _as_load_result(result)
        model_uri = getattr(ctx.model_config, "model_weights", None) or ctx.model_config.model

        logger.info(f"[Worker {ctx.global_rank}] Attempting model streamer loading from {model_uri}")
        try:
            weights_iter = self._stream_weights(model_uri, ctx, result.model)
        except Exception as e:
            logger.warning(
                f"[Worker {ctx.global_rank}] Model streamer loading failed, falling through: {e}"
            )
            raise StrategyFailed(str(e), mutated=False) from e

        try:
            result = ctx.adapter.apply_weight_iter(result, weights_iter)
            logger.info(f"[Worker {ctx.global_rank}] Model streamer weight loading complete")
            result = ctx.adapter.after_weight_iter_load(result)
        except Exception as e:
            logger.warning(
                f"[Worker {ctx.global_rank}] Model streamer loading failed, falling through: {e}"
            )
            raise StrategyFailed(str(e), mutated=True) from e

        register_tensors(result, ctx)
        return result

    def _stream_weights(
        self,
        model_uri: str,
        ctx: LoadContext,
        model: torch.nn.Module | None = None,
    ) -> Iterator[tuple[str, torch.Tensor]]:
        logger.info(f"[Worker {ctx.global_rank}] Streaming weights from {model_uri}")
        yield from ctx.adapter.build_model_streamer_weight_iter(
            model_uri,
            model=model,
        )