Skip to content

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

Contains pairs of key/value that can be calculated on demand

Source code in vllm_gaudi/extension/config.py 
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
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

Check if config is not empty

Source code in vllm_gaudi/extension/config.py 
28
29
30
def __bool__(self) -> bool:
    """Check if config is not empty"""
    return bool(self._data)

__getattr__ 

__getattr__(key: str)

Allow conveniently querying keys by using dot notation

Source code in vllm_gaudi/extension/config.py 
24
25
26
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)
Source code in vllm_gaudi/extension/config.py 
21
22
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

Trigger calculating all values

Source code in vllm_gaudi/extension/config.py 
46
47
48
def finalize(self) -> None:
    """Trigger calculating all values"""
    self.get_all()

get 

get(key: str)

Get key from internal structure, triggering calculation if needed

Source code in vllm_gaudi/extension/config.py 
32
33
34
35
36
37
38
39
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]

Return a dict with a subset of keys

Source code in vllm_gaudi/extension/config.py 
41
42
43
44
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

A callable that fetches values from env variables, applying conversions if necessary

Source code in vllm_gaudi/extension/config.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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__(_)
Source code in vllm_gaudi/extension/config.py 
157
158
159
160
161
162
163
164
165
166
167
168
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,
)
Source code in vllm_gaudi/extension/config.py 
152
153
154
155
def __init__(self, name: str, value_type: Constructor, check: Checker = skip_validation):
    self.name = name
    self.value_type = value_type
    self.check = check

Value

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

Source code in vllm_gaudi/extension/config.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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)

Return value from experimental flag if provided by user or calculate it based on dependencies

Source code in vllm_gaudi/extension/config.py 
196
197
198
199
200
201
202
203
204
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,
)
Source code in vllm_gaudi/extension/config.py 
174
175
176
177
178
179
180
181
182
183
184
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)

Check for errors and throw if any found

Source code in vllm_gaudi/extension/config.py 
190
191
192
193
194
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

Return associated experimental flag

Source code in vllm_gaudi/extension/config.py 
186
187
188
def to_env_flag(self) -> Env:
    """ Return associated experimental flag """
    return Env(self.env_var, self.env_var_type)

ValueFromList

Bases: Value

Helper class to create a value with a limited list of possible options

Source code in vllm_gaudi/extension/config.py 
207
208
209
210
211
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])
Source code in vllm_gaudi/extension/config.py 
210
211
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

Return True if all functions return True

Source code in vllm_gaudi/extension/config.py 
56
57
58
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

Return True when config[key] != True

Source code in vllm_gaudi/extension/config.py 
81
82
83
def Disabled(key: str) -> ValueFn:
    """Return True when config[key] != True"""
    return Not(Enabled(key))

Enabled 

Enabled(key: str) -> ValueFn

Return True when config[key] == True

Source code in vllm_gaudi/extension/config.py 
76
77
78
def Enabled(key: str) -> ValueFn:
    """Return True when config[key] == True"""
    return Eq(key, True)

Engine 

Engine(target_engine_version: str) -> ValueFn

Return True if current engine_version == target_engine_version

Source code in vllm_gaudi/extension/config.py 
115
116
117
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

Return True when config[key] == value

Source code in vllm_gaudi/extension/config.py 
71
72
73
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

Return first key for which config[key] is True

Source code in vllm_gaudi/extension/config.py 
86
87
88
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

Return True if current hardware == target_hw

Source code in vllm_gaudi/extension/config.py 
96
97
98
def Hardware(target_hw: str) -> ValueFn:
    """Return True if current hardware == target_hw"""
    return Eq('hw', target_hw)

Kernel 

Kernel(loader_fn: Callable) -> ValueFn

Return True if loader_fn result is not None and hardware != 'cpu'

Source code in vllm_gaudi/extension/config.py 
101
102
103
104
105
106
107
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

Return True if pt bridge is running in lazy mode

Source code in vllm_gaudi/extension/config.py 
91
92
93
def Lazy() -> ValueFn:
    """Return True if pt bridge is running in lazy mode"""
    return Eq('bridge_mode', 'lazy')

MinPackageVersion 

MinPackageVersion(
    package_name: str, min_version: str
) -> ValueFn
Source code in vllm_gaudi/extension/config.py 
126
127
128
129
130
131
132
def MinPackageVersion(package_name: str, min_version: str) -> ValueFn:

    def check(_cfg: Config):
        installed_version = get_version(package_name)
        return version.parse(installed_version) >= version.parse(min_version)

    return check

ModelType 

ModelType(target_model_type: str) -> ValueFn

Return True if current model_type == target_model_type

Source code in vllm_gaudi/extension/config.py 
110
111
112
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

Negate function result

Source code in vllm_gaudi/extension/config.py 
66
67
68
def Not(fn: ValueFn) -> ValueFn:
    """Negate function result"""
    return lambda cfg: not fn(cfg)

VersionRange 

VersionRange(*specifiers_str: str) -> ValueFn

Return True if any of the version specifiers matches current build

Source code in vllm_gaudi/extension/config.py 
120
121
122
123
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

Converts string representation of a bool to its value

Source code in vllm_gaudi/extension/config.py 
135
136
137
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]

Extract associated env flags from values

Source code in vllm_gaudi/extension/config.py 
222
223
224
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)

Converts a comma seperated string representation of a list of values

Source code in vllm_gaudi/extension/config.py 
140
141
142
143
144
145
146
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]]

Converts a list of values and returns dicts for both values and envs

Source code in vllm_gaudi/extension/config.py 
227
228
229
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]

Convert a list values/envs to a dict

Source code in vllm_gaudi/extension/config.py 
217
218
219
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}