用于在 LLM Compressor 中管理压缩生命周期的模块。
提供一个类来定义和管理压缩事件的生命周期，包括初始化、最终化和事件处理。
类
dataclass
CompressionLifecycle(
state: State = State(),
recipe: Recipe = Recipe(),
initialized_: bool = False,
finalized: bool = False,
_last_event_type: EventType
| None = EventType.BATCH_END,
_event_order: list[EventType] = (
lambda: [
EventType.BATCH_START,
EventType.LOSS_CALCULATED,
EventType.OPTIM_PRE_STEP,
EventType.OPTIM_POST_STEP,
EventType.BATCH_END,
]
)(),
global_step: int = 0,
)
一个用于在 LLM Compressor 中管理压缩事件生命周期的类。
参数
方法
event(
event_type: EventType,
global_step: int | None = 0,
**kwargs,
) -> list[Any]
处理压缩事件。
参数
返回
引发
-
ValueError – 如果在初始化之前、最终化之后或对于无效事件类型调用
源自 llmcompressor/core/lifecycle.py 的源代码
| def event(
self, event_type: EventType, global_step: int | None = 0, **kwargs
) -> list[Any]:
"""
Handle a compression event.
:param event_type: The type of event to handle
:type event_type: EventType
:param kwargs: Additional arguments to pass to the event handlers
:return: List of data returned from handling the event by modifiers
:rtype: List[Any]
:raises ValueError: If called before initialization, after finalization,
or for an invalid event type
"""
if not self.initialized_:
logger.error("Cannot invoke event before initializing")
raise ValueError("Cannot invoke event before initializing")
if self.finalized:
logger.error("Cannot invoke event after finalizing")
raise ValueError("Cannot invoke event after finalizing")
if event_type in [EventType.INITIALIZE, EventType.FINALIZE]:
logger.error(
"Cannot invoke {} event. Use the corresponding method instead.",
event_type,
)
raise ValueError(
f"Cannot invoke {event_type} event. "
f"Use the corresponding method instead."
)
if not self._validate_event_order(event_type):
raise ValueError(
f"Lifecycle events must appear following order: {self._event_order}. "
f"Instead, {self._last_event_type} was called before {event_type}"
)
if event_type == EventType.LOSS_CALCULATED and (
"loss" not in kwargs or kwargs["loss"] is None
):
logger.error("Loss must be provided for loss calculated event")
raise ValueError("Loss must be provided for loss calculated event")
logger.debug("Handling event: {}", event_type)
# update global step
if global_step is not None:
self.global_step = global_step
event = Event(type_=event_type)
mod_data = []
for mod in self.recipe.modifiers:
data = mod.update_event(state=self.state, event=event, **kwargs)
logger.debug("Updated event with modifier: {}", mod)
if data is not None:
mod_data.append(data)
assert (
event is not None
), f"Event lifecycle did not return an event for {event_type}"
return mod_data
|
finalize(**kwargs) -> list[Any]
最终化压缩生命周期。
参数
返回
引发
源自 llmcompressor/core/lifecycle.py 的源代码
| def finalize(self, **kwargs) -> list[Any]:
"""
Finalize the compression lifecycle.
:param kwargs: Additional arguments to update the state with
:return: List of data returned from finalizing modifiers
:rtype: List[Any]
:raises ValueError: If called before initialization or more than once
"""
if not self.initialized_:
logger.error("Cannot finalize before initializing")
raise ValueError("Cannot finalize before initializing")
if self.finalized:
logger.error("Cannot finalize more than once")
raise ValueError("Cannot finalize more than once")
logger.debug("Finalizing compression lifecycle")
mod_data = []
for mod in self.recipe.modifiers:
data = mod.finalize(state=self.state, **kwargs)
logger.debug("Finalized modifier: {}", mod)
if data is not None:
mod_data.append(data)
self.finalized = True
logger.info(
"Compression lifecycle finalized for {} modifiers",
len(self.recipe.modifiers),
)
return mod_data
|
initialize(
recipe: RecipeInput | None = None,
recipe_stage: RecipeStageInput | None = None,
recipe_args: RecipeArgsInput | None = None,
**kwargs,
) -> list[Any]
初始化压缩生命周期。
参数
返回
源自 llmcompressor/core/lifecycle.py 的源代码
| def initialize(
self,
recipe: RecipeInput | None = None,
recipe_stage: RecipeStageInput | None = None,
recipe_args: RecipeArgsInput | None = None,
**kwargs,
) -> list[Any]:
"""
Initialize the compression lifecycle.
:param kwargs: Additional arguments to update the state with
:return: List of data returned from initialization of modifiers
:rtype: List[Any]
"""
self.state.update(**kwargs)
if self.initialized_: # TODO: do not initialize twice
return
logger.debug("Initializing compression lifecycle")
if not recipe:
self.recipe = Recipe()
else:
self.recipe = Recipe.create_instance(
path_or_modifiers=recipe, target_stage=recipe_stage
)
if recipe_args:
self.recipe.args = {**recipe_args}
mod_data = []
for mod in self.recipe.modifiers:
data = mod.initialize(state=self.state, **kwargs)
logger.debug("Initialized modifier: {}", mod)
if data is not None:
mod_data.append(data)
self.initialized_ = True
logger.info(
"Compression lifecycle initialized for {} modifiers",
len(self.recipe.modifiers),
)
return mod_data
|
重置压缩生命周期，最终化所有活动的修饰符并重置所有属性。
源自 llmcompressor/core/lifecycle.py 的源代码
| def reset(self):
"""
Reset the compression lifecycle, finalizing any active modifiers
and resetting all attributes.
"""
logger.debug("Resetting compression lifecycle")
for mod in self.recipe.modifiers:
if not mod.initialized or mod.finalized:
continue
try:
mod.finalize(self.state)
logger.debug("Finalized modifier: {}", mod)
except Exception as e:
logger.warning(f"Exception during finalizing modifier: {e}")
self.__init__()
logger.info("Compression lifecycle reset")
|