vllm.v1.kv_offload.factory

Classes:

• OffloadingSpecFactory –

OffloadingSpecFactory

Methods:

• register_spec –

  Register a spec with a lazy-loading module and class name.

Source code in vllm/v1/kv_offload/factory.py

class OffloadingSpecFactory:
    _registry: dict[str, Callable[[], type[OffloadingSpec]]] = {}

    @classmethod
    def register_spec(cls, name: str, module_path: str, class_name: str) -> None:
        """Register a spec with a lazy-loading module and class name."""
        if name in cls._registry:
            raise ValueError(f"Connector '{name}' is already registered.")

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

        cls._registry[name] = loader

    @classmethod
    def get_spec_cls(cls, extra_config: Mapping[str, Any]) -> type[OffloadingSpec]:
        spec_name = extra_config.get("spec_name", "CPUOffloadingSpec")
        if spec_name in cls._registry:
            spec_cls = cls._registry[spec_name]()
        else:
            spec_module_path = extra_config.get("spec_module_path")
            if spec_module_path is None:
                raise ValueError(f"Unsupported spec type: {spec_name}")
            logger.warning_once(
                "Loading out-of-tree offloading spec. This API is "
                "experimental and subject to change in the future "
                "as we iterate the design."
            )
            logger.info(
                "Loading out-of-tree offloading spec '%s' from '%s'.",
                spec_name,
                spec_module_path,
            )
            spec_module = importlib.import_module(spec_module_path)
            spec_cls = getattr(spec_module, spec_name)
        assert issubclass(spec_cls, OffloadingSpec)
        return spec_cls

    @classmethod
    def create_spec(cls, config: OffloadingConfig) -> OffloadingSpec:
        spec_name = config.extra_config.get("spec_name", "CPUOffloadingSpec")
        spec_cls = cls.get_spec_cls(config.extra_config)
        logger.info("Creating offloading spec with name: %s", spec_name)
        return spec_cls(config)

register_spec(name, module_path, class_name) classmethod

Register a spec with a lazy-loading module and class name.

Source code in vllm/v1/kv_offload/factory.py

@classmethod
def register_spec(cls, name: str, module_path: str, class_name: str) -> None:
    """Register a spec with a lazy-loading module and class name."""
    if name in cls._registry:
        raise ValueError(f"Connector '{name}' is already registered.")

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

    cls._registry[name] = loader