Skip to content

vllm.v1.kv_offload.tiering.factory

Classes:

SecondaryTierFactory

Registry for SecondaryTierManager implementations, resolved by type.

Mirrors OffloadingSpecFactory (vllm/v1/kv_offload/factory.py): built-in tiers are pre-registered below. External tiers can either register_tier() a friendly short name up front, or skip registration entirely and pass a module_path at lookup time (out-of-tree, no vLLM fork/patch required) -- see get_tier_class.

Methods:

Source code in vllm/v1/kv_offload/tiering/factory.py 
 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
 97
 98
 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
139
140
141
142
143
class SecondaryTierFactory:
    """Registry for SecondaryTierManager implementations, resolved by type.

    Mirrors OffloadingSpecFactory (vllm/v1/kv_offload/factory.py): built-in
    tiers are pre-registered below. External tiers can either
    register_tier() a friendly short name up front, or skip registration
    entirely and pass a module_path at lookup time (out-of-tree, no vLLM
    fork/patch required) -- see get_tier_class.
    """

    _registry: dict[str, Callable[[], type[SecondaryTierManager]]] = {}

    @classmethod
    def register_tier(cls, tier_type: str, module_path: str, class_name: str) -> None:
        if tier_type in cls._registry:
            raise ValueError(f"Tier '{tier_type}' is already registered.")

        def loader() -> type[SecondaryTierManager]:
            module = importlib.import_module(module_path)
            return getattr(module, class_name)

        cls._registry[tier_type] = loader

    @classmethod
    def create_secondary_tier(
        cls,
        tier_config: dict,
        primary_kv_view: memoryview,
        offloading_spec: "OffloadingSpec",
    ) -> SecondaryTierManager:
        tier_cls = cls.get_tier_class(tier_config)
        config = tier_config.copy()
        tier_type = config.pop("type")
        config.pop("module_path", None)
        locality = config.get("locality")
        bp_config = config.pop("backpressure", None)
        bp_detector = None
        if bp_config is not None:
            bp_config = bp_config.copy()
            bp_cls_name = bp_config.pop("backpressure_cls", None)
            if bp_cls_name is None:
                raise ValueError(
                    "backpressure config must include 'backpressure_cls' "
                    "(e.g. 'EMABackpressureDetector')"
                )
            policy_cls_name = bp_config.pop("policy_cls", None)
            policy = _import_class(policy_cls_name)() if policy_cls_name else None
            detector_cls: type[BackpressureDetector] = _import_class(bp_cls_name)
            defaults = detector_cls.default_config(tier_type, locality=locality)
            if defaults:
                for k, v in defaults.items():
                    bp_config.setdefault(k, v)
            bp_detector = detector_cls(policy=policy, **bp_config)
        return tier_cls(
            offloading_spec=offloading_spec,
            primary_kv_view=primary_kv_view,
            tier_type=tier_type,
            backpressure_detector=bp_detector,
            **config,
        )

    @classmethod
    def get_tier_class(cls, tier_config: dict) -> type[SecondaryTierManager]:
        """Get a secondary tier class by type.

        Args:
            tier_config: Tier configuration dict. Must contain 'type' (the
                tier name or out-of-tree class name). If 'type' is not a
                registered tier and 'module_path' is given, the class is
                imported from there -- an out-of-tree tier needs no
                register_tier() call, just this module path in the config.

        Returns:
            The secondary tier manager class.

        Raises:
            ValueError: If 'type' is missing, or is neither registered nor
                resolvable via module_path.

        """
        tier_type = tier_config.get("type")
        if not tier_type:
            raise ValueError("Secondary tier configuration must include 'type'")
        if tier_type in cls._registry:
            return cls._registry[tier_type]()
        module_path = tier_config.get("module_path")
        if module_path is None:
            raise ValueError(
                f"Unknown secondary tier type: {tier_type!r}. "
                f"Supported types: {list(cls._registry)}. "
                "For an out-of-tree tier, also set 'module_path'."
            )
        logger.warning_once(
            "Loading out-of-tree secondary tier. This API is "
            "experimental and subject to change in the future "
            "as we iterate the design."
        )
        logger.info(
            "Loading out-of-tree secondary tier '%s' from '%s'.",
            tier_type,
            module_path,
        )
        module = importlib.import_module(module_path)
        tier_cls = getattr(module, tier_type)
        assert issubclass(tier_cls, SecondaryTierManager)
        return tier_cls

get_tier_class(tier_config) classmethod

Get a secondary tier class by type.

Parameters:

  • tier_config

    (dict) –

    Tier configuration dict. Must contain 'type' (the tier name or out-of-tree class name). If 'type' is not a registered tier and 'module_path' is given, the class is imported from there -- an out-of-tree tier needs no register_tier() call, just this module path in the config.

Returns:

Raises:

  • ValueError –

    If 'type' is missing, or is neither registered nor resolvable via module_path.

Source code in vllm/v1/kv_offload/tiering/factory.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
139
140
141
142
143
@classmethod
def get_tier_class(cls, tier_config: dict) -> type[SecondaryTierManager]:
    """Get a secondary tier class by type.

    Args:
        tier_config: Tier configuration dict. Must contain 'type' (the
            tier name or out-of-tree class name). If 'type' is not a
            registered tier and 'module_path' is given, the class is
            imported from there -- an out-of-tree tier needs no
            register_tier() call, just this module path in the config.

    Returns:
        The secondary tier manager class.

    Raises:
        ValueError: If 'type' is missing, or is neither registered nor
            resolvable via module_path.

    """
    tier_type = tier_config.get("type")
    if not tier_type:
        raise ValueError("Secondary tier configuration must include 'type'")
    if tier_type in cls._registry:
        return cls._registry[tier_type]()
    module_path = tier_config.get("module_path")
    if module_path is None:
        raise ValueError(
            f"Unknown secondary tier type: {tier_type!r}. "
            f"Supported types: {list(cls._registry)}. "
            "For an out-of-tree tier, also set 'module_path'."
        )
    logger.warning_once(
        "Loading out-of-tree secondary tier. This API is "
        "experimental and subject to change in the future "
        "as we iterate the design."
    )
    logger.info(
        "Loading out-of-tree secondary tier '%s' from '%s'.",
        tier_type,
        module_path,
    )
    module = importlib.import_module(module_path)
    tier_cls = getattr(module, tier_type)
    assert issubclass(tier_cls, SecondaryTierManager)
    return tier_cls

_import_class(name)

Import a class by name.

If name contains a dot it is treated as a fully qualified path (e.g. 'my_package.MyDetector'). Otherwise it is looked up in the default backpressure module (vllm.v1.kv_offload.tiering.backpressure), so short names like 'DropStorePolicy' or 'EMABackpressureDetector' work out of the box.

Source code in vllm/v1/kv_offload/tiering/factory.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def _import_class(name: str) -> type:
    """Import a class by name.

    If ``name`` contains a dot it is treated as a fully qualified path
    (e.g. ``'my_package.MyDetector'``).  Otherwise it is looked up in
    the default backpressure module
    (``vllm.v1.kv_offload.tiering.backpressure``), so short names like
    ``'DropStorePolicy'`` or ``'EMABackpressureDetector'`` work out of
    the box.
    """
    if "." in name:
        module_path, _, class_name = name.rpartition(".")
    else:
        module_path, class_name = _DEFAULT_PACKAGE, name
    module = importlib.import_module(module_path)
    return getattr(module, class_name)