vllm_gaudi.extension.config

Constructor module-attribute

Constructor: TypeAlias = Callable[[str], Any]

HasName module-attribute

HasName = TypeVar('HasName', bound=Value | Env)

T module-attribute

T = TypeVar('T')

ValueFn module-attribute

ValueFn: TypeAlias = Callable[[Config], Any]

Config

包含可以按需计算的键/值对

源代码在 vllm_gaudi/extension/config.py

class Config:
    """ Contains pairs of key/value that can be calculated on demand"""

    def __init__(self, *sources: dict[str, Any], **extra: Any):
        self._data = dict(itertools.chain(*[v.items() for v in sources], extra.items()))

    def __getattr__(self, key: str):
        """Allow conveniently querying keys by using dot notation"""
        return self.get(key)

    def __bool__(self) -> bool:
        """Check if config is not empty"""
        return bool(self._data)

    def get(self, key: str):
        """Get key from internal structure, triggering calculation if needed"""
        assert key in self._data, f'Unknown run-time configuration parameter: {key}!'
        value = self._data.get(key)
        if callable(value):
            value = value(self)
            self._data[key] = value
        return value

    def get_all(self, keys: Optional[list[str]] = None) -> dict[str, Any]:
        """Return a dict with a subset of keys"""
        keys = keys or list(self._data.keys())
        return {k: self.get(k) for k in keys}

    def finalize(self) -> None:
        """Trigger calculating all values"""
        self.get_all()

_data instance-attribute

_data = dict(chain(*[(items()) for v in sources], items()))

__bool__

__bool__() -> bool

检查配置是否为空

源代码在 vllm_gaudi/extension/config.py

def __bool__(self) -> bool:
    """Check if config is not empty"""
    return bool(self._data)

__getattr__

__getattr__(key: str)

允许通过点表示法方便地查询键

源代码在 vllm_gaudi/extension/config.py

def __getattr__(self, key: str):
    """Allow conveniently querying keys by using dot notation"""
    return self.get(key)

__init__

__init__(*sources: dict[str, Any], **extra: Any)

源代码在 vllm_gaudi/extension/config.py

def __init__(self, *sources: dict[str, Any], **extra: Any):
    self._data = dict(itertools.chain(*[v.items() for v in sources], extra.items()))

finalize

finalize() -> None

触发所有值的计算

源代码在 vllm_gaudi/extension/config.py

def finalize(self) -> None:
    """Trigger calculating all values"""
    self.get_all()

get

get(key: str)

从内部结构中获取键，并在需要时触发计算

源代码在 vllm_gaudi/extension/config.py

def get(self, key: str):
    """Get key from internal structure, triggering calculation if needed"""
    assert key in self._data, f'Unknown run-time configuration parameter: {key}!'
    value = self._data.get(key)
    if callable(value):
        value = value(self)
        self._data[key] = value
    return value

get_all

get_all(keys: Optional[list[str]] = None) -> dict[str, Any]

返回一个包含部分键的字典

源代码在 vllm_gaudi/extension/config.py

def get_all(self, keys: Optional[list[str]] = None) -> dict[str, Any]:
    """Return a dict with a subset of keys"""
    keys = keys or list(self._data.keys())
    return {k: self.get(k) for k in keys}

Env

一个可调用对象，用于从环境变量中获取值，并在必要时应用转换

源代码在 vllm_gaudi/extension/config.py

class Env:
    """A callable that fetches values from env variables, applying conversions if necessary"""

    def __init__(self, name: str, value_type: Constructor, check: Checker = skip_validation):
        self.name = name
        self.value_type = value_type
        self.check = check

    def __call__(self, _):
        value = os.environ.get(self.name)
        if value is not None:
            try:
                value = self.value_type(value)
                if error := self.check(value):
                    raise RuntimeError(error)
                return value
            except Exception as e:
                msg = f'{self.name}: exception during construction: {e}'
                raise RuntimeError(msg)
        return None

check instance-attribute

check = check

name instance-attribute

name = name

value_type instance-attribute

value_type = value_type

__call__

__call__(_)

源代码在 vllm_gaudi/extension/config.py

def __call__(self, _):
    value = os.environ.get(self.name)
    if value is not None:
        try:
            value = self.value_type(value)
            if error := self.check(value):
                raise RuntimeError(error)
            return value
        except Exception as e:
            msg = f'{self.name}: exception during construction: {e}'
            raise RuntimeError(msg)
    return None

__init__

__init__(
    name: str,
    value_type: Constructor,
    check: Checker = skip_validation,
)

源代码在 vllm_gaudi/extension/config.py

def __init__(self, name: str, value_type: Constructor, check: Checker = skip_validation):
    self.name = name
    self.value_type = value_type
    self.check = check

Value

一个可调用对象，通过其依赖项计算值，或由关联的实验标志覆盖

源代码在 vllm_gaudi/extension/config.py

class Value:
    """A callable that returns the value calculated through its dependencies or overriden by an associated experimental flag"""

    def __init__(self,
                 name: str,
                 dependencies: Any,
                 env_var: Optional[str] = None,
                 env_var_type: Constructor = boolean,
                 check: Checker = skip_validation):
        self.name = name
        self.env_var = env_var if env_var is not None else 'VLLM_' + name.upper()
        self.env_var_type = env_var_type
        self.dependencies = dependencies
        self.check = check

    def to_env_flag(self) -> Env:
        """ Return associated experimental flag """
        return Env(self.env_var, self.env_var_type)

    def _validate(self, value):
        """ Check for errors and throw if any found"""
        if error := self.check(value):
            raise RuntimeError(f'{self.name}: {error}')
        return value

    def __call__(self, config):
        """ Return value from experimental flag if provided by user or calculate it based on dependencies """
        if (override := config.get(self.env_var)) is not None:
            result = override
        elif callable(self.dependencies):
            result = self.dependencies(config)
        else:
            result = self.dependencies
        return self._validate(result)

check instance-attribute

check = check

dependencies instance-attribute

dependencies = dependencies

env_var instance-attribute

env_var = (
    env_var if env_var is not None else "VLLM_" + upper()
)

env_var_type instance-attribute

env_var_type = env_var_type

name instance-attribute

name = name

__call__

__call__(config)

如果用户提供了实验标志，则返回该标志的值，否则根据依赖项计算

源代码在 vllm_gaudi/extension/config.py

def __call__(self, config):
    """ Return value from experimental flag if provided by user or calculate it based on dependencies """
    if (override := config.get(self.env_var)) is not None:
        result = override
    elif callable(self.dependencies):
        result = self.dependencies(config)
    else:
        result = self.dependencies
    return self._validate(result)

__init__

__init__(
    name: str,
    dependencies: Any,
    env_var: Optional[str] = None,
    env_var_type: Constructor = boolean,
    check: Checker = skip_validation,
)

源代码在 vllm_gaudi/extension/config.py

def __init__(self,
             name: str,
             dependencies: Any,
             env_var: Optional[str] = None,
             env_var_type: Constructor = boolean,
             check: Checker = skip_validation):
    self.name = name
    self.env_var = env_var if env_var is not None else 'VLLM_' + name.upper()
    self.env_var_type = env_var_type
    self.dependencies = dependencies
    self.check = check

_validate

_validate(value)

检查错误并抛出

源代码在 vllm_gaudi/extension/config.py

def _validate(self, value):
    """ Check for errors and throw if any found"""
    if error := self.check(value):
        raise RuntimeError(f'{self.name}: {error}')
    return value

to_env_flag

to_env_flag() -> Env

返回关联的实验标志

源代码在 vllm_gaudi/extension/config.py

def to_env_flag(self) -> Env:
    """ Return associated experimental flag """
    return Env(self.env_var, self.env_var_type)

ValueFromList

Bases: Value

用于创建具有有限可能选项的值的辅助类

源代码在 vllm_gaudi/extension/config.py

class ValueFromList(Value):
    """ Helper class to create a value with a limited list of possible options """

    def __init__(self, name: str, options: list[str]):
        super().__init__(name, FirstEnabled(*options), env_var_type=str, check=choice(*options))

__init__

__init__(name: str, options: list[str])

源代码在 vllm_gaudi/extension/config.py

def __init__(self, name: str, options: list[str]):
    super().__init__(name, FirstEnabled(*options), env_var_type=str, check=choice(*options))

All

All(*parts: ValueFn) -> ValueFn

当所有函数都返回 True 时返回 True

源代码在 vllm_gaudi/extension/config.py

def All(*parts: ValueFn) -> ValueFn:
    """Return True if all functions return True"""
    return lambda cfg: all(p(cfg) for p in parts)

Disabled

Disabled(key: str) -> ValueFn

当 config[key] != True 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Disabled(key: str) -> ValueFn:
    """Return True when config[key] != True"""
    return Not(Enabled(key))

Enabled

Enabled(key: str) -> ValueFn

当 config[key] == True 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Enabled(key: str) -> ValueFn:
    """Return True when config[key] == True"""
    return Eq(key, True)

Engine

Engine(target_engine_version: str) -> ValueFn

当 current_engine_version == target_engine_version 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Engine(target_engine_version: str) -> ValueFn:
    """Return True if current engine_version == target_engine_version"""
    return Eq('engine_version', target_engine_version)

Eq

Eq(key: str, value: Any) -> ValueFn

当 config[key] == value 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Eq(key: str, value: Any) -> ValueFn:
    """Return True when config[key] == value"""
    return lambda cfg: (cfg.get(key) == value)

FirstEnabled

FirstEnabled(*keys: str) -> ValueFn

返回第一个 config[key] 为 True 的键

源代码在 vllm_gaudi/extension/config.py

def FirstEnabled(*keys: str) -> ValueFn:
    """Return first key for which config[key] is True"""
    return lambda cfg: next((k for k in keys if cfg.get(k)), None)

Hardware

Hardware(target_hw: str) -> ValueFn

当 current_hardware == target_hw 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Hardware(target_hw: str) -> ValueFn:
    """Return True if current hardware == target_hw"""
    return Eq('hw', target_hw)

Kernel

Kernel(loader_fn: Callable) -> ValueFn

当 loader_fn 结果不为 None 且 hardware != 'cpu' 时返回 True

源代码在 vllm_gaudi/extension/config.py

def Kernel(loader_fn: Callable) -> ValueFn:
    """Return True if loader_fn result is not None and hardware != 'cpu'"""

    def kernel_exists(_):
        return loader_fn() is not None

    return All(kernel_exists, Not(Hardware('cpu')))

Lazy

Lazy() -> ValueFn

当 pt bridge 处于懒惰模式运行时返回 True

源代码在 vllm_gaudi/extension/config.py

def Lazy() -> ValueFn:
    """Return True if pt bridge is running in lazy mode"""
    return Eq('bridge_mode', 'lazy')

ModelType

ModelType(target_model_type: str) -> ValueFn

当 current_model_type == target_model_type 时返回 True

源代码在 vllm_gaudi/extension/config.py

def ModelType(target_model_type: str) -> ValueFn:
    """Return True if current model_type == target_model_type"""
    return Eq('model_type', target_model_type)

Not

Not(fn: ValueFn) -> ValueFn

否定函数结果

源代码在 vllm_gaudi/extension/config.py

def Not(fn: ValueFn) -> ValueFn:
    """Negate function result"""
    return lambda cfg: not fn(cfg)

VersionRange

VersionRange(*specifiers_str: str) -> ValueFn

当任何版本说明符与当前构建匹配时返回 True

源代码在 vllm_gaudi/extension/config.py

def VersionRange(*specifiers_str: str) -> ValueFn:
    """Return True if any of the version specifiers matches current build"""
    specifiers = [SpecifierSet(s) for s in specifiers_str]
    return lambda cfg: any(Version(cfg.build) in s for s in specifiers)

boolean

boolean(x: str) -> bool

将布尔值的字符串表示转换为其值

源代码在 vllm_gaudi/extension/config.py

def boolean(x: str) -> bool:
    """Converts string representation of a bool to its value"""
    return x.lower() in ['true', 't', '1', 'yes', 'y', 'on']

env_flags

env_flags(values: list[Value]) -> list[Env]

从值中提取关联的环境标志

源代码在 vllm_gaudi/extension/config.py

def env_flags(values: list[Value]) -> list[Env]:
    """Extract associated env flags from values"""
    return [v.to_env_flag() for v in values]

list_of

list_of(t: Constructor)

将逗号分隔的字符串表示转换为值列表

源代码在 vllm_gaudi/extension/config.py

def list_of(t: Constructor):
    """Converts a comma seperated string representation of a list of values"""

    def list_of_impl(x: str) -> list[Any]:
        return [t(v) for v in x.split(',')]

    return list_of_impl

split_values_and_flags

split_values_and_flags(
    values: list[Value],
) -> tuple[dict[str, Value], dict[str, Env]]

将值列表转换为字典，并同时返回值的字典和环境标志的字典

源代码在 vllm_gaudi/extension/config.py

def split_values_and_flags(values: list[Value]) -> tuple[dict[str, Value], dict[str, Env]]:
    """Converts a list of values and returns dicts for both values and envs"""
    return to_dict(values), to_dict(env_flags(values))

to_dict

to_dict(collection: list[HasName]) -> dict[str, HasName]

将值/环境标志列表转换为字典

源代码在 vllm_gaudi/extension/config.py

def to_dict(collection: list[HasName]) -> dict[str, HasName]:
    """Convert a list values/envs to a dict"""
    return {c.name: c for c in collection}